metrics: fix input metrics double-counting with conditional routing #11948
+127
−21
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -1135,7 +1135,8 @@ static int split_and_append_route_payloads(struct flb_input_instance *ins, | |
| size_t tag_len, | ||
| const void *buf, | ||
| size_t buf_size, | ||
| int local_append, | ||
| int *any_appended) | ||
| { | ||
| int ret; | ||
| int appended; | ||
|
|
@@ -1397,18 +1398,21 @@ static int split_and_append_route_payloads(struct flb_input_instance *ins, | |
| continue; | ||
| } | ||
|
|
||
| /* internal route copy: skip metrics, the batch is counted once */ | ||
| if (local_append == FLB_TRUE) { | ||
| ret = flb_input_chunk_append_raw_local_flags(ins, | ||
| FLB_INPUT_LOGS, | ||
| FLB_INPUT_CHUNK_SKIP_INPUT_METRICS, | ||
|
There was a problem hiding this comment.
When conditional/per-record routing is active and a route-copy append succeeds but the later unconditional append fails (for example, the route copy pushes the input over Useful? React with 👍 / 👎. There was a problem hiding this comment. Codex Review: Didn't find any major issues. More of your lovely PRs please. Reviewed commit: ℹ️ About Codex in GitHubCodex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
If Codex has suggestions, it will comment; otherwise it will react with 👍. When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback". |
||
| payload->total_records, | ||
| payload->tag, | ||
| flb_sds_len(payload->tag), | ||
| payload->data, | ||
| payload->size); | ||
| } | ||
| else { | ||
| ret = flb_input_chunk_append_raw_flags(ins, | ||
| FLB_INPUT_LOGS, | ||
| FLB_INPUT_CHUNK_SKIP_INPUT_METRICS, | ||
| payload->total_records, | ||
| payload->tag, | ||
| flb_sds_len(payload->tag), | ||
|
|
@@ -1445,6 +1449,14 @@ static int split_and_append_route_payloads(struct flb_input_instance *ins, | |
| return -1; | ||
| } | ||
|
|
||
| /* | ||
| * Report acceptance so the caller counts the batch once, even if a | ||
| * later route copy in this loop fails. | ||
| */ | ||
| if (any_appended != NULL) { | ||
| *any_appended = FLB_TRUE; | ||
| } | ||
|
|
||
| appended++; | ||
| } | ||
|
|
||
|
|
@@ -1473,6 +1485,9 @@ static int input_log_append_processed_internal(struct flb_input_instance *ins, | |
| int ret; | ||
| int conditional_result; | ||
| int conditional_handled = FLB_FALSE; | ||
| int conditional_active = FLB_FALSE; | ||
| int base_flags = 0; | ||
| int accepted = FLB_FALSE; | ||
| size_t dummy = 0; | ||
| const char *base_tag = tag; | ||
| size_t base_tag_len = tag_len; | ||
|
|
@@ -1492,29 +1507,76 @@ static int input_log_append_processed_internal(struct flb_input_instance *ins, | |
| base_tag_len = strlen(base_tag); | ||
| } | ||
|
|
||
| /* | ||
| * With conditional routing the batch is fanned out into per-route copies | ||
| * plus a base append, all skipping input metrics. It is accounted once | ||
| * below, after at least one append is accepted, so a fully rejected batch | ||
| * is not counted. The non-routed path keeps counting in | ||
| * flb_input_chunk_append_raw(). | ||
| */ | ||
| conditional_active = (cfl_list_size(&ins->routes_direct) > 0 && | ||
| input_has_conditional_routes(ins) == FLB_TRUE); | ||
| if (conditional_active == FLB_TRUE) { | ||
| base_flags = FLB_INPUT_CHUNK_SKIP_INPUT_METRICS; | ||
| } | ||
|
|
||
| conditional_result = split_and_append_route_payloads(ins, records, tag, tag_len, | ||
| buf, buf_size, | ||
| local_append, &accepted); | ||
| if (conditional_result > 0) { | ||
| conditional_handled = FLB_TRUE; | ||
| } | ||
|
|
||
| /* | ||
| * Base append carries the full buffer for non-conditional routes (routing | ||
| * is additive). It is skipped on split failure. | ||
| */ | ||
| if (conditional_result >= 0) { | ||
| if (local_append == FLB_TRUE) { | ||
| 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_flags(ins, FLB_INPUT_LOGS, | ||
| base_flags, records, | ||
| tag, tag_len, buf, buf_size); | ||
| } | ||
| if (ret == 0) { | ||
| accepted = FLB_TRUE; | ||
| } | ||
| } | ||
| else { | ||
| ret = -1; | ||
| } | ||
|
|
||
| #ifdef FLB_HAVE_METRICS | ||
| /* count the batch once, now that at least one append was accepted */ | ||
| if (conditional_active == FLB_TRUE && accepted == FLB_TRUE && buf_size > 0) { | ||
| size_t effective_records = records; | ||
|
|
||
| /* mirror the zero-record recovery in input_chunk_append_raw() */ | ||
| if (effective_records == 0) { | ||
| effective_records = flb_mp_count_log_records(buf, buf_size); | ||
| } | ||
|
|
||
| if (effective_records > 0) { | ||
| uint64_t ts = cfl_time_now(); | ||
| char *name = (char *) flb_input_name(ins); | ||
|
|
||
| cmt_counter_add(ins->cmt_records, ts, effective_records, 1, | ||
| (char *[]) {name}); | ||
| cmt_counter_add(ins->cmt_bytes, ts, buf_size, 1, | ||
| (char *[]) {name}); | ||
|
|
||
| flb_metrics_sum(FLB_METRIC_N_RECORDS, effective_records, ins->metrics); | ||
| flb_metrics_sum(FLB_METRIC_N_BYTES, buf_size, ins->metrics); | ||
| } | ||
| } | ||
| #endif | ||
|
|
||
| if (conditional_result < 0) { | ||
| return -1; | ||
| } | ||
|
|
||
| if (ret == 0 && conditional_handled == FLB_TRUE && base_tag) { | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
In conditional routing under
mem_buf_limitor storage pause, one of these routed appends can successfully create a route-specific chunk and then pause the input viaflb_input_chunk_protect; because the route copy is markedFLB_INPUT_CHUNK_SKIP_INPUT_METRICS, the only planned input-counter update is the later unconditional append. That append runs after this loop and returns before counting whenflb_input_buf_paused()is true, so chunks already accepted for conditional routes are delivered without ever being reflected influentbit_input_records_total/_bytes_total.Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@codex review