- Added unit tests for the removed severity field from jsonPayload sent to stackdriver

- Added support for overriding the logName sent to stackdriver. By default, it's `projects/<project id>/logs/<tag>`, but `<tag>` should be replacable by a field from the log entry, because this will allow to group the logs properly and visualize in the logs dropdown in stackdriver.

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

Author: tpetrov-lp

plugins/out_stackdriver/stackdriver.c

@@ -994,30 +994,31 @@ static int get_msgpack_obj(msgpack_object * subobj, const msgpack_object * o,
     return -1;
 }
 
-static int get_severity_level(severity_t * s, const msgpack_object * o,
-                              const flb_sds_t key)
+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
-        && validate_severity_level(s, tmp.via.str.ptr, tmp.via.str.size) == 0) {
+    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_trace(flb_sds_t * t, const msgpack_object * o, const flb_sds_t key)
+static int get_severity_level(severity_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) {
-        *t = flb_sds_create_len(tmp.via.str.ptr, tmp.via.str.size);
+    if (get_msgpack_obj(&tmp, o, key, flb_sds_len(key), MSGPACK_OBJECT_STR) == 0
+        && validate_severity_level(s, tmp.via.str.ptr, tmp.via.str.size) == 0) {
         return 0;
     }
-
-    *t = 0;
+    *s = 0;
     return -1;
 }
 
+
 static int get_stream(msgpack_object_map map)
 {
     int i;
@@ -1118,8 +1119,11 @@ static int pack_json_payload(int insert_id_extracted,
         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) {
@@ -1152,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;
             }
@@ -1229,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;
             }
@@ -1278,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;
@@ -1294,6 +1299,10 @@ static int stackdriver_format(struct flb_config *config,
     int trace_extracted = FLB_FALSE;
     flb_sds_t 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;
@@ -1616,11 +1625,18 @@ static int stackdriver_format(struct flb_config *config,
         /* Extract trace */
         trace_extracted = FLB_FALSE;
         if (ctx->trace_key
-            && get_trace(&trace, obj, ctx->trace_key) == 0) {
+            && 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) {
@@ -1767,9 +1783,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

@@ -111,6 +111,7 @@ struct flb_stackdriver {
     flb_sds_t resource;
     flb_sds_t severity_key;
     flb_sds_t trace_key;
+    flb_sds_t log_name_key;
 
     /* oauth2 context */
     struct flb_oauth2 *o;

plugins/out_stackdriver/stackdriver_conf.c

@@ -280,6 +280,11 @@ struct flb_stackdriver *flb_stackdriver_conf_create(struct flb_output_instance *
         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);
+    }
+
     if (flb_sds_cmp(ctx->resource, "k8s_container",
                     flb_sds_len(ctx->resource)) == 0 || 
         flb_sds_cmp(ctx->resource, "k8s_node", 
@@ -356,6 +361,7 @@ int flb_stackdriver_conf_destroy(struct flb_stackdriver *ctx)
     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

@@ -2,4 +2,9 @@
 	"1591111124," \
 	"{"	\
         "\"custom_log_name_key\": \"custom_log_name\"" \
-	"}]"
+	"}]"
+
+#define LOG_NAME_NO_OVERRIDE	"[" \
+	"1591111124," \
+	"{"	\
+	"}]"

tests/runtime/out_stackdriver.c

@@ -35,6 +35,7 @@
 #include "data/stackdriver/stackdriver_test_k8s_resource.h"
 #include "data/stackdriver/stackdriver_test_labels.h"
 #include "data/stackdriver/stackdriver_test_trace.h"
+#include "data/stackdriver/stackdriver_test_log_name.h"
 #include "data/stackdriver/stackdriver_test_insert_id.h"
 #include "data/stackdriver/stackdriver_test_source_location.h"
 #include "data/stackdriver/stackdriver_test_http_request.h"
@@ -588,6 +589,38 @@ static void cb_check_trace_common_case(void *ctx, int ffd,
     flb_sds_destroy(res_data);
 }
 
+static void cb_check_log_name_override(void *ctx, int ffd,
+                                       int res_ret, void *res_data, size_t res_size,
+                                       void *data)
+{
+    int ret;
+
+    /* logName in the entries is created using the value under log_name_key */
+    ret = mp_kv_cmp(
+        res_data, res_size, "$entries[0]['logName']", "projects/fluent-bit-test/logs/custom_log_name");
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* log_name_key has been removed from jsonPayload */
+    ret = mp_kv_exists(res_data, res_size, "$entries[0]['jsonPayload']['custom_log_name_key']");
+    TEST_CHECK(ret == FLB_FALSE);
+
+    flb_sds_destroy(res_data);
+}
+
+static void cb_check_log_name_no_override(void *ctx, int ffd,
+                                          int res_ret, void *res_data, size_t res_size,
+                                          void *data)
+{
+    int ret;
+
+    /* logName in the entries is created using the tag */
+    ret = mp_kv_cmp(
+        res_data, res_size, "$entries[0]['logName']", "projects/fluent-bit-test/logs/test");
+    TEST_CHECK(ret == FLB_TRUE);
+
+    flb_sds_destroy(res_data);
+}
+
 static void cb_check_k8s_node_resource(void *ctx, int ffd,
                                        int res_ret, void *res_data, size_t res_size,
                                        void *data)
@@ -1058,12 +1091,20 @@ static void cb_check_multi_entries_severity(void *ctx, int ffd,
     ret = mp_kv_cmp(res_data, res_size, "$entries[0]['severity']", "INFO");
     TEST_CHECK(ret == FLB_TRUE);
 
+    // verifies that severity is removed from jsonPayload
+    ret = mp_kv_exists(res_data, res_size, "$entries[0]['jsonPayload']['severity']");
+    TEST_CHECK(ret == FLB_FALSE);
+
     ret = mp_kv_exists(res_data, res_size, "$entries[1]['severity']");
     TEST_CHECK(ret == FLB_FALSE);
 
     ret = mp_kv_cmp(res_data, res_size, "$entries[2]['severity']", "DEBUG");
     TEST_CHECK(ret == FLB_TRUE);
 
+    // verifies that severity is removed from jsonPayload
+    ret = mp_kv_exists(res_data, res_size, "$entries[2]['jsonPayload']['severity']");
+    TEST_CHECK(ret == FLB_FALSE);
+
     ret = mp_kv_exists(res_data, res_size, "$entries[3]['severity']");
     TEST_CHECK(ret == FLB_FALSE);
 
@@ -1913,6 +1954,88 @@ void flb_test_trace_common_case()
     flb_destroy(ctx);
 }
 
+void flb_test_log_name_override()
+{
+    int ret;
+    int size = sizeof(LOG_NAME_OVERRIDE) - 1;
+    flb_ctx_t *ctx;
+    int in_ffd;
+    int out_ffd;
+
+    /* Create context, flush every second (some checks omitted here) */
+    ctx = flb_create();
+    flb_service_set(ctx, "flush", "1", "grace", "1", NULL);
+
+    /* Lib input mode */
+    in_ffd = flb_input(ctx, (char *) "lib", NULL);
+    flb_input_set(ctx, in_ffd, "tag", "test", NULL);
+
+    /* Stackdriver output */
+    out_ffd = flb_output(ctx, (char *) "stackdriver", NULL);
+    flb_output_set(ctx, out_ffd,
+                   "match", "test",
+                   "resource", "gce_instance",
+                   "log_name_key", "custom_log_name_key",
+                   NULL);
+
+    /* Enable test mode */
+    ret = flb_output_set_test(ctx, out_ffd, "formatter",
+                              cb_check_log_name_override,
+                              NULL, NULL);
+
+    /* Start */
+    ret = flb_start(ctx);
+    TEST_CHECK(ret == 0);
+
+    /* Ingest data sample */
+    flb_lib_push(ctx, in_ffd, (char *) LOG_NAME_OVERRIDE, size);
+
+    sleep(2);
+    flb_stop(ctx);
+    flb_destroy(ctx);
+}
+
+void flb_test_log_name_no_override()
+{
+    int ret;
+    int size = sizeof(LOG_NAME_NO_OVERRIDE) - 1;
+    flb_ctx_t *ctx;
+    int in_ffd;
+    int out_ffd;
+
+    /* Create context, flush every second (some checks omitted here) */
+    ctx = flb_create();
+    flb_service_set(ctx, "flush", "1", "grace", "1", NULL);
+
+    /* Lib input mode */
+    in_ffd = flb_input(ctx, (char *) "lib", NULL);
+    flb_input_set(ctx, in_ffd, "tag", "test", NULL);
+
+    /* Stackdriver output */
+    out_ffd = flb_output(ctx, (char *) "stackdriver", NULL);
+    flb_output_set(ctx, out_ffd,
+                   "match", "test",
+                   "resource", "gce_instance",
+                   "log_name_key", "custom_log_name_key",
+                   NULL);
+
+    /* Enable test mode */
+    ret = flb_output_set_test(ctx, out_ffd, "formatter",
+                              cb_check_log_name_no_override,
+                              NULL, NULL);
+
+    /* Start */
+    ret = flb_start(ctx);
+    TEST_CHECK(ret == 0);
+
+    /* Ingest data sample */
+    flb_lib_push(ctx, in_ffd, (char *) LOG_NAME_NO_OVERRIDE, size);
+
+    sleep(2);
+    flb_stop(ctx);
+    flb_destroy(ctx);
+}
+
 void flb_test_resource_global_custom_prefix()
 {
     /* configuring tag_prefix for non-k8s resource type should have no effect at all */
@@ -3987,6 +4110,10 @@ TEST_LIST = {
     /* test trace */
     {"trace_common_case", flb_test_trace_common_case},
 
+    /* test log name */
+    {"log_name_override", flb_test_log_name_override},
+    {"log_name_no_override", flb_test_log_name_no_override},
+
     /* test insertId */
     {"insertId_common_case", flb_test_insert_id_common_case},
     {"empty_insertId", flb_test_empty_insert_id},