Refactor router config parser and add route precedence test

ROUTER.md

@@ -0,0 +1,108 @@
+# Telemetry Routing Overview
+
+Fluent Bit's telemetry router provides a unified configuration surface for
+logs, metrics, and traces.  Routes are defined on a per-input basis and apply to
+all signals produced by that input.  Each route declares an optional signal
+filter, a condition, per-route processors, and the list of output targets that
+should receive matching telemetry.
+
+## Configuration Structure
+
+Routes are declared inside `pipeline.inputs[]` entries in the Fluent Bit YAML
+configuration:
+
+```yaml
+pipeline:
+  inputs:
+    - name: opentelemetry
+      processors:
+        - name: parser
+          parser: json
+      routes:
+        logs:
+          - name: errors
+            condition:
+              rules:
+                - field: "$level"
+                  op: eq
+                  value: "error"
+            to:
+              outputs:
+                - name: loki
+                  fallback: s3_archive
+          - name: default
+            condition:
+              default: true
+            to:
+              outputs:
+                - name: elasticsearch
+        metrics:
+          - name: cpu_hot
+            condition:
+              rules:
+                - field: "$metric.name"
+                  op: regex
+                  value: "^cpu_"
+            to:
+              outputs:
+                - name: prometheus_remote
+```
+
+### Key elements
+
+* **Input processors** – shared processors executed before any routing logic.
+* **Routes** – grouped by telemetry signal.  Each key under `routes` must be a
+  signal label (`logs`, `metrics`, `traces`, or a comma-separated combination)
+  whose array value contains the ordered route definitions for that signal.  A
+  key of `any` targets all telemetry types.
+  * `condition.rules` contains record accessor comparisons evaluated against the
+    chunk.  Routes can also mark themselves as the default handler with
+    `condition.default: true`.
+  * `processors` (optional) define per-route processor chains executed after the
+    condition succeeds and before dispatching to outputs.
+  * `to.outputs` lists the primary output targets.  Entries may be simple names
+    or objects with an optional `fallback` output used when the primary target
+    fails.
+
+## Evaluation Order
+
+1. Execute input-level processors.
+2. For each route whose signal mask matches the chunk type:
+   1. Evaluate the route condition.  Routes flagged as `default` bypass rule
+      evaluation.
+   2. If the condition succeeds, run the route processors and then attempt to
+      send the chunk to each configured output.
+3. When an output write fails permanently, the router retries once using the
+   configured fallback output (if any).
+
+## Conditions and Field Resolution
+
+Route conditions rely on record accessors that are aware of the telemetry type:
+
+* **Logs** – record keys, metadata, `exists`, `eq`, `regex`, `contains`, `in`.
+* **Metrics** – metric name/value, resource and attribute keys, numeric
+  comparison operators (`gt`, `lt`, `gte`, `lte`).
+* **Traces** – span fields and resource/scope attributes with equality, regex,
+  and duration comparisons.
+
+The loader validates that each referenced field is supported by the selected
+signals, providing early feedback during configuration parsing.
+
+## Output Fallbacks
+
+Outputs may declare a secondary target that receives the chunk when the primary
+write fails with a permanent error.  Fallback dispatch happens once per output
+attempt and produces debug logs as well as `fluentbit_routing_fallback_total`
+metrics to aid troubleshooting.
+
+## Metrics and Observability
+
+The router exposes Prometheus counters using the `fluentbit_routing_*` prefix
+covering routed records, bytes, condition failures, and fallback events.  The
+labels capture the input name, route name, target output, and signal type.
+
+## Cleanup Helpers
+
+The router exposes `flb_router_routes_destroy()` to release all resources
+allocated during YAML parsing.  Callers should destroy the list after invoking
+`flb_router_config_parse()` once the configuration has been consumed.

include/fluent-bit/flb_router.h

@@ -23,6 +23,11 @@
 #include <fluent-bit/flb_info.h>
 #include <fluent-bit/flb_input.h>
 #include <fluent-bit/flb_output.h>
+#include <fluent-bit/flb_event.h>
+#include <fluent-bit/flb_sds.h>
+#include <fluent-bit/flb_config.h>
+#include <cfl/cfl.h>
+#include <monkey/mk_core.h>
 
 struct flb_router_path {
     struct flb_output_instance *ins;
@@ -56,6 +61,61 @@ static inline int flb_router_match_type(int in_event_type,
     return FLB_TRUE;
 }
 
+enum flb_router_signal {
+    FLB_ROUTER_SIGNAL_LOGS    = (1U << 0),
+    FLB_ROUTER_SIGNAL_METRICS = (1U << 1),
+    FLB_ROUTER_SIGNAL_TRACES  = (1U << 2),
+    FLB_ROUTER_SIGNAL_ANY     = (FLB_ROUTER_SIGNAL_LOGS |
+                                 FLB_ROUTER_SIGNAL_METRICS |
+                                 FLB_ROUTER_SIGNAL_TRACES)
+};
+
+struct flb_route_condition_rule {
+    flb_sds_t field;
+    flb_sds_t op;
+    flb_sds_t value;
+    struct cfl_list _head;
+};
+
+struct flb_route_condition {
+    struct cfl_list rules;
+    int is_default;
+};
+
+struct flb_route_output {
+    flb_sds_t name;
+    flb_sds_t fallback;
+    struct cfl_list _head;
+};
+
+struct flb_route_processor_property {
+    flb_sds_t key;
+    flb_sds_t value;
+    struct cfl_list _head;
+};
+
+struct flb_route_processor {
+    flb_sds_t name;
+    struct cfl_list properties;
+    struct cfl_list _head;
+};
+
+struct flb_route {
+    flb_sds_t name;
+    uint32_t signals;
+    struct flb_route_condition *condition;
+    struct cfl_list outputs;
+    struct cfl_list processors;
+    struct cfl_list _head;
+};
+
+struct flb_input_routes {
+    flb_sds_t input_name;
+    struct cfl_list processors;
+    struct cfl_list routes;
+    struct cfl_list _head;
+};
+
 int flb_router_connect(struct flb_input_instance *in,
                        struct flb_output_instance *out);
 int flb_router_connect_direct(struct flb_input_instance *in,
@@ -65,4 +125,24 @@ int flb_router_match(const char *tag, int tag_len,
                      const char *match, void *match_regex);
 int flb_router_io_set(struct flb_config *config);
 void flb_router_exit(struct flb_config *config);
+
+uint32_t flb_router_signal_from_chunk(struct flb_event_chunk *chunk);
+
+int flb_route_condition_eval(struct flb_event_chunk *chunk,
+                             struct flb_route *route);
+int flb_condition_eval_logs(struct flb_event_chunk *chunk,
+                            struct flb_route *route);
+int flb_condition_eval_metrics(struct flb_event_chunk *chunk,
+                               struct flb_route *route);
+int flb_condition_eval_traces(struct flb_event_chunk *chunk,
+                              struct flb_route *route);
+
+struct flb_cf;
+
+int flb_router_config_parse(struct flb_cf *cf,
+                            struct cfl_list *input_routes,
+                            struct flb_config *config);
+void flb_router_routes_destroy(struct cfl_list *input_routes);
+
 #endif
+

src/CMakeLists.txt

@@ -54,6 +54,8 @@ set(src
   flb_upstream_ha.c
   flb_upstream_node.c
   flb_router.c
+  flb_router_condition.c
+  flb_router_config.c
   flb_worker.c
   flb_coro.c
   flb_time.c

src/flb_router_condition.c

@@ -0,0 +1,123 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2024 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.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#include <fluent-bit/flb_mem.h>
+#include <fluent-bit/flb_log.h>
+#include <fluent-bit/flb_router.h>
+
+uint32_t flb_router_signal_from_chunk(struct flb_event_chunk *chunk)
+{
+    if (!chunk) {
+        return 0;
+    }
+
+    switch (chunk->type) {
+    case FLB_EVENT_TYPE_LOGS:
+        return FLB_ROUTER_SIGNAL_LOGS;
+    case FLB_EVENT_TYPE_METRICS:
+        return FLB_ROUTER_SIGNAL_METRICS;
+    case FLB_EVENT_TYPE_TRACES:
+        return FLB_ROUTER_SIGNAL_TRACES;
+    default:
+        break;
+    }
+
+    return 0;
+}
+
+int flb_condition_eval_logs(struct flb_event_chunk *chunk,
+                            struct flb_route *route)
+{
+    (void) chunk;
+    (void) route;
+
+    /*
+     * The full condition evaluation engine requires field resolvers that map
+     * record accessors to the different telemetry payload shapes.  The wiring
+     * of those resolvers is part of a bigger effort and will be implemented in
+     * follow-up changes.  For the time being we simply report that the
+     * condition failed so that the runtime can rely on explicit default
+     * routes.
+     */
+    return FLB_FALSE;
+}
+
+int flb_condition_eval_metrics(struct flb_event_chunk *chunk,
+                               struct flb_route *route)
+{
+    (void) chunk;
+    (void) route;
+
+    return FLB_FALSE;
+}
+
+int flb_condition_eval_traces(struct flb_event_chunk *chunk,
+                              struct flb_route *route)
+{
+    (void) chunk;
+    (void) route;
+
+    return FLB_FALSE;
+}
+
+int flb_route_condition_eval(struct flb_event_chunk *chunk,
+                             struct flb_route *route)
+{
+    uint32_t signal;
+
+    if (!route) {
+        return FLB_FALSE;
+    }
+
+    if (!route->condition) {
+        return FLB_TRUE;
+    }
+
+    signal = flb_router_signal_from_chunk(chunk);
+    if (signal == 0) {
+        return FLB_FALSE;
+    }
+
+    if ((route->signals != 0) && (route->signals != FLB_ROUTER_SIGNAL_ANY) &&
+        ((route->signals & signal) == 0)) {
+        return FLB_FALSE;
+    }
+
+    if (route->condition->is_default) {
+        return FLB_TRUE;
+    }
+
+    if (cfl_list_is_empty(&route->condition->rules) == 0) {
+        return FLB_TRUE;
+    }
+
+    switch (signal) {
+    case FLB_ROUTER_SIGNAL_LOGS:
+        return flb_condition_eval_logs(chunk, route);
+    case FLB_ROUTER_SIGNAL_METRICS:
+        return flb_condition_eval_metrics(chunk, route);
+    case FLB_ROUTER_SIGNAL_TRACES:
+        return flb_condition_eval_traces(chunk, route);
+    default:
+        break;
+    }
+
+    return FLB_FALSE;
+}
+