input_log: account input metrics once before route fan-out

Counting the input metrics on the base append while the per-route copies
are skipped leaves a gap under backpressure. A route copy can succeed,
push the input over its memory or storage limit and pause it via
flb_input_chunk_protect(), or the base append can fail after some copies
were already delivered. In both cases the base append returns before
counting, so records already queued for a route go unreflected in
fluentbit_input_records_total and fluentbit_input_bytes_total.

Account the original ingestion exactly once at the routing boundary,
before the split, gated by the entry pause state, and mark the base
append with FLB_INPUT_CHUNK_SKIP_INPUT_METRICS as the per-route copies
already are. Counting up front keeps the totals correct regardless of
which downstream append later pauses or fails. The plain non-routed path
is unchanged and keeps counting inside flb_input_chunk_append_raw().

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: zshuang0316 <zshuang0316@163.com>

Author: zshuang0316

src/flb_input_log.c

@@ -1482,6 +1482,7 @@ static int input_log_append_processed_internal(struct flb_input_instance *ins,
     int ret;
     int conditional_result;
     int conditional_handled = FLB_FALSE;
+    int base_flags = 0;
     size_t dummy = 0;
     const char *base_tag = tag;
     size_t base_tag_len = tag_len;
@@ -1501,6 +1502,34 @@ static int input_log_append_processed_internal(struct flb_input_instance *ins,
         base_tag_len = strlen(base_tag);
     }
 
+#ifdef FLB_HAVE_METRICS
+    /*
+     * Conditional / per-record routing fans this batch out into one chunk per
+     * matching route (each appended below with
+     * FLB_INPUT_CHUNK_SKIP_INPUT_METRICS) plus a base append that is also
+     * skipped. Account the original ingestion exactly once here, BEFORE the
+     * split, so the input counters are neither inflated by the route count nor
+     * lost when a route copy pauses the input or the base append fails after
+     * some copies were already delivered. The plain, non-routed path keeps
+     * counting inside flb_input_chunk_append_raw().
+     */
+    if (cfl_list_size(&ins->routes_direct) > 0 &&
+        input_has_conditional_routes(ins) == FLB_TRUE &&
+        flb_input_buf_paused(ins) == FLB_FALSE &&
+        records > 0 && buf_size > 0) {
+        uint64_t ts = cfl_time_now();
+        char *name = (char *) flb_input_name(ins);
+
+        cmt_counter_add(ins->cmt_records, ts, records, 1, (char *[]) {name});
+        cmt_counter_add(ins->cmt_bytes, ts, buf_size, 1, (char *[]) {name});
+
+        flb_metrics_sum(FLB_METRIC_N_RECORDS, records, ins->metrics);
+        flb_metrics_sum(FLB_METRIC_N_BYTES, buf_size, ins->metrics);
+
+        base_flags = FLB_INPUT_CHUNK_SKIP_INPUT_METRICS;
+    }
+#endif
+
     conditional_result = split_and_append_route_payloads(ins, records, tag, tag_len,
                                                          buf, buf_size,
                                                          local_append);
@@ -1515,15 +1544,19 @@ static int input_log_append_processed_internal(struct flb_input_instance *ins,
     /*
      * Always call flb_input_chunk_append_raw to ensure non-conditional routes
      * receive data even when conditional routes exist. The conditional routing
-     * should be additive, not exclusive.
+     * should be additive, not exclusive. When the ingestion was already counted
+     * above, base_flags carries FLB_INPUT_CHUNK_SKIP_INPUT_METRICS so this
+     * append does not count it a second time.
      */
     if (local_append == FLB_TRUE) {
-        ret = flb_input_chunk_append_raw_local(ins, FLB_INPUT_LOGS, records,
-                                               tag, tag_len, buf, buf_size);
+        ret = flb_input_chunk_append_raw_local_flags(ins, FLB_INPUT_LOGS,
+                                                     base_flags, records,
+                                                     tag, tag_len, buf, buf_size);
     }
     else {
-        ret = flb_input_chunk_append_raw(ins, FLB_INPUT_LOGS, records,
-                                         tag, tag_len, buf, buf_size);
+        ret = flb_input_chunk_append_raw_flags(ins, FLB_INPUT_LOGS,
+                                               base_flags, records,
+                                               tag, tag_len, buf, buf_size);
     }
 
     if (ret == 0 && conditional_handled == FLB_TRUE && base_tag) {