filter_kubernetes: new configuration property 'kube_tag_prefix'

Kubernetes filter in Fluent Bit uses incoming record Tags to resolve
local metadata such as pod name and namespace. When used in conjunction
with in_tail a configured/expected tag could be changed leading to make
the filter regular expressions to fail.

The new 'kube_tag_prefix' configuration property aims to provide a way
for the user to specify what's the Tag prefix used in in_tail. It can be
used as follows:

  [INPUT]
      Name   tail
      Path   /var/log/containers/*.log
      Tag    kube.service.*

  [FILTER]
      Name             kubernetes
      Match            *
      Kube_Tag_Prefix  kube.service.

The default value for Kube_Tag_Prefix is 'kube.var.log.containers.'

As a side note, here is a simple explanation of the workflow:

  - A container log is stored as /var/log/container/apache-logs-annotated_default_apache-2367.log

  - Tail plugin read the file and emit a record with it content with the following tag:

    => kube.service.var.log.containers.apache-logs-annotated_default_apache-2367.log

    as you can see the Tag is expanded with the absolute path of the monitored file and
    also all slashes (/) are replaced with dots (.).

  - Filter Kubernetes needs to get the basename of the monitored file, so it uses the Tag
    associated for such purpose. Here Kube_Tag_Prefix is used to remove the prefix from the Tag
    appended previously in the Tail section:

    => apache-logs-annotated_default_apache-2367.log

  - Filter Kubernetes apply a Regex to lookup podname and namespace from the new content generated
    above ending in: podname = 'apache-logs' and namespace = 'default'.

With this new approach is possible to define custom Tags on Tail and make Filter Kubernetes still
compatible with that.

Signed-off-by: Eduardo Silva <[email protected]>

plugins/filter_kubernetes/kube_conf.c

@@ -189,6 +189,15 @@ struct flb_kube *flb_kube_conf_create(struct flb_filter_instance *i,
  }
  }
 
+ /* Kubernetes Tag prefix */
+ tmp = flb_filter_get_property("kube_tag_prefix", i);
+ if (tmp) {
+ ctx->kube_tag_prefix = flb_sds_create(tmp);
+ }
+ else {
+ ctx->kube_tag_prefix = flb_sds_create(FLB_KUBE_TAG_PREFIX);
+ }
+
  /* Kubernetes Token file */
  tmp = flb_filter_get_property("kube_token_file", i);
  if (!tmp) {
@@ -327,6 +336,8 @@ void flb_kube_conf_destroy(struct flb_kube *ctx)
  flb_free(ctx->merge_log_key);
  }
 
+ flb_sds_destroy(ctx->kube_tag_prefix);
+
  /* Destroy regex content only if a parser was not defined */
  if (ctx->parser == NULL && ctx->regex) {
  flb_regex_destroy(ctx->regex);

plugins/filter_kubernetes/kube_conf.h

@@ -27,6 +27,7 @@
 #include <fluent-bit/flb_upstream.h>
 #include <fluent-bit/flb_macros.h>
 #include <fluent-bit/flb_io.h>
+#include <fluent-bit/flb_sds.h>
 #include <fluent-bit/flb_regex.h>
 
 /*
@@ -54,6 +55,12 @@
 #define FLB_API_PORT 443
 #define FLB_API_TLS FLB_TRUE
 
+/*
+ * Default expected Kubernetes tag prefix, this is used mostly when source
+ * data comes from in_tail with custom tags like: kube.service.*
+ */
+#define FLB_KUBE_TAG_PREFIX "kube.var.log.containers."
+
 struct kube_meta;
 
 /* Filter context */
@@ -101,6 +108,9 @@ struct flb_kube {
  /* API Server end point */
  char kube_url[1024];
 
+ /* Kubernetes tag prefix */
+ flb_sds_t kube_tag_prefix;
+
  /* Regex context to parse records */
  struct flb_regex *regex;
  struct flb_parser *parser;

plugins/filter_kubernetes/kube_meta.c

@@ -629,6 +629,8 @@ static inline int extract_meta(struct flb_kube *ctx, char *tag, int tag_len,
  int i;
  size_t off = 0;
  ssize_t n;
+ int kube_tag_len;
+ unsigned char *kube_tag_str;
  char *container = NULL;
  int container_found = FLB_FALSE;
  int container_length = 0;
@@ -684,8 +686,22 @@ static inline int extract_meta(struct flb_kube *ctx, char *tag, int tag_len,
  msgpack_unpacked_destroy(&mp_result);
  }
  else {
- /* Lookup using Tag string */
- n = flb_regex_do(ctx->regex, (unsigned char *) tag, tag_len, &result);
+ /*
+ * Lookup metadata using regular expression. In order to let the
+ * regex work we need to know before hand what's the Tag prefix
+ * set and make sure the adjustment can be done.
+ */
+ kube_tag_len = flb_sds_len(ctx->kube_tag_prefix);
+ if (kube_tag_len + 1 >= tag_len) {
+ flb_error("[filter_kube] incoming record tag (%s) is shorter "
+ "than kube_tag_prefix value (%s)",
+ tag, ctx->kube_tag_prefix);
+ return -1;
+ }
+ kube_tag_str = (unsigned char *) tag + kube_tag_len;
+ kube_tag_len = tag_len - kube_tag_len;
+ n = flb_regex_do(ctx->regex, kube_tag_str, kube_tag_len, &result);
+
  }
 
  if (n <= 0) {

plugins/filter_kubernetes/kube_regex.h

@@ -23,7 +23,7 @@
 
 #include "kube_conf.h"
 
-#define KUBE_TAG_TO_REGEX "(?<tag>[^.]+)?\\.?(?<pod_name>[a-z0-9](?:[-a-z0-9]*[a-z0-9])?(?:\\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*)_(?<namespace_name>[^_]+)_(?<container_name>.+)-(?<docker_id>[a-z0-9]{64})\\.log$"
+#define KUBE_TAG_TO_REGEX "(?<pod_name>[a-z0-9](?:[-a-z0-9]*[a-z0-9])?(?:\\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*)_(?<namespace_name>[^_]+)_(?<container_name>.+)-(?<docker_id>[a-z0-9]{64})\\.log$"
 
 #define KUBE_JOURNAL_TO_REGEX "^(?<name_prefix>[^_]+)_(?<container_name>[^\\._]+)(\\.(?<container_hash>[^_]+))?_(?<pod_name>[^_]+)_(?<namespace_name>[^_]+)_[^_]+_[^_]+$"