Docker instrumentation finished in grunt

resource_system/src/grunt/grunt.c

@@ -338,9 +338,9 @@ int main(int argc, char** argv)
                 JSON_Print_Command command = {0};
                 while (instrumentCommandDequeue(&command)) {
                     // TODO(naman): Enable this after proper testing
-                    /* if (!kafkaWrite(kafka.writer, command.topic, "resource_daemon", command.msg)) { */
-                    /*     return -1; */
-                    /* } */
+                    if (!kafkaWrite(kafka.writer, command.topic, "resource_daemon", command.msg)) {
+                        return -1;
+                    }
                 }
             }
 

resource_system/src/grunt/instrument_docker.c

@@ -12,7 +12,7 @@ void* dockerProcessLoop (void *arg)
     while (true) {
         // TODO(naman): Get data
         Char *data_cmd = NULL;
-        sbufPrint(data_cmd, "docker stats %s", (Char*)arg);
+        sbufPrint(data_cmd, "docker stats --no-stream --format \"{{ json . }}\" %s", (Char*)arg);
         FILE* data_file = popen(data_cmd, "r");
 
         fseek(data_file, 0, SEEK_END);
@@ -22,17 +22,49 @@ void* dockerProcessLoop (void *arg)
         fread(data, 1, (Size)size + 1, data_file);
         fclose(data_file);
 
+        const Char *json_error = NULL;
+        cJSON *data_json = cJSON_ParseWithOpts(data, &json_error, true);
+
         Char *json = NULL;
+        sbufPrint(json, "{\"cpu_percentage\": %f",
+                  atof(cJSON_GetObjectItem(data_json, "CPUPerc")->valuestring));
+        sbufPrint(json, ",\n\"memory_percentage\": %f",
+                  atof(cJSON_GetObjectItem(data_json, "MemPerc")->valuestring));
+        sbufPrint(json, ",\n\"memory_used\": %f",
+                  atof(cJSON_GetObjectItem(data_json, "MemUsage")->valuestring));
+        sbufPrint(json, ",\n\"memory_limit\": %f",
+                  atof(strchr(cJSON_GetObjectItem(data_json, "MemUsage")->valuestring, '/') + 1));
+        sbufPrint(json, ",\n\"block_input\": %f",
+                  atof(cJSON_GetObjectItem(data_json, "BlockIO")->valuestring));
+        sbufPrint(json, ",\n\"block_output\": %f",
+                  atof(strchr(cJSON_GetObjectItem(data_json, "BlockIO")->valuestring, '/') + 1));
+        sbufPrint(json, ",\n\"net_down\": %f",
+                  atof(cJSON_GetObjectItem(data_json, "NetIO")->valuestring));
+        sbufPrint(json, ",\n\"net_up\": %f",
+                  atof(strchr(cJSON_GetObjectItem(data_json, "NetIO")->valuestring, '/') + 1));
+        sbufPrint(json, "\n}\n");
+
+        cJSON *json_parse = cJSON_Parse(json);
+        Char *json_pretty = cJSON_Print(json_parse);
 
         Char *output = NULL;
         sbufPrint(output, "{\"node_id\": \"%s\"", node_name);
-        sbufPrint(output, ",\n\"type\": \"%s\"", "docker");
-        sbufPrint(output, ",\n\"data\": %s", json ? json : "{}");
+        sbufPrint(output, ",\n\"entity_id\": \"%s\"", (Char*)arg);
+        sbufPrint(output, ",\n\"type\": \"%s\"", "instrumentation");
+        sbufPrint(output, ",\n\"backend\": \"%s\"", "docker");
+        sbufPrint(output, ",\n\"data\": %s", json_pretty);
         sbufPrint(output, "\n}\n");
 
         JSON_Print_Command jpc = {.msg = output,
                                   .topic = "LOG_CHANNEL"};
 
+        cJSON_free(json_pretty);
+        cJSON_Delete(json_parse);
+        sbufDelete(output);
+        sbufDelete(json);
+        free(data);
+        sbufDelete(data_cmd);
+
         U64 time_before = timeMilli();
         instrumentCommandEnqueue(jpc);
         U64 time_after = timeMilli();