config_format: yaml: fix error logic for input processors and cfl_array.

src/config_format/flb_cf_yaml.c

@@ -429,7 +429,7 @@ static int read_glob(struct flb_cf *conf, struct local_ctx *ctx,
     for (idx = 0; idx < glb.gl_pathc; idx++) {
         ret = read_config(conf, ctx, state->file, glb.gl_pathv[idx]);
 
-    if (ret < 0) {
+        if (ret < 0) {
             break;
         }
     }
@@ -648,11 +648,12 @@ static enum status state_copy_into_config_group(struct parser_state *state, stru
         case CFL_VARIANT_ARRAY:
             carr = cfl_array_create(kvp->val->data.as_array->entry_count);
 
-            if (carr) {
+            if (carr == NULL) {
                 flb_error("unable to allocate array");
                 cfl_kvlist_destroy(copy);
                 return YAML_FAILURE;
             }
+
             for (idx = 0; idx < kvp->val->data.as_array->entry_count; idx++) {
                 var = cfl_array_fetch_by_index(kvp->val->data.as_array, idx);
 

tests/internal/config_format_yaml.c

@@ -14,6 +14,7 @@
 #define FLB_TESTS_CONF_PATH FLB_TESTS_DATA_PATH "/data/config_format/yaml"
 #define FLB_000 FLB_TESTS_CONF_PATH "/fluent-bit.yaml"
 #define FLB_001 FLB_TESTS_CONF_PATH "/issue_7559.yaml"
+#define FLB_002 FLB_TESTS_CONF_PATH "/processors.yaml"
 
 /*
  * Configurations to test:
@@ -47,7 +48,7 @@ static void test_basic()
     /* Total number of sections */
     TEST_CHECK(mk_list_size(&cf->sections) == 9);
 
-	/* SERVICE check */
+    /* SERVICE check */
     TEST_CHECK(cf->service != NULL);
     if (cf->service) {
         TEST_CHECK(cfl_list_size(&cf->service->properties->list) == 3);
@@ -262,8 +263,8 @@ static void test_parser_conf()
     /* Total number of inputs */
     if(!TEST_CHECK(mk_list_size(&config->parsers) == cnt+1)) {
         TEST_MSG("Section number error. Got=%d expect=%d", 
-	         mk_list_size(&config->parsers),
-		 cnt+1);
+            mk_list_size(&config->parsers),
+            cnt+1);
     }
 
     flb_cf_dump(cf);
@@ -313,12 +314,127 @@ static void test_camel_case_key()
 
 }
 
+/* data/config_format/processors.yaml */
+static void test_processors()
+{
+    struct mk_list *head;
+    struct flb_cf *cf;
+    struct flb_cf_section *s;
+    struct flb_cf_group *g;
+    struct cfl_variant *v;
+    struct cfl_variant *logs;
+    struct cfl_variant *record_modifier_filter;
+    struct cfl_variant *records;
+    struct cfl_variant *record;
+    int idx = 0;
+
+    cf = flb_cf_yaml_create(NULL, FLB_002, NULL, 0);
+    TEST_CHECK(cf != NULL);
+    if (!cf) {
+        exit(EXIT_FAILURE);
+    }
+
+    /* Total number of sections */
+    TEST_CHECK(mk_list_size(&cf->sections) == 2);
+
+    /* Check number sections per list */
+    TEST_CHECK(mk_list_size(&cf->parsers) == 0);
+    TEST_CHECK(mk_list_size(&cf->multiline_parsers) == 0);
+    TEST_CHECK(mk_list_size(&cf->customs) == 0);
+    TEST_CHECK(mk_list_size(&cf->inputs) == 1);
+    TEST_CHECK(mk_list_size(&cf->filters) == 0);
+    TEST_CHECK(mk_list_size(&cf->outputs) == 1);
+    TEST_CHECK(mk_list_size(&cf->others) == 0);
+
+    /* check inputs */
+    idx = 0;
+    mk_list_foreach(head, &cf->inputs) {
+        s = mk_list_entry(head, struct flb_cf_section, _head_section);
+        switch (idx) {
+        case 0:
+            v = flb_cf_section_property_get(cf, s, "name");
+            TEST_CHECK(v->type == CFL_VARIANT_STRING);
+            TEST_CHECK(strcmp(v->data.as_string, "dummy") == 0);
+            break;
+        }
+        idx++;
+    }
+
+    /* check outputs */
+    idx = 0;
+    mk_list_foreach(head, &cf->outputs) {
+        s = mk_list_entry(head, struct flb_cf_section, _head_section);
+        switch (idx) {
+        case 0:
+            v = flb_cf_section_property_get(cf, s, "name");
+            TEST_CHECK(v->type == CFL_VARIANT_STRING);
+            TEST_CHECK(strcmp(v->data.as_string, "stdout") == 0);
+            break;
+        }
+        idx++;
+    }
+
+    /* groups */
+    s = flb_cf_section_get_by_name(cf, "input");
+    TEST_CHECK(s != NULL);
+    TEST_CHECK(mk_list_size(&s->groups) == 1);
+
+    mk_list_foreach(head, &s->groups) {
+        g = mk_list_entry(head, struct flb_cf_group, _head);
+        TEST_CHECK(cfl_list_size(&g->properties->list) == 1);
+        TEST_CHECK(strcmp(g->name, "processors") == 0);
+
+        logs = cfl_kvlist_fetch(g->properties, "logs");
+        TEST_CHECK(logs != NULL);
+        if (logs == NULL) {
+            continue;
+        }
+
+        TEST_CHECK(logs->type == CFL_VARIANT_ARRAY);
+        if (logs->type == CFL_VARIANT_ARRAY) {
+            TEST_CHECK(logs->data.as_array->entry_count == 1);
+
+            record_modifier_filter = cfl_array_fetch_by_index(logs->data.as_array, 0);
+            TEST_CHECK(record_modifier_filter != NULL);
+
+            if (record_modifier_filter) {
+                TEST_CHECK(record_modifier_filter->type == CFL_VARIANT_KVLIST);
+
+                records = cfl_kvlist_fetch(record_modifier_filter->data.as_kvlist, "record");
+                TEST_CHECK(records->type == CFL_VARIANT_ARRAY);
+                TEST_CHECK(records->data.as_array->entry_count == 2);
+
+                for (idx = 0; idx < 2; idx++) {
+                    record = cfl_array_fetch_by_index(records->data.as_array, idx);
+                    TEST_CHECK(record->type == CFL_VARIANT_STRING);
+
+                    if (record->type != CFL_VARIANT_STRING) {
+                        continue;
+                    }
+
+                    switch (idx) {
+                    case 0:
+                        TEST_CHECK(strcmp(record->data.as_string, "filtered_by record_modifier") == 0);
+                        break;
+                    case 1:
+                        TEST_CHECK(strcmp(record->data.as_string, "powered_by calyptia") == 0);
+                        break;
+                    }
+                }
+            }
+        }
+    }
+
+    flb_cf_destroy(cf);
+}
+
 TEST_LIST = {
     { "basic"    , test_basic},
     { "customs section", test_customs_section},
     { "slist odd", test_slist_odd},
     { "slist even", test_slist_even},
     { "parsers file conf", test_parser_conf},
     { "camel_case_key", test_camel_case_key},
+    { "processors", test_processors},
     { 0 }
 };

tests/internal/data/config_format/yaml/processors.yaml

@@ -0,0 +1,14 @@
+---
+pipeline:
+  inputs:
+    - name: dummy
+      processors:
+        logs:
+          - name: record_modifier
+            record:
+              - filtered_by record_modifier
+              - powered_by calyptia
+  outputs:
+    - name: stdout
+      match: "*"
+      format: json_lines