record_accessor: add flb_ra_translate_check that explicitly errors wh…

…en keys are not found Signed-off-by: Wesley Pettit <[email protected]>
fluent · Jun 30, 2022 · c5fec54 · c5fec54
1 parent 0c41d2b
commit c5fec54
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 4 deletions.
diff --git a/include/fluent-bit/flb_record_accessor.h b/include/fluent-bit/flb_record_accessor.h
@@ -38,6 +38,10 @@ void flb_ra_dump(struct flb_record_accessor *ra);
 flb_sds_t flb_ra_translate(struct flb_record_accessor *ra,
                            char *tag, int tag_len,
                            msgpack_object map, struct flb_regex_search *result);
+flb_sds_t flb_ra_translate_check(struct flb_record_accessor *ra,
+                                 char *tag, int tag_len,
+                                 msgpack_object map, struct flb_regex_search *result,
+                                 int check);
 int flb_ra_is_static(struct flb_record_accessor *ra);
 int flb_ra_strcmp(struct flb_record_accessor *ra, msgpack_object map,
                   char *str, int len);

diff --git a/src/flb_record_accessor.c b/src/flb_record_accessor.c
@@ -487,11 +487,30 @@ flb_sds_t flb_ra_translate(struct flb_record_accessor *ra,
                            char *tag, int tag_len,
                            msgpack_object map, struct flb_regex_search *result)
 {
-    int found;
+    int check = FLB_FALSE;
+
+    return flb_ra_translate_check(ra, tag, tag_len, map, result, check);
+}
+
+/*
+ * Translate a record accessor buffer, tag and records are optional
+ * parameters.
+ *
+ * For safety, the function returns a newly created string that needs
+ * to be destroyed by the caller.
+ * 
+ * Returns NULL if `check` is FLB_TRUE and any key loopup in the record failed
+ */
+flb_sds_t flb_ra_translate_check(struct flb_record_accessor *ra,
+                                 char *tag, int tag_len,
+                                 msgpack_object map, struct flb_regex_search *result,
+                                 int check)
+{
     flb_sds_t tmp = NULL;
     flb_sds_t buf;
     struct mk_list *head;
     struct flb_ra_parser *rp;
+    int found = FLB_FALSE;
 
     buf = flb_sds_create_size(ra->size_hint);
     if (!buf) {
@@ -506,6 +525,11 @@ flb_sds_t flb_ra_translate(struct flb_record_accessor *ra,
         }
         else if (rp->type == FLB_RA_PARSER_KEYMAP) {
             tmp = ra_translate_keymap(rp, buf, map, &found);
+            if (check == FLB_TRUE && found == FLB_FALSE) {
+                flb_warn("[record accessor] translation failed, root key=%s", rp->key->name);
+                flb_sds_destroy(buf);
+                return NULL;
+            }
         }
         else if (rp->type == FLB_RA_PARSER_REGEX_ID && result) {
             tmp = ra_translate_regex_id(rp, result, buf);
@@ -516,9 +540,6 @@ flb_sds_t flb_ra_translate(struct flb_record_accessor *ra,
         else if (rp->type == FLB_RA_PARSER_TAG_PART && tag) {
             tmp = ra_translate_tag_part(rp, buf, tag, tag_len);
         }
-        else {
-
-        }
 
         //else if (rp->type == FLB_RA_PARSER_FUNC) {
             //tmp = ra_translate_func(rp, buf, tag, tag_len);