From 52799ff7703d656f9be0fafbff18a96663ab11da Mon Sep 17 00:00:00 2001 From: Yu Yi Date: Tue, 30 Jun 2026 19:33:08 -0400 Subject: [PATCH 1/4] out_stackdriver: make resource metadata extraction thread-safe When the output runs with workers >= 2, multiple flush threads call stackdriver_format() concurrently on the shared plugin context and overwrite the same ctx->pod_name / namespace_name / container_name / node_name / local_resource_id strings (flb_sds_destroy + flb_sds_create), causing use-after-free / double-free crashes (SIGSEGV) under load. Move these per-flush fields into a stack-local struct stackdriver_format_ctx owned by stackdriver_format(), thread the resource helpers through it, and release them in a single cleanup path. The shared context no longer holds mutable per-record resource state, so the format path is safe for any worker count without locking. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Yu Yi --- plugins/out_stackdriver/stackdriver.c | 495 +++++++++++---------- plugins/out_stackdriver/stackdriver.h | 5 - plugins/out_stackdriver/stackdriver_conf.c | 9 +- 3 files changed, 251 insertions(+), 258 deletions(-) diff --git a/plugins/out_stackdriver/stackdriver.c b/plugins/out_stackdriver/stackdriver.c index 212deadad8c..8f4ef69b08e 100644 --- a/plugins/out_stackdriver/stackdriver.c +++ b/plugins/out_stackdriver/stackdriver.c @@ -46,6 +46,15 @@ #include "stackdriver_helper.h" #include "stackdriver_resource_types.h" +struct stackdriver_format_ctx { + struct flb_stackdriver *ctx; + flb_sds_t local_resource_id; + flb_sds_t namespace_name; + flb_sds_t pod_name; + flb_sds_t container_name; + flb_sds_t node_name; +}; + pthread_key_t oauth2_type; pthread_key_t oauth2_token; pthread_key_t oauth2_token_expires; @@ -718,12 +727,13 @@ static struct mk_list *parse_local_resource_id_to_list(char *local_resource_id, * - if local_resource_id is missing from the payLoad, use the tag of the log */ static int extract_local_resource_id(const void *data, size_t bytes, - struct flb_stackdriver *ctx, const char *tag) { + struct stackdriver_format_ctx *fmt_ctx, const char *tag) { msgpack_object_map map; flb_sds_t local_resource_id; struct flb_log_event_decoder log_decoder; struct flb_log_event log_event; int ret; + struct flb_stackdriver *ctx = fmt_ctx->ctx; ret = flb_log_event_decoder_init(&log_decoder, (char *) data, bytes); @@ -749,11 +759,11 @@ static int extract_local_resource_id(const void *data, size_t bytes, } /* we need to create up the local_resource_id from previous log */ - if (ctx->local_resource_id) { - flb_sds_destroy(ctx->local_resource_id); + if (fmt_ctx->local_resource_id) { + flb_sds_destroy(fmt_ctx->local_resource_id); } - ctx->local_resource_id = flb_sds_create(local_resource_id); + fmt_ctx->local_resource_id = flb_sds_create(local_resource_id); flb_sds_destroy(local_resource_id); @@ -775,7 +785,7 @@ static int extract_local_resource_id(const void *data, size_t bytes, * - use the extracted local_resource_id to assign the label keys for different * resource types that are specified in the configuration of stackdriver_out plugin */ -static int set_monitored_resource_labels(struct flb_stackdriver *ctx, char *type) +static int set_monitored_resource_labels(struct stackdriver_format_ctx *fmt_ctx, char *type) { int ret = -1; int first = FLB_TRUE; @@ -788,8 +798,9 @@ static int set_monitored_resource_labels(struct flb_stackdriver *ctx, char *type struct mk_list *list = NULL; struct mk_list *head; flb_sds_t new_local_resource_id; + struct flb_stackdriver *ctx = fmt_ctx->ctx; - if (!ctx->local_resource_id) { + if (!fmt_ctx->local_resource_id) { flb_plg_error(ctx->ins, "local_resource_is is not assigned"); return -1; } @@ -799,15 +810,19 @@ static int set_monitored_resource_labels(struct flb_stackdriver *ctx, char *type len_k8s_pod = sizeof(K8S_POD) - 1; prefix_len = flb_sds_len(ctx->tag_prefix); - if (flb_sds_casecmp(ctx->tag_prefix, ctx->local_resource_id, prefix_len) != 0) { - flb_plg_error(ctx->ins, "tag_prefix [%s] doesn't match the prefix of" - " local_resource_id [%s]", ctx->tag_prefix, - ctx->local_resource_id); + if (flb_sds_len(fmt_ctx->local_resource_id) < prefix_len) { + flb_plg_error(ctx->ins, "invalid local_resource_id"); + return -1; + } + + if (flb_sds_casecmp(ctx->tag_prefix, fmt_ctx->local_resource_id, prefix_len) != 0) { + flb_plg_error(ctx->ins, "prefix doesn't match for local_resource_id %s", + fmt_ctx->local_resource_id); return -1; } - new_local_resource_id = flb_sds_create_len(ctx->local_resource_id, - flb_sds_len(ctx->local_resource_id)); + new_local_resource_id = flb_sds_create_len(fmt_ctx->local_resource_id, + flb_sds_len(fmt_ctx->local_resource_id)); replace_prefix_dot(new_local_resource_id, prefix_len - 1); if (strncmp(type, K8S_CONTAINER, len_k8s_container) == 0) { @@ -826,28 +841,28 @@ static int set_monitored_resource_labels(struct flb_stackdriver *ctx, char *type /* Follow the order of fields in local_resource_id */ if (counter == 0) { - if (ctx->namespace_name) { - flb_sds_destroy(ctx->namespace_name); + if (fmt_ctx->namespace_name) { + flb_sds_destroy(fmt_ctx->namespace_name); } - ctx->namespace_name = flb_sds_create(ptr->val); + fmt_ctx->namespace_name = flb_sds_create(ptr->val); } else if (counter == 1) { - if (ctx->pod_name) { - flb_sds_destroy(ctx->pod_name); + if (fmt_ctx->pod_name) { + flb_sds_destroy(fmt_ctx->pod_name); } - ctx->pod_name = flb_sds_create(ptr->val); + fmt_ctx->pod_name = flb_sds_create(ptr->val); } else if (counter == 2) { - if (ctx->container_name) { - flb_sds_destroy(ctx->container_name); + if (fmt_ctx->container_name) { + flb_sds_destroy(fmt_ctx->container_name); } - ctx->container_name = flb_sds_create(ptr->val); + fmt_ctx->container_name = flb_sds_create(ptr->val); } counter++; } - if (!ctx->namespace_name || !ctx->pod_name || !ctx->container_name) { + if (!fmt_ctx->namespace_name || !fmt_ctx->pod_name || !fmt_ctx->container_name) { goto error; } } @@ -865,14 +880,14 @@ static int set_monitored_resource_labels(struct flb_stackdriver *ctx, char *type } if (ptr != NULL) { - if (ctx->node_name) { - flb_sds_destroy(ctx->node_name); + if (fmt_ctx->node_name) { + flb_sds_destroy(fmt_ctx->node_name); } - ctx->node_name = flb_sds_create(ptr->val); + fmt_ctx->node_name = flb_sds_create(ptr->val); } } - if (!ctx->node_name) { + if (!fmt_ctx->node_name) { goto error; } } @@ -891,22 +906,22 @@ static int set_monitored_resource_labels(struct flb_stackdriver *ctx, char *type /* Follow the order of fields in local_resource_id */ if (counter == 0) { - if (ctx->namespace_name) { - flb_sds_destroy(ctx->namespace_name); + if (fmt_ctx->namespace_name) { + flb_sds_destroy(fmt_ctx->namespace_name); } - ctx->namespace_name = flb_sds_create(ptr->val); + fmt_ctx->namespace_name = flb_sds_create(ptr->val); } else if (counter == 1) { - if (ctx->pod_name) { - flb_sds_destroy(ctx->pod_name); + if (fmt_ctx->pod_name) { + flb_sds_destroy(fmt_ctx->pod_name); } - ctx->pod_name = flb_sds_create(ptr->val); + fmt_ctx->pod_name = flb_sds_create(ptr->val); } counter++; } - if (!ctx->namespace_name || !ctx->pod_name) { + if (!fmt_ctx->namespace_name || !fmt_ctx->pod_name) { goto error; } } @@ -928,30 +943,36 @@ static int set_monitored_resource_labels(struct flb_stackdriver *ctx, char *type } if (strncmp(type, K8S_CONTAINER, len_k8s_container) == 0) { - if (ctx->namespace_name) { - flb_sds_destroy(ctx->namespace_name); + if (fmt_ctx->namespace_name) { + flb_sds_destroy(fmt_ctx->namespace_name); + fmt_ctx->namespace_name = NULL; } - if (ctx->pod_name) { - flb_sds_destroy(ctx->pod_name); + if (fmt_ctx->pod_name) { + flb_sds_destroy(fmt_ctx->pod_name); + fmt_ctx->pod_name = NULL; } - if (ctx->container_name) { - flb_sds_destroy(ctx->container_name); + if (fmt_ctx->container_name) { + flb_sds_destroy(fmt_ctx->container_name); + fmt_ctx->container_name = NULL; } } else if (strncmp(type, K8S_NODE, len_k8s_node) == 0) { - if (ctx->node_name) { - flb_sds_destroy(ctx->node_name); + if (fmt_ctx->node_name) { + flb_sds_destroy(fmt_ctx->node_name); + fmt_ctx->node_name = NULL; } } else if (strncmp(type, K8S_POD, len_k8s_pod) == 0) { - if (ctx->namespace_name) { - flb_sds_destroy(ctx->namespace_name); + if (fmt_ctx->namespace_name) { + flb_sds_destroy(fmt_ctx->namespace_name); + fmt_ctx->namespace_name = NULL; } - if (ctx->pod_name) { - flb_sds_destroy(ctx->pod_name); + if (fmt_ctx->pod_name) { + flb_sds_destroy(fmt_ctx->pod_name); + fmt_ctx->pod_name = NULL; } } @@ -983,21 +1004,22 @@ static int is_tag_match_regex(struct flb_stackdriver *ctx, return ret; } -static int is_local_resource_id_match_regex(struct flb_stackdriver *ctx) +static int is_local_resource_id_match_regex(struct stackdriver_format_ctx *fmt_ctx) { int ret; int prefix_len; int len_to_be_matched; const char *str_to_be_matcheds; + struct flb_stackdriver *ctx = fmt_ctx->ctx; - if (!ctx->local_resource_id) { + if (!fmt_ctx->local_resource_id) { flb_plg_warn(ctx->ins, "local_resource_id not found in the payload"); return -1; } prefix_len = flb_sds_len(ctx->tag_prefix); - str_to_be_matcheds = ctx->local_resource_id + prefix_len; - len_to_be_matched = flb_sds_len(ctx->local_resource_id) - prefix_len; + str_to_be_matcheds = fmt_ctx->local_resource_id + prefix_len; + len_to_be_matched = flb_sds_len(fmt_ctx->local_resource_id) - prefix_len; ret = flb_regex_match(ctx->regex, (unsigned char *) str_to_be_matcheds, @@ -1013,7 +1035,7 @@ static void cb_results(const char *name, const char *value, * extract_resource_labels_from_regex(4) will only be called if the * tag or local_resource_id field matches the regex rule */ -static int extract_resource_labels_from_regex(struct flb_stackdriver *ctx, +static int extract_resource_labels_from_regex(struct stackdriver_format_ctx *fmt_ctx, const char *tag, int tag_len, int from_tag) { @@ -1022,6 +1044,7 @@ static int extract_resource_labels_from_regex(struct flb_stackdriver *ctx, int len_to_be_matched; int local_resource_id_len; const char *str_to_be_matcheds; + struct flb_stackdriver *ctx = fmt_ctx->ctx; struct flb_regex_search result; prefix_len = flb_sds_len(ctx->tag_prefix); @@ -1031,8 +1054,8 @@ static int extract_resource_labels_from_regex(struct flb_stackdriver *ctx, } else { // this will be called only if the payload contains local_resource_id - local_resource_id_len = flb_sds_len(ctx->local_resource_id); - str_to_be_matcheds = ctx->local_resource_id + prefix_len; + local_resource_id_len = flb_sds_len(fmt_ctx->local_resource_id); + str_to_be_matcheds = fmt_ctx->local_resource_id + prefix_len; } len_to_be_matched = local_resource_id_len - prefix_len; @@ -1043,25 +1066,28 @@ static int extract_resource_labels_from_regex(struct flb_stackdriver *ctx, return -1; } - flb_regex_parse(ctx->regex, &result, cb_results, ctx); + flb_regex_parse(ctx->regex, &result, cb_results, fmt_ctx); return ret; } -static int process_local_resource_id(struct flb_stackdriver *ctx, +static int process_local_resource_id(struct stackdriver_format_ctx *fmt_ctx, const char *tag, int tag_len, char *type) { int ret; + struct flb_stackdriver *ctx = fmt_ctx->ctx; // parsing local_resource_id from tag takes higher priority if (is_tag_match_regex(ctx, tag, tag_len) > 0) { - ret = extract_resource_labels_from_regex(ctx, tag, tag_len, FLB_TRUE); - } - else if (is_local_resource_id_match_regex(ctx) > 0) { - ret = extract_resource_labels_from_regex(ctx, tag, tag_len, FLB_FALSE); + ret = extract_resource_labels_from_regex(fmt_ctx, tag, tag_len, FLB_TRUE); } else { - ret = set_monitored_resource_labels(ctx, type); + if (is_local_resource_id_match_regex(fmt_ctx) > 0) { + ret = extract_resource_labels_from_regex(fmt_ctx, tag, tag_len, FLB_FALSE); + } + else { + ret = set_monitored_resource_labels(fmt_ctx, type); + } } return ret; @@ -1247,35 +1273,35 @@ static void pack_labels(struct flb_stackdriver *ctx, static void cb_results(const char *name, const char *value, size_t vlen, void *data) { - struct flb_stackdriver *ctx = data; + struct stackdriver_format_ctx *fmt_ctx = data; if (vlen == 0) { return; } if (strcmp(name, "pod_name") == 0) { - if (ctx->pod_name != NULL) { - flb_sds_destroy(ctx->pod_name); + if (fmt_ctx->pod_name != NULL) { + flb_sds_destroy(fmt_ctx->pod_name); } - ctx->pod_name = flb_sds_create_len(value, vlen); + fmt_ctx->pod_name = flb_sds_create_len(value, vlen); } else if (strcmp(name, "namespace_name") == 0) { - if (ctx->namespace_name != NULL) { - flb_sds_destroy(ctx->namespace_name); + if (fmt_ctx->namespace_name != NULL) { + flb_sds_destroy(fmt_ctx->namespace_name); } - ctx->namespace_name = flb_sds_create_len(value, vlen); + fmt_ctx->namespace_name = flb_sds_create_len(value, vlen); } else if (strcmp(name, "container_name") == 0) { - if (ctx->container_name != NULL) { - flb_sds_destroy(ctx->container_name); + if (fmt_ctx->container_name != NULL) { + flb_sds_destroy(fmt_ctx->container_name); } - ctx->container_name = flb_sds_create_len(value, vlen); + fmt_ctx->container_name = flb_sds_create_len(value, vlen); } else if (strcmp(name, "node_name") == 0) { - if (ctx->node_name != NULL) { - flb_sds_destroy(ctx->node_name); + if (fmt_ctx->node_name != NULL) { + flb_sds_destroy(fmt_ctx->node_name); } - ctx->node_name = flb_sds_create_len(value, vlen); + fmt_ctx->node_name = flb_sds_create_len(value, vlen); } return; @@ -1341,6 +1367,8 @@ static int cb_stackdriver_init(struct flb_output_instance *ins, } ctx->token_mutex_initialized = FLB_TRUE; + + /* Create Upstream context for Stackdriver Logging (no oauth2 service) */ ctx->u = flb_upstream_create_url(config, ctx->cloud_logging_write_url, io_flags, ins->tls); @@ -1417,6 +1445,7 @@ static int cb_stackdriver_init(struct flb_output_instance *ins, return 0; error: + if (ctx->token_mutex_initialized == FLB_TRUE) { pthread_mutex_destroy(&ctx->token_mutex); ctx->token_mutex_initialized = FLB_FALSE; @@ -1509,7 +1538,7 @@ static int get_msgpack_obj(msgpack_object * subobj, const msgpack_object * o, continue; } - if (flb_sds_cmp(key, p->key.via.str.ptr, p->key.via.str.size) == 0) { + if (flb_sds_casecmp(key, p->key.via.str.ptr, p->key.via.str.size) == 0) { *subobj = p->val; return 0; } @@ -1721,8 +1750,9 @@ static int pack_payload(int insert_id_extracted, } /* processing logging.googleapis.com/operation */ - if (validate_key(kv->key, OPERATION_FIELD_IN_JSON, - OPERATION_KEY_SIZE) + if (operation_extracted == FLB_TRUE + && validate_key(kv->key, OPERATION_FIELD_IN_JSON, + OPERATION_KEY_SIZE) && kv->val.type == MSGPACK_OBJECT_MAP) { if (operation_extra_size > 0) { msgpack_pack_object(mp_pck, kv->key); @@ -1731,8 +1761,9 @@ static int pack_payload(int insert_id_extracted, continue; } - if (validate_key(kv->key, SOURCELOCATION_FIELD_IN_JSON, - SOURCE_LOCATION_SIZE) + if (source_location_extracted == FLB_TRUE + && validate_key(kv->key, SOURCELOCATION_FIELD_IN_JSON, + SOURCE_LOCATION_SIZE) && kv->val.type == MSGPACK_OBJECT_MAP) { if (source_location_extra_size > 0) { @@ -1743,8 +1774,9 @@ static int pack_payload(int insert_id_extracted, continue; } - if (validate_key(kv->key, ctx->http_request_key, - ctx->http_request_key_size) + if (http_request_extracted == FLB_TRUE + && validate_key(kv->key, ctx->http_request_key, + ctx->http_request_key_size) && kv->val.type == MSGPACK_OBJECT_MAP) { if(http_request_extra_size > 0) { @@ -1807,98 +1839,74 @@ static int pack_payload(int insert_id_extracted, static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, int total_records, - const char *tag, int tag_len, + char *tag, int tag_len, const void *data, size_t bytes, struct flb_config *config, int *formatted_records) { - int len; int ret; - int array_size = 0; - /* The default value is 3: timestamp, jsonPayload, logName. */ - int entry_size = 3; - size_t s; - // size_t off = 0; - char path[PATH_MAX]; - char time_formatted[255]; - const char *newtag; - const char *new_log_name; - msgpack_object *obj; - msgpack_sbuffer mp_sbuf; - msgpack_packer mp_pck; - flb_sds_t out_buf; - struct flb_mp_map_header mh; - - /* Parameters for project_id_key */ - int project_id_extracted = FLB_FALSE; - flb_sds_t project_id_key; - - /* Parameters for severity */ - int severity_extracted = FLB_FALSE; + int len; + int entry_size; + int labels_size; + int records_count = 0; + int tms_status; + int trace_extracted; + int span_id_extracted; + int trace_sampled_extracted; + int trace_sampled; + int project_id_extracted; + int log_name_extracted; + int severity_extracted; severity_t severity; - - /* Parameters for trace */ - int trace_extracted = FLB_FALSE; + int in_status; + int insert_id_extracted; + int operation_extracted; + int operation_first; + int operation_last; + int operation_extra_size; + int source_location_extracted; + int source_location_extra_size; + int http_request_extracted; + int http_request_extra_size; + char *new_trace; + char *newtag; + char *new_log_name; + char *operation_id; + char *operation_producer; + char *source_location_file; + char *source_location_function; + char path[1024]; + char time_formatted[32]; + char stackdriver_trace[256]; flb_sds_t trace = NULL; - char stackdriver_trace[PATH_MAX]; - const char *new_trace; - - /* Parameters for span id */ - int span_id_extracted = FLB_FALSE; - flb_sds_t span_id; - - /* Parameters for trace sampled */ - int trace_sampled_extracted = FLB_FALSE; - int trace_sampled = FLB_FALSE; - - /* Parameters for log name */ - int log_name_extracted = FLB_FALSE; + flb_sds_t span_id = NULL; + flb_sds_t project_id_key = NULL; flb_sds_t log_name = NULL; + flb_sds_t stream_key = NULL; flb_sds_t stream = NULL; - flb_sds_t stream_key; - - /* Parameters for insertId */ + flb_sds_t out_buf = NULL; + struct flb_log_event_decoder log_decoder; + struct flb_log_event log_event; msgpack_object insert_id_obj; - insert_id_status in_status; - int insert_id_extracted; - - /* Parameters in Operation */ - flb_sds_t operation_id; - flb_sds_t operation_producer; - int operation_first = FLB_FALSE; - int operation_last = FLB_FALSE; - int operation_extracted = FLB_FALSE; - int operation_extra_size = 0; - - /* Parameters for sourceLocation */ - flb_sds_t source_location_file; - int64_t source_location_line = 0; - flb_sds_t source_location_function; - int source_location_extracted = FLB_FALSE; - int source_location_extra_size = 0; - - /* Parameters for httpRequest */ + msgpack_sbuffer mp_sbuf; + msgpack_packer mp_pck; + msgpack_object *obj; + msgpack_object *payload_labels_ptr = NULL; struct http_request_field http_request; - int http_request_extracted = FLB_FALSE; - int http_request_extra_size = 0; - - /* Parameters for Timestamp */ + struct flb_mp_map_header mh; + struct flb_mp_map_header entries_mh; struct tm tm; - // struct flb_time tms; - timestamp_status tms_status; - /* Count number of records */ - array_size = total_records; - - if (formatted_records != NULL) { - *formatted_records = 0; - } + size_t s; + long source_location_line; - /* Parameters for labels */ - msgpack_object *payload_labels_ptr; - int labels_size = 0; + /* consolidated cleanup variables */ + int decoder_initialized = FLB_FALSE; + int sbuf_initialized = FLB_FALSE; + struct stackdriver_format_ctx fmt_ctx; - struct flb_log_event_decoder log_decoder; - struct flb_log_event log_event; + /* Initialize fmt_ctx */ + memset(&fmt_ctx, 0, sizeof(struct stackdriver_format_ctx)); + fmt_ctx.ctx = ctx; ret = flb_log_event_decoder_init(&log_decoder, (char *) data, bytes); @@ -1908,39 +1916,11 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, return NULL; } - - /* - * Search each entry and validate insertId. - * Reject the entry if insertId is invalid. - * If all the entries are rejected, stop formatting. - * - */ - while ((ret = flb_log_event_decoder_next( - &log_decoder, - &log_event)) == FLB_EVENT_DECODER_SUCCESS) { - /* Extract insertId */ - in_status = validate_insert_id(&insert_id_obj, log_event.body); - - if (in_status == INSERTID_INVALID) { - flb_plg_error(ctx->ins, - "Incorrect insertId received. InsertId should be non-empty string."); - array_size -= 1; - } - } - - flb_log_event_decoder_destroy(&log_decoder); - - /* Sounds like this should compare to -1 instead of zero */ - if (array_size == 0) { - return NULL; - } - - if (formatted_records != NULL) { - *formatted_records = array_size; - } + decoder_initialized = FLB_TRUE; /* Create temporal msgpack buffer */ msgpack_sbuffer_init(&mp_sbuf); + sbuf_initialized = FLB_TRUE; msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write); /* @@ -1975,11 +1955,11 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, ret = pack_resource_labels(ctx, &mh, &mp_pck, data, bytes); if (ret != 0) { if (ctx->resource_type == RESOURCE_TYPE_K8S) { - ret = extract_local_resource_id(data, bytes, ctx, tag); + ret = extract_local_resource_id(data, bytes, &fmt_ctx, tag); if (ret != 0) { flb_plg_error(ctx->ins, "fail to construct local_resource_id"); - msgpack_sbuffer_destroy(&mp_sbuf); - return NULL; + out_buf = NULL; + goto cleanup; } } ret = parse_monitored_resource(ctx, data, bytes, &mp_pck); @@ -2095,12 +2075,12 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, * k8s_container... */ - ret = process_local_resource_id(ctx, tag, tag_len, K8S_CONTAINER); + ret = process_local_resource_id(&fmt_ctx, tag, tag_len, K8S_CONTAINER); if (ret == -1) { flb_plg_error(ctx->ins, "fail to extract resource labels " "for k8s_container resource type"); - msgpack_sbuffer_destroy(&mp_sbuf); - return NULL; + out_buf = NULL; + goto cleanup; } flb_mp_map_header_init(&mh, &mp_pck); @@ -2133,33 +2113,33 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, ctx->cluster_name, flb_sds_len(ctx->cluster_name)); } - if (ctx->namespace_name) { + if (fmt_ctx.namespace_name) { flb_mp_map_header_append(&mh); msgpack_pack_str(&mp_pck, 14); msgpack_pack_str_body(&mp_pck, "namespace_name", 14); - msgpack_pack_str(&mp_pck, flb_sds_len(ctx->namespace_name)); + msgpack_pack_str(&mp_pck, flb_sds_len(fmt_ctx.namespace_name)); msgpack_pack_str_body(&mp_pck, - ctx->namespace_name, - flb_sds_len(ctx->namespace_name)); + fmt_ctx.namespace_name, + flb_sds_len(fmt_ctx.namespace_name)); } - if (ctx->pod_name) { + if (fmt_ctx.pod_name) { flb_mp_map_header_append(&mh); msgpack_pack_str(&mp_pck, 8); msgpack_pack_str_body(&mp_pck, "pod_name", 8); - msgpack_pack_str(&mp_pck, flb_sds_len(ctx->pod_name)); + msgpack_pack_str(&mp_pck, flb_sds_len(fmt_ctx.pod_name)); msgpack_pack_str_body(&mp_pck, - ctx->pod_name, flb_sds_len(ctx->pod_name)); + fmt_ctx.pod_name, flb_sds_len(fmt_ctx.pod_name)); } - if (ctx->container_name) { + if (fmt_ctx.container_name) { flb_mp_map_header_append(&mh); msgpack_pack_str(&mp_pck, 14); msgpack_pack_str_body(&mp_pck, "container_name", 14); - msgpack_pack_str(&mp_pck, flb_sds_len(ctx->container_name)); + msgpack_pack_str(&mp_pck, flb_sds_len(fmt_ctx.container_name)); msgpack_pack_str_body(&mp_pck, - ctx->container_name, - flb_sds_len(ctx->container_name)); + fmt_ctx.container_name, + flb_sds_len(fmt_ctx.container_name)); } flb_mp_map_header_end(&mh); @@ -2171,12 +2151,12 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, * k8s_node. */ - ret = process_local_resource_id(ctx, tag, tag_len, K8S_NODE); + ret = process_local_resource_id(&fmt_ctx, tag, tag_len, K8S_NODE); if (ret == -1) { flb_plg_error(ctx->ins, "fail to process local_resource_id from " "log entry for k8s_node"); - msgpack_sbuffer_destroy(&mp_sbuf); - return NULL; + out_buf = NULL; + goto cleanup; } flb_mp_map_header_init(&mh, &mp_pck); @@ -2209,13 +2189,13 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, ctx->cluster_name, flb_sds_len(ctx->cluster_name)); } - if (ctx->node_name) { + if (fmt_ctx.node_name) { flb_mp_map_header_append(&mh); msgpack_pack_str(&mp_pck, 9); msgpack_pack_str_body(&mp_pck, "node_name", 9); - msgpack_pack_str(&mp_pck, flb_sds_len(ctx->node_name)); + msgpack_pack_str(&mp_pck, flb_sds_len(fmt_ctx.node_name)); msgpack_pack_str_body(&mp_pck, - ctx->node_name, flb_sds_len(ctx->node_name)); + fmt_ctx.node_name, flb_sds_len(fmt_ctx.node_name)); } flb_mp_map_header_end(&mh); @@ -2228,12 +2208,12 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, * k8s_pod.. */ - ret = process_local_resource_id(ctx, tag, tag_len, K8S_POD); + ret = process_local_resource_id(&fmt_ctx, tag, tag_len, K8S_POD); if (ret == -1) { flb_plg_error(ctx->ins, "fail to process local_resource_id from " "log entry for k8s_pod"); - msgpack_sbuffer_destroy(&mp_sbuf); - return NULL; + out_buf = NULL; + goto cleanup; } flb_mp_map_header_init(&mh, &mp_pck); @@ -2266,23 +2246,23 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, ctx->cluster_name, flb_sds_len(ctx->cluster_name)); } - if (ctx->namespace_name) { + if (fmt_ctx.namespace_name) { flb_mp_map_header_append(&mh); msgpack_pack_str(&mp_pck, 14); msgpack_pack_str_body(&mp_pck, "namespace_name", 14); - msgpack_pack_str(&mp_pck, flb_sds_len(ctx->namespace_name)); + msgpack_pack_str(&mp_pck, flb_sds_len(fmt_ctx.namespace_name)); msgpack_pack_str_body(&mp_pck, - ctx->namespace_name, - flb_sds_len(ctx->namespace_name)); + fmt_ctx.namespace_name, + flb_sds_len(fmt_ctx.namespace_name)); } - if (ctx->pod_name) { + if (fmt_ctx.pod_name) { flb_mp_map_header_append(&mh); msgpack_pack_str(&mp_pck, 8); msgpack_pack_str_body(&mp_pck, "pod_name", 8); - msgpack_pack_str(&mp_pck, flb_sds_len(ctx->pod_name)); + msgpack_pack_str(&mp_pck, flb_sds_len(fmt_ctx.pod_name)); msgpack_pack_str_body(&mp_pck, - ctx->pod_name, flb_sds_len(ctx->pod_name)); + fmt_ctx.pod_name, flb_sds_len(fmt_ctx.pod_name)); } flb_mp_map_header_end(&mh); @@ -2329,26 +2309,16 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, else { flb_plg_error(ctx->ins, "unsupported resource type '%s'", ctx->resource); - msgpack_sbuffer_destroy(&mp_sbuf); - return NULL; + out_buf = NULL; + goto cleanup; } } } msgpack_pack_str(&mp_pck, 7); msgpack_pack_str_body(&mp_pck, "entries", 7); - /* Append entries */ - msgpack_pack_array(&mp_pck, array_size); - - ret = flb_log_event_decoder_init(&log_decoder, (char *) data, bytes); - - if (ret != FLB_EVENT_DECODER_SUCCESS) { - flb_plg_error(ctx->ins, - "Log event decoder initialization error : %d", ret); - msgpack_sbuffer_destroy(&mp_sbuf); - - return NULL; - } + /* Append entries dynamic array */ + flb_mp_array_header_init(&entries_mh, &mp_pck); while ((ret = flb_log_event_decoder_next( &log_decoder, @@ -2454,8 +2424,8 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, operation_last = FLB_FALSE; operation_extra_size = 0; operation_extracted = extract_operation(&operation_id, &operation_producer, - &operation_first, &operation_last, obj, - &operation_extra_size); + &operation_first, &operation_last, + obj, &operation_extra_size); if (operation_extracted == FLB_TRUE) { entry_size += 1; @@ -2469,8 +2439,7 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, source_location_extracted = extract_source_location(&source_location_file, &source_location_line, &source_location_function, - obj, - &source_location_extra_size); + obj, &source_location_extra_size); if (source_location_extracted == FLB_TRUE) { entry_size += 1; @@ -2515,10 +2484,8 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, destroy_http_request(&http_request); - flb_log_event_decoder_destroy(&log_decoder); - msgpack_sbuffer_destroy(&mp_sbuf); - - return NULL; + out_buf = NULL; + goto cleanup; } /* Number of parsed labels */ @@ -2694,18 +2661,52 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, msgpack_pack_str(&mp_pck, s); msgpack_pack_str_body(&mp_pck, time_formatted, s); + + flb_mp_array_header_append(&entries_mh); + records_count++; } - flb_log_event_decoder_destroy(&log_decoder); + flb_mp_array_header_end(&entries_mh); + + if (records_count == 0) { + out_buf = NULL; + goto cleanup; + } + + if (formatted_records != NULL) { + *formatted_records = records_count; + } /* Convert from msgpack to JSON */ out_buf = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size, config->json_escape_unicode); - msgpack_sbuffer_destroy(&mp_sbuf); - if (!out_buf) { flb_plg_error(ctx->ins, "error formatting JSON payload"); - return NULL; + } + +cleanup: + if (sbuf_initialized) { + msgpack_sbuffer_destroy(&mp_sbuf); + } + if (decoder_initialized) { + flb_log_event_decoder_destroy(&log_decoder); + } + + /* Destroy local dynamic resource strings */ + if (fmt_ctx.local_resource_id) { + flb_sds_destroy(fmt_ctx.local_resource_id); + } + if (fmt_ctx.namespace_name) { + flb_sds_destroy(fmt_ctx.namespace_name); + } + if (fmt_ctx.pod_name) { + flb_sds_destroy(fmt_ctx.pod_name); + } + if (fmt_ctx.container_name) { + flb_sds_destroy(fmt_ctx.container_name); + } + if (fmt_ctx.node_name) { + flb_sds_destroy(fmt_ctx.node_name); } return out_buf; @@ -3375,6 +3376,8 @@ static struct flb_config_map config_map[] = { 0, FLB_TRUE, offsetof(struct flb_stackdriver, cloud_logging_base_url), "The base Cloud Logging API URL to use for the /v2/entries:write API request. Default: https://logging.googleapis.com" }, + + /* EOF */ {0} }; diff --git a/plugins/out_stackdriver/stackdriver.h b/plugins/out_stackdriver/stackdriver.h index f54faa9d339..1d7a295e4dc 100644 --- a/plugins/out_stackdriver/stackdriver.h +++ b/plugins/out_stackdriver/stackdriver.h @@ -136,12 +136,7 @@ struct flb_stackdriver { /* kubernetes specific */ flb_sds_t cluster_name; flb_sds_t cluster_location; - flb_sds_t namespace_name; - flb_sds_t pod_name; - flb_sds_t container_name; - flb_sds_t node_name; - flb_sds_t local_resource_id; flb_sds_t tag_prefix; /* shadow tag_prefix for safe deallocation */ flb_sds_t tag_prefix_k8s; diff --git a/plugins/out_stackdriver/stackdriver_conf.c b/plugins/out_stackdriver/stackdriver_conf.c index 3bf55865ff1..d4413898950 100644 --- a/plugins/out_stackdriver/stackdriver_conf.c +++ b/plugins/out_stackdriver/stackdriver_conf.c @@ -707,13 +707,7 @@ int flb_stackdriver_conf_destroy(struct flb_stackdriver *ctx) flb_sds_destroy(ctx->metadata_server); } - if (ctx->resource_type == RESOURCE_TYPE_K8S){ - flb_sds_destroy(ctx->namespace_name); - flb_sds_destroy(ctx->pod_name); - flb_sds_destroy(ctx->container_name); - flb_sds_destroy(ctx->node_name); - flb_sds_destroy(ctx->local_resource_id); - } + if (ctx->metadata_server_auth) { flb_sds_destroy(ctx->zone); @@ -750,6 +744,7 @@ int flb_stackdriver_conf_destroy(struct flb_stackdriver *ctx) flb_kv_release(&ctx->config_labels); flb_kv_release(&ctx->resource_labels_kvs); + if (ctx->token_mutex_initialized) { pthread_mutex_destroy(&ctx->token_mutex); } From eb0694a8bf079f4309bef008d889f2cb638e5696 Mon Sep 17 00:00:00 2001 From: Yu Yi Date: Tue, 30 Jun 2026 19:33:08 -0400 Subject: [PATCH 2/4] tests: cover out_stackdriver multi-worker format-context safety Add resource_k8s_container_concurrency: two output workers across five input streams formatting k8s_container records concurrently, exercising the thread-local format context and guarding against the resource-string data race. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Yu Yi --- tests/runtime/out_stackdriver.c | 56 +++++++++++++++++++++++++++++++++ 1 file changed, 56 insertions(+) diff --git a/tests/runtime/out_stackdriver.c b/tests/runtime/out_stackdriver.c index 2bd79f31aaf..8e3de97cec7 100644 --- a/tests/runtime/out_stackdriver.c +++ b/tests/runtime/out_stackdriver.c @@ -621,6 +621,8 @@ static void cb_check_k8s_container_resource(void *ctx, int ffd, flb_sds_destroy(res_data); } + + static void cb_check_k8s_container_resource_diff_tag(void *ctx, int ffd, int res_ret, void *res_data, size_t res_size, void *data) @@ -3606,6 +3608,8 @@ void flb_test_resource_k8s_container_common() flb_destroy(ctx); } + + void flb_test_resource_k8s_container_multi_tag_value() { int ret; @@ -3660,6 +3664,56 @@ void flb_test_resource_k8s_container_multi_tag_value() flb_destroy(ctx); } +void flb_test_resource_k8s_container_concurrency() +{ + int ret; + int i; + int k; + flb_ctx_t *ctx; + int in_ffd[5]; + int out_ffd; + char payload[512]; + char tag[32]; + + ctx = flb_create(); + flb_service_set(ctx, "flush", "1", "grace", "1", NULL); + + for (k = 0; k < 5; k++) { + in_ffd[k] = flb_input(ctx, (char *) "lib", NULL); + snprintf(tag, sizeof(tag), "test.%d", k); + flb_input_set(ctx, in_ffd[k], "tag", tag, NULL); + } + + out_ffd = flb_output(ctx, (char *) "stackdriver", NULL); + flb_output_set(ctx, out_ffd, + "match", "test.*", + "resource", "k8s_container", + "google_service_credentials", SERVICE_CREDENTIALS, + "k8s_cluster_name", "test_cluster_name", + "k8s_cluster_location", "test_cluster_location", + "workers", "2", + NULL); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + + for (i = 0; i < 500; i++) { + for (k = 0; k < 5; k++) { + snprintf(payload, sizeof(payload), + "[1591649196, {" + "\"message\": \"concurrency_test_inp%d_rec%d\"," + "\"logging.googleapis.com/local_resource_id\": \"k8s_container.ns_%d.pod_%d.ctr_%d\"" + "}]", k, i, i + (k * 1000), i + (k * 1000), i + (k * 1000)); + flb_lib_push(ctx, in_ffd[k], payload, strlen(payload)); + } + } + + sleep(4); + flb_stop(ctx); + flb_destroy(ctx); +} + + void flb_test_resource_k8s_container_custom_tag_prefix() { int ret; @@ -6601,8 +6655,10 @@ TEST_LIST = { /* test k8s */ {"resource_k8s_container_common", flb_test_resource_k8s_container_common }, + {"resource_k8s_container_no_local_resource_id", flb_test_resource_k8s_container_no_local_resource_id }, {"resource_k8s_container_multi_tag_value", flb_test_resource_k8s_container_multi_tag_value } , + {"resource_k8s_container_concurrency", flb_test_resource_k8s_container_concurrency }, {"resource_k8s_container_custom_tag_prefix", flb_test_resource_k8s_container_custom_tag_prefix }, {"resource_k8s_container_custom_tag_prefix_with_dot", flb_test_resource_k8s_container_custom_tag_prefix_with_dot }, {"resource_k8s_container_default_tag_regex", flb_test_resource_k8s_container_default_tag_regex }, From 48091b81ba8249f0a94db9c3c6706eba2cf4be88 Mon Sep 17 00:00:00 2001 From: Yu Yi Date: Wed, 1 Jul 2026 18:51:01 -0400 Subject: [PATCH 3/4] out_stackdriver: add trust_payload_local_resource_id option The logging.googleapis.com/local_resource_id field is read from the log payload to derive the k8s_container/k8s_pod/k8s_node monitored resource labels. In multi-tenant pipelines the payload is workload-controlled, so a compromised or malicious workload can forge this field and attribute its log entries to another namespace/pod/container across trust boundaries. Add a trust_payload_local_resource_id option (default: true, preserving existing behavior). When set to false: - the payload local_resource_id is ignored; resource labels are only derived from the tag (via tag_prefix + regex), configured resource_labels, or a payload monitored resource map. - when none of those trusted sources is available, entries fall back to the cluster scoped k8s_cluster resource instead of packing workload labels forged from the payload. This mirrors the hardening previously applied to the Go based logging agent in GoogleCloudPlatform/k8s-stackdriver#1186. Signed-off-by: Yu Yi --- plugins/out_stackdriver/stackdriver.c | 116 +++++++++++++++++++++++--- plugins/out_stackdriver/stackdriver.h | 3 + 2 files changed, 108 insertions(+), 11 deletions(-) diff --git a/plugins/out_stackdriver/stackdriver.c b/plugins/out_stackdriver/stackdriver.c index 8f4ef69b08e..65a16ac6392 100644 --- a/plugins/out_stackdriver/stackdriver.c +++ b/plugins/out_stackdriver/stackdriver.c @@ -671,6 +671,61 @@ static int parse_monitored_resource(struct flb_stackdriver *ctx, const void *dat return ret; } +/* + * Check if the payload carries a monitored resource map that + * parse_monitored_resource() would consume, without packing anything + */ +static int payload_has_monitored_resource(const void *data, size_t bytes) +{ + int found = FLB_FALSE; + int ret; + msgpack_object *obj; + msgpack_object_kv *kv; + msgpack_object_kv *kvend; + msgpack_object_kv *p; + msgpack_object_kv *pend; + struct flb_log_event_decoder log_decoder; + struct flb_log_event log_event; + + ret = flb_log_event_decoder_init(&log_decoder, (char *) data, bytes); + if (ret != FLB_EVENT_DECODER_SUCCESS) { + return FLB_FALSE; + } + + while (found == FLB_FALSE && + flb_log_event_decoder_next(&log_decoder, + &log_event) == FLB_EVENT_DECODER_SUCCESS) { + obj = log_event.body; + kv = obj->via.map.ptr; + kvend = obj->via.map.ptr + obj->via.map.size; + for (; kv < kvend; ++kv) { + if (kv->val.type != MSGPACK_OBJECT_MAP || + kv->key.type != MSGPACK_OBJECT_STR || + strncmp(MONITORED_RESOURCE_KEY, + kv->key.via.str.ptr, kv->key.via.str.size) != 0) { + continue; + } + + p = kv->val.via.map.ptr; + pend = kv->val.via.map.ptr + kv->val.via.map.size; + for (; p < pend; ++p) { + if (p->key.type == MSGPACK_OBJECT_STR && + p->val.type == MSGPACK_OBJECT_MAP && + p->val.via.map.size > 0 && + strncmp("labels", p->key.via.str.ptr, + p->key.via.str.size) == 0) { + found = FLB_TRUE; + break; + } + } + } + } + + flb_log_event_decoder_destroy(&log_decoder); + + return found; +} + /* * Given a local_resource_id, split the content using the proper separator generating * a linked list to store the spliited string @@ -748,8 +803,13 @@ static int extract_local_resource_id(const void *data, size_t bytes, &log_decoder, &log_event)) == FLB_EVENT_DECODER_SUCCESS) { map = log_event.body->via.map; - local_resource_id = get_str_value_from_msgpack_map(map, LOCAL_RESOURCE_ID_KEY, - LEN_LOCAL_RESOURCE_ID_KEY); + if (ctx->trust_payload_local_resource_id == FLB_TRUE) { + local_resource_id = get_str_value_from_msgpack_map(map, LOCAL_RESOURCE_ID_KEY, + LEN_LOCAL_RESOURCE_ID_KEY); + } + else { + local_resource_id = NULL; + } if (local_resource_id == NULL) { /* if local_resource_id is not found, use the tag of the log */ @@ -1081,7 +1141,7 @@ static int process_local_resource_id(struct stackdriver_format_ctx *fmt_ctx, if (is_tag_match_regex(ctx, tag, tag_len) > 0) { ret = extract_resource_labels_from_regex(fmt_ctx, tag, tag_len, FLB_TRUE); } - else { + else if (ctx->trust_payload_local_resource_id == FLB_TRUE) { if (is_local_resource_id_match_regex(fmt_ctx) > 0) { ret = extract_resource_labels_from_regex(fmt_ctx, tag, tag_len, FLB_FALSE); } @@ -1089,6 +1149,14 @@ static int process_local_resource_id(struct stackdriver_format_ctx *fmt_ctx, ret = set_monitored_resource_labels(fmt_ctx, type); } } + else { + /* + * The payload cannot be trusted and the tag carries no resource + * labels; stackdriver_format() already fell back to the cluster + * scoped resource before reaching this path. + */ + ret = -1; + } return ret; } @@ -1871,6 +1939,7 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, char *new_trace; char *newtag; char *new_log_name; + const char *resource_type; char *operation_id; char *operation_producer; char *source_location_file; @@ -1942,12 +2011,31 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, /* type & labels */ msgpack_pack_map(&mp_pck, 2); + /* + * When the payload local_resource_id is untrusted and no trusted source + * of resource labels is available (tag regex, configured resource labels + * or a payload monitored resource), attribute the entries to the cluster + * scoped resource instead of forgeable namespace/pod/container labels. + */ + resource_type = ctx->resource; + if (ctx->trust_payload_local_resource_id == FLB_FALSE && + ctx->resource_type == RESOURCE_TYPE_K8S && + strcmp(ctx->resource, K8S_CLUSTER) != 0 && + (ctx->should_skip_resource_labels_api == FLB_TRUE || + mk_list_size(&ctx->resource_labels_kvs) == 0) && + payload_has_monitored_resource(data, bytes) == FLB_FALSE && + is_tag_match_regex(ctx, tag, tag_len) <= 0) { + resource_type = K8S_CLUSTER; + flb_plg_debug(ctx->ins, "untrusted payload local_resource_id and tag " + "without resource labels; falling back to k8s_cluster " + "resource for tag %s", tag); + } + /* type */ msgpack_pack_str(&mp_pck, 4); msgpack_pack_str_body(&mp_pck, "type", 4); - msgpack_pack_str(&mp_pck, flb_sds_len(ctx->resource)); - msgpack_pack_str_body(&mp_pck, ctx->resource, - flb_sds_len(ctx->resource)); + msgpack_pack_str(&mp_pck, strlen(resource_type)); + msgpack_pack_str_body(&mp_pck, resource_type, strlen(resource_type)); msgpack_pack_str(&mp_pck, 6); msgpack_pack_str_body(&mp_pck, "labels", 6); @@ -2067,7 +2155,7 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, } flb_mp_map_header_end(&mh); } - else if (strcmp(ctx->resource, K8S_CONTAINER) == 0) { + else if (strcmp(resource_type, K8S_CONTAINER) == 0) { /* k8s_container resource has fields project_id, location, cluster_name, * namespace_name, pod_name, container_name * @@ -2144,7 +2232,7 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, flb_mp_map_header_end(&mh); } - else if (strcmp(ctx->resource, K8S_NODE) == 0) { + else if (strcmp(resource_type, K8S_NODE) == 0) { /* k8s_node resource has fields project_id, location, cluster_name, node_name * * The local_resource_id for k8s_node is in format: @@ -2200,7 +2288,7 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, flb_mp_map_header_end(&mh); } - else if (strcmp(ctx->resource, K8S_POD) == 0) { + else if (strcmp(resource_type, K8S_POD) == 0) { /* k8s_pod resource has fields project_id, location, cluster_name, * namespace_name, pod_name. * @@ -2267,7 +2355,7 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx, flb_mp_map_header_end(&mh); } - else if (strcmp(ctx->resource, K8S_CLUSTER) == 0) { + else if (strcmp(resource_type, K8S_CLUSTER) == 0) { /* k8s_cluster resource has fields project_id, location, cluster_name * * There is no local_resource_id for k8s_cluster as we get all info @@ -3376,7 +3464,13 @@ static struct flb_config_map config_map[] = { 0, FLB_TRUE, offsetof(struct flb_stackdriver, cloud_logging_base_url), "The base Cloud Logging API URL to use for the /v2/entries:write API request. Default: https://logging.googleapis.com" }, - + { + FLB_CONFIG_MAP_BOOL, "trust_payload_local_resource_id", "true", + 0, FLB_TRUE, offsetof(struct flb_stackdriver, trust_payload_local_resource_id), + "Trust the logging.googleapis.com/local_resource_id field supplied in the " + "log payload. When disabled and the tag carries no resource labels, " + "entries are attributed to the k8s_cluster resource instead" + }, /* EOF */ {0} diff --git a/plugins/out_stackdriver/stackdriver.h b/plugins/out_stackdriver/stackdriver.h index 1d7a295e4dc..8bbdf68df2b 100644 --- a/plugins/out_stackdriver/stackdriver.h +++ b/plugins/out_stackdriver/stackdriver.h @@ -217,6 +217,9 @@ struct flb_stackdriver { flb_sds_t cloud_logging_base_url; flb_sds_t cloud_logging_write_url; + /* trust the local_resource_id supplied in the log payload */ + int trust_payload_local_resource_id; + #ifdef FLB_HAVE_METRICS /* metrics */ struct cmt_counter *cmt_successful_requests; From 59255d273cf8602a415e9c396070e932af0faf87 Mon Sep 17 00:00:00 2001 From: Yu Yi Date: Wed, 1 Jul 2026 18:51:01 -0400 Subject: [PATCH 4/4] tests: cover out_stackdriver untrusted payload local_resource_id Cover the trust_payload_local_resource_id=false behavior: - resource_k8s_container_untrusted: a payload supplied local_resource_id with a non-matching tag falls back to the cluster scoped k8s_cluster resource and packs no workload labels. - resource_k8s_container_untrusted_tag_wins: a tag matching the default regex still resolves the full k8s_container resource; the untrusted payload value is ignored. Signed-off-by: Yu Yi --- tests/runtime/out_stackdriver.c | 137 +++++++++++++++++++++++++++++++- 1 file changed, 136 insertions(+), 1 deletion(-) diff --git a/tests/runtime/out_stackdriver.c b/tests/runtime/out_stackdriver.c index 8e3de97cec7..b1a13af1bab 100644 --- a/tests/runtime/out_stackdriver.c +++ b/tests/runtime/out_stackdriver.c @@ -621,7 +621,52 @@ static void cb_check_k8s_container_resource(void *ctx, int ffd, flb_sds_destroy(res_data); } +static void cb_check_k8s_container_untrusted(void *ctx, int ffd, + int res_ret, void *res_data, size_t res_size, + void *data) +{ + int ret; + + /* payload local_resource_id is untrusted and the tag does not match the + * regex, so entries fall back to the cluster scoped resource */ + ret = mp_kv_cmp(res_data, res_size, "$resource['type']", "k8s_cluster"); + TEST_CHECK(ret == FLB_TRUE); + + /* project id */ + ret = mp_kv_cmp(res_data, res_size, + "$resource['labels']['project_id']", "fluent-bit"); + TEST_CHECK(ret == FLB_TRUE); + + /* location */ + ret = mp_kv_cmp(res_data, res_size, + "$resource['labels']['location']", "test_cluster_location"); + TEST_CHECK(ret == FLB_TRUE); + + /* cluster name */ + ret = mp_kv_cmp(res_data, res_size, + "$resource['labels']['cluster_name']", "test_cluster_name"); + TEST_CHECK(ret == FLB_TRUE); + /* no workload scoped labels may be derived from the untrusted payload */ + ret = mp_kv_exists(res_data, res_size, + "$resource['labels']['namespace_name']"); + TEST_CHECK(ret == FLB_FALSE); + + ret = mp_kv_exists(res_data, res_size, + "$resource['labels']['pod_name']"); + TEST_CHECK(ret == FLB_FALSE); + + ret = mp_kv_exists(res_data, res_size, + "$resource['labels']['container_name']"); + TEST_CHECK(ret == FLB_FALSE); + + /* check `local_resource_id` has been removed from jsonPayload */ + ret = mp_kv_exists(res_data, res_size, + "$entries[0]['jsonPayload']['logging.googleapis.com/local_resource_id']"); + TEST_CHECK(ret == FLB_FALSE); + + flb_sds_destroy(res_data); +} static void cb_check_k8s_container_resource_diff_tag(void *ctx, int ffd, int res_ret, void *res_data, size_t res_size, @@ -3608,7 +3653,96 @@ void flb_test_resource_k8s_container_common() flb_destroy(ctx); } +void flb_test_resource_k8s_container_untrusted() +{ + int ret; + int size = sizeof(K8S_CONTAINER_COMMON) - 1; + flb_ctx_t *ctx; + int in_ffd; + int out_ffd; + + /* Create context, flush every second (some checks omitted here) */ + ctx = flb_create(); + flb_service_set(ctx, "flush", "1", "grace", "1", NULL); + + /* Lib input mode */ + in_ffd = flb_input(ctx, (char *) "lib", NULL); + flb_input_set(ctx, in_ffd, "tag", "test", NULL); + + /* Stackdriver output */ + out_ffd = flb_output(ctx, (char *) "stackdriver", NULL); + flb_output_set(ctx, out_ffd, + "match", "test", + "resource", "k8s_container", + "google_service_credentials", SERVICE_CREDENTIALS, + "k8s_cluster_name", "test_cluster_name", + "k8s_cluster_location", "test_cluster_location", + "trust_payload_local_resource_id", "false", + NULL); + + /* Enable test mode */ + ret = flb_output_set_test(ctx, out_ffd, "formatter", + cb_check_k8s_container_untrusted, + NULL, NULL); + + /* Start */ + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + + /* Ingest data sample carrying a forged local_resource_id */ + flb_lib_push(ctx, in_ffd, (char *) K8S_CONTAINER_COMMON, size); + + sleep(2); + flb_stop(ctx); + flb_destroy(ctx); +} + +void flb_test_resource_k8s_container_untrusted_tag_wins() +{ + int ret; + int size = sizeof(K8S_CONTAINER_COMMON_DIFF_TAGS) - 1; + flb_ctx_t *ctx; + int in_ffd; + int out_ffd; + /* Create context, flush every second (some checks omitted here) */ + ctx = flb_create(); + flb_service_set(ctx, "flush", "1", "grace", "1", NULL); + + /* Lib input mode */ + in_ffd = flb_input(ctx, (char *) "lib", NULL); + flb_input_set(ctx, in_ffd, "tag", + "kube.var.log.containers.apache-logs-annotated_default_apache-aeeccc7a9f00f6e4e066aeff0434cf80621215071f1b20a51e8340aa7c35eac6.log", NULL); + + /* Stackdriver output */ + out_ffd = flb_output(ctx, (char *) "stackdriver", NULL); + flb_output_set(ctx, out_ffd, + "match", "kube.var.log.containers.*", + "resource", "k8s_container", + "google_service_credentials", SERVICE_CREDENTIALS, + "k8s_cluster_name", "test_cluster_name", + "k8s_cluster_location", "test_cluster_location", + "tag_prefix", "kube.var.log.containers.", + "trust_payload_local_resource_id", "false", + NULL); + + /* Enable test mode */ + ret = flb_output_set_test(ctx, out_ffd, "formatter", + cb_check_k8s_container_resource_default_regex, + NULL, NULL); + + /* Start */ + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + + /* Ingest data sample with a different local_resource_id in the payload; + * the trusted tag must win over the untrusted payload value */ + flb_lib_push(ctx, in_ffd, (char *) K8S_CONTAINER_COMMON_DIFF_TAGS, size); + + sleep(2); + flb_stop(ctx); + flb_destroy(ctx); +} void flb_test_resource_k8s_container_multi_tag_value() { @@ -6655,7 +6789,8 @@ TEST_LIST = { /* test k8s */ {"resource_k8s_container_common", flb_test_resource_k8s_container_common }, - + {"resource_k8s_container_untrusted", flb_test_resource_k8s_container_untrusted }, + {"resource_k8s_container_untrusted_tag_wins", flb_test_resource_k8s_container_untrusted_tag_wins }, {"resource_k8s_container_no_local_resource_id", flb_test_resource_k8s_container_no_local_resource_id }, {"resource_k8s_container_multi_tag_value", flb_test_resource_k8s_container_multi_tag_value } , {"resource_k8s_container_concurrency", flb_test_resource_k8s_container_concurrency },