out_stackdriver: add trace and Custom Log Name Support (#2683)

- Add trace to stackdriver output. This is necessary for stackdriver to correlate request logs with app logs. See https://medium.com/google-cloud/combining-correlated-log-lines-in-google-stackdriver-dd23284aeb29

- Added a default label for trace_key which is needed for the proper removal from the jsonPayload.

Signed-off-by: Todor Petrov <[email protected]>

Author: tpetrov-lp

plugins/out_stackdriver/stackdriver.c

@@ -994,6 +994,18 @@ static int get_msgpack_obj(msgpack_object * subobj, const msgpack_object * o,
     return -1;
 }
 
+static int get_string(flb_sds_t * s, const msgpack_object * o, const flb_sds_t key)
+{
+    msgpack_object tmp;
+    if (get_msgpack_obj(&tmp, o, key, flb_sds_len(key), MSGPACK_OBJECT_STR) == 0) {
+        *s = flb_sds_create_len(tmp.via.str.ptr, tmp.via.str.size);
+        return 0;
+    }
+
+    *s = 0;
+    return -1;
+}
+
 static int get_severity_level(severity_t * s, const msgpack_object * o,
                               const flb_sds_t key)
 {
@@ -1006,6 +1018,7 @@ static int get_severity_level(severity_t * s, const msgpack_object * o,
     return -1;
 }
 
+
 static int get_stream(msgpack_object_map map)
 {
     int i;
@@ -1104,8 +1117,13 @@ static int pack_json_payload(int insert_id_extracted,
         monitored_resource_key,
         local_resource_id_key,
         ctx->labels_key,
+        ctx->severity_key,
+        ctx->trace_key,
+        ctx->log_name_key,
         stream
-        /* more special fields are required to be added */
+        /* more special fields are required to be added, but, if this grows with more 
+           than a few records, it might need to be converted to flb_hash
+         */
     };
 
     if (insert_id_extracted == FLB_TRUE) {
@@ -1138,7 +1156,7 @@ static int pack_json_payload(int insert_id_extracted,
              * check length of key to avoid partial matching
              * e.g. labels key = labels && kv->key = labelss
              */
-            if (flb_sds_cmp(removed, kv->key.via.str.ptr, len) == 0) {
+            if (removed && flb_sds_cmp(removed, kv->key.via.str.ptr, len) == 0) {
                 to_remove += 1;
                 break;
             }
@@ -1215,7 +1233,7 @@ static int pack_json_payload(int insert_id_extracted,
         len = kv->key.via.str.size;
         for (j = 0; j < len_to_be_removed; j++) {
             removed = to_be_removed[j];
-            if (flb_sds_cmp(removed, kv->key.via.str.ptr, len) == 0) {
+            if (removed && flb_sds_cmp(removed, kv->key.via.str.ptr, len) == 0) {
                 key_not_found = 0;
                 break;
             }
@@ -1264,6 +1282,7 @@ static int stackdriver_format(struct flb_config *config,
     char path[PATH_MAX];
     char time_formatted[255];
     const char *newtag;
+    const char *new_log_name;
     msgpack_object *obj;
     msgpack_object *labels_ptr;
     msgpack_unpacked result;
@@ -1276,6 +1295,16 @@ static int stackdriver_format(struct flb_config *config,
     int severity_extracted = FLB_FALSE;
     severity_t severity;
 
+    /* Parameters for trace */
+    int trace_extracted = FLB_FALSE;
+    flb_sds_t trace;
+    char stackdriver_trace[PATH_MAX];
+    const char *new_trace;
+
+    /* Parameters for log name */
+    int log_name_extracted = FLB_FALSE;
+    flb_sds_t log_name;
+
     /* Parameters for insertId */
     msgpack_object insert_id_obj;
     insert_id_status in_status;
@@ -1581,7 +1610,8 @@ static int stackdriver_format(struct flb_config *config,
          *  "labels": "...",
          *  "logName": "...",
          *  "jsonPayload": {...},
-         *  "timestamp": "..."
+         *  "timestamp": "...",
+         *  "trace": "..."
          * }
          */
         entry_size = 3;
@@ -1594,6 +1624,21 @@ static int stackdriver_format(struct flb_config *config,
             entry_size += 1;
         }
 
+        /* Extract trace */
+        trace_extracted = FLB_FALSE;
+        if (ctx->trace_key
+            && get_string(&trace, obj, ctx->trace_key) == 0) {
+            trace_extracted = FLB_TRUE;
+            entry_size += 1;
+        }
+
+        /* Extract log name */
+        log_name_extracted = FLB_FALSE;
+        if (ctx->log_name_key
+            && get_string(&log_name, obj, ctx->log_name_key) == 0) {
+            log_name_extracted = FLB_TRUE;
+        }
+
         /* Extract insertId */
         in_status = validate_insert_id(&insert_id_obj, obj);
         if (in_status == INSERTID_VALID) {
@@ -1668,6 +1713,26 @@ static int stackdriver_format(struct flb_config *config,
             msgpack_pack_int(&mp_pck, severity);
         }
 
+        /* Add trace into the log entry */
+        if (trace_extracted == FLB_TRUE) {
+            msgpack_pack_str(&mp_pck, 5);
+            msgpack_pack_str_body(&mp_pck, "trace", 5);
+
+            if (ctx->autoformat_stackdriver_trace) {
+                len = snprintf(stackdriver_trace, sizeof(stackdriver_trace) - 1,
+                    "projects/%s/traces/%s", ctx->project_id, trace);
+                new_trace = stackdriver_trace;
+            }
+            else {
+                len = flb_sds_len(trace);
+                new_trace = trace;
+            }
+
+            msgpack_pack_str(&mp_pck, len);
+            msgpack_pack_str_body(&mp_pck, new_trace, len);
+            flb_sds_destroy(trace); 
+        }               
+
         /* Add insertId field into the log entry */
         if (insert_id_extracted == FLB_TRUE) {
             msgpack_pack_str(&mp_pck, 8);
@@ -1729,9 +1794,21 @@ static int stackdriver_format(struct flb_config *config,
                 newtag = "stderr";
             }
         }
+
+        if (log_name_extracted == FLB_FALSE) {
+            new_log_name = newtag;
+        }  
+        else {
+            new_log_name = log_name;
+        }
+
         /* logName */
         len = snprintf(path, sizeof(path) - 1,
-                       "projects/%s/logs/%s", ctx->project_id, newtag);
+                       "projects/%s/logs/%s", ctx->project_id, new_log_name);
+
+        if (log_name_extracted == FLB_TRUE) {
+            flb_sds_destroy(log_name); 
+        }
 
         msgpack_pack_str(&mp_pck, 7);
         msgpack_pack_str_body(&mp_pck, "logName", 7);

plugins/out_stackdriver/stackdriver.h

@@ -50,6 +50,9 @@
 #define MONITORED_RESOURCE_KEY "logging.googleapis.com/monitored_resource"
 #define LOCAL_RESOURCE_ID_KEY "logging.googleapis.com/local_resource_id"
 #define DEFAULT_LABELS_KEY "logging.googleapis.com/labels"
+#define DEFAULT_SEVERITY_KEY "logging.googleapis.com/severity"
+#define DEFAULT_TRACE_KEY "logging.googleapis.com/trace"
+#define DEFAULT_LOG_NAME_KEY "logging.googleapis.com/logName"
 #define DEFAULT_INSERT_ID_KEY "logging.googleapis.com/insertId"
 #define SOURCELOCATION_FIELD_IN_JSON "logging.googleapis.com/sourceLocation"
 #define HTTPREQUEST_FIELD_IN_JSON "logging.googleapis.com/http_request"
@@ -108,6 +111,9 @@ struct flb_stackdriver {
     /* other */
     flb_sds_t resource;
     flb_sds_t severity_key;
+    flb_sds_t trace_key;
+    flb_sds_t log_name_key;
+    bool autoformat_stackdriver_trace;
 
     /* oauth2 context */
     struct flb_oauth2 *o;

plugins/out_stackdriver/stackdriver_conf.c

@@ -22,6 +22,7 @@
 #include <fluent-bit/flb_compat.h>
 #include <fluent-bit/flb_info.h>
 #include <fluent-bit/flb_unescape.h>
+#include <fluent-bit/flb_utils.h>
 
 #include <jsmn/jsmn.h>
 
@@ -268,8 +269,35 @@ struct flb_stackdriver *flb_stackdriver_conf_create(struct flb_output_instance *
     if (tmp) {
         ctx->severity_key = flb_sds_create(tmp);
     }
+    else {
+        ctx->severity_key = flb_sds_create(DEFAULT_SEVERITY_KEY);
+    }
+
+    tmp = flb_output_get_property("autoformat_stackdriver_trace", ins);
+    if (tmp) {
+        ctx->autoformat_stackdriver_trace = flb_utils_bool(tmp);
+    }
+    else {
+        ctx->autoformat_stackdriver_trace = FLB_FALSE;
+    }
+    
+    tmp = flb_output_get_property("trace_key", ins);
+    if (tmp) {
+        ctx->trace_key = flb_sds_create(tmp);
+    }
+    else {
+        ctx->trace_key = flb_sds_create(DEFAULT_TRACE_KEY);
+    }
+
+    tmp = flb_output_get_property("log_name_key", ins);
+    if (tmp) {
+        ctx->log_name_key = flb_sds_create(tmp);
+    }
+    else {
+        ctx->log_name_key = flb_sds_create(DEFAULT_LOG_NAME_KEY);
+    }
 
-    if (flb_sds_cmp(ctx->resource, "k8s_container", 
+    if (flb_sds_cmp(ctx->resource, "k8s_container",
                     flb_sds_len(ctx->resource)) == 0 || 
         flb_sds_cmp(ctx->resource, "k8s_node", 
                     flb_sds_len(ctx->resource)) == 0 ||
@@ -344,6 +372,8 @@ int flb_stackdriver_conf_destroy(struct flb_stackdriver *ctx)
     flb_sds_destroy(ctx->token_uri);
     flb_sds_destroy(ctx->resource);
     flb_sds_destroy(ctx->severity_key);
+    flb_sds_destroy(ctx->trace_key);
+    flb_sds_destroy(ctx->log_name_key);
     flb_sds_destroy(ctx->labels_key);
     flb_sds_destroy(ctx->tag_prefix);
 

tests/runtime/data/stackdriver/stackdriver_test_log_name.h

@@ -0,0 +1,10 @@
+#define LOG_NAME_OVERRIDE	"[" \
+	"1591111124," \
+	"{"	\
+        "\"custom_log_name_key\": \"custom_log_name\"" \
+	"}]"
+
+#define LOG_NAME_NO_OVERRIDE	"[" \
+	"1591111124," \
+	"{"	\
+	"}]"

tests/runtime/data/stackdriver/stackdriver_test_trace.h

@@ -0,0 +1,6 @@
+#define TRACE_COMMON_CASE	"[" \
+	"1591111124," \
+	"{"	\
+        "\"trace\": \"test-trace-id-xyz\"" \
+	"}]"
+