out_stackdriver: add k8s_cluster resource type (#8349)

* out_stackdriver: add k8s_cluster resource type this can be used in combination with the in_kubernetes_events to write k8s_cluster type events to stackdriver(such as replicaset events) Signed-off-by: ryanohnemus <[email protected]> * out_stackdriver: add k8s_cluster resource test Signed-off-by: ryanohnemus <[email protected]> --------- Signed-off-by: ryanohnemus <[email protected]>
fluent · Jan 25, 2024 · 00c7262 · 00c7262
1 parent 28aae40
commit 00c7262
Show file tree

Hide file tree

Showing 5 changed files with 119 additions and 1 deletion.
diff --git a/plugins/out_stackdriver/stackdriver.c b/plugins/out_stackdriver/stackdriver.c
@@ -2135,6 +2135,45 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx,
 
                 flb_mp_map_header_end(&mh);
             }
+            else if (strcmp(ctx->resource, K8S_CLUSTER) == 0) {
+                /* k8s_cluster resource has fields project_id, location, cluster_name
+                *
+                * There is no local_resource_id for k8s_cluster as we get all info
+                *      from plugin config
+                */
+
+                flb_mp_map_header_init(&mh, &mp_pck);
+
+                if (ctx->project_id) {
+                    flb_mp_map_header_append(&mh);
+                    msgpack_pack_str(&mp_pck, 10);
+                    msgpack_pack_str_body(&mp_pck, "project_id", 10);
+                    msgpack_pack_str(&mp_pck, flb_sds_len(ctx->project_id));
+                    msgpack_pack_str_body(&mp_pck,
+                                        ctx->project_id, flb_sds_len(ctx->project_id));
+                }
+
+                if (ctx->cluster_location) {
+                    flb_mp_map_header_append(&mh);
+                    msgpack_pack_str(&mp_pck, 8);
+                    msgpack_pack_str_body(&mp_pck, "location", 8);
+                    msgpack_pack_str(&mp_pck, flb_sds_len(ctx->cluster_location));
+                    msgpack_pack_str_body(&mp_pck,
+                                        ctx->cluster_location,
+                                        flb_sds_len(ctx->cluster_location));
+                }
+
+                if (ctx->cluster_name) {
+                    flb_mp_map_header_append(&mh);
+                    msgpack_pack_str(&mp_pck, 12);
+                    msgpack_pack_str_body(&mp_pck, "cluster_name", 12);
+                    msgpack_pack_str(&mp_pck, flb_sds_len(ctx->cluster_name));
+                    msgpack_pack_str_body(&mp_pck,
+                                        ctx->cluster_name, flb_sds_len(ctx->cluster_name));
+                }
+
+                flb_mp_map_header_end(&mh);
+            }
             else {
                 flb_plg_error(ctx->ins, "unsupported resource type '%s'",
                             ctx->resource);

diff --git a/plugins/out_stackdriver/stackdriver.h b/plugins/out_stackdriver/stackdriver.h
@@ -72,6 +72,7 @@
  */
 #define STACKDRIVER_NET_ERROR  502
 
+#define K8S_CLUSTER   "k8s_cluster"
 #define K8S_CONTAINER "k8s_container"
 #define K8S_NODE      "k8s_node"
 #define K8S_POD       "k8s_pod"

diff --git a/plugins/out_stackdriver/stackdriver_resource_types.c b/plugins/out_stackdriver/stackdriver_resource_types.c
@@ -27,7 +27,7 @@
 static const struct resource_type resource_types[] = {
     { 
         .id = RESOURCE_TYPE_K8S,
-        .resources = {"k8s_container", "k8s_node", "k8s_pod"},
+        .resources = {"k8s_container", "k8s_node", "k8s_pod", "k8s_cluster"},
         .required_labels = {"cluster_name", "location"}
     },
     { 

diff --git a/tests/runtime/data/stackdriver/stackdriver_test_k8s_resource.h b/tests/runtime/data/stackdriver/stackdriver_test_k8s_resource.h
@@ -86,4 +86,9 @@
 	"\"END_KEY\": \"JSON_END\""		\
 	"}]"
 
+#define K8S_CLUSTER_NO_LOCAL_RESOURCE_ID	"["		\
+	"1591649196,"			\
+	"{"				\
+    "\"message\": \"K8S_CLUSTER_NO_LOCAL_RESOURCE_ID\""		\
+	"}]"
 
diff --git a/tests/runtime/out_stackdriver.c b/tests/runtime/out_stackdriver.c
@@ -841,6 +841,34 @@ static void cb_check_log_name_no_override(void *ctx, int ffd,
     flb_sds_destroy(res_data);
 }
 
+static void cb_check_k8s_cluster_resource(void *ctx, int ffd,
+                                       int res_ret, void *res_data, size_t res_size,
+                                       void *data)
+{
+    int ret;
+
+    /* resource type */
+    ret = mp_kv_cmp(res_data, res_size, "$resource['type']", "k8s_cluster");
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* project id */
+    ret = mp_kv_cmp(res_data, res_size,
+                    "$resource['labels']['project_id']", "fluent-bit");
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* location */
+    ret = mp_kv_cmp(res_data, res_size,
+                    "$resource['labels']['location']", "test_cluster_location");
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* cluster name */
+    ret = mp_kv_cmp(res_data, res_size,
+                    "$resource['labels']['cluster_name']", "test_cluster_name");
+    TEST_CHECK(ret == FLB_TRUE);
+
+    flb_sds_destroy(res_data);
+}
+
 static void cb_check_project_key_override(void *ctx, int ffd,
                                           int res_ret, void *res_data, size_t res_size,
                                           void *data)
@@ -3777,6 +3805,50 @@ void flb_test_resource_k8s_container_custom_k8s_regex_custom_prefix()
     flb_destroy(ctx);
 }
 
+void flb_test_resource_k8s_cluster_no_local_resource_id()
+{
+    int ret;
+    int size = sizeof(K8S_CLUSTER_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", "test", NULL);
+
+    /* Stackdriver output */
+    out_ffd = flb_output(ctx, (char *) "stackdriver", NULL);
+    flb_output_set(ctx, out_ffd,
+                   "match", "test",
+                   "resource", "k8s_cluster",
+                   "google_service_credentials", SERVICE_CREDENTIALS,
+                   "k8s_cluster_name", "test_cluster_name",
+                   "k8s_cluster_location", "test_cluster_location",
+                   NULL);
+
+    /* Enable test mode */
+    ret = flb_output_set_test(ctx, out_ffd, "formatter",
+                              cb_check_k8s_cluster_resource,
+                              NULL, NULL);
+
+    /* Start */
+    ret = flb_start(ctx);
+    TEST_CHECK(ret == 0);
+
+    /* Ingest data sample */
+    flb_lib_push(ctx, in_ffd, (char *) K8S_CLUSTER_NO_LOCAL_RESOURCE_ID, size);
+
+    sleep(2);
+    flb_stop(ctx);
+    flb_destroy(ctx);
+}
+
+
 void flb_test_resource_k8s_node_common()
 {
     int ret;
@@ -6293,6 +6365,7 @@ TEST_LIST = {
     {"resource_k8s_container_default_tag_regex", flb_test_resource_k8s_container_default_tag_regex },
     {"resource_k8s_container_custom_k8s_regex", flb_test_resource_k8s_container_custom_k8s_regex },
     {"resource_k8s_container_custom_k8s_regex_custom_prefix", flb_test_resource_k8s_container_custom_k8s_regex_custom_prefix },
+    {"resource_k8s_cluster_no_local_resource_id", flb_test_resource_k8s_cluster_no_local_resource_id },
     {"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_node_custom_k8s_regex_with_dot", flb_test_resource_k8s_node_custom_k8s_regex_with_dot },