out_splunk: reduce noise and fix hec_token handling (fix fluent#8859)

The following patch perform 2 changes in the code that helps to fix the
problems found with Splunk hec token handling:

1. In the recent PR fluent#8793, when using the record accessor API flb_ra_translate_check()
   to validate if the hec_token field exists, leads to noisy log messages since
   that function warns the issue if the field is not found. Most of users are not
   using hec_token set by Splunk input plugin, so their logging gets noisy.

   This patch replaces that call with flb_ra_translate() which fixes the problem.

2. If hec_token was set in the record metadata, it was being store in the main
   context of the plugin, however the flush callbacks that formats and deliver the
   data runs in separate/parallel threads that could lead to a race condition if
   more than onen thread tries to manipulate the value.

   This patch adds protection to the context value so it becomes thread safe.

Signed-off-by: Eduardo Silva <eduardo@calyptia.com>
Signed-off-by: Markus Bergholz <git@osuv.de>

plugins/out_splunk/splunk.c

@@ -354,12 +354,27 @@ static flb_sds_t extract_hec_token(struct flb_splunk *ctx, msgpack_object map,
 
     /* Extract HEC token (map which is from metadata lookup) */
     if (ctx->metadata_auth_key) {
-        hec_token = flb_ra_translate_check(ctx->ra_metadata_auth_key, tag, tag_len,
-                                     map, NULL, FLB_TRUE);
-        if (hec_token) {
+        hec_token = flb_ra_translate(ctx->ra_metadata_auth_key, tag, tag_len,
+                                     map, NULL);
+        /*
+         * record accessor translation can return an empty string buffer if the
+         * translation was not successfull or the value was not found. We consider
+         * a valid token any string which length is greater than 0.
+         *
+         * note: flb_ra_translate_check() is not used here because it will print
+         * an error message if the translation fails:
+         *
+         * ref: https://github.com/fluent/fluent-bit/issues/8859
+         */
+        if (hec_token && flb_sds_len(hec_token) > 0) {
             return hec_token;
         }
 
+        /* destroy empty string */
+        if (hec_token) {
+            flb_sds_destroy(hec_token);
+        }
+
         flb_plg_debug(ctx->ins, "Could not find hec_token in metadata");
         return NULL;
     }
@@ -368,21 +383,45 @@ static flb_sds_t extract_hec_token(struct flb_splunk *ctx, msgpack_object map,
     return NULL;
 }
 
+static void set_metadata_auth_header(struct flb_splunk *ctx, flb_sds_t hec_token)
+{
+    pthread_mutex_lock(&ctx->mutex_hec_token);
+
+    if (ctx->metadata_auth_header != NULL) {
+        flb_sds_destroy(ctx->metadata_auth_header);
+    }
+    ctx->metadata_auth_header = hec_token;
+
+    pthread_mutex_unlock(&ctx->mutex_hec_token);
+}
+
+static flb_sds_t get_metadata_auth_header(struct flb_splunk *ctx)
+{
+    flb_sds_t auth_header = NULL;
+
+    pthread_mutex_lock(&ctx->mutex_hec_token);
+    auth_header = flb_sds_create(ctx->metadata_auth_header);
+    pthread_mutex_unlock(&ctx->mutex_hec_token);
+
+    return auth_header;
+}
+
 static inline int splunk_format(const void *in_buf, size_t in_bytes,
                                 char *tag, int tag_len,
                                 char **out_buf, size_t *out_size,
                                 struct flb_splunk *ctx)
 {
     int ret;
+    char *err;
     msgpack_object map;
     msgpack_object metadata;
     msgpack_sbuffer mp_sbuf;
     msgpack_packer mp_pck;
-    char *err;
     flb_sds_t tmp;
     flb_sds_t record;
     flb_sds_t json_out;
     flb_sds_t metadata_hec_token = NULL;
+
     struct flb_log_event_decoder log_decoder;
     struct flb_log_event log_event;
 
@@ -403,8 +442,6 @@ static inline int splunk_format(const void *in_buf, size_t in_bytes,
         return -1;
     }
 
-    ctx->metadata_auth_header = NULL;
-
     while ((ret = flb_log_event_decoder_next(
                     &log_decoder,
                     &log_event)) == FLB_EVENT_DECODER_SUCCESS) {
@@ -422,10 +459,7 @@ static inline int splunk_format(const void *in_buf, size_t in_bytes,
              * specify only one splunk token per one instance.
              * So, it should be valid if storing only last value of
              * splunk token per one chunk. */
-            if (ctx->metadata_auth_header != NULL) {
-                cfl_sds_destroy(ctx->metadata_auth_header);
-            }
-            ctx->metadata_auth_header = metadata_hec_token;
+            set_metadata_auth_header(ctx, metadata_hec_token);
         }
 
         if (ctx->event_key) {
@@ -598,6 +632,7 @@ static void cb_splunk_flush(struct flb_event_chunk *event_chunk,
     size_t payload_size;
     (void) i_ins;
     (void) config;
+    flb_sds_t metadata_auth_header = NULL;
 
     /* Get upstream connection */
     u_conn = flb_upstream_conn_get(ctx->u);
@@ -677,16 +712,19 @@ static void cb_splunk_flush(struct flb_event_chunk *event_chunk,
         flb_http_buffer_size(c, resp_size);
     }
 
+    metadata_auth_header = get_metadata_auth_header(ctx);
+
     /* HTTP Client */
     flb_http_add_header(c, "User-Agent", 10, "Fluent-Bit", 10);
 
     /* Try to use http_user and http_passwd if not, fallback to auth_header */
     if (ctx->http_user && ctx->http_passwd) {
         flb_http_basic_auth(c, ctx->http_user, ctx->http_passwd);
     }
-    else if (ctx->metadata_auth_header) {
+    else if (metadata_auth_header) {
         flb_http_add_header(c, "Authorization", 13,
-                            ctx->metadata_auth_header, flb_sds_len(ctx->metadata_auth_header));
+                            metadata_auth_header,
+                            flb_sds_len(metadata_auth_header));
     }
     else if (ctx->auth_header) {
         flb_http_add_header(c, "Authorization", 13,
@@ -754,10 +792,10 @@ static void cb_splunk_flush(struct flb_event_chunk *event_chunk,
         flb_sds_destroy(buf_data);
     }
 
-    /* Cleanup */
-    if (ctx->metadata_auth_header != NULL) {
-        cfl_sds_destroy(ctx->metadata_auth_header);
+    if (metadata_auth_header) {
+        flb_sds_destroy(metadata_auth_header);
     }
+
     flb_http_client_destroy(c);
     flb_upstream_conn_release(u_conn);
     FLB_OUTPUT_RETURN(ret);

plugins/out_splunk/splunk.h

@@ -120,6 +120,8 @@ struct flb_splunk {
 
     /* Plugin instance */
     struct flb_output_instance *ins;
+
+    pthread_mutex_t mutex_hec_token;
 };
 
 #endif

plugins/out_splunk/splunk_conf.c

@@ -262,6 +262,8 @@ struct flb_splunk *flb_splunk_conf_create(struct flb_output_instance *ins,
         }
     }
 
+    pthread_mutex_init(&ctx->mutex_hec_token, NULL);
+
     /* Currently, Splunk HEC token is stored in a fixed key, hec_token. */
     ctx->metadata_auth_key = "$hec_token";
     if (ctx->metadata_auth_key) {
@@ -325,6 +327,10 @@ int flb_splunk_conf_destroy(struct flb_splunk *ctx)
         flb_ra_destroy(ctx->ra_metadata_auth_key);
     }
 
+    if (ctx->metadata_auth_header) {
+        flb_sds_destroy(ctx->metadata_auth_header);
+    }
+
     event_fields_destroy(ctx);
 
     flb_free(ctx);