From 808e5eef68526ea65103e012664ac040af44ac3b Mon Sep 17 00:00:00 2001 From: Povilas Vaitkus Date: Tue, 28 Jul 2026 19:41:41 +0300 Subject: [PATCH] filter_kubernetes: bound metadata fetch I/O and negative-cache failures The pod metadata fetch is synchronous with async disabled and no read timeout, on the pipeline/event-loop thread. A Kubernetes API server / kubelet connection that establishes then stalls blocks recv() indefinitely, so in_tail (same event loop) stops collecting and the tail input flatlines while the pod stays healthy. Failed lookups are not cached, so every record for an unseen pod re-issues the blocking request. Add two safeguards (defaults on, 0 = legacy): - kube_meta_io_timeout (30s): read timeout on the API/kubelet upstreams so a stalled read errors instead of wedging the pipeline forever. - kube_meta_negative_cache_ttl (60s): remember a failed lookup briefly so records for that pod pass through instead of re-blocking on every line. Fixes #12168 Signed-off-by: Povilas Vaitkus --- plugins/filter_kubernetes/kube_conf.c | 16 ++++++++++ plugins/filter_kubernetes/kube_conf.h | 7 ++++ plugins/filter_kubernetes/kube_meta.c | 44 ++++++++++++++++++++++++++ plugins/filter_kubernetes/kubernetes.c | 14 ++++++++ 4 files changed, 81 insertions(+) diff --git a/plugins/filter_kubernetes/kube_conf.c b/plugins/filter_kubernetes/kube_conf.c index 73e73495e92..38b8a852178 100644 --- a/plugins/filter_kubernetes/kube_conf.c +++ b/plugins/filter_kubernetes/kube_conf.c @@ -152,6 +152,18 @@ struct flb_kube *flb_kube_conf_create(struct flb_filter_instance *ins, } + if (ctx->kube_meta_negative_cache_ttl > 0) { + ctx->neg_hash_table = flb_hash_table_create_with_ttl( + ctx->kube_meta_negative_cache_ttl, + FLB_HASH_TABLE_EVICT_OLDER, + FLB_HASH_TABLE_SIZE, + FLB_HASH_TABLE_SIZE); + if (!ctx->neg_hash_table) { + flb_kube_conf_destroy(ctx); + return NULL; + } + } + if (!ctx->hash_table || !ctx->namespace_hash_table) { flb_kube_conf_destroy(ctx); return NULL; @@ -215,6 +227,10 @@ void flb_kube_conf_destroy(struct flb_kube *ctx) flb_hash_table_destroy(ctx->namespace_hash_table); } + if (ctx->neg_hash_table) { + flb_hash_table_destroy(ctx->neg_hash_table); + } + if (ctx->aws_pod_service_hash_table) { flb_hash_table_destroy(ctx->aws_pod_service_hash_table); } diff --git a/plugins/filter_kubernetes/kube_conf.h b/plugins/filter_kubernetes/kube_conf.h index 781053c9274..001f64b7366 100644 --- a/plugins/filter_kubernetes/kube_conf.h +++ b/plugins/filter_kubernetes/kube_conf.h @@ -198,6 +198,12 @@ struct flb_kube { int kube_meta_cache_ttl; int kube_meta_namespace_cache_ttl; + /* Bound the synchronous metadata fetch and remember recently-failed + * lookups so a stalled API server / kubelet connection cannot block the + * pipeline thread (and therefore in_tail) indefinitely. */ + int kube_meta_io_timeout; + int kube_meta_negative_cache_ttl; + /* Configuration used for enabling pod to service name mapping*/ int aws_use_pod_association; char *aws_pod_association_host; @@ -239,6 +245,7 @@ struct flb_kube { struct flb_config *config; struct flb_hash_table *hash_table; struct flb_hash_table *namespace_hash_table; + struct flb_hash_table *neg_hash_table; struct flb_upstream *kubelet_upstream; struct flb_upstream *kube_api_upstream; struct flb_filter_instance *ins; diff --git a/plugins/filter_kubernetes/kube_meta.c b/plugins/filter_kubernetes/kube_meta.c index f8371b1a826..cb3e50d5a82 100644 --- a/plugins/filter_kubernetes/kube_meta.c +++ b/plugins/filter_kubernetes/kube_meta.c @@ -2221,6 +2221,10 @@ static int flb_kubelet_network_init(struct flb_kube *ctx, struct flb_config *con /* Remove async flag from upstream */ flb_stream_disable_async_mode(&ctx->kubelet_upstream->base); + if (ctx->kube_meta_io_timeout > 0) { + ctx->kubelet_upstream->base.net.io_timeout = ctx->kube_meta_io_timeout; + } + return 0; } @@ -2276,6 +2280,12 @@ static int flb_kube_network_init(struct flb_kube *ctx, struct flb_config *config /* Remove async flag from upstream */ flb_stream_disable_async_mode(&ctx->kube_api_upstream->base); + /* Bound the (synchronous) metadata read so a stalled connection cannot + * block the pipeline thread -- and in_tail -- forever (0 = legacy). */ + if (ctx->kube_meta_io_timeout > 0) { + ctx->kube_api_upstream->base.net.io_timeout = ctx->kube_meta_io_timeout; + } + /* Continue the filter kubernetes plugin functionality if the pod_association fails */ if (ctx->aws_use_pod_association) { flb_kube_pod_association_init(ctx, config); @@ -2424,10 +2434,44 @@ static inline int lookup_pod_meta(struct flb_kube *ctx, meta->cache_key, meta->cache_key_len, (void *) &hash_meta_buf, &hash_meta_size); if (ret == -1) { + /* + * Skip the blocking request if this pod's metadata lookup failed + * recently, so records for an unseen pod do not each re-issue a + * synchronous request while the control plane is unreachable. + */ + if (ctx->neg_hash_table) { + const char *neg_buf; + size_t neg_size; + if (flb_hash_table_get(ctx->neg_hash_table, + meta->cache_key, meta->cache_key_len, + (void *) &neg_buf, &neg_size) != -1) { + flb_plg_debug(ctx->ins, + "negative cache hit for %.*s, skipping metadata " + "lookup (record left un-enriched)", + (int) meta->cache_key_len, meta->cache_key); + *out_buf = NULL; + *out_size = 0; + return 0; + } + } + /* Retrieve API server meta and merge with local meta */ ret = get_and_merge_pod_meta(ctx, meta, &tmp_hash_meta_buf, &hash_meta_size); if (ret == -1) { + /* + * Only negatively-cache network lookup failures. With + * use_tag_for_meta the metadata is derived from the tag (no + * request), so a failure there must not suppress future work. + */ + if (ctx->neg_hash_table && !ctx->use_tag_for_meta) { + flb_plg_debug(ctx->ins, + "metadata lookup failed for %.*s, negative-caching", + (int) meta->cache_key_len, meta->cache_key); + flb_hash_table_add(ctx->neg_hash_table, + meta->cache_key, meta->cache_key_len, + "1", 1); + } *out_buf = NULL; *out_size = 0; return 0; diff --git a/plugins/filter_kubernetes/kubernetes.c b/plugins/filter_kubernetes/kubernetes.c index 24548e5fa68..18d6c2d1150 100644 --- a/plugins/filter_kubernetes/kubernetes.c +++ b/plugins/filter_kubernetes/kubernetes.c @@ -1126,6 +1126,20 @@ static struct flb_config_map config_map[] = { 0, FLB_TRUE, offsetof(struct flb_kube, kube_token_ttl), "kubernetes token ttl, until it is reread from the token file. Default: 10m" }, + { + FLB_CONFIG_MAP_TIME, "kube_meta_io_timeout", "30s", + 0, FLB_TRUE, offsetof(struct flb_kube, kube_meta_io_timeout), + "network read timeout for the API server / kubelet metadata request. " + "The request is synchronous on the pipeline thread, so without a bound a " + "stalled connection blocks in_tail indefinitely. Default: 30s (0 = legacy)." + }, + { + FLB_CONFIG_MAP_TIME, "kube_meta_negative_cache_ttl", "60s", + 0, FLB_TRUE, offsetof(struct flb_kube, kube_meta_negative_cache_ttl), + "how long a failed pod metadata lookup is remembered so records for that " + "pod pass through un-enriched instead of re-issuing a blocking request per " + "line while the control plane is unreachable. Default: 60s (0 = disabled)." + }, /* * Set TTL for K8s cached metadata */