#14 Build counted activity.

src/main/java/org/valuereporter/activity/CountedActivity.java

@@ -0,0 +1,17 @@
+package org.valuereporter.activity;
+
+import java.util.Map;
+
+/**
+ * Created by baardl on 01.08.17.
+ */
+public class CountedActivity extends ObservedActivity{
+    private long count = 1;
+    public CountedActivity(String name, long startTime, Map<String, Object> data) {
+        super(name, startTime, data);
+    }
+
+    public long getCount() {
+        return count;
+    }
+}

src/main/java/org/valuereporter/activity/timeseries/CommandSendActivities.java

@@ -4,9 +4,14 @@
 import com.netflix.hystrix.HystrixCommand;
 import com.netflix.hystrix.HystrixCommandGroupKey;
 import org.slf4j.Logger;
+import org.valuereporter.activity.CountedActivity;
 import org.valuereporter.activity.ObservedActivity;
 
+import java.net.URI;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.StringJoiner;
 
 import static org.slf4j.LoggerFactory.getLogger;
 
@@ -29,52 +34,75 @@ public class CommandSendActivities extends HystrixCommand<String>  {
     public static final String APPLICATION_BINARY = "application/octet-stream";
     private final String observedActivitiesJson;
     private final int no_of_activities;
+    private final URI influxDbUri;
+    private final String databaseName;
 
+    /*
     public CommandSendActivities(final String reporterHost, final String reporterPort, final String prefix, final List<ObservedActivity> observedActivities) {
         super(HystrixCommandGroupKey.Factory.asKey("ValueReporter-Activities-group"));
-        observedActivitiesJson = buildJson(observedActivities);
+        observedActivitiesJson = buildBody(observedActivities);
         no_of_activities = observedActivities.size();
         this.reporterHost = reporterHost;
         this.reporterPort = reporterPort;
         this.prefix = prefix;
         this.observedActivities = observedActivities;
     }
+    */
 
-    public CommandSendActivities(String influxDbUri, String databaseName, List<ObservedActivity> observedActivities) {
+    public CommandSendActivities(URI influxDbUri, String databaseName, List<ObservedActivity> observedActivities) {
         super(HystrixCommandGroupKey.Factory.asKey("ValueReporter-Activities-group"));
-        observedActivitiesJson = buildJson(observedActivities);
+        this.influxDbUri = influxDbUri;
+        this.databaseName = databaseName;
+        observedActivitiesJson = buildBody(observedActivities);
         no_of_activities = observedActivities.size();
         this.reporterHost = null;
         this.reporterPort = null;
         this.prefix = null;
         this.observedActivities = observedActivities;
     }
 
-    protected String buildJson(List<ObservedActivity> observedActivities)  {
+    protected String buildBody(List<ObservedActivity> observedActivities)  {
         String json = "client-access,host=dev.shareproc.com,service=api,function=login,ip=127.0.0.1 count=1\n";
-//        try {
-//            json = mapper.writeValueAsString(observedActivities);
-//        } catch (IOException e) {
-//            e.printStackTrace();
-//        }
-        return json;
+        String body = "";
+        String line = "";
+        for (ObservedActivity activity : observedActivities) {
+            if (activity instanceof CountedActivity) {
+                line = activity.getName() + "," + buildTags(activity) + " " + buildMeasurement((CountedActivity)activity) + "\n";
+            } else {
+                line = activity.getName() + "," + buildTags(activity) + " count=1\n";
+
+            }
+            log.trace("Line: [{}]", line);
+            body += line;
+        }
+        return body;
+    }
+
+    protected String buildMeasurement(CountedActivity activity) {
+        String measurement = "";
+        if (activity != null) {
+            measurement = "count=" + activity.getCount();
+        }
+        return measurement;
+    }
+
+    String buildTags(ObservedActivity activity) {
+        StringJoiner tags = new StringJoiner(",");
+        if (activity != null) {
+            Map tagMap = activity.getData();
+            Set<Map.Entry> entries = tagMap.entrySet();
+            for (Map.Entry entry : entries) {
+                tags.add(entry.getKey() + "=" +entry.getValue());
+            }
+        }
+        return tags.toString();
     }
 
     @Override
     protected String run() {
-//        Client client = ClientBuilder.newClient();
-//        String observationUrl = "http://"+reporterHost + ":" + reporterPort +"/reporter/observe/activities";
-//        log.info("Connection to ValueReporter on {}" , observationUrl);
-//        final WebTarget observationTarget = client.target(observationUrl);
-//        WebTarget webResource = observationTarget.path(prefix);
-//        log.trace("Forwarding observedActivities as Json \n{}", observedActivitiesJson);
-//        Response response = webResource.request(MediaType.APPLICATION_JSON).post(Entity.entity(observedActivitiesJson, MediaType.APPLICATION_JSON));
-//        int statusCode = statusCode;
-
-        String observationUrl = "http://influxdb-component-ox6b3xp9td0-772793266.eu-west-1.elb.amazonaws.com:8086/write?db=shareproc";
-        //http://"+reporterHost + ":" + reporterPort +"/reporter/observe" + "/activities/" + prefix;
-        //hei
-        log.info("Connection to ValueReporter on {} num of activities: {}" , observationUrl,no_of_activities);
+        String observationUrl = influxDbUri + "/write?db=" + databaseName;
+//        http://influxdb-component-ox6b3xp9td0-772793266.eu-west-1.elb.amazonaws.com:8086/write?db=shareproc";
+        log.info("Connection to InfluxDb on {} num of activities: {}" , observationUrl,no_of_activities);
         HttpRequest request = HttpRequest.post(observationUrl ).acceptJson().contentType(APPLICATION_BINARY).send(observedActivitiesJson);
         int statusCode = request.code();
         String responseBody = request.body();
@@ -86,17 +114,17 @@ protected String run() {
                 log.trace("Updated via http ok. No content in response, as expected.");
                 break;
             case STATUS_FORBIDDEN:
-                log.warn("Can not access ValueReporter. The application will function as normally, though Observation statistics will not be stored. URL {}, HttpStatus {}, Response {}, ", observationUrl,statusCode, responseBody);
+                log.warn("Can not access InfluxDb. The application will function as normally, though Observation statistics will not be stored. URL {}, HttpStatus {}, Response {}, ", observationUrl,statusCode, responseBody);
                 break;
             default:
-                log.trace("Retrying access to ValueReporter");
+                log.trace("Retrying access to InfluxDb");
                 request = HttpRequest.post(observationUrl ).acceptJson().contentType(APPLICATION_BINARY).send(observedActivitiesJson);
                 statusCode = request.code();
                 responseBody = request.body();
                 if (statusCode == STATUS_OK) {
                     log.trace("Retry via http ok. Response is {}", responseBody);
                 } else {
-                    log.error("Error while accessing ValueReporter. The application will function as normally, though Observation statistics will not be stored. URL {}, HttpStatus {},Response from ValueReporter {}", observationUrl, statusCode, responseBody);
+                    log.error("Error while accessing InfluxDb. The application will function as normally, though Observation statistics will not be stored. URL {}, HttpStatus {},Response from InfluxDb {}", observationUrl, statusCode, responseBody);
                 }
         }
         return "OK";

src/test/java/org/valuereporter/activity/timeseries/CommandSendActivitiesTest.java

@@ -0,0 +1,59 @@
+package org.valuereporter.activity.timeseries;
+
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+import org.valuereporter.activity.CountedActivity;
+import org.valuereporter.activity.ObservedActivity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.testng.Assert.assertEquals;
+
+/**
+ * Created by baardl on 01.08.17.
+ */
+public class CommandSendActivitiesTest {
+    List<ObservedActivity> observedActivities = null;
+    ObservedActivity activity = null;
+    CommandSendActivities commandSendActivities = null;
+    @BeforeMethod
+    public void setUp() throws Exception {
+        String name = "client-access";
+        long startTime = System.nanoTime();
+        Map<String, Object> data = new HashMap<>();
+        data.put("host", "whydahdev.cantara.no");
+        data.put("service", "sts");
+        data.put("function", "login");
+        data.put("ip", "127.0.0.1");
+        activity = new CountedActivity(name,startTime,data);
+        observedActivities = new ArrayList<>();
+        observedActivities.add(activity);
+
+        URI uri = URI.create("http://localhost");
+        String databaseName = "testdb";
+        commandSendActivities = new CommandSendActivities(uri, databaseName, observedActivities);
+    }
+
+    @Test
+    public void testBuildBody() throws Exception {
+    }
+
+    @Test
+    public void testBuildMeasurement() throws Exception {
+
+    }
+
+    @Test
+    public void testBuildTags() throws Exception {
+        String expected = "service=sts,function=login,ip=127.0.0.1,host=whydahdev.cantara.no";
+        assertEquals(commandSendActivities.buildTags(activity), expected);
+        assertEquals(commandSendActivities.buildTags(null), "");
+
+
+    }
+
+}

src/test/java/org/valuereporter/activity/timeseries/ManualSendActivitiesTest.java

@@ -1,20 +1,34 @@
 package org.valuereporter.activity.timeseries;
 
+import org.valuereporter.activity.CountedActivity;
 import org.valuereporter.activity.ObservedActivity;
 
+import java.net.URI;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 /**
  * Created by baardl on 31.07.17.
  */
 public class ManualSendActivitiesTest {
 
     public static void main(String[] args) {
-        String influxDbUri = "http://influxdb-component-ox6b3xp9td0-772793266.eu-west-1.elb.amazonaws.com:8086/write?db=";
+        URI influxDbUri = URI.create("http://influxdb-component-ox6b3xp9td0-772793266.eu-west-1.elb.amazonaws.com:8086"); ///write?db=";
         String database = "shareproc";
 
         List<ObservedActivity> observedActivities = new ArrayList<>();
+        //client-access,host=dev.shareproc.com,service=api,function=login,ip=127.0.0.1 count=1
+        String name = "client-access";
+        long startTime = System.nanoTime();
+        Map<String, Object> data = new HashMap<>();
+        data.put("host", "whydahdev.cantara.no");
+        data.put("service", "sts");
+        data.put("function", "login");
+        data.put("ip", "127.0.0.1");
+        ObservedActivity activity = new CountedActivity(name,startTime,data);
+        observedActivities.add(activity);
         CommandSendActivities sendActivities = new CommandSendActivities(influxDbUri, database, observedActivities);
         sendActivities.execute();
     }