out_stackdriver: support dot '.' in tag_prefix option (#2468)

Signed-off-by: Jeff Luo <[email protected]>

Author: JeffLuoo

plugins/out_stackdriver/stackdriver.c

@@ -288,6 +288,31 @@ static bool validate_msgpack_unpacked_data(msgpack_object root)
            root.via.array.ptr[1].type == MSGPACK_OBJECT_MAP;
 }
 
+void replace_prefix_dot(flb_sds_t s, int tag_prefix_len)
+{
+    int i;
+    int str_len;
+    char c;
+
+    if (!s) {
+        return;
+    }
+
+    str_len = flb_sds_len(s);
+    if (tag_prefix_len > str_len) {
+        flb_error("[output] tag_prefix shouldn't be longer than local_resource_id");
+        return;
+    }
+
+    for (i = 0; i < tag_prefix_len; i++) {
+        c = s[i];
+
+        if (c == '.') {
+            s[i] = '_';
+        }
+    }
+}
+
 static flb_sds_t get_str_value_from_msgpack_map(msgpack_object_map map,
                                                 const char *key, int key_size)
 {
@@ -428,9 +453,11 @@ static int process_local_resource_id(struct flb_stackdriver *ctx, char *type)
     int len_k8s_container;
     int len_k8s_node;
     int len_k8s_pod;
+    int prefix_len;
     struct local_resource_id_list *ptr;
     struct mk_list *list = NULL;
     struct mk_list *head;
+    flb_sds_t new_local_resource_id;
 
     if (!ctx->local_resource_id) {
         flb_plg_error(ctx->ins, "local_resource_is is not assigned");
@@ -441,8 +468,20 @@ static int process_local_resource_id(struct flb_stackdriver *ctx, char *type)
     len_k8s_node = sizeof(K8S_NODE) - 1;
     len_k8s_pod = sizeof(K8S_POD) - 1;
 
+    prefix_len = flb_sds_len(ctx->tag_prefix);
+    if (flb_sds_casecmp(ctx->tag_prefix, ctx->local_resource_id, prefix_len) != 0) {
+        flb_plg_error(ctx->ins, "tag_prefix [%s] doesn't match the prefix of"
+                      " local_resource_id [%s]", ctx->tag_prefix,
+                      ctx->local_resource_id);
+        return -1;
+    }
+
+    new_local_resource_id = flb_sds_create_len(ctx->local_resource_id,
+                                               flb_sds_len(ctx->local_resource_id));
+    replace_prefix_dot(new_local_resource_id, prefix_len);
+
     if (strncmp(type, K8S_CONTAINER, len_k8s_container) == 0) {
-        list = parse_local_resource_id_to_list(ctx->local_resource_id, K8S_CONTAINER);
+        list = parse_local_resource_id_to_list(new_local_resource_id, K8S_CONTAINER);
         if (!list) {
             goto error;
         }
@@ -451,11 +490,6 @@ static int process_local_resource_id(struct flb_stackdriver *ctx, char *type)
         mk_list_foreach(head, list) {
             ptr = mk_list_entry(head, struct local_resource_id_list, _head);
             if (first) {
-                /* check the prefix */
-                if (flb_sds_len(ptr->val) != flb_sds_len(ctx->tag_prefix) ||
-                    strncmp(ptr->val, ctx->tag_prefix, flb_sds_len(ctx->tag_prefix)) != 0) {
-                    goto error;
-                }
                 first = FLB_FALSE;
                 continue;
             }
@@ -488,19 +522,14 @@ static int process_local_resource_id(struct flb_stackdriver *ctx, char *type)
         }
     }
     else if (strncmp(type, K8S_NODE, len_k8s_node) == 0) {
-        list = parse_local_resource_id_to_list(ctx->local_resource_id, K8S_NODE);
+        list = parse_local_resource_id_to_list(new_local_resource_id, K8S_NODE);
         if (!list) {
             goto error;
         }
 
         mk_list_foreach(head, list) {
             ptr = mk_list_entry(head, struct local_resource_id_list, _head);
             if (first) {
-                /* check the prefix */
-                if (flb_sds_len(ptr->val) != flb_sds_len(ctx->tag_prefix) ||
-                    strncmp(ptr->val, ctx->tag_prefix, flb_sds_len(ctx->tag_prefix)) != 0) {
-                    goto error;
-                }
                 first = FLB_FALSE;
                 continue;
             }
@@ -518,19 +547,14 @@ static int process_local_resource_id(struct flb_stackdriver *ctx, char *type)
         }
     }
     else if (strncmp(type, K8S_POD, len_k8s_pod) == 0) {
-        list = parse_local_resource_id_to_list(ctx->local_resource_id, K8S_POD);
+        list = parse_local_resource_id_to_list(new_local_resource_id, K8S_POD);
         if (!list) {
             goto error;
         }
 
         mk_list_foreach(head, list) {
             ptr = mk_list_entry(head, struct local_resource_id_list, _head);
             if (first) {
-                /* check the prefix */
-                if (flb_sds_len(ptr->val) != flb_sds_len(ctx->tag_prefix) ||
-                    strncmp(ptr->val, ctx->tag_prefix, flb_sds_len(ctx->tag_prefix)) != 0) {
-                    goto error;
-                }
                 first = FLB_FALSE;
                 continue;
             }
@@ -563,6 +587,7 @@ static int process_local_resource_id(struct flb_stackdriver *ctx, char *type)
         flb_slist_destroy(list);
         flb_free(list);
     }
+    flb_sds_destroy(new_local_resource_id);
 
     return ret;
 
@@ -584,6 +609,8 @@ static int process_local_resource_id(struct flb_stackdriver *ctx, char *type)
         flb_sds_destroy(ctx->namespace_name);
         flb_sds_destroy(ctx->pod_name);
     }
+
+    flb_sds_destroy(new_local_resource_id);
     return -1;
 }
 

tests/runtime/out_stackdriver.c

@@ -2168,6 +2168,50 @@ void flb_test_resource_k8s_container_custom_tag_prefix()
     flb_destroy(ctx);
 }
 
+void flb_test_resource_k8s_container_custom_tag_prefix_with_dot()
+{
+    int ret;
+    int size = sizeof(K8S_CONTAINER_NO_LOCAL_RESOURCE_ID) - 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", "kube.custom.tag.testnamespace.testpod.testctr", NULL);
+
+    /* Stackdriver output */
+    out_ffd = flb_output(ctx, (char *) "stackdriver", NULL);
+    flb_output_set(ctx, out_ffd,
+                   "match", "kube.custom.tag.*",
+                   "resource", "k8s_container",
+                   "google_service_credentials", SERVICE_CREDENTIALS,
+                   "k8s_cluster_name", "test_cluster_name",
+                   "k8s_cluster_location", "test_cluster_location",
+                   "tag_prefix", "kube.custom.tag",
+                   NULL);
+
+    /* Enable test mode */
+    ret = flb_output_set_test(ctx, out_ffd, "formatter",
+                              cb_check_k8s_container_resource,
+                              NULL, NULL);
+
+    /* Start */
+    ret = flb_start(ctx);
+    TEST_CHECK(ret == 0);
+
+    /* Ingest data sample */
+    flb_lib_push(ctx, in_ffd, (char *) K8S_CONTAINER_NO_LOCAL_RESOURCE_ID, size);
+
+    sleep(2);
+    flb_stop(ctx);
+    flb_destroy(ctx);
+}
+
 void flb_test_resource_k8s_node_common()
 {
     int ret;
@@ -3592,6 +3636,7 @@ TEST_LIST = {
     {"resource_k8s_container_no_local_resource_id", flb_test_resource_k8s_container_no_local_resource_id },
     {"resource_k8s_container_multi_tag_value", flb_test_resource_k8s_container_multi_tag_value } ,
     {"resource_k8s_container_custom_tag_prefix", flb_test_resource_k8s_container_custom_tag_prefix },
+    {"resource_k8s_container_custom_tag_prefix_with_dot", flb_test_resource_k8s_container_custom_tag_prefix_with_dot },
     {"resource_k8s_node_common", flb_test_resource_k8s_node_common },
     {"resource_k8s_node_no_local_resource_id", flb_test_resource_k8s_node_no_local_resource_id },
     {"resource_k8s_pod_common", flb_test_resource_k8s_pod_common },