in_http: in_splunk: in_elasticsearch: fix tag_key lookup [Backport to 4.0]

plugins/in_elasticsearch/in_elasticsearch.h

@@ -25,6 +25,7 @@
 #include <fluent-bit/flb_input.h>
 #include <fluent-bit/flb_utils.h>
 #include <fluent-bit/flb_log_event_encoder.h>
+#include <fluent-bit/flb_record_accessor.h>
 
 #include <monkey/monkey.h>
 #include <fluent-bit/http_server/flb_http_server.h>
@@ -35,14 +36,15 @@
 struct flb_in_elasticsearch {
     flb_sds_t listen;
     flb_sds_t tcp_port;
-    const char *tag_key;
-    const char *meta_key;
+    flb_sds_t tag_key;
+    flb_sds_t meta_key;
     flb_sds_t hostname;
     flb_sds_t es_version;
     char cluster_name[16];
     char node_name[12];
 
     struct flb_log_event_encoder *log_encoder;
+    struct flb_record_accessor *ra_tag_key;
 
     struct flb_input_instance *ins;
 

plugins/in_elasticsearch/in_elasticsearch_bulk_prot.c

@@ -22,6 +22,8 @@
 #include <fluent-bit/flb_error.h>
 #include <fluent-bit/flb_pack.h>
 #include <fluent-bit/flb_gzip.h>
+#include <fluent-bit/flb_record_accessor.h>
+#include <fluent-bit/flb_ra_key.h>
 
 #include <monkey/monkey.h>
 #include <monkey/mk_core.h>
@@ -245,67 +247,32 @@ static int send_response(struct in_elasticsearch_bulk_conn *conn, int http_statu
 /* implements functionality to get tag from key in record */
 static flb_sds_t tag_key(struct flb_in_elasticsearch *ctx, msgpack_object *map)
 {
-    size_t map_size = map->via.map.size;
-    msgpack_object_kv *kv;
-    msgpack_object  key;
-    msgpack_object  val;
-    char *key_str = NULL;
-    char *val_str = NULL;
-    size_t key_str_size = 0;
-    size_t val_str_size = 0;
-    int j;
-    int check = FLB_FALSE;
-    int found = FLB_FALSE;
-    flb_sds_t tag;
-
-    kv = map->via.map.ptr;
+    flb_sds_t tag = NULL;
+    struct flb_ra_value *ra_val;
 
-    for(j=0; j < map_size; j++) {
-        check = FLB_FALSE;
-        found = FLB_FALSE;
-        key = (kv+j)->key;
-        if (key.type == MSGPACK_OBJECT_BIN) {
-            key_str  = (char *) key.via.bin.ptr;
-            key_str_size = key.via.bin.size;
-            check = FLB_TRUE;
-        }
-        if (key.type == MSGPACK_OBJECT_STR) {
-            key_str  = (char *) key.via.str.ptr;
-            key_str_size = key.via.str.size;
-            check = FLB_TRUE;
-        }
-
-        if (check == FLB_TRUE) {
-            if (strncmp(ctx->tag_key, key_str, key_str_size) == 0) {
-                val = (kv+j)->val;
-                if (val.type == MSGPACK_OBJECT_BIN) {
-                    val_str  = (char *) val.via.bin.ptr;
-                    val_str_size = val.via.str.size;
-                    found = FLB_TRUE;
-                    break;
-                }
-                if (val.type == MSGPACK_OBJECT_STR) {
-                    val_str  = (char *) val.via.str.ptr;
-                    val_str_size = val.via.str.size;
-                    found = FLB_TRUE;
-                    break;
-                }
-            }
-        }
+    /* If no record accessor is configured, return NULL */
+    if (!ctx->ra_tag_key) {
+        return NULL;
     }
 
-    if (found == FLB_TRUE) {
-        tag = flb_sds_create_len(val_str, val_str_size);
-        if (!tag) {
-            flb_errno();
-            return NULL;
-        }
-        return tag;
+    /* Use record accessor to get the value */
+    ra_val = flb_ra_get_value_object(ctx->ra_tag_key, *map);
+    if (!ra_val) {
+        flb_plg_warn(ctx->ins, "Could not find tag_key %s in record", ctx->tag_key);
+        return NULL;
     }
 
+    /* Convert the value to string */
+    if (ra_val->type == FLB_RA_STRING) {
+        tag = flb_sds_create_len(ra_val->o.via.str.ptr, ra_val->o.via.str.size);
+    }
+    else {
+        flb_plg_error(ctx->ins, "tag_key %s value is not a string or binary", ctx->tag_key);
+    }
 
-    flb_plg_error(ctx->ins, "Could not find tag_key %s in record", ctx->tag_key);
-    return NULL;
+    /* Clean up the record accessor value */
+    flb_ra_key_value_destroy(ra_val);
+    return tag;
 }
 
 static int get_write_op(struct flb_in_elasticsearch *ctx, msgpack_object *map, flb_sds_t *out_write_op, size_t *out_key_size)

plugins/in_elasticsearch/in_elasticsearch_config.c

@@ -67,12 +67,25 @@ struct flb_in_elasticsearch *in_elasticsearch_config_create(struct flb_input_ins
         return NULL;
     }
 
+    /* Create record accessor for tag_key if specified */
+    if (ctx->tag_key) {
+        ctx->ra_tag_key = flb_ra_create(ctx->tag_key, FLB_TRUE);
+        if (!ctx->ra_tag_key) {
+            flb_plg_error(ctx->ins, "invalid record accessor pattern for tag_key: %s", ctx->tag_key);
+            in_elasticsearch_config_destroy(ctx);
+            return NULL;
+        }
+    }
 
     return ctx;
 }
 
 int in_elasticsearch_config_destroy(struct flb_in_elasticsearch *ctx)
 {
+    if (ctx->ra_tag_key) {
+        flb_ra_destroy(ctx->ra_tag_key);
+    }
+
     flb_log_event_encoder_destroy(ctx->log_encoder);
 
     /* release all connections */

plugins/in_http/http.h

@@ -25,6 +25,7 @@
 #include <fluent-bit/flb_input.h>
 #include <fluent-bit/flb_utils.h>
 #include <fluent-bit/flb_log_event_encoder.h>
+#include <fluent-bit/flb_record_accessor.h>
 
 #include <monkey/monkey.h>
 #include <fluent-bit/http_server/flb_http_server.h>
@@ -36,7 +37,8 @@ struct flb_http {
     int successful_response_code;
     flb_sds_t listen;
     flb_sds_t tcp_port;
-    const char *tag_key;
+    flb_sds_t tag_key;
+    struct flb_record_accessor *ra_tag_key;
 
     /* Success HTTP headers */
     struct mk_list *success_headers;

plugins/in_http/http_config.c

@@ -75,20 +75,27 @@ struct flb_http *http_config_create(struct flb_input_instance *ins)
 
     if (ret != FLB_EVENT_ENCODER_SUCCESS) {
         flb_plg_error(ctx->ins, "error initializing event encoder : %d", ret);
-
         http_config_destroy(ctx);
-
         return NULL;
     }
 
     ctx->success_headers_str = flb_sds_create_size(1);
 
     if (ctx->success_headers_str == NULL) {
         http_config_destroy(ctx);
-
         return NULL;
     }
 
+    /* Create record accessor for tag_key if specified */
+    if (ctx->tag_key) {
+        ctx->ra_tag_key = flb_ra_create(ctx->tag_key, FLB_TRUE);
+        if (!ctx->ra_tag_key) {
+            flb_plg_error(ctx->ins, "invalid record accessor pattern for tag_key: %s", ctx->tag_key);
+            http_config_destroy(ctx);
+            return NULL;
+        }
+    }
+
     flb_config_map_foreach(header_iterator, header_pair, ctx->success_headers) {
         header_name = mk_list_entry_first(header_pair->val.list,
                                           struct flb_slist_entry,
@@ -132,6 +139,10 @@ struct flb_http *http_config_create(struct flb_input_instance *ins)
 
 int http_config_destroy(struct flb_http *ctx)
 {
+    if (ctx->ra_tag_key) {
+        flb_ra_destroy(ctx->ra_tag_key);
+    }
+
     /* release all connections */
     http_conn_release_all(ctx);
 

plugins/in_http/http_prot.c

@@ -27,6 +27,8 @@
 #include <fluent-bit/flb_gzip.h>
 #include <fluent-bit/flb_zstd.h>
 #include <fluent-bit/flb_snappy.h>
+#include <fluent-bit/flb_record_accessor.h>
+#include <fluent-bit/flb_ra_key.h>
 
 #include <monkey/monkey.h>
 #include <monkey/mk_core.h>
@@ -42,8 +44,8 @@ static inline char hex2nibble(char c)
     if ((c >= 0x30) && (c <= '9')) {
         return c - 0x30;
     }
-    // 0x30-0x39 are digits, 0x41-0x46 A-F,
-    // so there is a gap at 0x40
+
+    /* 0x30-0x39 are digits, 0x41-0x46 A-F, so there is a gap at 0x40 */
     if ((c >= 'A') && (c <= 'F')) {
         return (c - 'A') + 10;
     }
@@ -163,70 +165,54 @@ static int send_response(struct http_conn *conn, int http_status, char *message)
     return 0;
 }
 
-/* implements functionality to get tag from key in record */
-static flb_sds_t tag_key(struct flb_http *ctx, msgpack_object *map)
+static void sanitize_tag(flb_sds_t tag)
 {
-    size_t map_size = map->via.map.size;
-    msgpack_object_kv *kv;
-    msgpack_object  key;
-    msgpack_object  val;
-    char *key_str = NULL;
-    char *val_str = NULL;
-    size_t key_str_size = 0;
-    size_t val_str_size = 0;
-    int j;
-    int check = FLB_FALSE;
-    int found = FLB_FALSE;
-    flb_sds_t tag;
+    size_t i;
 
-    kv = map->via.map.ptr;
+    if (!tag) {
+        return;
+    }
 
-    for(j=0; j < map_size; j++) {
-        check = FLB_FALSE;
-        found = FLB_FALSE;
-        key = (kv+j)->key;
-        if (key.type == MSGPACK_OBJECT_BIN) {
-            key_str  = (char *) key.via.bin.ptr;
-            key_str_size = key.via.bin.size;
-            check = FLB_TRUE;
-        }
-        if (key.type == MSGPACK_OBJECT_STR) {
-            key_str  = (char *) key.via.str.ptr;
-            key_str_size = key.via.str.size;
-            check = FLB_TRUE;
+    for (i = 0; i < flb_sds_len(tag); i++) {
+        if (!isalnum(tag[i]) && tag[i] != '_' && tag[i] != '.') {
+            tag[i] = '_';
         }
+    }
+}
 
-        if (check == FLB_TRUE) {
-            if (strncmp(ctx->tag_key, key_str, key_str_size) == 0) {
-                val = (kv+j)->val;
-                if (val.type == MSGPACK_OBJECT_BIN) {
-                    val_str  = (char *) val.via.bin.ptr;
-                    val_str_size = val.via.bin.size;
-                    found = FLB_TRUE;
-                    break;
-                }
-                if (val.type == MSGPACK_OBJECT_STR) {
-                    val_str  = (char *) val.via.str.ptr;
-                    val_str_size = val.via.str.size;
-                    found = FLB_TRUE;
-                    break;
-                }
-            }
-        }
+/* implements functionality to get tag from key in record */
+static flb_sds_t tag_key(struct flb_http *ctx, msgpack_object *map)
+{
+    struct flb_ra_value *ra_val;
+    flb_sds_t tag = NULL;
+
+    /* If no record accessor is configured, return NULL */
+    if (!ctx->ra_tag_key) {
+        return NULL;
     }
 
-    if (found == FLB_TRUE) {
-        tag = flb_sds_create_len(val_str, val_str_size);
-        if (!tag) {
-            flb_errno();
-            return NULL;
+    /* Use record accessor to get the value */
+    ra_val = flb_ra_get_value_object(ctx->ra_tag_key, *map);
+    if (!ra_val) {
+        flb_plg_debug(ctx->ins, "Could not find tag_key %s in record", ctx->tag_key);
+        return NULL;
+    }
+
+    /* Convert the value to string */
+    if (ra_val->type == FLB_RA_STRING) {
+        tag = flb_sds_create_len(ra_val->o.via.str.ptr, ra_val->o.via.str.size);
+        if (tag) {
+            sanitize_tag(tag);
         }
-        return tag;
+    }
+    else {
+        flb_plg_debug(ctx->ins, "tag_key %s value is not a string", ctx->tag_key);
     }
 
+    /* Clean up the record accessor value */
+    flb_ra_key_value_destroy(ra_val);
 
-    flb_plg_warn(ctx->ins, "Could not find tag_key %s in record", ctx->tag_key);
-    return NULL;
+    return tag;
 }
 
 static int process_pack_record(struct flb_http *ctx, struct flb_time *tm,
@@ -878,7 +864,6 @@ int http_prot_handle(struct flb_http *ctx, struct http_conn *conn,
                      struct mk_http_session *session,
                      struct mk_http_request *request)
 {
-    int i;
     int ret;
     int len;
     char *uri;
@@ -927,12 +912,7 @@ int http_prot_handle(struct flb_http *ctx, struct http_conn *conn,
         /* New tag skipping the URI '/' */
         flb_sds_cat_safe(&tag, uri + 1, len - 1);
 
-        /* Sanitize, only allow alphanum chars */
-        for (i = 0; i < flb_sds_len(tag); i++) {
-            if (!isalnum(tag[i]) && tag[i] != '_' && tag[i] != '.') {
-                tag[i] = '_';
-            }
-        }
+        sanitize_tag(tag);
     }
 
     mk_mem_free(uri);

plugins/in_splunk/splunk.c

@@ -135,6 +135,9 @@ static int in_splunk_init(struct flb_input_instance *ins,
         ctx->http_server.request_callback = splunk_prot_handle_ng;
 
         flb_input_downstream_set(ctx->http_server.downstream, ctx->ins);
+
+        flb_plg_info(ctx->ins, "listening on %s:%u",
+                     ins->host.listen, ins->host.port);
     }
     else {
         ctx->downstream = flb_downstream_create(FLB_TRANSPORT_TCP,
@@ -157,6 +160,8 @@ static int in_splunk_init(struct flb_input_instance *ins,
 
         flb_input_downstream_set(ctx->downstream, ctx->ins);
 
+        flb_plg_info(ctx->ins, "listening on %s:%s", ctx->listen, ctx->tcp_port);
+
         /* Collect upon data available on the standard input */
         ret = flb_input_set_collector_socket(ins,
                                             in_splunk_collect,
@@ -172,6 +177,7 @@ static int in_splunk_init(struct flb_input_instance *ins,
         ctx->collector_id = ret;
     }
 
+
     return 0;
 }