filter_kubernetes: Plug SEGV occurence on enabling Use_Pod_Association

plugins/filter_kubernetes/kube_conf.h

@@ -29,6 +29,7 @@
 #include <fluent-bit/flb_sds.h>
 #include <fluent-bit/flb_regex.h>
 #include <fluent-bit/flb_hash_table.h>
+#include <fluent-bit/flb_pthread.h>
 
 /*
  * Since this filter might get a high number of request per second,
@@ -213,6 +214,13 @@ struct flb_kube {
     int aws_pod_service_map_refresh_interval;
     flb_sds_t aws_pod_service_preload_cache_path;
     struct flb_upstream *aws_pod_association_upstream;
+    pthread_mutex_t aws_pod_service_mutex;
+    pthread_cond_t aws_pod_service_cond;
+    pthread_t aws_pod_service_thread;
+    int aws_pod_service_sync_initialized;
+    int aws_pod_service_thread_created;
+    int aws_pod_service_shutdown;
+    struct mk_event_loop *aws_pod_service_event_loop;
     /*
      * This variable holds the Kubernetes platform type
      * Current checks for EKS or Native Kuberentes

plugins/filter_kubernetes/kubernetes.c

@@ -35,6 +35,7 @@
 #include "kubernetes_aws.h"
 
 #include <stdio.h>
+#include <errno.h>
 #include <msgpack.h>
 #include <sys/stat.h>
 
@@ -44,31 +45,68 @@
 #define MERGE_MAP         2 /* merge direct binary object (v)            */
 #define FLB_KUBE_LOCAL_LOGS_INPUT "fluentbit_logs"
 
-struct task_args {
-    struct flb_kube *ctx;
-    char *api_server_url;
-};
+static int wait_for_pod_service_map_refresh(struct flb_kube *ctx)
+{
+    int ret;
+    int shutdown;
+    struct flb_time current_time;
+    struct timespec deadline;
+
+    pthread_mutex_lock(&ctx->aws_pod_service_mutex);
+
+    flb_time_get(&current_time);
+    deadline = current_time.tm;
+    deadline.tv_sec += ctx->aws_pod_service_map_refresh_interval;
 
-pthread_mutex_t metadata_mutex;
-pthread_t background_thread;
-struct task_args *task_args = {0};
-struct mk_event_loop *evl;
+    while (!ctx->aws_pod_service_shutdown) {
+        ret = pthread_cond_timedwait(&ctx->aws_pod_service_cond,
+                                     &ctx->aws_pod_service_mutex,
+                                     &deadline);
+        if (ret == ETIMEDOUT) {
+            break;
+        }
+        else if (ret != 0) {
+            flb_plg_error(ctx->ins,
+                          "Failed waiting for pod service map refresh");
+            break;
+        }
+    }
 
-void *update_pod_service_map(void *arg)
+    shutdown = ctx->aws_pod_service_shutdown;
+    pthread_mutex_unlock(&ctx->aws_pod_service_mutex);
+
+    return shutdown;
+}
+
+static void *update_pod_service_map(void *arg)
 {
+    struct flb_kube *ctx;
+
+    ctx = arg;
+
     flb_engine_evl_init();
-    evl = mk_event_loop_create(256);
-    if (evl == NULL) {
-        flb_plg_error(task_args->ctx->ins,
+    ctx->aws_pod_service_event_loop = mk_event_loop_create(256);
+    if (ctx->aws_pod_service_event_loop == NULL) {
+        flb_plg_error(ctx->ins,
                       "Failed to create event loop for pod service map");
         return NULL;
     }
-    flb_engine_evl_set(evl);
+    flb_engine_evl_set(ctx->aws_pod_service_event_loop);
+
     while (1) {
-        fetch_pod_service_map(task_args->ctx,task_args->api_server_url,&metadata_mutex);
-        flb_plg_debug(task_args->ctx->ins, "Updating pod to service map after %d seconds", task_args->ctx->aws_pod_service_map_refresh_interval);
-        sleep(task_args->ctx->aws_pod_service_map_refresh_interval);
+        fetch_pod_service_map(ctx,
+                              ctx->aws_pod_association_endpoint,
+                              &ctx->aws_pod_service_mutex);
+        flb_plg_debug(ctx->ins,
+                      "Updating pod to service map after %d seconds",
+                      ctx->aws_pod_service_map_refresh_interval);
+
+        if (wait_for_pod_service_map_refresh(ctx)) {
+            break;
+        }
     }
+
+    return NULL;
 }
 
 static int get_stream(msgpack_object_map map)
@@ -239,24 +277,38 @@ static int cb_kube_init(struct flb_filter_instance *f_ins,
      */
     flb_kube_meta_init(ctx, config);
 
-/*
- * Init separate thread for calling pod to
- * service map
- */
-    pthread_mutex_init(&metadata_mutex, NULL);
-
     if (ctx->aws_use_pod_association) {
-        task_args = flb_malloc(sizeof(struct task_args));
-        if (!task_args) {
-            flb_errno();
+        ret = pthread_mutex_init(&ctx->aws_pod_service_mutex, NULL);
+        if (ret != 0) {
+            flb_plg_error(ctx->ins,
+                          "Failed to initialize pod service map mutex");
+            flb_kube_conf_destroy(ctx);
             return -1;
         }
-        task_args->ctx = ctx;
-        task_args->api_server_url = ctx->aws_pod_association_endpoint;
-        if (pthread_create(&background_thread, NULL, update_pod_service_map, NULL) != 0) {
-            flb_error("Failed to create background thread");
-            background_thread = 0;
-            flb_free(task_args);
+
+        ret = pthread_cond_init(&ctx->aws_pod_service_cond, NULL);
+        if (ret != 0) {
+            flb_plg_error(ctx->ins,
+                          "Failed to initialize pod service map condition");
+            pthread_mutex_destroy(&ctx->aws_pod_service_mutex);
+            flb_kube_conf_destroy(ctx);
+            return -1;
+        }
+        ctx->aws_pod_service_sync_initialized = FLB_TRUE;
+
+        ret = pthread_create(&ctx->aws_pod_service_thread,
+                             NULL,
+                             update_pod_service_map,
+                             ctx);
+        if (ret != 0) {
+            flb_plg_error(ctx->ins,
+                          "Failed to create pod service map background thread");
+            pthread_cond_destroy(&ctx->aws_pod_service_cond);
+            pthread_mutex_destroy(&ctx->aws_pod_service_mutex);
+            ctx->aws_pod_service_sync_initialized = FLB_FALSE;
+        }
+        else {
+            ctx->aws_pod_service_thread_created = FLB_TRUE;
         }
     }
 
@@ -824,20 +876,30 @@ static int cb_kube_exit(void *data, struct flb_config *config)
     struct flb_kube *ctx;
 
     ctx = data;
-
-    flb_kube_conf_destroy(ctx);
-    if (background_thread) {
-        pthread_cancel(background_thread);
-        pthread_join(background_thread, NULL);
+
+    if (ctx->aws_pod_service_thread_created) {
+        pthread_mutex_lock(&ctx->aws_pod_service_mutex);
+        ctx->aws_pod_service_shutdown = FLB_TRUE;
+        pthread_cond_signal(&ctx->aws_pod_service_cond);
+        pthread_mutex_unlock(&ctx->aws_pod_service_mutex);
+
+        pthread_join(ctx->aws_pod_service_thread, NULL);
+        ctx->aws_pod_service_thread_created = FLB_FALSE;
     }
-    pthread_mutex_destroy(&metadata_mutex);
 
-    if (task_args) {
-        flb_free(task_args);
+    if (ctx->aws_pod_service_event_loop) {
+        mk_event_loop_destroy(ctx->aws_pod_service_event_loop);
+        ctx->aws_pod_service_event_loop = NULL;
     }
-    if (evl) {
-        mk_event_loop_destroy(evl);
+
+    if (ctx->aws_pod_service_sync_initialized) {
+        pthread_cond_destroy(&ctx->aws_pod_service_cond);
+        pthread_mutex_destroy(&ctx->aws_pod_service_mutex);
+        ctx->aws_pod_service_sync_initialized = FLB_FALSE;
     }
+
+    flb_kube_conf_destroy(ctx);
+
     return 0;
 }
 

tests/runtime/data/kubernetes/pod-service.map

@@ -0,0 +1 @@
+{}

tests/runtime/filter_kubernetes.c

@@ -499,6 +499,75 @@ static void flb_test_local_fluentbit_logs()
     flb_destroy(ctx.flb);
 }
 
+static void flb_test_pod_association_multiple_instances()
+{
+    int ret;
+    int in_ffd;
+    int filter_ffd;
+    int out_ffd;
+    flb_ctx_t *ctx;
+
+    ctx = flb_create();
+    TEST_CHECK_(ctx != NULL, "initialising service");
+    if (!ctx) {
+        return;
+    }
+
+    ret = flb_service_set(ctx,
+                          "Flush", "0.2",
+                          "Grace", "1",
+                          "Log_Level", "error",
+                          NULL);
+    TEST_CHECK_(ret == 0, "setting service options");
+
+    in_ffd = flb_input(ctx, "dummy", NULL);
+    TEST_CHECK_(in_ffd >= 0, "initialising input");
+    ret = flb_input_set(ctx, in_ffd,
+                        "Tag", "application.test",
+                        "Samples", "1",
+                        NULL);
+    TEST_CHECK_(ret == 0, "setting input options");
+
+    filter_ffd = flb_filter(ctx, "kubernetes", NULL);
+    TEST_CHECK_(filter_ffd >= 0, "initialising first filter");
+    ret = flb_filter_set(ctx, filter_ffd,
+                         "Match", "application.*",
+                         "Dummy_Meta", "On",
+                         "Use_Pod_Association", "On",
+                         "AWS_Pod_Service_Preload_Cache_Dir",
+                         DPATH "/pod-service",
+                         "AWS_Pod_Service_Map_Refresh_Interval", "1",
+                         NULL);
+    TEST_CHECK_(ret == 0, "setting first filter options");
+
+    filter_ffd = flb_filter(ctx, "kubernetes", NULL);
+    TEST_CHECK_(filter_ffd >= 0, "initialising second filter");
+    ret = flb_filter_set(ctx, filter_ffd,
+                         "Match", "dataplane.*",
+                         "Dummy_Meta", "On",
+                         "Use_Pod_Association", "On",
+                         "AWS_Pod_Service_Preload_Cache_Dir",
+                         DPATH "/pod-service",
+                         "AWS_Pod_Service_Map_Refresh_Interval", "1",
+                         NULL);
+    TEST_CHECK_(ret == 0, "setting second filter options");
+
+    out_ffd = flb_output(ctx, "null", NULL);
+    TEST_CHECK_(out_ffd >= 0, "initialising output");
+    ret = flb_output_set(ctx, out_ffd, "Match", "*", NULL);
+    TEST_CHECK_(ret == 0, "setting output options");
+
+    ret = flb_start(ctx);
+    TEST_CHECK_(ret == 0, "starting engine");
+    if (ret == 0) {
+        flb_time_msleep(100);
+        ret = flb_stop(ctx);
+        TEST_CHECK_(ret == 0, "stopping engine");
+    }
+
+    flb_destroy(ctx);
+}
+
 
 #define flb_test_core(target, suffix, nExpected) \
     kube_test("core/" target, KUBE_TAIL, suffix, nExpected, NULL);
@@ -1159,6 +1228,7 @@ TEST_LIST = {
     {"kube_core_base_with_namespace_labels_and_annotations", flb_test_core_base_with_namespace_labels_and_annotations},
     {"kube_core_base_with_owner_references", flb_test_core_base_with_owner_references},
     {"kube_local_fluentbit_logs", flb_test_local_fluentbit_logs},
+    {"kube_pod_association_multiple_instances", flb_test_pod_association_multiple_instances},
     {"kube_options_use-kubelet_enabled_json", flb_test_options_use_kubelet_enabled_json},
     {"kube_options_use-kubelet_disabled_json", flb_test_options_use_kubelet_disabled_json},
     {"kube_options_merge_log_enabled_text", flb_test_options_merge_log_enabled_text},