From be561a35003685c8fc1c900a773d6495e72c8fdd Mon Sep 17 00:00:00 2001 From: Takahiro Yamashita Date: Sun, 18 Feb 2024 14:12:36 +0900 Subject: [PATCH] out_opentelemetry: use record accessor to look up value Signed-off-by: Takahiro Yamashita --- plugins/out_opentelemetry/opentelemetry.c | 165 +++++++++--------- plugins/out_opentelemetry/opentelemetry.h | 16 ++ .../out_opentelemetry/opentelemetry_conf.c | 65 +++++++ 3 files changed, 162 insertions(+), 84 deletions(-) diff --git a/plugins/out_opentelemetry/opentelemetry.c b/plugins/out_opentelemetry/opentelemetry.c index 77ad5bfb594..73c86a339ac 100644 --- a/plugins/out_opentelemetry/opentelemetry.c +++ b/plugins/out_opentelemetry/opentelemetry.c @@ -24,6 +24,7 @@ #include #include #include +#include #include #include @@ -943,31 +944,6 @@ static int flush_to_otel(struct opentelemetry_context *ctx, return ret; } -static msgpack_object *get_msgpack_object_from_map(msgpack_object *obj, flb_sds_t key) -{ - msgpack_object *ret = NULL; - msgpack_object key_obj; - int i; - - if (obj == NULL || obj->type != MSGPACK_OBJECT_MAP|| flb_sds_len(key) == 0) { - return NULL; - } - for (i=0; i< obj->via.map.size; i++) { - key_obj = obj->via.map.ptr[i].key; - if (key_obj.type != MSGPACK_OBJECT_STR) { - continue; - } - if (flb_sds_len(key) != key_obj.via.str.size) { - continue; - } - if (memcmp(key_obj.via.str.ptr, key, flb_sds_len(key)) == 0) { - ret = &obj->via.map.ptr[i].val; - break; - } - } - return ret; -} - /* https://opentelemetry.io/docs/specs/otel/logs/data-model/#field-severitynumber */ static int is_valid_severity_text(const char *str, size_t str_len) { @@ -1000,85 +976,105 @@ static int append_v1_logs_metadata(struct opentelemetry_context *ctx, struct flb_log_event *event, Opentelemetry__Proto__Logs__V1__LogRecord *log_record) { - msgpack_object *obj = NULL; + struct flb_ra_value *ra_val; if (ctx == NULL || event == NULL || log_record == NULL) { return -1; } /* ObservedTimestamp */ - obj = get_msgpack_object_from_map(event->metadata, - ctx->logs_observed_timestamp_metadata_key); - if (obj != NULL && obj->type == MSGPACK_OBJECT_POSITIVE_INTEGER) { - log_record->observed_time_unix_nano = obj->via.u64; + if (ctx->ra_observed_timestamp_metadata) { + ra_val = flb_ra_get_value_object(ctx->ra_observed_timestamp_metadata, *event->metadata); + if (ra_val != NULL && ra_val->o.type == MSGPACK_OBJECT_POSITIVE_INTEGER) { + log_record->observed_time_unix_nano = ra_val->o.via.u64; + flb_ra_key_value_destroy(ra_val); + } } /* Timestamp */ - obj = get_msgpack_object_from_map(event->metadata, - ctx->logs_timestamp_metadata_key); - if (obj != NULL && obj->type == MSGPACK_OBJECT_POSITIVE_INTEGER) { - log_record->time_unix_nano = obj->via.u64; - } - else { - log_record->time_unix_nano = flb_time_to_nanosec(&event->timestamp); + if (ctx->ra_timestamp_metadata) { + ra_val = flb_ra_get_value_object(ctx->ra_timestamp_metadata, *event->metadata); + if (ra_val != NULL && ra_val->o.type == MSGPACK_OBJECT_POSITIVE_INTEGER) { + log_record->time_unix_nano = ra_val->o.via.u64; + flb_ra_key_value_destroy(ra_val); + } + else { + log_record->time_unix_nano = flb_time_to_nanosec(&event->timestamp); + } } /* SeverityText */ - obj = get_msgpack_object_from_map(event->metadata, - ctx->logs_severity_text_metadata_key); - if (obj != NULL && obj->type == MSGPACK_OBJECT_STR && - is_valid_severity_text(obj->via.str.ptr, obj->via.str.size) == FLB_TRUE) { - log_record->severity_text = flb_calloc(1, obj->via.str.size+1); - if (log_record->severity_text) { - strncpy(log_record->severity_text, obj->via.str.ptr, obj->via.str.size); + if (ctx->ra_severity_text_metadata) { + ra_val = flb_ra_get_value_object(ctx->ra_severity_text_metadata, *event->metadata); + if (ra_val != NULL && ra_val->o.type == MSGPACK_OBJECT_STR && + is_valid_severity_text(ra_val->o.via.str.ptr, ra_val->o.via.str.size) == FLB_TRUE) { + log_record->severity_text = flb_calloc(1, ra_val->o.via.str.size+1); + if (log_record->severity_text) { + strncpy(log_record->severity_text, ra_val->o.via.str.ptr, ra_val->o.via.str.size); + } + flb_ra_key_value_destroy(ra_val); + } + else { + /* To prevent invalid free */ + log_record->severity_text = NULL; } - } - else { - /* To prevent invalid free */ - log_record->severity_text = NULL; } /* SeverityNumber */ - obj = get_msgpack_object_from_map(event->metadata, - ctx->logs_severity_number_metadata_key); - if (obj != NULL && obj->type == MSGPACK_OBJECT_POSITIVE_INTEGER && - is_valid_severity_number(obj->via.u64) == FLB_TRUE) { - log_record->severity_number = obj->via.u64; + if (ctx->ra_severity_number_metadata) { + ra_val = flb_ra_get_value_object(ctx->ra_severity_number_metadata, *event->metadata); + if (ra_val != NULL && ra_val->o.type == MSGPACK_OBJECT_POSITIVE_INTEGER && + is_valid_severity_number(ra_val->o.via.u64) == FLB_TRUE) { + log_record->severity_number = ra_val->o.via.u64; + flb_ra_key_value_destroy(ra_val); + } } /* TraceFlags */ - obj = get_msgpack_object_from_map(event->metadata, - ctx->logs_trace_flags_metadata_key); - if (obj != NULL && obj->type == MSGPACK_OBJECT_POSITIVE_INTEGER) { - log_record->flags = (uint32_t)obj->via.u64; + if (ctx->ra_trace_flags_metadata) { + ra_val = flb_ra_get_value_object(ctx->ra_trace_flags_metadata, *event->metadata); + if (ra_val != NULL && ra_val->o.type == MSGPACK_OBJECT_POSITIVE_INTEGER) { + log_record->flags = (uint32_t)ra_val->o.via.u64; + flb_ra_key_value_destroy(ra_val); + } } /* SpanId */ - obj = get_msgpack_object_from_map(event->metadata, - ctx->logs_span_id_metadata_key); - if (obj != NULL && obj->type == MSGPACK_OBJECT_BIN) { - log_record->span_id.data = flb_calloc(1, obj->via.bin.size); - if (log_record->span_id.data) { - memcpy(log_record->span_id.data, obj->via.bin.ptr, obj->via.bin.size); - log_record->span_id.len = obj->via.bin.size; + if (ctx->ra_span_id_metadata) { + ra_val = flb_ra_get_value_object(ctx->ra_span_id_metadata, *event->metadata); + if (ra_val != NULL && ra_val->o.type == MSGPACK_OBJECT_BIN) { + log_record->span_id.data = flb_calloc(1, ra_val->o.via.bin.size); + if (log_record->span_id.data) { + memcpy(log_record->span_id.data, ra_val->o.via.bin.ptr, ra_val->o.via.bin.size); + log_record->span_id.len = ra_val->o.via.bin.size; + } + flb_ra_key_value_destroy(ra_val); } } /* TraceId */ - obj = get_msgpack_object_from_map(event->metadata, - ctx->logs_trace_id_metadata_key); - if (obj != NULL && obj->type == MSGPACK_OBJECT_BIN) { - log_record->trace_id.data = flb_calloc(1, obj->via.bin.size); - if (log_record->trace_id.data) { - memcpy(log_record->trace_id.data, obj->via.bin.ptr, obj->via.bin.size); - log_record->trace_id.len = obj->via.bin.size; + if (ctx->ra_trace_id_metadata) { + ra_val = flb_ra_get_value_object(ctx->ra_trace_id_metadata, *event->metadata); + if (ra_val != NULL && ra_val->o.type == MSGPACK_OBJECT_BIN) { + log_record->trace_id.data = flb_calloc(1, ra_val->o.via.bin.size); + if (log_record->trace_id.data) { + memcpy(log_record->trace_id.data, ra_val->o.via.bin.ptr, ra_val->o.via.bin.size); + log_record->trace_id.len = ra_val->o.via.bin.size; + } + flb_ra_key_value_destroy(ra_val); } } /* Attributes */ - obj = get_msgpack_object_from_map(event->metadata, - ctx->logs_attributes_metadata_key); - if (obj != NULL && obj->type == MSGPACK_OBJECT_MAP) { - log_record->attributes = msgpack_map_to_otlp_kvarray(obj, &log_record->n_attributes); + if (ctx->ra_attributes_metadata) { + ra_val = flb_ra_get_value_object(ctx->ra_attributes_metadata, *event->metadata); + if (ra_val != NULL && ra_val->o.type == MSGPACK_OBJECT_MAP) { + if (log_record->attributes != NULL) { + otlp_kvarray_destroy(log_record->attributes, + log_record->n_attributes); + } + log_record->attributes = msgpack_map_to_otlp_kvarray(&ra_val->o, &log_record->n_attributes); + flb_ra_key_value_destroy(ra_val); + } } return 0; @@ -1136,7 +1132,6 @@ static int process_logs(struct flb_event_chunk *event_chunk, while (flb_log_event_decoder_next(decoder, &event) == FLB_EVENT_DECODER_SUCCESS) { ra_match = NULL; opentelemetry__proto__logs__v1__log_record__init(&log_records[log_record_count]); - append_v1_logs_metadata(ctx, &event, &log_records[log_record_count]); /* * Set the record body by using the logic defined in the configuration by @@ -1161,6 +1156,8 @@ static int process_logs(struct flb_event_chunk *event_chunk, break; } + append_v1_logs_metadata(ctx, &event, &log_records[log_record_count]); + ret = FLB_OK; log_records[log_record_count].time_unix_nano = flb_time_to_nanosec(&event.timestamp); @@ -1498,42 +1495,42 @@ static struct flb_config_map config_map[] = { "Specify if the response paylod should be logged or not" }, { - FLB_CONFIG_MAP_STR, "logs_observed_timestamp_metadata_key", "ObservedTimestamp", + FLB_CONFIG_MAP_STR, "logs_observed_timestamp_metadata_key", "$ObservedTimestamp", 0, FLB_TRUE, offsetof(struct opentelemetry_context, logs_observed_timestamp_metadata_key), "Specify an ObservedTimestamp key" }, { - FLB_CONFIG_MAP_STR, "logs_timestamp_metadata_key", "Timestamp", + FLB_CONFIG_MAP_STR, "logs_timestamp_metadata_key", "$Timestamp", 0, FLB_TRUE, offsetof(struct opentelemetry_context, logs_timestamp_metadata_key), "Specify an Timestamp key" }, { - FLB_CONFIG_MAP_STR, "logs_severity_text_metadata_key", "SeverityText", + FLB_CONFIG_MAP_STR, "logs_severity_text_metadata_key", "$SeverityText", 0, FLB_TRUE, offsetof(struct opentelemetry_context, logs_severity_text_metadata_key), "Specify an SeverityText key" }, { - FLB_CONFIG_MAP_STR, "logs_severity_number_metadata_key", "SeverityNumber", + FLB_CONFIG_MAP_STR, "logs_severity_number_metadata_key", "$SeverityNumber", 0, FLB_TRUE, offsetof(struct opentelemetry_context, logs_severity_number_metadata_key), "Specify an SeverityNumber key" }, { - FLB_CONFIG_MAP_STR, "logs_trace_flags_metadata_key", "TraceFlags", + FLB_CONFIG_MAP_STR, "logs_trace_flags_metadata_key", "$TraceFlags", 0, FLB_TRUE, offsetof(struct opentelemetry_context, logs_trace_flags_metadata_key), "Specify an TraceFlags key" }, { - FLB_CONFIG_MAP_STR, "logs_span_id_metadata_key", "SpanId", + FLB_CONFIG_MAP_STR, "logs_span_id_metadata_key", "$SpanId", 0, FLB_TRUE, offsetof(struct opentelemetry_context, logs_span_id_metadata_key), "Specify an SpanId key" }, { - FLB_CONFIG_MAP_STR, "logs_trace_id_metadata_key", "TraceId", + FLB_CONFIG_MAP_STR, "logs_trace_id_metadata_key", "$TraceId", 0, FLB_TRUE, offsetof(struct opentelemetry_context, logs_trace_id_metadata_key), "Specify an TraceId key" }, { - FLB_CONFIG_MAP_STR, "logs_attributes_metadata_key", "Attributes", + FLB_CONFIG_MAP_STR, "logs_attributes_metadata_key", "$Attributes", 0, FLB_TRUE, offsetof(struct opentelemetry_context, logs_attributes_metadata_key), "Specify an Attributes key" }, diff --git a/plugins/out_opentelemetry/opentelemetry.h b/plugins/out_opentelemetry/opentelemetry.h index b0a4d903b11..d3b62e0d819 100644 --- a/plugins/out_opentelemetry/opentelemetry.h +++ b/plugins/out_opentelemetry/opentelemetry.h @@ -61,13 +61,29 @@ struct opentelemetry_context { /* metadata keys */ flb_sds_t logs_observed_timestamp_metadata_key; + struct flb_record_accessor *ra_observed_timestamp_metadata; + flb_sds_t logs_timestamp_metadata_key; + struct flb_record_accessor *ra_timestamp_metadata; + flb_sds_t logs_severity_text_metadata_key; + struct flb_record_accessor *ra_severity_text_metadata; + flb_sds_t logs_severity_number_metadata_key; + struct flb_record_accessor *ra_severity_number_metadata; + flb_sds_t logs_trace_flags_metadata_key; + struct flb_record_accessor *ra_trace_flags_metadata; + flb_sds_t logs_span_id_metadata_key; + struct flb_record_accessor *ra_span_id_metadata; + flb_sds_t logs_trace_id_metadata_key; + struct flb_record_accessor *ra_trace_id_metadata; + flb_sds_t logs_attributes_metadata_key; + struct flb_record_accessor *ra_attributes_metadata; + flb_sds_t logs_instrumentation_scope_metadata_key; flb_sds_t logs_resource_metadata_key; diff --git a/plugins/out_opentelemetry/opentelemetry_conf.c b/plugins/out_opentelemetry/opentelemetry_conf.c index 83f14bbe153..13680178c28 100644 --- a/plugins/out_opentelemetry/opentelemetry_conf.c +++ b/plugins/out_opentelemetry/opentelemetry_conf.c @@ -380,6 +380,47 @@ struct opentelemetry_context *flb_opentelemetry_context_create(struct flb_output } } + ctx->ra_observed_timestamp_metadata = flb_ra_create((char*)ctx->logs_observed_timestamp_metadata_key, + FLB_FALSE); + if (ctx->ra_observed_timestamp_metadata == NULL) { + flb_plg_error(ins, "failed to create ra for observed timestamp"); + } + ctx->ra_timestamp_metadata = flb_ra_create((char*)ctx->logs_timestamp_metadata_key, + FLB_FALSE); + if (ctx->ra_timestamp_metadata == NULL) { + flb_plg_error(ins, "failed to create ra for timestamp"); + } + ctx->ra_severity_text_metadata = flb_ra_create((char*)ctx->logs_severity_text_metadata_key, + FLB_FALSE); + if (ctx->ra_severity_text_metadata == NULL) { + flb_plg_error(ins, "failed to create ra for severity text"); + } + ctx->ra_severity_number_metadata = flb_ra_create((char*)ctx->logs_severity_number_metadata_key, + FLB_FALSE); + if (ctx->ra_severity_number_metadata == NULL) { + flb_plg_error(ins, "failed to create ra for severity number"); + } + ctx->ra_trace_flags_metadata = flb_ra_create((char*)ctx->logs_trace_flags_metadata_key, + FLB_FALSE); + if (ctx->ra_trace_flags_metadata == NULL) { + flb_plg_error(ins, "failed to create ra for trace flags"); + } + ctx->ra_span_id_metadata = flb_ra_create((char*)ctx->logs_span_id_metadata_key, + FLB_FALSE); + if (ctx->ra_span_id_metadata == NULL) { + flb_plg_error(ins, "failed to create ra for span id"); + } + ctx->ra_trace_id_metadata = flb_ra_create((char*)ctx->logs_trace_id_metadata_key, + FLB_FALSE); + if (ctx->ra_trace_id_metadata == NULL) { + flb_plg_error(ins, "failed to create ra for trace id"); + } + ctx->ra_attributes_metadata = flb_ra_create((char*)ctx->logs_attributes_metadata_key, + FLB_FALSE); + if (ctx->ra_attributes_metadata == NULL) { + flb_plg_error(ins, "failed to create ra for attributes"); + } + return ctx; } @@ -401,6 +442,30 @@ void flb_opentelemetry_context_destroy(struct opentelemetry_context *ctx) if (ctx->mp_accessor) { flb_mp_accessor_destroy(ctx->mp_accessor); } + if (ctx->ra_observed_timestamp_metadata) { + flb_ra_destroy(ctx->ra_observed_timestamp_metadata); + } + if (ctx->ra_timestamp_metadata) { + flb_ra_destroy(ctx->ra_timestamp_metadata); + } + if (ctx->ra_severity_text_metadata) { + flb_ra_destroy(ctx->ra_severity_text_metadata); + } + if (ctx->ra_severity_number_metadata) { + flb_ra_destroy(ctx->ra_severity_number_metadata); + } + if (ctx->ra_trace_flags_metadata) { + flb_ra_destroy(ctx->ra_trace_flags_metadata); + } + if (ctx->ra_span_id_metadata) { + flb_ra_destroy(ctx->ra_span_id_metadata); + } + if (ctx->ra_trace_id_metadata) { + flb_ra_destroy(ctx->ra_trace_id_metadata); + } + if (ctx->ra_attributes_metadata) { + flb_ra_destroy(ctx->ra_attributes_metadata); + } flb_free(ctx->proxy_host); flb_free(ctx);