Skip to content

out_cloudwatch_logs: reuse entity record accessors across records #12256

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

Open
ParakhJaggi wants to merge 1 commit into
base: master
Choose a base branch
from
+97 −24
Open

out_cloudwatch_logs: reuse entity record accessors across records #12256

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
89 changes: 65 additions & 24 deletions plugins/out_cloudwatch_logs/cloudwatch_api.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1236,57 +1236,98 @@ static void set_entity_field(char **field, struct flb_ra_value *val,
}
}

/*
* Paths for the entity record accessors, indexed by the ENTITY_RA_* slots.
* Keep in sync with the field map in parse_entity().
*/
static const char *entity_ra_paths[ENTITY_RA_MAX] = {
[ENTITY_RA_SERVICE_NAME] = "$kubernetes['aws_entity_service_name']",
[ENTITY_RA_ENVIRONMENT] = "$kubernetes['aws_entity_environment']",
[ENTITY_RA_NAMESPACE] = "$kubernetes['namespace_name']",
[ENTITY_RA_NODE] = "$kubernetes['host']",
[ENTITY_RA_CLUSTER] = "$kubernetes['aws_entity_cluster']",
[ENTITY_RA_WORKLOAD] = "$kubernetes['aws_entity_workload']",
[ENTITY_RA_NAME_SOURCE] = "$kubernetes['aws_entity_name_source']",
[ENTITY_RA_PLATFORM] = "$kubernetes['aws_entity_platform']",
[ENTITY_RA_INSTANCE_ID] = "$aws_entity_ec2_instance_id",
[ENTITY_RA_ACCOUNT_ID] = "$aws_entity_account_id"
};

void entity_ra_destroy(struct flb_cloudwatch *ctx)
{
int i;

for (i = 0; i < ENTITY_RA_MAX; i++) {
if (ctx->entity_ra[i]) {
flb_ra_destroy(ctx->entity_ra[i]);
ctx->entity_ra[i] = NULL;
}
}
}

int entity_ra_init(struct flb_cloudwatch *ctx)
{
int i;

for (i = 0; i < ENTITY_RA_MAX; i++) {
ctx->entity_ra[i] = flb_ra_create((char *) entity_ra_paths[i],
FLB_FALSE);
if (ctx->entity_ra[i] == NULL) {
flb_plg_error(ctx->ins, "Could not parse entity record accessor %s",
entity_ra_paths[i]);
entity_ra_destroy(ctx);
return -1;
}
}

return 0;
}

void parse_entity(struct flb_cloudwatch *ctx, entity *entity,
msgpack_object map, int map_size)
{
struct flb_record_accessor *ra;
struct flb_ra_value *val;
int i;

struct {
const char *path;
char **field;
int *filter_count;
int *found_flag;
} field_map[] = {
{"$kubernetes['aws_entity_service_name']", &entity->key_attributes->name,
} field_map[ENTITY_RA_MAX] = {
[ENTITY_RA_SERVICE_NAME] = {&entity->key_attributes->name,
&entity->filter_count, &entity->service_name_found},
{"$kubernetes['aws_entity_environment']", &entity->key_attributes->environment,
[ENTITY_RA_ENVIRONMENT] = {&entity->key_attributes->environment,
&entity->filter_count, &entity->environment_found},
{"$kubernetes['namespace_name']", &entity->attributes->namespace,
[ENTITY_RA_NAMESPACE] = {&entity->attributes->namespace,
NULL, NULL},
{"$kubernetes['host']", &entity->attributes->node, NULL, NULL},
{"$kubernetes['aws_entity_cluster']", &entity->attributes->cluster_name,
[ENTITY_RA_NODE] = {&entity->attributes->node, NULL, NULL},
[ENTITY_RA_CLUSTER] = {&entity->attributes->cluster_name,
&entity->filter_count, NULL},
{"$kubernetes['aws_entity_workload']", &entity->attributes->workload,
[ENTITY_RA_WORKLOAD] = {&entity->attributes->workload,
&entity->filter_count, NULL},
{"$kubernetes['aws_entity_name_source']", &entity->attributes->name_source,
[ENTITY_RA_NAME_SOURCE] = {&entity->attributes->name_source,
&entity->filter_count, &entity->name_source_found},
{"$kubernetes['aws_entity_platform']", &entity->attributes->platform_type,
[ENTITY_RA_PLATFORM] = {&entity->attributes->platform_type,
&entity->filter_count, NULL},
{"$aws_entity_ec2_instance_id", &entity->attributes->instance_id,
[ENTITY_RA_INSTANCE_ID] = {&entity->attributes->instance_id,
&entity->root_filter_count, NULL},
{"$aws_entity_account_id", &entity->key_attributes->account_id,
&entity->root_filter_count, NULL},
{NULL, NULL, NULL, NULL}
[ENTITY_RA_ACCOUNT_ID] = {&entity->key_attributes->account_id,
&entity->root_filter_count, NULL}
};

for (i = 0; field_map[i].path; i++) {
ra = flb_ra_create((char *) field_map[i].path, FLB_FALSE);
if (!ra) {

for (i = 0; i < ENTITY_RA_MAX; i++) {
if (ctx->entity_ra[i] == NULL) {
continue;
}
val = flb_ra_get_value_object(ra, map);

val = flb_ra_get_value_object(ctx->entity_ra[i], map);
if (val) {
set_entity_field(field_map[i].field, val, field_map[i].filter_count,
field_map[i].found_flag);
flb_ra_key_value_destroy(val);
}

flb_ra_destroy(ra);
}
Comment on lines +1268 to 1329

@coderabbitai coderabbitai Bot Aug 10, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

rg -n -C 4 --glob '*.{c,h}' \
  '\b(entity_ra_init|entity_ra_destroy|parse_entity|flb_ra_create)\s*\(' \
  plugins/out_cloudwatch_logs tests

rg -n -C 4 --glob '*.{c,h}' \
  'add_entity|aws_entity_service_name|aws_entity_environment|FLB_RA_STRING' \
  plugins/out_cloudwatch_logs tests

Repository: fluent/fluent-bit

Length of output: 50373

🏁 Script executed:

#!/bin/bash
set -euo pipefail

printf 'Files matching cloudwatch tests:\n'
fd -i 'cloudwatch' tests | sed -n '1,80p'

printf '\nReferences to entity RA/entity fields/services in out_cloudwatch_logs:\n'
rg -n --glob '*.{c,h}' \
  'entity_ra_init|entity_ra_destroy|parse_entity|set_entity_field|add_entity|aws_entity_service_name|aws_entity_environment|aws_entity_cluster|aws_entity_workload|aws_entity_ec2_instance_id|aws_entity_account_id|ENTITY_RA_|filter_count|found_flag' \
  plugins/out_cloudwatch_logs tests | sed -n '1,220p'

printf '\nRelevant init/cleanup code:\n'
sed -n '140,172p;508,526p' plugins/out_cloudwatch_logs/cloudwatch_logs.c
sed -n '1200,1235p;1238,1345p' plugins/out_cloudwatch_logs/cloudwatch_api.c

Repository: fluent/fluent-bit

Length of output: 15857

Add unit coverage for entity accessor lifecycle.

tests/runtime/out_cloudwatch.c does not cover the new entity_ra_init()/entity_ra_destroy() path or parse_entity() field handling. Add tests for successful initialization, partial accessor initialization failure, missing fields, and non-string entity fields, including cleanup that releases only initialized accessors.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@plugins/out_cloudwatch_logs/cloudwatch_api.c` around lines 1268 - 1329, Add
unit tests in the existing CloudWatch test suite covering entity_ra_init()
success, partial flb_ra_create() failure, and cleanup through
entity_ra_destroy() that releases only initialized accessors. Exercise
parse_entity() with missing fields and non-string values, verifying fields and
counters remain correct without crashes or invalid updates. Reuse the existing
CloudWatch test helpers and accessor-path symbols rather than adding unrelated
production changes.

Source: Coding guidelines


if (entity->key_attributes->name == NULL &&
entity->attributes->name_source == NULL &&
entity->attributes->workload != NULL) {
Expand Down
2 changes: 2 additions & 0 deletions plugins/out_cloudwatch_logs/cloudwatch_api.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,8 @@ int put_log_events(struct flb_cloudwatch *ctx, struct cw_flush *buf,
struct log_stream *stream,
size_t payload_size);
int create_log_group(struct flb_cloudwatch *ctx, struct log_stream *stream);
int entity_ra_init(struct flb_cloudwatch *ctx);
void entity_ra_destroy(struct flb_cloudwatch *ctx);
int compare_events(const void *a_arg, const void *b_arg);
void reset_flush_buf(struct flb_cloudwatch *ctx, struct cw_flush *buf);
void cloudwatch_mock_call_count_reset(void);
Expand Down
8 changes: 8 additions & 0 deletions plugins/out_cloudwatch_logs/cloudwatch_logs.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -162,6 +162,12 @@ static int cb_cloudwatch_init(struct flb_output_instance *ins,
}
}

if (ctx->add_entity == FLB_TRUE) {
if (entity_ra_init(ctx) != 0) {
goto error;
}
}

tmp = flb_output_get_property("log_format", ins);
if (tmp) {
ctx->log_format = tmp;
Expand Down Expand Up @@ -514,6 +520,8 @@ void flb_cloudwatch_ctx_destroy(struct flb_cloudwatch *ctx)
flb_ra_destroy(ctx->ra_stream);
}

entity_ra_destroy(ctx);

if (ctx->group_name) {
flb_sds_destroy(ctx->group_name);
}
Expand Down
22 changes: 22 additions & 0 deletions plugins/out_cloudwatch_logs/cloudwatch_logs.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,6 +146,25 @@ struct log_stream {
struct mk_list _head;
};

/*
* Slots for the record accessors used to extract entity fields. The paths are
* constant, so the accessors are compiled once at init time and reused for
* every record instead of being rebuilt per record.
*/
enum {
ENTITY_RA_SERVICE_NAME = 0,
ENTITY_RA_ENVIRONMENT,
ENTITY_RA_NAMESPACE,
ENTITY_RA_NODE,
ENTITY_RA_CLUSTER,
ENTITY_RA_WORKLOAD,
ENTITY_RA_NAME_SOURCE,
ENTITY_RA_PLATFORM,
ENTITY_RA_INSTANCE_ID,
ENTITY_RA_ACCOUNT_ID,
ENTITY_RA_MAX
};

struct flb_cloudwatch {
/*
* TLS instances can not be re-used. So we have one for:
Expand Down Expand Up @@ -217,6 +236,9 @@ struct flb_cloudwatch {
int kubernete_metadata_enabled;

int add_entity;

/* record accessors for entity fields, compiled once at init time */
struct flb_record_accessor *entity_ra[ENTITY_RA_MAX];
};

void flb_cloudwatch_ctx_destroy(struct flb_cloudwatch *ctx);
Expand Down
Loading