ra_key: add support for BINARY type values

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

Author: edsiper

include/fluent-bit/flb_ra_key.h

@@ -32,7 +32,13 @@ enum ra_types {
     FLB_RA_INT,
     FLB_RA_FLOAT,
     FLB_RA_STRING,
-    FLB_RA_NULL
+    FLB_RA_NULL,
+    FLB_RA_BINARY
+};
+
+enum ra_storage_type {
+    FLB_RA_COPY = 0,
+    FLB_RA_REF
 };
 
 /* condition value types */
@@ -41,15 +47,27 @@ typedef union {
     int64_t i64;
     double f64;
     flb_sds_t string;
+    flb_sds_t binary;
+    struct {
+        const char *buf;
+        size_t len;
+    } ref;
 } ra_val;
 
 /* Represent any value object */
 struct flb_ra_value {
     int type;
+    int storage; /* FLB_RA_COPY or FLB_RA_REF */
     msgpack_object o;
     ra_val val;
 };
 
+const char *flb_ra_value_buffer(struct flb_ra_value *v, size_t *len);
+
+struct flb_ra_value *flb_ra_key_to_value_ext(flb_sds_t ckey,
+                                             msgpack_object map,
+                                             struct mk_list *subkeys,
+                                             int copy);
 struct flb_ra_value *flb_ra_key_to_value(flb_sds_t ckey,
                                          msgpack_object map,
                                          struct mk_list *subkeys);

src/flb_ra_key.c

@@ -2,7 +2,7 @@
 
 /*  Fluent Bit
  *  ==========
- *  Copyright (C) 2015-2024 The Fluent Bit Authors
+ *  Copyright (C) 2015-2025 The Fluent Bit Authors
  *
  *  Licensed under the Apache License, Version 2.0 (the "License");
  *  you may not use this file except in compliance with the License.
@@ -29,47 +29,75 @@
 
 /* Map msgpack object into flb_ra_value representation */
 static int msgpack_object_to_ra_value(msgpack_object o,
-                                      struct flb_ra_value *result)
+                                      struct flb_ra_value *result,
+                                      int copy)
 {
     result->o = o;
 
     /* Compose result with found value */
     if (o.type == MSGPACK_OBJECT_BOOLEAN) {
         result->type = FLB_RA_BOOL;
+        result->storage = FLB_RA_COPY;
         result->val.boolean = o.via.boolean;
         return 0;
     }
     else if (o.type == MSGPACK_OBJECT_POSITIVE_INTEGER ||
              o.type == MSGPACK_OBJECT_NEGATIVE_INTEGER) {
         result->type = FLB_RA_INT;
+        result->storage = FLB_RA_COPY;
         result->val.i64 = o.via.i64;
         return 0;
     }
     else if (o.type == MSGPACK_OBJECT_FLOAT32 ||
              o.type == MSGPACK_OBJECT_FLOAT) {
         result->type = FLB_RA_FLOAT;
+        result->storage = FLB_RA_COPY;
         result->val.f64 = o.via.f64;
         return 0;
     }
     else if (o.type == MSGPACK_OBJECT_STR) {
         result->type = FLB_RA_STRING;
-        result->val.string = flb_sds_create_len((char *) o.via.str.ptr,
-                                                o.via.str.size);
-
-        /* Handle cases where flb_sds_create_len fails */
-        if (result->val.string == NULL) {
-            return -1;
+        if (copy) {
+            result->storage = FLB_RA_COPY;
+            result->val.string = flb_sds_create_len((char *) o.via.str.ptr, o.via.str.size);
+            if (result->val.string == NULL) {
+                return -1;
+            }
+        }
+        else {
+            result->storage = FLB_RA_REF;
+            result->val.ref.buf = (const char *) o.via.str.ptr;
+            result->val.ref.len = o.via.str.size;
         }
         return 0;
     }
     else if (o.type == MSGPACK_OBJECT_MAP) {
         /* return boolean 'true', just denoting the existence of the key */
         result->type = FLB_RA_BOOL;
+        result->storage = FLB_RA_COPY;
         result->val.boolean = true;
         return 0;
     }
+    else if (o.type == MSGPACK_OBJECT_BIN) {
+        result->type = FLB_RA_BINARY;
+        if (copy) {
+            result->storage = FLB_RA_COPY;
+            result->val.binary = flb_sds_create_len((char *) o.via.bin.ptr, o.via.bin.size);
+            if (result->val.binary == NULL) {
+                flb_errno();
+                return -1;
+            }
+        }
+        else {
+            result->storage = FLB_RA_REF;
+            result->val.ref.buf = (const char *) o.via.bin.ptr;
+            result->val.ref.len = o.via.bin.size;
+        }
+        return 0;
+    }
     else if (o.type == MSGPACK_OBJECT_NIL) {
         result->type = FLB_RA_NULL;
+        result->storage = FLB_RA_COPY;
         return 0;
     }
 
@@ -121,7 +149,7 @@ static int msgpack_object_strcmp(msgpack_object o, char *str, int len)
 
 /* Lookup perfect match of sub-keys and map content */
 static int subkey_to_object(msgpack_object *map, struct mk_list *subkeys,
-                           msgpack_object **out_key, msgpack_object **out_val)
+                            msgpack_object **out_key, msgpack_object **out_val)
 {
     int i = 0;
     int levels;
@@ -207,9 +235,10 @@ static int subkey_to_object(msgpack_object *map, struct mk_list *subkeys,
     return 0;
 }
 
-struct flb_ra_value *flb_ra_key_to_value(flb_sds_t ckey,
-                                         msgpack_object map,
-                                         struct mk_list *subkeys)
+struct flb_ra_value *flb_ra_key_to_value_ext(flb_sds_t ckey,
+                                             msgpack_object map,
+                                             struct mk_list *subkeys,
+                                             int copy)
 {
     int i;
     int ret;
@@ -240,7 +269,7 @@ struct flb_ra_value *flb_ra_key_to_value(flb_sds_t ckey,
 
         ret = subkey_to_object(&val, subkeys, &out_key, &out_val);
         if (ret == 0) {
-            ret = msgpack_object_to_ra_value(*out_val, result);
+            ret = msgpack_object_to_ra_value(*out_val, result, copy);
             if (ret == -1) {
                 flb_free(result);
                 return NULL;
@@ -253,7 +282,7 @@ struct flb_ra_value *flb_ra_key_to_value(flb_sds_t ckey,
         }
     }
     else {
-        ret = msgpack_object_to_ra_value(val, result);
+        ret = msgpack_object_to_ra_value(val, result, copy);
         if (ret == -1) {
             flb_error("[ra key] cannot process key value");
             flb_free(result);
@@ -264,6 +293,13 @@ struct flb_ra_value *flb_ra_key_to_value(flb_sds_t ckey,
     return result;
 }
 
+struct flb_ra_value *flb_ra_key_to_value(flb_sds_t ckey,
+                                         msgpack_object map,
+                                         struct mk_list *subkeys)
+{
+    return flb_ra_key_to_value_ext(ckey, map, subkeys, FLB_TRUE);
+}
+
 int flb_ra_key_value_get(flb_sds_t ckey, msgpack_object map,
                          struct mk_list *subkeys,
                          msgpack_object **start_key,
@@ -429,7 +465,7 @@ static int update_subkey_array(msgpack_object *obj, struct mk_list *subkeys,
     }
 
     msgpack_pack_array(mp_pck, size);
-    for (i=0; i<size; i++) {
+    for (i = 0; i < size; i++) {
         if (i != entry->array_id) {
             msgpack_pack_object(mp_pck, obj->via.array.ptr[i]);
             continue;
@@ -483,7 +519,7 @@ static int update_subkey_map(msgpack_object *obj, struct mk_list *subkeys,
     }
 
     msgpack_pack_map(mp_pck, size);
-    for (i=0; i<size; i++) {
+    for (i = 0; i < size; i++) {
         if (i != ret_id) {
             msgpack_pack_object(mp_pck, obj->via.map.ptr[i].key);
             msgpack_pack_object(mp_pck, obj->via.map.ptr[i].val);
@@ -561,7 +597,7 @@ int flb_ra_key_value_update(struct flb_ra_parser *rp, msgpack_object map,
     msgpack_pack_map(mp_pck, map_size);
     if (levels == 0) {
         /* no subkeys */
-        for (i=0; i<map_size; i++) {
+        for (i = 0; i < map_size; i++) {
             if (i != kv_id) {
                 /* pack original key/val */
                 msgpack_pack_object(mp_pck, map.via.map.ptr[i].key);
@@ -586,7 +622,7 @@ int flb_ra_key_value_update(struct flb_ra_parser *rp, msgpack_object map,
         return 0;
     }
 
-    for (i=0; i<map_size; i++) {
+    for (i = 0; i < map_size; i++) {
         msgpack_pack_object(mp_pck, map.via.map.ptr[i].key);
         if (i != kv_id) {
             msgpack_pack_object(mp_pck, map.via.map.ptr[i].val);
@@ -647,7 +683,7 @@ static int append_subkey_array(msgpack_object *obj, struct mk_list *subkeys,
     }
 
     msgpack_pack_array(mp_pck, size);
-    for (i=0; i<size; i++) {
+    for (i = 0; i < size; i++) {
         if (i != entry->array_id) {
             msgpack_pack_object(mp_pck, obj->via.array.ptr[i]);
             continue;
@@ -691,7 +727,7 @@ static int append_subkey_map(msgpack_object *obj, struct mk_list *subkeys,
     if (levels == *matched) {
         /* append val */
         msgpack_pack_map(mp_pck, size+1);
-        for (i=0; i<size; i++) {
+        for (i = 0; i < size; i++) {
             msgpack_pack_object(mp_pck, obj->via.map.ptr[i].key);
             msgpack_pack_object(mp_pck, obj->via.map.ptr[i].val);
         }
@@ -703,15 +739,14 @@ static int append_subkey_map(msgpack_object *obj, struct mk_list *subkeys,
         return 0;
     }
 
-
     ret_id = ra_key_val_id(entry->str, *obj);
     if (ret_id < 0) {
         flb_trace("%s: not found", __FUNCTION__);
         return -1;
     }
 
     msgpack_pack_map(mp_pck, size);
-    for (i=0; i<size; i++) {
+    for (i = 0; i < size; i++) {
         if (i != ret_id) {
             msgpack_pack_object(mp_pck, obj->via.map.ptr[i].key);
             msgpack_pack_object(mp_pck, obj->via.map.ptr[i].val);
@@ -770,7 +805,7 @@ int flb_ra_key_value_append(struct flb_ra_parser *rp, msgpack_object map,
     if (ref_level < 0) {
         /* no subkeys */
         msgpack_pack_map(mp_pck, map_size+1);
-        for (i=0; i<map_size; i++) {
+        for (i = 0; i < map_size; i++) {
             msgpack_pack_object(mp_pck, map.via.map.ptr[i].key);
             msgpack_pack_object(mp_pck, map.via.map.ptr[i].val);
         }
@@ -787,7 +822,7 @@ int flb_ra_key_value_append(struct flb_ra_parser *rp, msgpack_object map,
     }
 
     msgpack_pack_map(mp_pck, map_size);
-    for (i=0; i<map_size; i++) {
+    for (i = 0; i < map_size; i++) {
         msgpack_pack_object(mp_pck, map.via.map.ptr[i].key);
         if (i != kv_id) {
             msgpack_pack_object(mp_pck, map.via.map.ptr[i].val);
@@ -806,8 +841,34 @@ int flb_ra_key_value_append(struct flb_ra_parser *rp, msgpack_object map,
 
 void flb_ra_key_value_destroy(struct flb_ra_value *v)
 {
-    if (v->type == FLB_RA_STRING) {
+    if (v->type == FLB_RA_STRING && v->storage == FLB_RA_COPY) {
         flb_sds_destroy(v->val.string);
     }
+    else if (v->type == FLB_RA_BINARY && v->storage == FLB_RA_COPY) {
+        flb_sds_destroy(v->val.binary);
+    }
     flb_free(v);
 }
+
+const char *flb_ra_value_buffer(struct flb_ra_value *v, size_t *len)
+{
+    if (v->storage == FLB_RA_REF) {
+        if (len) {
+            *len = v->val.ref.len;
+        }
+        return v->val.ref.buf;
+    }
+
+    if (len) {
+        if (v->type == FLB_RA_STRING) {
+            *len = flb_sds_len(v->val.string);
+        }
+        else {
+            *len = flb_sds_len(v->val.binary);
+        }
+    }
+    if (v->type == FLB_RA_STRING) {
+        return v->val.string;
+    }
+    return v->val.binary;
+}