Skip to content

filter_kubernetes: add entity attribute retrieval logics #10586

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
edsiper merged 1 commit into from
Sep 22, 2025
Merged

filter_kubernetes: add entity attribute retrieval logics #10586

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions plugins/filter_kubernetes/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ set(src
kube_regex.c
kube_property.c
kubernetes.c
kubernetes_aws.c
)

FLB_PLUGIN(filter_kubernetes "${src}" "")
Expand Down
28 changes: 28 additions & 0 deletions plugins/filter_kubernetes/kube_conf.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -189,6 +189,15 @@ struct flb_kube *flb_kube_conf_create(struct flb_filter_instance *ins,
flb_plg_info(ctx->ins, "https=%i host=%s port=%i",
ctx->api_https, ctx->api_host, ctx->api_port);
}

ctx->aws_pod_service_hash_table = flb_hash_table_create_with_ttl(ctx->aws_pod_service_map_ttl,
FLB_HASH_TABLE_EVICT_OLDER,
FLB_HASH_TABLE_SIZE,
FLB_HASH_TABLE_SIZE);
if (!ctx->aws_pod_service_hash_table) {
flb_kube_conf_destroy(ctx);
return NULL;
}
Comment on lines +193 to +200
Copy link

@coderabbitai coderabbitai bot Aug 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Align eviction policy with TTL (match existing caches and docs).

When aws_pod_service_map_ttl == 0 you still create a TTL table with EVICT_OLDER. For consistency with other caches and the option help text (“evicted at random when capacity is reached”), use EVICT_RANDOM without TTL when ttl == 0.

-    ctx->aws_pod_service_hash_table = flb_hash_table_create_with_ttl(ctx->aws_pod_service_map_ttl,
-                                       FLB_HASH_TABLE_EVICT_OLDER,
-                                       FLB_HASH_TABLE_SIZE,
-                                       FLB_HASH_TABLE_SIZE);
+    if (ctx->aws_pod_service_map_ttl > 0) {
+        ctx->aws_pod_service_hash_table =
+            flb_hash_table_create_with_ttl(ctx->aws_pod_service_map_ttl,
+                                           FLB_HASH_TABLE_EVICT_OLDER,
+                                           FLB_HASH_TABLE_SIZE,
+                                           FLB_HASH_TABLE_SIZE);
+    }
+    else {
+        ctx->aws_pod_service_hash_table =
+            flb_hash_table_create(FLB_HASH_TABLE_EVICT_RANDOM,
+                                  FLB_HASH_TABLE_SIZE,
+                                  FLB_HASH_TABLE_SIZE);
+    }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
ctx->aws_pod_service_hash_table = flb_hash_table_create_with_ttl(ctx->aws_pod_service_map_ttl,
FLB_HASH_TABLE_EVICT_OLDER,
FLB_HASH_TABLE_SIZE,
FLB_HASH_TABLE_SIZE);
if (!ctx->aws_pod_service_hash_table) {
flb_kube_conf_destroy(ctx);
return NULL;
}
if (ctx->aws_pod_service_map_ttl > 0) {
ctx->aws_pod_service_hash_table =
flb_hash_table_create_with_ttl(ctx->aws_pod_service_map_ttl,
FLB_HASH_TABLE_EVICT_OLDER,
FLB_HASH_TABLE_SIZE,
FLB_HASH_TABLE_SIZE);
}
else {
ctx->aws_pod_service_hash_table =
flb_hash_table_create(FLB_HASH_TABLE_EVICT_RANDOM,
FLB_HASH_TABLE_SIZE,
FLB_HASH_TABLE_SIZE);
}
if (!ctx->aws_pod_service_hash_table) {
flb_kube_conf_destroy(ctx);
return NULL;
}
🤖 Prompt for AI Agents 
In plugins/filter_kubernetes/kube_conf.c around lines 193 to 200, the
aws_pod_service_hash_table is always created with a TTL and the EVICT_OLDER
policy; change this so when ctx->aws_pod_service_map_ttl == 0 you create the
hash table with no TTL and use FLB_HASH_TABLE_EVICT_RANDOM (to match other
caches and docs), otherwise keep the current flb_hash_table_create_with_ttl call
and FLB_HASH_TABLE_EVICT_OLDER; ensure the conditional creates the appropriate
table and retains the existing error check and cleanup path if creation fails.

return ctx;
}

Expand All @@ -206,6 +215,10 @@ void flb_kube_conf_destroy(struct flb_kube *ctx)
flb_hash_table_destroy(ctx->namespace_hash_table);
}

if (ctx->aws_pod_service_hash_table) {
flb_hash_table_destroy(ctx->aws_pod_service_hash_table);
}

if (ctx->merge_log == FLB_TRUE) {
flb_free(ctx->unesc_buf);
}
Expand All @@ -214,6 +227,9 @@ void flb_kube_conf_destroy(struct flb_kube *ctx)
if (ctx->parser == NULL && ctx->regex) {
flb_regex_destroy(ctx->regex);
}
if (ctx->deploymentRegex) {
flb_regex_destroy(ctx->deploymentRegex);
}

flb_free(ctx->api_host);
flb_free(ctx->token);
Expand All @@ -228,6 +244,18 @@ void flb_kube_conf_destroy(struct flb_kube *ctx)
flb_upstream_destroy(ctx->kube_api_upstream);
}

if (ctx->aws_pod_association_tls) {
flb_tls_destroy(ctx->aws_pod_association_tls);
}

if (ctx->aws_pod_association_upstream) {
flb_upstream_destroy(ctx->aws_pod_association_upstream);
}

if (ctx->platform) {
flb_free(ctx->platform);
}
Comment on lines +247 to +257
Copy link

@coderabbitai coderabbitai bot Aug 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Free newly-added config strings in destroy path (memory leaks).

aws_pod_association_* strings, aws_pod_service_preload_cache_path (sds), and set_platform are not freed.

     if (ctx->aws_pod_association_tls) {
         flb_tls_destroy(ctx->aws_pod_association_tls);
     }

     if (ctx->aws_pod_association_upstream) {
         flb_upstream_destroy(ctx->aws_pod_association_upstream);
     }
+
+    /* Free AWS pod association config strings */
+    flb_free(ctx->aws_pod_association_host);
+    flb_free(ctx->aws_pod_association_endpoint);
+    if (ctx->aws_pod_service_preload_cache_path) {
+        flb_sds_destroy(ctx->aws_pod_service_preload_cache_path);
+    }
+    flb_free(ctx->aws_pod_association_host_server_ca_file);
+    flb_free(ctx->aws_pod_association_host_client_cert_file);
+    flb_free(ctx->aws_pod_association_host_client_key_file);
+    flb_free(ctx->set_platform);


#ifdef FLB_HAVE_TLS
if (ctx->tls) {
flb_tls_destroy(ctx->tls);
Expand Down
68 changes: 68 additions & 0 deletions plugins/filter_kubernetes/kube_conf.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,8 +65,40 @@
#define FLB_KUBE_TAG_PREFIX "kube.var.log.containers."
#endif

/*
* Maximum attribute length for Entity's KeyAttributes
* values
* https://docs.aws.amazon.com/applicationsignals/latest/APIReference/API_Service.html#:~:text=Maximum%20length%20of%201024.
*/
#define KEY_ATTRIBUTES_MAX_LEN 1024
#define SERVICE_NAME_SOURCE_MAX_LEN 64

/*
* Configmap used for verifying whether if FluentBit is
* on EKS or native Kubernetes
*/
#define KUBE_SYSTEM_NAMESPACE "kube-system"
#define AWS_AUTH_CONFIG_MAP "aws-auth"

/*
* Possible platform values for Kubernetes plugin
*/
#define NATIVE_KUBERNETES_PLATFORM "k8s"
#define EKS_PLATFORM "eks"

struct kube_meta;

struct service_attributes {
char name[KEY_ATTRIBUTES_MAX_LEN];
int name_len;
char environment[KEY_ATTRIBUTES_MAX_LEN];
int environment_len;
char name_source[SERVICE_NAME_SOURCE_MAX_LEN];
int name_source_len;
int fields;

};

/* Filter context */
struct flb_kube {
/* Configuration parameters */
Expand Down Expand Up @@ -124,6 +156,7 @@ struct flb_kube {

/* Regex context to parse records */
struct flb_regex *regex;
struct flb_regex *deploymentRegex;
struct flb_parser *parser;

/* TLS CA certificate file */
Expand Down Expand Up @@ -165,6 +198,41 @@ struct flb_kube {
int kube_meta_cache_ttl;
int kube_meta_namespace_cache_ttl;

/* Configuration used for enabling pod to service name mapping*/
int aws_use_pod_association;
char *aws_pod_association_host;
char *aws_pod_association_endpoint;
int aws_pod_association_port;

/*
* TTL is used to check how long should the mapped entry
* remain in the hash table
*/
struct flb_hash_table *aws_pod_service_hash_table;
int aws_pod_service_map_ttl;
int aws_pod_service_map_refresh_interval;
flb_sds_t aws_pod_service_preload_cache_path;
struct flb_upstream *aws_pod_association_upstream;
/*
* This variable holds the Kubernetes platform type
* Current checks for EKS or Native Kuberentes
*/
char *platform;
/*
* This value is used for holding the platform config
* value. Platform will be overriden with this variable
* if it's set
*/
char *set_platform;

//Agent TLS certs
struct flb_tls *aws_pod_association_tls;
char *aws_pod_association_host_server_ca_file;
char *aws_pod_association_host_client_cert_file;
char *aws_pod_association_host_client_key_file;
int aws_pod_association_host_tls_debug;
int aws_pod_association_host_tls_verify;

Comment on lines +201 to +235
Copy link

@coderabbitai coderabbitai bot Aug 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Make background-map state per-instance (avoid globals).

Move the background thread state into struct flb_kube: mutex, thread handle, args, and event loop pointer. This prevents cross-instance interference.

 #include <fluent-bit/flb_hash_table.h>
+/* For background map refresh */
+#include <pthread.h>
+struct mk_event_loop;

 struct flb_kube {
@@
     int aws_pod_association_host_tls_verify;
 
+    /* Background pod→service map worker (per instance) */
+    pthread_mutex_t metadata_mutex;
+    pthread_t       background_thread;
+    void           *task_args;      /* struct task_args* from kubernetes.c */
+    struct mk_event_loop *evl;
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
/* Configuration used for enabling pod to service name mapping*/
int aws_use_pod_association;
char *aws_pod_association_host;
char *aws_pod_association_endpoint;
int aws_pod_association_port;
/*
* TTL is used to check how long should the mapped entry
* remain in the hash table
*/
struct flb_hash_table *aws_pod_service_hash_table;
int aws_pod_service_map_ttl;
int aws_pod_service_map_refresh_interval;
flb_sds_t aws_pod_service_preload_cache_path;
struct flb_upstream *aws_pod_association_upstream;
/*
* This variable holds the Kubernetes platform type
* Current checks for EKS or Native Kuberentes
*/
char *platform;
/*
* This value is used for holding the platform config
* value. Platform will be overriden with this variable
* if it's set
*/
char *set_platform;
//Agent TLS certs
struct flb_tls *aws_pod_association_tls;
char *aws_pod_association_host_server_ca_file;
char *aws_pod_association_host_client_cert_file;
char *aws_pod_association_host_client_key_file;
int aws_pod_association_host_tls_debug;
int aws_pod_association_host_tls_verify;
++ b/plugins/filter_kubernetes/kube_conf.h
@@ -1,6 +1,10 @@
#ifndef FLB_KUBE_CONF_H
#define FLB_KUBE_CONF_H
#include <fluent-bit/flb_hash_table.h>
/* For background map refresh */
#include <pthread.h>
struct mk_event_loop;
struct flb_kube {
@@ -231,6 +235,11 @@ struct flb_kube {
int aws_pod_association_host_tls_verify;
/* Background pod→service map worker (per instance) */
pthread_mutex_t metadata_mutex;
pthread_t background_thread;
void *task_args; /* struct task_args* from kubernetes.c */
struct mk_event_loop *evl;
/* …remaining fields… */
};
🤖 Prompt for AI Agents 
In plugins/filter_kubernetes/kube_conf.h around lines 201 to 235, the
background-map thread state is currently global and can interfere across filter
instances; add per-instance fields to struct flb_kube instead: a pthread_mutex
(or flb_mutex type), a thread handle (pthread_t or equivalent), a pointer/struct
for thread args, and a pointer to the instance-specific event loop (struct
flb_event_loop *), then remove/replace any global background-map state
references to use these new struct flb_kube members so each filter instance owns
its own mutex, thread, args, and loop.

struct flb_tls *tls;
struct flb_tls *kubelet_tls;

Expand Down
Loading
Loading