filter_parser: support record accessor key

Author: agup006

plugins/filter_parser/filter_parser.c

@@ -102,6 +102,7 @@ static int configure(struct filter_parser_ctx *ctx,
     struct flb_kv *kv;
 
     ctx->key_name = NULL;
+    ctx->ra_key = NULL;
     ctx->reserve_data = FLB_FALSE;
     ctx->preserve_key = FLB_FALSE;
     mk_list_init(&ctx->parsers);
@@ -118,6 +119,15 @@ static int configure(struct filter_parser_ctx *ctx,
     }
     ctx->key_name_len = flb_sds_len(ctx->key_name);
 
+    if (ctx->key_name && ctx->key_name[0] == '$') {
+        ctx->ra_key = flb_ra_create(ctx->key_name, FLB_TRUE);
+        if (!ctx->ra_key) {
+            flb_plg_error(ctx->ins, "invalid record accessor pattern '%s'",
+                          ctx->key_name);
+            return -1;
+        }
+    }
+
     /* Read all Parsers */
     mk_list_foreach(head, &f_ins->properties) {
         kv = mk_list_entry(head, struct flb_kv, _head);
@@ -246,20 +256,11 @@ static int cb_parser_filter(const void *data, size_t bytes,
                 }
             }
 
-            /* Process the target key */
-            for (i = 0; i < map_num; i++) {
-                kv = &obj->via.map.ptr[i];
-                if (msgpackobj2char(&kv->key, &key_str, &key_len) < 0) {
-                    continue;
-                }
+            if (ctx->ra_key) {
+                struct flb_ra_value *rval;
 
-                if (key_len == ctx->key_name_len &&
-                    !strncmp(key_str, ctx->key_name, key_len)) {
-                    if (msgpackobj2char(&kv->val, &val_str, &val_len) < 0) {
-                        continue;
-                    }
-
-                    /* Lookup parser */
+                rval = flb_ra_get_value_object(ctx->ra_key, *obj);
+                if (rval && msgpackobj2char(&rval->o, &val_str, &val_len) == 0) {
                     mk_list_foreach(head, &ctx->parsers) {
                         fp = mk_list_entry(head, struct filter_parser, _head);
                         flb_time_zero(&parsed_time);
@@ -271,17 +272,53 @@ static int cb_parser_filter(const void *data, size_t bytes,
                             if (flb_time_to_nanosec(&parsed_time) != 0L) {
                                 flb_time_copy(&tm, &parsed_time);
                             }
+                            break;
+                        }
+                    }
+                }
 
-                            if (append_arr != NULL) {
-                                if (!ctx->preserve_key) {
-                                    append_arr[i] = NULL;
+                if (rval) {
+                    flb_ra_key_value_destroy(rval);
+                }
+            }
+            else {
+                /* Process the target key */
+                for (i = 0; i < map_num; i++) {
+                    kv = &obj->via.map.ptr[i];
+                    if (msgpackobj2char(&kv->key, &key_str, &key_len) < 0) {
+                        continue;
+                    }
+
+                    if (key_len == ctx->key_name_len &&
+                        !strncmp(key_str, ctx->key_name, key_len)) {
+                        if (msgpackobj2char(&kv->val, &val_str, &val_len) < 0) {
+                            continue;
+                        }
+
+                        /* Lookup parser */
+                        mk_list_foreach(head, &ctx->parsers) {
+                            fp = mk_list_entry(head, struct filter_parser, _head);
+                            flb_time_zero(&parsed_time);
+
+                            parse_ret = flb_parser_do(fp->parser, val_str, val_len,
+                                                      (void **) &out_buf, &out_size,
+                                                      &parsed_time);
+                            if (parse_ret >= 0) {
+                                if (flb_time_to_nanosec(&parsed_time) != 0L) {
+                                    flb_time_copy(&tm, &parsed_time);
                                 }
-                                else if (!ctx->reserve_data) {
-                                    /* Store only the key being preserved */
-                                    append_arr[0] = kv;
+
+                                if (append_arr != NULL) {
+                                    if (!ctx->preserve_key) {
+                                        append_arr[i] = NULL;
+                                    }
+                                    else if (!ctx->reserve_data) {
+                                        /* Store only the key being preserved */
+                                        append_arr[0] = kv;
+                                    }
                                 }
+                                break;
                             }
-                            break;
                         }
                     }
                 }
@@ -413,6 +450,9 @@ static int cb_parser_exit(void *data, struct flb_config *config)
     }
 
     delete_parsers(ctx);
+    if (ctx->ra_key) {
+        flb_ra_destroy(ctx->ra_key);
+    }
     flb_free(ctx);
     return 0;
 }

plugins/filter_parser/filter_parser.h

@@ -24,6 +24,8 @@
 #include <fluent-bit/flb_filter.h>
 #include <fluent-bit/flb_parser.h>
 #include <fluent-bit/flb_sds.h>
+#include <fluent-bit/flb_record_accessor.h>
+#include <fluent-bit/flb_ra_key.h>
 
 struct filter_parser {
     struct flb_parser *parser;
@@ -33,6 +35,7 @@ struct filter_parser {
 struct filter_parser_ctx {
     flb_sds_t key_name;
     int    key_name_len;
+    struct flb_record_accessor *ra_key;
     int    reserve_data;
     int    preserve_key;
     struct mk_list parsers;