Skip to content

conditionals: resolve TAG record accessor in processor and filter conditions #12090

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
zshuang0316 wants to merge 5 commits into
base: master
Choose a base branch
from
Draft

conditionals: resolve TAG record accessor in processor and filter conditions #12090

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
b0820ca
conditionals: resolve TAG record accessor in condition rules
zshuang0316 Jul 13, 2026
00f5671
mp: pass the chunk tag when evaluating record conditions
zshuang0316 Jul 13, 2026
218ea60
processor: expose the chunk tag to condition rules
zshuang0316 Jul 13, 2026
aee44ad
router_condition: pass NULL tag to condition evaluation
zshuang0316 Jul 13, 2026
6142006
tests: conditionals: cover TAG rule resolution
zshuang0316 Jul 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 12 additions & 4 deletions include/fluent-bit/flb_conditionals.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,11 +97,19 @@ int flb_condition_add_rule(struct flb_condition *cond,

void flb_condition_destroy(struct flb_condition *cond);

/* Evaluation function */
/*
* Evaluation functions.
*
* tag/tag_len reference the tag of the chunk being evaluated so that rules
* using the $TAG record accessor can be resolved. Callers that have no tag
* available (e.g. the router, which matches by tag natively) may pass NULL/0.
*/
int flb_condition_evaluate_ex(struct flb_condition *cond,
void *ctx,
flb_condition_get_variant_fn get_variant);
flb_condition_get_variant_fn get_variant,
const char *tag, int tag_len);
int flb_condition_evaluate(struct flb_condition *cond,
struct flb_mp_chunk_record *record);
struct flb_mp_chunk_record *record,
const char *tag, int tag_len);

#endif /* FLB_CONDITIONS_H */
#endif /* FLB_CONDITIONS_H */
4 changes: 4 additions & 0 deletions include/fluent-bit/flb_mp_chunk.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,10 @@ struct flb_mp_chunk_cobj {

/* Condition for filtering records during processing */
struct flb_condition *condition;

/* Tag of the chunk being processed, so conditions can reference $TAG */
const char *tag;
int tag_len;
};


Expand Down
17 changes: 11 additions & 6 deletions src/flb_conditionals.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -278,7 +278,8 @@ void flb_condition_destroy(struct flb_condition *cond)
}

static int evaluate_rule(struct flb_condition_rule *rule,
struct cfl_variant *record_variant)
struct cfl_variant *record_variant,
const char *tag, int tag_len)
{
flb_sds_t str_val;
int i;
Expand All @@ -291,7 +292,8 @@ static int evaluate_rule(struct flb_condition_rule *rule,
}

flb_trace("[condition] evaluating rule with record accessor");
str_val = flb_cfl_ra_translate(rule->ra, NULL, 0, *record_variant, NULL);
str_val = flb_cfl_ra_translate(rule->ra, (char *) tag, tag_len,
*record_variant, NULL);
if (!str_val) {
flb_trace("[condition] record accessor translation failed");
return FLB_FALSE;
Expand Down Expand Up @@ -362,7 +364,8 @@ static int evaluate_rule(struct flb_condition_rule *rule,

int flb_condition_evaluate_ex(struct flb_condition *cond,
void *ctx,
flb_condition_get_variant_fn get_variant)
flb_condition_get_variant_fn get_variant,
const char *tag, int tag_len)
{
struct mk_list *head;
struct flb_condition_rule *rule;
Expand Down Expand Up @@ -406,7 +409,7 @@ int flb_condition_evaluate_ex(struct flb_condition *cond,
}

flb_trace("[condition] evaluating rule against record");
result = evaluate_rule(rule, record_variant);
result = evaluate_rule(rule, record_variant, tag, tag_len);
flb_trace("[condition] rule evaluation result: %d", result);

if (cond->op == FLB_COND_OP_AND && result == FLB_FALSE) {
Expand Down Expand Up @@ -438,7 +441,8 @@ int flb_condition_evaluate_ex(struct flb_condition *cond,
}

int flb_condition_evaluate(struct flb_condition *cond,
struct flb_mp_chunk_record *record)
struct flb_mp_chunk_record *record,
const char *tag, int tag_len)
{
if (!cond) {
return FLB_TRUE;
Expand All @@ -448,5 +452,6 @@ int flb_condition_evaluate(struct flb_condition *cond,
return FLB_FALSE;
}

return flb_condition_evaluate_ex(cond, record, default_get_record_variant);
return flb_condition_evaluate_ex(cond, record, default_get_record_variant,
tag, tag_len);
}
9 changes: 6 additions & 3 deletions src/flb_mp.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1803,7 +1803,8 @@ int flb_mp_chunk_cobj_record_next(struct flb_mp_chunk_cobj *chunk_cobj,
/* If there's a condition, check if the record matches */
if (condition != NULL && record != NULL) {
flb_trace("[mp] evaluating condition for record");
ret = flb_condition_evaluate(condition, record);
ret = flb_condition_evaluate(condition, record,
chunk_cobj->tag, chunk_cobj->tag_len);
flb_trace("[mp] condition evaluation result: %s", ret ? "TRUE" : "FALSE");
if (ret == FLB_FALSE) {
flb_trace("[mp] record didn't match condition, skipping");
Expand All @@ -1828,7 +1829,8 @@ int flb_mp_chunk_cobj_record_next(struct flb_mp_chunk_cobj *chunk_cobj,
/* If there's a condition, check if the record matches */
if (condition != NULL && record != NULL) {
flb_trace("[mp] evaluating condition for next record");
ret = flb_condition_evaluate(condition, record);
ret = flb_condition_evaluate(condition, record,
chunk_cobj->tag, chunk_cobj->tag_len);
flb_trace("[mp] next record condition evaluation result: %s", ret ? "TRUE" : "FALSE");
if (ret == FLB_FALSE) {
flb_trace("[mp] next record didn't match condition, skipping");
Expand All @@ -1852,7 +1854,8 @@ int flb_mp_chunk_cobj_record_next(struct flb_mp_chunk_cobj *chunk_cobj,
/* If there's a condition, check if the record matches */
if (condition != NULL && record != NULL) {
flb_trace("[mp] evaluating condition for first record");
ret = flb_condition_evaluate(condition, record);
ret = flb_condition_evaluate(condition, record,
chunk_cobj->tag, chunk_cobj->tag_len);
flb_trace("[mp] first record condition evaluation result: %s", ret ? "TRUE" : "FALSE");
if (ret == FLB_FALSE) {
flb_trace("[mp] first record didn't match condition, skipping");
Expand Down
14 changes: 11 additions & 3 deletions src/flb_processor.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -328,7 +328,8 @@ static int append_buffer_records_to_encoder(struct flb_log_event_encoder *encode
}

static int evaluate_condition_for_log_event(struct flb_condition *condition,
struct flb_log_event *log_event)
struct flb_log_event *log_event,
const char *tag, int tag_len)
{
int ret;
struct flb_mp_chunk_record record = {0};
Expand All @@ -344,7 +345,7 @@ static int evaluate_condition_for_log_event(struct flb_condition *condition,
return FLB_FALSE;
}

ret = flb_condition_evaluate(condition, &record);
ret = flb_condition_evaluate(condition, &record, tag, tag_len);

cfl_object_destroy(record.cobj_record);
cfl_object_destroy(record.cobj_metadata);
Expand Down Expand Up @@ -393,7 +394,8 @@ static int run_filter_with_condition(struct flb_filter_instance *f_ins,

while ((ret = flb_log_event_decoder_next(&decoder, &log_event)) ==
FLB_EVENT_DECODER_SUCCESS) {
condition_match = evaluate_condition_for_log_event(condition, &log_event);
condition_match = evaluate_condition_for_log_event(condition, &log_event,
tag, tag_len);

if (condition_match == FLB_FALSE) {
ret = append_raw_record_to_encoder(&final_encoder,
Expand Down Expand Up @@ -1531,6 +1533,10 @@ int flb_processor_run(struct flb_processor *proc,
flb_debug("[processor] no condition set for processor unit (pu=%s)", pu->name);
}

/* Expose the chunk tag so conditions can reference $TAG */
chunk_cobj->tag = tag;
chunk_cobj->tag_len = tag_len;

/* Invoke processor plugin callback */
ret = p_ins->p->cb_process_logs(p_ins, chunk_cobj, tag, tag_len);
if (ret != FLB_PROCESSOR_SUCCESS) {
Expand All @@ -1541,6 +1547,8 @@ int flb_processor_run(struct flb_processor *proc,

/* Clear the condition after processing */
chunk_cobj->condition = NULL;
chunk_cobj->tag = NULL;
chunk_cobj->tag_len = 0;
finalize = FLB_FALSE;

/* is this processing_unit the last one from the list ? */
Expand Down
4 changes: 2 additions & 2 deletions src/flb_router_condition.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -304,7 +304,7 @@ int flb_condition_eval_logs(struct flb_event_chunk *chunk,
cfl_list_foreach(head, &context->chunk_cobj->records) {
record = cfl_list_entry(head, struct flb_mp_chunk_record, _head);

if (flb_condition_evaluate_ex(compiled, record, route_logs_get_variant) == FLB_TRUE) {
if (flb_condition_evaluate_ex(compiled, record, route_logs_get_variant, NULL, 0) == FLB_TRUE) {
result = FLB_TRUE;
break;
}
Expand Down Expand Up @@ -432,7 +432,7 @@ int flb_router_condition_evaluate_record(struct flb_route *route,
return FLB_FALSE;
}

return flb_condition_evaluate_ex(compiled, record, route_logs_get_variant);
return flb_condition_evaluate_ex(compiled, record, route_logs_get_variant, NULL, 0);
}

static int parse_rule_operator(const flb_sds_t op_str,
Expand Down
Loading
Loading