processor: moved the processor stack initialization to the right stage

This PR moves the initialization of the processor stack (and its components) to the
correct stage which means processor stacks are initialized as part of the initialization 
process of the plugin instance they are attached to.

This fixes an issue where certain filters such as `rewrite_tag` would cause a crash while 
being initialized because they depend on the rest of the system being initialized.

Signed-off-by: Hiroshi Hatake <[email protected]>

src/flb_input.c

@@ -1206,6 +1206,12 @@ int flb_input_instance_init(struct flb_input_instance *ins,
         }
     }
 
+    /* initialize processors */
+    ret = flb_processor_init(ins->processor);
+    if (ret == -1) {
+        return -1;
+    }
+
     return 0;
 }
 

src/flb_output.c

@@ -1300,6 +1300,12 @@ int flb_output_init_all(struct flb_config *config)
                       flb_output_name(ins));
             return -1;
         }
+
+        /* initialize processors */
+        ret = flb_processor_init(ins->processor);
+        if (ret == -1) {
+            return -1;
+        }
     }
 
     return 0;

src/flb_processor.c

@@ -526,6 +526,7 @@ static int load_from_config_format_group(struct flb_processor *proc, int type, s
     struct cfl_kvpair *pair = NULL;
     struct cfl_list *head;
     struct flb_processor_unit *pu;
+    struct flb_filter_instance *f_ins;
 
     if (val->type != CFL_VARIANT_ARRAY) {
         return -1;
@@ -566,6 +567,18 @@ static int load_from_config_format_group(struct flb_processor *proc, int type, s
             if (pair->val->type != CFL_VARIANT_STRING) {
                 continue;
             }
+            /* If filter plugin in processor unit has its own match rule,
+             * we must release the pre-allocated '*' match at first.
+             */
+            if (pu->unit_type == FLB_PROCESSOR_UNIT_FILTER) {
+                if (strcmp(pair->key, "match") == 0) {
+                    f_ins = (struct flb_filter_instance *)pu->ctx;
+                    if (f_ins->match != NULL) {
+                        flb_sds_destroy(f_ins->match);
+                        f_ins->match = NULL;
+                    }
+                }
+            }
 
             ret = flb_processor_unit_set_property(pu, pair->key, pair->val->data.as_string);
             if (ret == -1) {
@@ -615,12 +628,6 @@ int flb_processors_load_from_config_format_group(struct flb_processor *proc, str
         }
     }
 
-    /* initialize processors */
-    ret = flb_processor_init(proc);
-    if (ret == -1) {
-        return -1;
-    }
-
     return 0;
 }