filter_kubernetes: bound metadata fetch I/O and negative-cache failed lookups

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);
     }

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;

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;

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 
      */