diff --git a/plugins/filter_kubernetes/kube_conf.h b/plugins/filter_kubernetes/kube_conf.h index 781053c9274..85f90928c83 100644 --- a/plugins/filter_kubernetes/kube_conf.h +++ b/plugins/filter_kubernetes/kube_conf.h @@ -29,6 +29,7 @@ #include #include #include +#include /* * 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 diff --git a/plugins/filter_kubernetes/kubernetes.c b/plugins/filter_kubernetes/kubernetes.c index 24548e5fa68..e272b3d321b 100644 --- a/plugins/filter_kubernetes/kubernetes.c +++ b/plugins/filter_kubernetes/kubernetes.c @@ -35,6 +35,7 @@ #include "kubernetes_aws.h" #include +#include #include #include @@ -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(¤t_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; } diff --git a/tests/runtime/data/kubernetes/pod-service.map b/tests/runtime/data/kubernetes/pod-service.map new file mode 100644 index 00000000000..0967ef424bc --- /dev/null +++ b/tests/runtime/data/kubernetes/pod-service.map @@ -0,0 +1 @@ +{} diff --git a/tests/runtime/filter_kubernetes.c b/tests/runtime/filter_kubernetes.c index 8eef7ee41c2..751041a0851 100644 --- a/tests/runtime/filter_kubernetes.c +++ b/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},