From 95823e4bc6f80f8cd31792efa38ea601f917e7c5 Mon Sep 17 00:00:00 2001 From: ParakhJaggi Date: Mon, 10 Aug 2026 12:01:02 -0400 Subject: [PATCH] out_cloudwatch_logs: reuse entity record accessors across records parse_entity() built and destroyed ten record accessors for every single record when add_entity is enabled. The paths are compile time constants, so this repeated the same parse work per record and produced a steady churn of allocations and frees proportional to the ingested log volume. Compile the accessors once during plugin init and store them on the context, following the same pattern already used for ra_group and ra_stream, then release them in flb_cloudwatch_ctx_destroy(). The accessors are only built when add_entity is enabled, so deployments that do not use entity support are unaffected. The paths and the field map are indexed by the same ENTITY_RA_* slots so the two tables cannot drift out of order. Signed-off-by: ParakhJaggi --- plugins/out_cloudwatch_logs/cloudwatch_api.c | 89 ++++++++++++++----- plugins/out_cloudwatch_logs/cloudwatch_api.h | 2 + plugins/out_cloudwatch_logs/cloudwatch_logs.c | 8 ++ plugins/out_cloudwatch_logs/cloudwatch_logs.h | 22 +++++ 4 files changed, 97 insertions(+), 24 deletions(-) diff --git a/plugins/out_cloudwatch_logs/cloudwatch_api.c b/plugins/out_cloudwatch_logs/cloudwatch_api.c index d3aa39d3232..5f0e30d7c61 100644 --- a/plugins/out_cloudwatch_logs/cloudwatch_api.c +++ b/plugins/out_cloudwatch_logs/cloudwatch_api.c @@ -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); } - + if (entity->key_attributes->name == NULL && entity->attributes->name_source == NULL && entity->attributes->workload != NULL) { diff --git a/plugins/out_cloudwatch_logs/cloudwatch_api.h b/plugins/out_cloudwatch_logs/cloudwatch_api.h index c8f9540d4db..15da78a1b8a 100644 --- a/plugins/out_cloudwatch_logs/cloudwatch_api.h +++ b/plugins/out_cloudwatch_logs/cloudwatch_api.h @@ -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); diff --git a/plugins/out_cloudwatch_logs/cloudwatch_logs.c b/plugins/out_cloudwatch_logs/cloudwatch_logs.c index 2f8acf188d6..0331fd6e304 100644 --- a/plugins/out_cloudwatch_logs/cloudwatch_logs.c +++ b/plugins/out_cloudwatch_logs/cloudwatch_logs.c @@ -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; @@ -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); } diff --git a/plugins/out_cloudwatch_logs/cloudwatch_logs.h b/plugins/out_cloudwatch_logs/cloudwatch_logs.h index 4011c434b0d..7d46b3260fc 100644 --- a/plugins/out_cloudwatch_logs/cloudwatch_logs.h +++ b/plugins/out_cloudwatch_logs/cloudwatch_logs.h @@ -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: @@ -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);