filter_lua: support new option "type_array_key" (#3343)

    If set, the value of the key will be handled as array.
    It is useful to handle empty array.

Signed-off-by: Takahiro Yamashita <[email protected]>

Author: nokute78

plugins/filter_lua/lua.c

@@ -142,6 +142,23 @@ static int lua_arraylength(lua_State *l)
 }
 
 static void lua_tomsgpack(struct lua_filter *lf, msgpack_packer *pck, int index);
+static void lua_toarray(struct lua_filter *lf, msgpack_packer *pck, int index)
+{
+    int len;
+    int i;
+    lua_State *l = lf->lua->state;
+
+    lua_pushnumber(l, (lua_Number)lua_objlen(l, -1)); // lua_len
+    len = (int)lua_tointeger(l, -1);
+    lua_pop(l, 1);
+
+    msgpack_pack_array(pck, len);
+    for (i = 1; i <= len; i++) {
+        lua_rawgeti(l, -1, i);
+        lua_tomsgpack(lf, pck, 0);
+        lua_pop(l, 1);
+    }
+}
 static void try_to_convert_data_type(struct lua_filter *lf,
                                      msgpack_packer *pck,
                                      int index)
@@ -154,19 +171,33 @@ static void try_to_convert_data_type(struct lua_filter *lf,
     struct mk_list  *head     = NULL;
     struct l2c_type *l2c      = NULL;
 
+    // convert to int
     if ((lua_type(l, -2) == LUA_TSTRING)
         && lua_type(l, -1) == LUA_TNUMBER){
         tmp = lua_tolstring(l, -2, &len);
 
         mk_list_foreach_safe(head, tmp_list, &lf->l2c_types) {
             l2c = mk_list_entry(head, struct l2c_type, _head);
-            if (!strncmp(l2c->key, tmp, len)) {
+            if (!strncmp(l2c->key, tmp, len) && l2c->type == L2C_TYPE_INT) {
                 lua_tomsgpack(lf, pck, -1);
                 msgpack_pack_int64(pck, (int64_t)lua_tonumber(l, -1));
                 return;
             }
         }
     }
+    else if ((lua_type(l, -2) == LUA_TSTRING)
+             && lua_type(l, -1) == LUA_TTABLE){
+        tmp = lua_tolstring(l, -2, &len);
+
+        mk_list_foreach_safe(head, tmp_list, &lf->l2c_types) {
+            l2c = mk_list_entry(head, struct l2c_type, _head);
+            if (!strncmp(l2c->key, tmp, len) && l2c->type == L2C_TYPE_ARRAY) {
+                lua_tomsgpack(lf, pck, -1);
+                lua_toarray(lf, pck, 0);
+                return;
+            }
+        }
+    }
 
     /* not matched */
     lua_tomsgpack(lf, pck, -1);
@@ -585,6 +616,12 @@ static struct flb_config_map config_map[] = {
      "If these keys are matched, the fields are converted to integer. "
      "If more than one key, delimit by space."
     },
+    {
+     FLB_CONFIG_MAP_STR, "type_array_key", NULL,
+     0, FLB_FALSE, 0,
+     "If these keys are matched, the fields are converted to array. "
+     "If more than one key, delimit by space."
+    },
     {
      FLB_CONFIG_MAP_BOOL, "protected_mode", "true",
      0, FLB_TRUE, offsetof(struct lua_filter, protected_mode),

plugins/filter_lua/lua_config.c

@@ -131,6 +131,26 @@ struct lua_filter *lua_config_create(struct flb_filter_instance *ins,
 
             tmp_key = flb_strndup(sentry->value, sentry->len);
             l2c->key = flb_sds_create(tmp_key);
+            l2c->type = L2C_TYPE_INT;
+            flb_free(tmp_key);
+
+            mk_list_add(&l2c->_head, &lf->l2c_types);
+            lf->l2c_types_num++;
+        }
+        flb_utils_split_free(split);
+    }
+
+    tmp = flb_filter_get_property("type_array_key", ins);
+    if (tmp) {
+        split = flb_utils_split(tmp, ' ', L2C_TYPES_NUM_MAX);
+        mk_list_foreach_safe(head, tmp_list, split) {
+            l2c = flb_malloc(sizeof(struct l2c_type));
+
+            sentry = mk_list_entry(head, struct flb_split_entry, _head);
+
+            tmp_key = flb_strndup(sentry->value, sentry->len);
+            l2c->key = flb_sds_create(tmp_key);
+            l2c->type = L2C_TYPE_ARRAY;
             flb_free(tmp_key);
 
             mk_list_add(&l2c->_head, &lf->l2c_types);

plugins/filter_lua/lua_config.h

@@ -29,8 +29,14 @@
 #define LUA_BUFFER_CHUNK    1024 * 8  /* 8K should be enough to get started */
 #define L2C_TYPES_NUM_MAX   16
 
+enum l2c_type_enum {
+    L2C_TYPE_INT,
+    L2C_TYPE_ARRAY
+};
+
 struct l2c_type {
     flb_sds_t key;
+    int type;
     struct mk_list _head;
 };