in_splunk: use record accessor for tag key extraction and add listening info

Replace manual key lookup with record accessor pattern for better
performance and support for nested/complex key patterns. Add log
messages showing listening interface and port for both HTTP versions.

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

Author: edsiper

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

plugins/in_splunk/splunk.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>
@@ -41,7 +42,8 @@ struct flb_splunk_tokens {
 struct flb_splunk {
     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_splunk/splunk_config.c

@@ -229,11 +229,25 @@ struct flb_splunk *splunk_config_create(struct flb_input_instance *ins)
         }
     }
 
+    /* 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);
+            splunk_config_destroy(ctx);
+            return NULL;
+        }
+    }
+
     return ctx;
 }
 
 int splunk_config_destroy(struct flb_splunk *ctx)
 {
+    if (ctx->ra_tag_key) {
+        flb_ra_destroy(ctx->ra_tag_key);
+    }
+
     /* release all connections */
     splunk_conn_release_all(ctx);
 

plugins/in_splunk/splunk_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>
@@ -156,67 +158,33 @@ static int send_json_message_response(struct splunk_conn *conn, int http_status,
 /* implements functionality to get tag from key in record */
 static flb_sds_t tag_key(struct flb_splunk *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;
+    flb_sds_t tag = NULL;
+    struct flb_ra_value *ra_val;
 
-    kv = map->via.map.ptr;
-
-    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 no record accessor is configured, return NULL */
+    if (!ctx->ra_tag_key) {
+        return NULL;
+    }
 
-        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;
-                }
-            }
-        }
+    /* 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;
     }
 
-    if (found == FLB_TRUE) {
-        tag = flb_sds_create_len(val_str, val_str_size);
-        if (!tag) {
-            flb_errno();
-            return NULL;
-        }
-        return tag;
+    /* 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_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_error(ctx->ins, "Could not find tag_key %s in record", ctx->tag_key);
-    return NULL;
+    return tag;
 }
 
 /*