in_elasticsearch: reimplement tag_key lookup with record accessor api

Replace manual key lookup with record accessor pattern for better
performance and support for nested/complex key patterns.

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

Author: edsiper

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