diff --git a/src/opentelemetry/flb_opentelemetry_otlp_json.c b/src/opentelemetry/flb_opentelemetry_otlp_json.c index 9e11f10fc6a..65ce09b484b 100644 --- a/src/opentelemetry/flb_opentelemetry_otlp_json.c +++ b/src/opentelemetry/flb_opentelemetry_otlp_json.c @@ -41,6 +41,8 @@ #include #include +#include + #include #include @@ -50,12 +52,14 @@ struct otlp_logs_scope_state { int64_t scope_id; + uint64_t scope_hash; struct flb_json_mut_val *scope_log; struct flb_json_mut_val *log_records; }; struct otlp_logs_resource_state { int64_t resource_id; + uint64_t resource_hash; struct flb_json_mut_val *resource_log; struct flb_json_mut_val *scope_logs; struct otlp_logs_scope_state *scopes; @@ -69,6 +73,99 @@ struct otlp_metrics_scope_state { static msgpack_object *msgpack_map_get_object(msgpack_object_map *map, const char *key); +static uint64_t msgpack_object_hash(msgpack_object *object) +{ + uint64_t hash; + msgpack_sbuffer buffer; + msgpack_packer packer; + + if (object == NULL) { + return cfl_hash_64bits("null", 4); + } + + msgpack_sbuffer_init(&buffer); + msgpack_packer_init(&packer, &buffer, msgpack_sbuffer_write); + + if (msgpack_pack_object(&packer, *object) != 0) { + msgpack_sbuffer_destroy(&buffer); + return 0; + } + + hash = cfl_hash_64bits(buffer.data, buffer.size); + msgpack_sbuffer_destroy(&buffer); + + return hash; +} + +static uint64_t msgpack_object_pair_hash(msgpack_object *left, + msgpack_object *right) +{ + uint64_t hash; + msgpack_sbuffer buffer; + msgpack_packer packer; + + msgpack_sbuffer_init(&buffer); + msgpack_packer_init(&packer, &buffer, msgpack_sbuffer_write); + + if (msgpack_pack_array(&packer, 2) != 0) { + msgpack_sbuffer_destroy(&buffer); + return 0; + } + + if (left == NULL) { + msgpack_pack_nil(&packer); + } + else if (msgpack_pack_object(&packer, *left) != 0) { + msgpack_sbuffer_destroy(&buffer); + return 0; + } + + if (right == NULL) { + msgpack_pack_nil(&packer); + } + else if (msgpack_pack_object(&packer, *right) != 0) { + msgpack_sbuffer_destroy(&buffer); + return 0; + } + + hash = cfl_hash_64bits(buffer.data, buffer.size); + msgpack_sbuffer_destroy(&buffer); + + return hash; +} + +static msgpack_object *resource_schema_url_object(msgpack_object *resource_object, + msgpack_object *resource_body) +{ + msgpack_object *schema_url; + + if (resource_body != NULL && resource_body->type == MSGPACK_OBJECT_MAP) { + schema_url = msgpack_map_get_object(&resource_body->via.map, "schema_url"); + if (schema_url != NULL) { + return schema_url; + } + } + + if (resource_object != NULL && resource_object->type == MSGPACK_OBJECT_MAP) { + schema_url = msgpack_map_get_object(&resource_object->via.map, "schema_url"); + if (schema_url != NULL) { + return schema_url; + } + } + + return NULL; +} + +static uint64_t resource_identity_hash(msgpack_object *resource_object, + msgpack_object *resource_body) +{ + msgpack_object *schema_url; + + schema_url = resource_schema_url_object(resource_object, resource_body); + + return msgpack_object_pair_hash(resource_object, schema_url); +} + static void set_result(int *result, int value) { if (result != NULL) { @@ -890,12 +987,14 @@ static msgpack_object *msgpack_map_get_object(msgpack_object_map *map, static struct otlp_logs_resource_state *find_logs_resource_state( struct otlp_logs_resource_state *states, size_t state_count, - int64_t resource_id) + int64_t resource_id, + uint64_t resource_hash) { size_t index; for (index = 0; index < state_count; index++) { - if (states[index].resource_id == resource_id) { + if (states[index].resource_id == resource_id && + states[index].resource_hash == resource_hash) { return &states[index]; } } @@ -905,12 +1004,14 @@ static struct otlp_logs_resource_state *find_logs_resource_state( static struct otlp_logs_scope_state *find_logs_scope_state( struct otlp_logs_resource_state *resource, - int64_t scope_id) + int64_t scope_id, + uint64_t scope_hash) { size_t index; for (index = 0; index < resource->scope_count; index++) { - if (resource->scopes[index].scope_id == scope_id) { + if (resource->scopes[index].scope_id == scope_id && + resource->scopes[index].scope_hash == scope_hash) { return &resource->scopes[index]; } } @@ -1036,6 +1137,7 @@ static struct otlp_logs_resource_state *append_logs_resource_state( struct otlp_logs_resource_state **states, size_t *state_count, int64_t resource_id, + uint64_t resource_hash, msgpack_object *resource_object, msgpack_object *resource_body) { @@ -1060,16 +1162,14 @@ static struct otlp_logs_resource_state *append_logs_resource_state( return NULL; } - if (resource_body != NULL && resource_body->type == MSGPACK_OBJECT_MAP) { - schema_url = msgpack_map_get_object(&resource_body->via.map, "schema_url"); - if (schema_url != NULL && schema_url->type == MSGPACK_OBJECT_STR) { - if (!flb_json_mut_obj_add_strncpy(doc, - resource_log, - "schemaUrl", - schema_url->via.str.ptr, - schema_url->via.str.size)) { - return NULL; - } + schema_url = resource_schema_url_object(resource_object, resource_body); + if (schema_url != NULL && schema_url->type == MSGPACK_OBJECT_STR) { + if (!flb_json_mut_obj_add_strncpy(doc, + resource_log, + "schemaUrl", + schema_url->via.str.ptr, + schema_url->via.str.size)) { + return NULL; } } @@ -1086,6 +1186,7 @@ static struct otlp_logs_resource_state *append_logs_resource_state( memset(state, 0, sizeof(struct otlp_logs_resource_state)); state->resource_id = resource_id; + state->resource_hash = resource_hash; state->resource_log = resource_log; state->scope_logs = scope_logs; @@ -1098,6 +1199,7 @@ static struct otlp_logs_scope_state *append_logs_scope_state( struct flb_json_mut_doc *doc, struct otlp_logs_resource_state *resource_state, int64_t scope_id, + uint64_t scope_hash, msgpack_object *scope_object) { struct otlp_logs_scope_state *new_scopes; @@ -1147,6 +1249,7 @@ static struct otlp_logs_scope_state *append_logs_scope_state( memset(state, 0, sizeof(struct otlp_logs_scope_state)); state->scope_id = scope_id; + state->scope_hash = scope_hash; state->scope_log = scope_log; state->log_records = log_records; @@ -1163,15 +1266,23 @@ static int ensure_default_logs_scope_state( struct otlp_logs_resource_state **current_resource, struct otlp_logs_scope_state **current_scope) { + uint64_t resource_hash; + uint64_t scope_hash; + + resource_hash = resource_identity_hash(NULL, NULL); + scope_hash = msgpack_object_hash(NULL); + *current_resource = find_logs_resource_state(*resource_states, *resource_state_count, - 0); + 0, + resource_hash); if (*current_resource == NULL) { *current_resource = append_logs_resource_state(doc, resource_logs, resource_states, resource_state_count, 0, + resource_hash, NULL, NULL); if (*current_resource == NULL) { @@ -1179,11 +1290,12 @@ static int ensure_default_logs_scope_state( } } - *current_scope = find_logs_scope_state(*current_resource, 0); + *current_scope = find_logs_scope_state(*current_resource, 0, scope_hash); if (*current_scope == NULL) { *current_scope = append_logs_scope_state(doc, *current_resource, 0, + scope_hash, NULL); if (*current_scope == NULL) { return -1; @@ -1483,6 +1595,8 @@ static flb_sds_t flb_opentelemetry_logs_to_otlp_json_render( msgpack_object *group_metadata; msgpack_object *resource_object; msgpack_object *scope_object; + uint64_t resource_hash; + uint64_t scope_hash; flb_sds_t json; struct flb_log_event event; struct flb_log_event_decoder decoder; @@ -1600,15 +1714,20 @@ static flb_sds_t flb_opentelemetry_logs_to_otlp_json_render( scope_object = msgpack_map_get_object(&group_body->via.map, "scope"); } + resource_hash = resource_identity_hash(resource_object, group_body); + scope_hash = msgpack_object_hash(scope_object); + current_resource = find_logs_resource_state(resource_states, resource_state_count, - resource_id); + resource_id, + resource_hash); if (current_resource == NULL) { current_resource = append_logs_resource_state(doc, resource_logs, &resource_states, &resource_state_count, resource_id, + resource_hash, resource_object, group_body); if (current_resource == NULL) { @@ -1620,11 +1739,14 @@ static flb_sds_t flb_opentelemetry_logs_to_otlp_json_render( } } - current_scope = find_logs_scope_state(current_resource, scope_id); + current_scope = find_logs_scope_state(current_resource, + scope_id, + scope_hash); if (current_scope == NULL) { current_scope = append_logs_scope_state(doc, current_resource, scope_id, + scope_hash, scope_object); if (current_scope == NULL) { flb_log_event_decoder_destroy(&decoder); diff --git a/src/opentelemetry/flb_opentelemetry_otlp_proto.c b/src/opentelemetry/flb_opentelemetry_otlp_proto.c index 6523f789dc3..93cdbd16960 100644 --- a/src/opentelemetry/flb_opentelemetry_otlp_proto.c +++ b/src/opentelemetry/flb_opentelemetry_otlp_proto.c @@ -30,6 +30,8 @@ #include #include +#include + #include #include @@ -44,16 +46,21 @@ struct otlp_proto_logs_scope_state { int64_t scope_id; + uint64_t scope_hash; Opentelemetry__Proto__Logs__V1__ScopeLogs *scope_log; }; struct otlp_proto_logs_resource_state { int64_t resource_id; + uint64_t resource_hash; Opentelemetry__Proto__Logs__V1__ResourceLogs *resource_log; struct otlp_proto_logs_scope_state *scopes; size_t scope_count; }; +static msgpack_object *msgpack_map_get_object(msgpack_object_map *map, + const char *key); + static void set_result(int *result, int value) { if (result != NULL) { @@ -67,6 +74,99 @@ static void set_error(int *result, int value, int err) errno = err; } +static uint64_t msgpack_object_hash(msgpack_object *object) +{ + uint64_t hash; + msgpack_sbuffer buffer; + msgpack_packer packer; + + if (object == NULL) { + return cfl_hash_64bits("null", 4); + } + + msgpack_sbuffer_init(&buffer); + msgpack_packer_init(&packer, &buffer, msgpack_sbuffer_write); + + if (msgpack_pack_object(&packer, *object) != 0) { + msgpack_sbuffer_destroy(&buffer); + return 0; + } + + hash = cfl_hash_64bits(buffer.data, buffer.size); + msgpack_sbuffer_destroy(&buffer); + + return hash; +} + +static uint64_t msgpack_object_pair_hash(msgpack_object *left, + msgpack_object *right) +{ + uint64_t hash; + msgpack_sbuffer buffer; + msgpack_packer packer; + + msgpack_sbuffer_init(&buffer); + msgpack_packer_init(&packer, &buffer, msgpack_sbuffer_write); + + if (msgpack_pack_array(&packer, 2) != 0) { + msgpack_sbuffer_destroy(&buffer); + return 0; + } + + if (left == NULL) { + msgpack_pack_nil(&packer); + } + else if (msgpack_pack_object(&packer, *left) != 0) { + msgpack_sbuffer_destroy(&buffer); + return 0; + } + + if (right == NULL) { + msgpack_pack_nil(&packer); + } + else if (msgpack_pack_object(&packer, *right) != 0) { + msgpack_sbuffer_destroy(&buffer); + return 0; + } + + hash = cfl_hash_64bits(buffer.data, buffer.size); + msgpack_sbuffer_destroy(&buffer); + + return hash; +} + +static msgpack_object *resource_schema_url_object(msgpack_object *resource_object, + msgpack_object *resource_body) +{ + msgpack_object *schema_url; + + if (resource_body != NULL && resource_body->type == MSGPACK_OBJECT_MAP) { + schema_url = msgpack_map_get_object(&resource_body->via.map, "schema_url"); + if (schema_url != NULL) { + return schema_url; + } + } + + if (resource_object != NULL && resource_object->type == MSGPACK_OBJECT_MAP) { + schema_url = msgpack_map_get_object(&resource_object->via.map, "schema_url"); + if (schema_url != NULL) { + return schema_url; + } + } + + return NULL; +} + +static uint64_t resource_identity_hash(msgpack_object *resource_object, + msgpack_object *resource_body) +{ + msgpack_object *schema_url; + + schema_url = resource_schema_url_object(resource_object, resource_body); + + return msgpack_object_pair_hash(resource_object, schema_url); +} + static msgpack_object *msgpack_map_get_object(msgpack_object_map *map, const char *key) { @@ -573,12 +673,14 @@ static Opentelemetry__Proto__Common__V1__AnyValue *msgpack_object_to_otlp_any_va static struct otlp_proto_logs_resource_state *find_logs_resource_state( struct otlp_proto_logs_resource_state *states, size_t state_count, - int64_t resource_id) + int64_t resource_id, + uint64_t resource_hash) { size_t index; for (index = 0; index < state_count; index++) { - if (states[index].resource_id == resource_id) { + if (states[index].resource_id == resource_id && + states[index].resource_hash == resource_hash) { return &states[index]; } } @@ -588,12 +690,14 @@ static struct otlp_proto_logs_resource_state *find_logs_resource_state( static struct otlp_proto_logs_scope_state *find_logs_scope_state( struct otlp_proto_logs_resource_state *resource, - int64_t scope_id) + int64_t scope_id, + uint64_t scope_hash) { size_t index; for (index = 0; index < resource->scope_count; index++) { - if (resource->scopes[index].scope_id == scope_id) { + if (resource->scopes[index].scope_id == scope_id && + resource->scopes[index].scope_hash == scope_hash) { return &resource->scopes[index]; } } @@ -874,6 +978,7 @@ static struct otlp_proto_logs_resource_state *append_logs_resource_state( struct otlp_proto_logs_resource_state **states, size_t *state_count, int64_t resource_id, + uint64_t resource_hash, msgpack_object *resource_object, msgpack_object *resource_body) { @@ -902,17 +1007,15 @@ static struct otlp_proto_logs_resource_state *append_logs_resource_state( return NULL; } - if (resource_body != NULL && resource_body->type == MSGPACK_OBJECT_MAP) { - schema_url = msgpack_map_get_object(&resource_body->via.map, "schema_url"); - if (schema_url != NULL && schema_url->type == MSGPACK_OBJECT_STR) { - resource_log->schema_url = flb_strndup(schema_url->via.str.ptr, - schema_url->via.str.size); - if (resource_log->schema_url == NULL) { - otlp_kvarray_destroy(resource->attributes, resource->n_attributes); - flb_free(resource); - flb_free(resource_log); - return NULL; - } + schema_url = resource_schema_url_object(resource_object, resource_body); + if (schema_url != NULL && schema_url->type == MSGPACK_OBJECT_STR) { + resource_log->schema_url = flb_strndup(schema_url->via.str.ptr, + schema_url->via.str.size); + if (resource_log->schema_url == NULL) { + otlp_kvarray_destroy(resource->attributes, resource->n_attributes); + flb_free(resource); + flb_free(resource_log); + return NULL; } } @@ -945,6 +1048,7 @@ static struct otlp_proto_logs_resource_state *append_logs_resource_state( memset(state, 0, sizeof(struct otlp_proto_logs_resource_state)); state->resource_id = resource_id; + state->resource_hash = resource_hash; state->resource_log = resource_log; (*state_count)++; @@ -954,6 +1058,7 @@ static struct otlp_proto_logs_resource_state *append_logs_resource_state( static struct otlp_proto_logs_scope_state *append_logs_scope_state( struct otlp_proto_logs_resource_state *resource_state, int64_t scope_id, + uint64_t scope_hash, msgpack_object *scope_object) { struct otlp_proto_logs_scope_state *new_scopes; @@ -1032,6 +1137,7 @@ static struct otlp_proto_logs_scope_state *append_logs_scope_state( memset(state, 0, sizeof(struct otlp_proto_logs_scope_state)); state->scope_id = scope_id; + state->scope_hash = scope_hash; state->scope_log = scope_log; resource_state->scope_count++; @@ -1045,14 +1151,22 @@ static int ensure_default_logs_scope_state( struct otlp_proto_logs_resource_state **current_resource, struct otlp_proto_logs_scope_state **current_scope) { + uint64_t resource_hash; + uint64_t scope_hash; + + resource_hash = resource_identity_hash(NULL, NULL); + scope_hash = msgpack_object_hash(NULL); + *current_resource = find_logs_resource_state(*resource_states, *resource_state_count, - 0); + 0, + resource_hash); if (*current_resource == NULL) { *current_resource = append_logs_resource_state(export_logs, resource_states, resource_state_count, 0, + resource_hash, NULL, NULL); if (*current_resource == NULL) { @@ -1060,9 +1174,12 @@ static int ensure_default_logs_scope_state( } } - *current_scope = find_logs_scope_state(*current_resource, 0); + *current_scope = find_logs_scope_state(*current_resource, 0, scope_hash); if (*current_scope == NULL) { - *current_scope = append_logs_scope_state(*current_resource, 0, NULL); + *current_scope = append_logs_scope_state(*current_resource, + 0, + scope_hash, + NULL); if (*current_scope == NULL) { return -1; } @@ -1485,6 +1602,8 @@ flb_sds_t flb_opentelemetry_logs_to_otlp_proto(const void *event_chunk_data, msgpack_object *group_body; msgpack_object *resource_object; msgpack_object *scope_object; + uint64_t resource_hash; + uint64_t scope_hash; struct otlp_proto_logs_scope_state *current_scope; struct otlp_proto_logs_resource_state *current_resource; struct otlp_proto_logs_resource_state *resource_states; @@ -1577,14 +1696,19 @@ flb_sds_t flb_opentelemetry_logs_to_otlp_proto(const void *event_chunk_data, scope_object = msgpack_map_get_object(&group_body->via.map, "scope"); } + resource_hash = resource_identity_hash(resource_object, group_body); + scope_hash = msgpack_object_hash(scope_object); + current_resource = find_logs_resource_state(resource_states, resource_state_count, - resource_id); + resource_id, + resource_hash); if (current_resource == NULL) { current_resource = append_logs_resource_state(&export_logs, &resource_states, &resource_state_count, resource_id, + resource_hash, resource_object, group_body); if (current_resource == NULL) { @@ -1596,10 +1720,13 @@ flb_sds_t flb_opentelemetry_logs_to_otlp_proto(const void *event_chunk_data, } } - current_scope = find_logs_scope_state(current_resource, scope_id); + current_scope = find_logs_scope_state(current_resource, + scope_id, + scope_hash); if (current_scope == NULL) { current_scope = append_logs_scope_state(current_resource, scope_id, + scope_hash, scope_object); if (current_scope == NULL) { flb_log_event_decoder_destroy(&decoder); diff --git a/tests/integration/scenarios/in_opentelemetry/config/stdout-otlp-json-slow-flush.yaml b/tests/integration/scenarios/in_opentelemetry/config/stdout-otlp-json-slow-flush.yaml new file mode 100644 index 00000000000..789297e97c7 --- /dev/null +++ b/tests/integration/scenarios/in_opentelemetry/config/stdout-otlp-json-slow-flush.yaml @@ -0,0 +1,15 @@ +service: + flush: 5 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: opentelemetry + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + + outputs: + - name: stdout + match: '*' + format: otlp_json diff --git a/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py b/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py index 957d59e9105..71936ffc201 100644 --- a/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py +++ b/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py @@ -251,6 +251,81 @@ def maybe_read_prometheus_metric_value(metrics_text, metric_name, input_name): except AssertionError: return None + +def build_resource_collision_logs_payload(user_id, body, schema_url=None): + payload = { + "resourceLogs": [ + { + "resource": { + "attributes": [ + { + "key": "user.id", + "value": { + "stringValue": user_id, + }, + } + ], + }, + "scopeLogs": [ + { + "scope": {}, + "logRecords": [ + { + "timeUnixNano": "1640995200000000000", + "body": { + "stringValue": body, + }, + } + ], + } + ], + } + ], + } + + if schema_url is not None: + payload["resourceLogs"][0]["schemaUrl"] = schema_url + + return json_format.Parse(json.dumps(payload), ExportLogsServiceRequest()).SerializeToString() + + +def build_resource_collision_logs_json_payload(user_id, body, schema_url=None): + payload = { + "resourceLogs": [ + { + "resource": { + "attributes": [ + { + "key": "user.id", + "value": { + "stringValue": user_id, + }, + } + ], + }, + "scopeLogs": [ + { + "scope": {}, + "logRecords": [ + { + "timeUnixNano": "1640995200000000000", + "body": { + "stringValue": body, + }, + } + ], + } + ], + } + ], + } + + if schema_url is not None: + payload["resourceLogs"][0]["schemaUrl"] = schema_url + + return json.dumps(payload).encode("utf-8") + + class Service: def __init__(self, config_file, *, use_auth_server=False): # Compose the absolute path for the Fluent Bit configuration file @@ -644,6 +719,79 @@ def test_in_opentelemetry_stdout_otlp_json_logs(): assert records[0]["resource_attributes"]["service.name"] == "example-service" +def test_in_opentelemetry_stdout_otlp_json_logs_preserve_resources_across_requests(): + service = Service("stdout-otlp-json-slow-flush.yaml") + service.start() + + response = service.send_raw_request( + "/v1/logs", + build_resource_collision_logs_payload("user-a", "event-a"), + ) + assert 200 <= response.status_code < 300 + + response = service.send_raw_request( + "/v1/logs", + build_resource_collision_logs_payload("user-b", "event-b"), + ) + assert 200 <= response.status_code < 300 + + output = read_stdout_otlp_json(service, "resourceLogs", timeout=10) + service.stop() + + records = list(iter_log_records(output)) + body_to_user = { + record["body"]: record["resource_attributes"]["user.id"] + for record in records + } + + assert body_to_user["event-a"] == "user-a" + assert body_to_user["event-b"] == "user-b" + assert len(output["resourceLogs"]) == 2 + + +@pytest.mark.parametrize( + "content_type,payload_builder", + [ + ("application/x-protobuf", build_resource_collision_logs_payload), + ("application/json", build_resource_collision_logs_json_payload), + ], +) +def test_in_opentelemetry_stdout_otlp_json_logs_preserve_resource_schema_urls( + content_type, + payload_builder, +): + service = Service("stdout-otlp-json-slow-flush.yaml") + service.start() + + response = service.send_raw_request( + "/v1/logs", + payload_builder("same-user", "event-a", "schema-a"), + content_type=content_type, + ) + assert 200 <= response.status_code < 300 + + response = service.send_raw_request( + "/v1/logs", + payload_builder("same-user", "event-b", "schema-b"), + content_type=content_type, + ) + assert 200 <= response.status_code < 300 + + output = read_stdout_otlp_json(service, "resourceLogs", timeout=10) + service.stop() + + body_to_schema_url = { + record["body"]["stringValue"]: resource_log["schemaUrl"] + for resource_log in output["resourceLogs"] + for scope_log in resource_log["scopeLogs"] + for record in scope_log["logRecords"] + } + + assert body_to_schema_url["event-a"] == "schema-a" + assert body_to_schema_url["event-b"] == "schema-b" + assert len(output["resourceLogs"]) == 2 + + def test_in_opentelemetry_stdout_otlp_json_metrics(): service = Service("003-stdout-otlp-json.yaml") service.start() diff --git a/tests/integration/scenarios/out_kafka/config/out_kafka_otlp_json_slow_flush.yaml b/tests/integration/scenarios/out_kafka/config/out_kafka_otlp_json_slow_flush.yaml new file mode 100644 index 00000000000..80b1be35d0f --- /dev/null +++ b/tests/integration/scenarios/out_kafka/config/out_kafka_otlp_json_slow_flush.yaml @@ -0,0 +1,24 @@ +service: + flush: 5 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: opentelemetry + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + + outputs: + - name: kafka + match: "*" + brokers: 127.0.0.1:${TEST_SUITE_KAFKA_PORT} + topics: otlp-topic + format: otlp_json + message_key: static-otlp-key + raw_log_key: ignored_raw_key + message_key_field: ignored_message_key_field + topic_key: ignored_topic_key + queue_full_retries: 1 + rdkafka.api.version.request: false + rdkafka.broker.version.fallback: 0.8.2.0 diff --git a/tests/integration/scenarios/out_kafka/config/out_kafka_otlp_proto_slow_flush.yaml b/tests/integration/scenarios/out_kafka/config/out_kafka_otlp_proto_slow_flush.yaml new file mode 100644 index 00000000000..101fd274bc5 --- /dev/null +++ b/tests/integration/scenarios/out_kafka/config/out_kafka_otlp_proto_slow_flush.yaml @@ -0,0 +1,24 @@ +service: + flush: 5 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: opentelemetry + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + + outputs: + - name: kafka + match: "*" + brokers: 127.0.0.1:${TEST_SUITE_KAFKA_PORT} + topics: otlp-topic + format: otlp_proto + message_key: static-otlp-key + raw_log_key: ignored_raw_key + message_key_field: ignored_message_key_field + topic_key: ignored_topic_key + queue_full_retries: 1 + rdkafka.api.version.request: false + rdkafka.broker.version.fallback: 0.8.2.0 diff --git a/tests/integration/scenarios/out_kafka/tests/test_out_kafka_001.py b/tests/integration/scenarios/out_kafka/tests/test_out_kafka_001.py index 50a670b6ead..abe2f93a284 100644 --- a/tests/integration/scenarios/out_kafka/tests/test_out_kafka_001.py +++ b/tests/integration/scenarios/out_kafka/tests/test_out_kafka_001.py @@ -218,6 +218,43 @@ def _build_multi_resource_payload(service, signal_type, json_file): return payload +def _build_resource_collision_payload(user_id, body, schema_url=None): + payload = { + "resource_logs": [ + { + "resource": { + "attributes": [ + { + "key": "user.id", + "value": { + "string_value": user_id, + }, + } + ], + }, + "scope_logs": [ + { + "scope": {}, + "log_records": [ + { + "time_unix_nano": "1640995200000000000", + "body": { + "string_value": body, + }, + } + ], + } + ], + } + ], + } + + if schema_url is not None: + payload["resource_logs"][0]["schema_url"] = schema_url + + return payload + + def _decode_kafka_payload(message, format_name, signal_type): if format_name == "otlp_json": return json.loads(message["value"].decode("utf-8")) @@ -522,3 +559,89 @@ def test_out_kafka_otlp_formats_preserve_multiple_resources( assert "checkout-span" in span_names assert "bulk-trace-span" in span_names assert "checkout-bulk" in service_names + + +@pytest.mark.parametrize( + "format_name,config_file", + [ + ("otlp_json", "out_kafka_otlp_json_slow_flush.yaml"), + ("otlp_proto", "out_kafka_otlp_proto_slow_flush.yaml"), + ], +) +def test_out_kafka_otlp_logs_preserve_resources_across_requests_in_same_chunk( + format_name, + config_file, +): + service = Service(config_file) + service.start() + service.send_payload_dict( + _build_resource_collision_payload("user-a", "event-a"), + "logs", + ) + service.send_payload_dict( + _build_resource_collision_payload("user-b", "event-b"), + "logs", + ) + + messages = service.wait_for_messages(1, timeout=10) + service.stop() + + assert len(messages) == 1 + + resources = _collect_resources(messages[:1], format_name, "logs") + body_to_user = { + record["body"]["stringValue"]: next( + attribute["value"]["stringValue"] + for attribute in resource["resource"]["attributes"] + if attribute["key"] == "user.id" + ) + for resource in resources + for scope in resource["scopeLogs"] + for record in scope["logRecords"] + } + + assert "event-a" in body_to_user + assert "event-b" in body_to_user + assert body_to_user["event-a"] == "user-a" + assert body_to_user["event-b"] == "user-b" + assert len(resources) == 2 + + +@pytest.mark.parametrize( + "format_name,config_file", + [ + ("otlp_json", "out_kafka_otlp_json_slow_flush.yaml"), + ("otlp_proto", "out_kafka_otlp_proto_slow_flush.yaml"), + ], +) +def test_out_kafka_otlp_logs_preserve_resource_schema_urls_across_requests( + format_name, + config_file, +): + service = Service(config_file) + service.start() + service.send_payload_dict( + _build_resource_collision_payload("same-user", "event-a", "schema-a"), + "logs", + ) + service.send_payload_dict( + _build_resource_collision_payload("same-user", "event-b", "schema-b"), + "logs", + ) + + messages = service.wait_for_messages(1, timeout=10) + service.stop() + + assert len(messages) == 1 + + resources = _collect_resources(messages[:1], format_name, "logs") + body_to_schema_url = { + record["body"]["stringValue"]: resource["schemaUrl"] + for resource in resources + for scope in resource["scopeLogs"] + for record in scope["logRecords"] + } + + assert body_to_schema_url["event-a"] == "schema-a" + assert body_to_schema_url["event-b"] == "schema-b" + assert len(resources) == 2 diff --git a/tests/integration/scenarios/out_opentelemetry/config/out_otel_grpc_logs_otlp_input_slow_flush.yaml b/tests/integration/scenarios/out_opentelemetry/config/out_otel_grpc_logs_otlp_input_slow_flush.yaml new file mode 100644 index 00000000000..fb7038d1bc5 --- /dev/null +++ b/tests/integration/scenarios/out_opentelemetry/config/out_otel_grpc_logs_otlp_input_slow_flush.yaml @@ -0,0 +1,18 @@ +service: + flush: 5 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: opentelemetry + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + + outputs: + - name: opentelemetry + match: "*" + host: 127.0.0.1 + port: ${TEST_SUITE_HTTP_PORT} + http2: on + grpc: on diff --git a/tests/integration/scenarios/out_opentelemetry/config/out_otel_http_logs_otlp_input_slow_flush.yaml b/tests/integration/scenarios/out_opentelemetry/config/out_otel_http_logs_otlp_input_slow_flush.yaml new file mode 100644 index 00000000000..7e4106e6e9b --- /dev/null +++ b/tests/integration/scenarios/out_opentelemetry/config/out_otel_http_logs_otlp_input_slow_flush.yaml @@ -0,0 +1,17 @@ +service: + flush: 5 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: opentelemetry + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + + outputs: + - name: opentelemetry + match: "*" + host: 127.0.0.1 + port: ${TEST_SUITE_HTTP_PORT} + logs_uri: /v1/logs diff --git a/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py b/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py index 3d86154c44b..832c85e59bf 100644 --- a/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py +++ b/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py @@ -34,13 +34,17 @@ def _repo_relative(*parts): def iter_log_records(output): for resource_log in output.get("resourceLogs", []): + resource_attributes = { + item["key"]: next(iter(item["value"].values())) + for item in resource_log.get("resource", {}).get("attributes", []) + } for scope_log in resource_log.get("scopeLogs", []): for record in scope_log.get("logRecords", []): attributes = { item["key"]: next(iter(item["value"].values())) for item in record.get("attributes", []) } - yield record, attributes + yield record, attributes, resource_attributes def iter_metric_attributes(output): @@ -199,6 +203,21 @@ def send_json_logs_payload(self, json_file): ) response.raise_for_status() + def send_payload_dict(self, payload_dict, signal_type): + payload = self._build_signal_payload_from_dict(payload_dict, signal_type) + endpoints = { + "logs": "/v1/logs", + "metrics": "/v1/metrics", + "traces": "/v1/traces", + } + response = requests.post( + f"http://127.0.0.1:{self.flb_listener_port}{endpoints[signal_type]}", + data=payload.SerializeToString(), + headers={"Content-Type": "application/x-protobuf"}, + timeout=5, + ) + response.raise_for_status() + def send_json_traces_payload(self, json_file): payload = self._build_signal_payload(json_file, "traces") response = requests.post( @@ -227,6 +246,59 @@ def _build_signal_payload(self, json_file, signal_type): messages[signal_type], ) + def _build_signal_payload_from_dict(self, payload_dict, signal_type): + messages = { + "logs": ExportLogsServiceRequest(), + "metrics": ExportMetricsServiceRequest(), + "traces": ExportTraceServiceRequest(), + } + return json_format.Parse(json.dumps(payload_dict), messages[signal_type]) + + +def _build_resource_collision_payload(user_id, body): + return { + "resource_logs": [ + { + "resource": { + "attributes": [ + { + "key": "user.id", + "value": { + "string_value": user_id, + }, + } + ], + }, + "scope_logs": [ + { + "scope": {}, + "log_records": [ + { + "time_unix_nano": "1640995200000000000", + "body": { + "string_value": body, + }, + } + ], + } + ], + } + ], + } + + +def _assert_log_resource_attribution(logs_seen): + output = json.loads(json_format.MessageToJson(logs_seen[0])) + records = list(iter_log_records(output)) + body_to_user = { + record["body"]["stringValue"]: resource_attributes["user.id"] + for record, _, resource_attributes in records + } + + assert body_to_user["event-a"] == "user-a" + assert body_to_user["event-b"] == "user-b" + assert len(output["resourceLogs"]) == 2 + def test_out_opentelemetry_http_logs_uri_headers_and_basic_auth(): service = Service("out_otel_http_logs.yaml") @@ -336,7 +408,7 @@ def test_out_opentelemetry_gzip_and_logs_body_key_attributes(): request_seen = requests_seen[0] output = json.loads(json_format.MessageToJson(logs_seen[0])) - record, attributes = next(iter_log_records(output)) + record, attributes, _ = next(iter_log_records(output)) assert request_seen["headers"]["Content-Encoding"] == "gzip" assert record["body"]["stringValue"] == "body only" @@ -354,7 +426,7 @@ def test_out_opentelemetry_zstd_and_logs_body_key_attributes(): request_seen = requests_seen[0] output = json.loads(json_format.MessageToJson(logs_seen[0])) - record, attributes = next(iter_log_records(output)) + record, attributes, _ = next(iter_log_records(output)) assert request_seen["headers"]["Content-Encoding"] == "zstd" assert record["body"]["stringValue"] == "zstd body" @@ -399,6 +471,34 @@ def test_out_opentelemetry_grpc_custom_logs_uri(): assert record["body"]["stringValue"] == "hello via grpc" +@pytest.mark.parametrize( + "config_file,receiver_mode", + [ + ("out_otel_http_logs_otlp_input_slow_flush.yaml", "http"), + ("out_otel_grpc_logs_otlp_input_slow_flush.yaml", "grpc"), + ], + ids=["http", "grpc"], +) +def test_out_opentelemetry_logs_preserve_resources_across_otlp_input_requests( + config_file, + receiver_mode, +): + service = Service(config_file, receiver_mode=receiver_mode) + service.start() + service.send_payload_dict( + _build_resource_collision_payload("user-a", "event-a"), + "logs", + ) + service.send_payload_dict( + _build_resource_collision_payload("user-b", "event-b"), + "logs", + ) + logs_seen = service.wait_for_signal("logs", minimum_count=1, timeout=10) + service.stop() + + _assert_log_resource_attribution(logs_seen) + + def test_out_opentelemetry_metrics_uri_and_add_label(): service = Service("out_otel_http_metrics.yaml") service.start() @@ -498,7 +598,7 @@ def test_out_opentelemetry_custom_metadata_key_accessors(): request_seen = requests_seen[0] output = json.loads(json_format.MessageToJson(logs_seen[0])) - record, attributes = next(iter_log_records(output)) + record, attributes, _ = next(iter_log_records(output)) assert request_seen["path"] == "/metadata/logs" assert record["severityText"] == "WARN" diff --git a/tests/internal/opentelemetry.c b/tests/internal/opentelemetry.c index cc080c1878f..404cca8e1b9 100644 --- a/tests/internal/opentelemetry.c +++ b/tests/internal/opentelemetry.c @@ -2056,6 +2056,208 @@ void test_opentelemetry_logs_otlp_json_roundtrip() flb_log_event_encoder_destroy(&encoder); } +void test_opentelemetry_logs_otlp_json_preserves_appended_resources() +{ + int ret; + int result; + char *input_a; + char *input_b; + char *expected; + flb_sds_t actual; + flb_sds_t normalized_expected; + struct flb_log_event_encoder encoder; + struct flb_opentelemetry_otlp_logs_options options; + + input_a = + "{\"resourceLogs\":[{\"resource\":{\"attributes\":[{\"key\":\"user.id\"," + "\"value\":{\"stringValue\":\"user-a\"}}]},\"scopeLogs\":[{\"scope\":{}," + "\"logRecords\":[{\"timeUnixNano\":\"1640995200000000000\"," + "\"body\":{\"stringValue\":\"event-a\"}}]}]}]}"; + + input_b = + "{\"resourceLogs\":[{\"resource\":{\"attributes\":[{\"key\":\"user.id\"," + "\"value\":{\"stringValue\":\"user-b\"}}]},\"scopeLogs\":[{\"scope\":{}," + "\"logRecords\":[{\"timeUnixNano\":\"1640995201000000000\"," + "\"body\":{\"stringValue\":\"event-b\"}}]}]}]}"; + + expected = + "{\"resourceLogs\":[{\"resource\":{\"attributes\":[{\"key\":\"user.id\"," + "\"value\":{\"stringValue\":\"user-a\"}}]},\"scopeLogs\":[{\"scope\":{}," + "\"logRecords\":[{\"timeUnixNano\":\"1640995200000000000\"," + "\"body\":{\"stringValue\":\"event-a\"}}]}]},{\"resource\":{\"attributes\":[{" + "\"key\":\"user.id\",\"value\":{\"stringValue\":\"user-b\"}}]},\"scopeLogs\":[{" + "\"scope\":{},\"logRecords\":[{\"timeUnixNano\":\"1640995201000000000\"," + "\"body\":{\"stringValue\":\"event-b\"}}]}]}]}"; + + ret = flb_log_event_encoder_init(&encoder, + FLB_LOG_EVENT_FORMAT_DEFAULT); + TEST_CHECK(ret == FLB_EVENT_ENCODER_SUCCESS); + if (ret != FLB_EVENT_ENCODER_SUCCESS) { + return; + } + + ret = flb_opentelemetry_logs_json_to_msgpack(&encoder, + input_a, + strlen(input_a), + "log", + &result); + TEST_CHECK(ret == 0); + TEST_CHECK(result == 0); + + ret = flb_opentelemetry_logs_json_to_msgpack(&encoder, + input_b, + strlen(input_b), + "log", + &result); + TEST_CHECK(ret == 0); + TEST_CHECK(result == 0); + if (ret != 0 || result != 0) { + flb_log_event_encoder_destroy(&encoder); + return; + } + + memset(&options, 0, sizeof(options)); + options.logs_require_otel_metadata = FLB_TRUE; + options.logs_body_key = "log"; + + actual = flb_opentelemetry_logs_to_otlp_json(encoder.output_buffer, + encoder.output_length, + &options, + &result); + TEST_CHECK(actual != NULL); + TEST_CHECK(result == FLB_OPENTELEMETRY_OTLP_JSON_SUCCESS); + if (actual == NULL) { + flb_log_event_encoder_destroy(&encoder); + return; + } + + normalized_expected = test_normalize_json(expected); + TEST_CHECK(normalized_expected != NULL); + if (normalized_expected == NULL) { + flb_sds_destroy(actual); + flb_log_event_encoder_destroy(&encoder); + return; + } + + TEST_CHECK(strcmp(normalized_expected, actual) == 0); + + flb_sds_destroy(normalized_expected); + flb_sds_destroy(actual); + flb_log_event_encoder_destroy(&encoder); +} + +void test_opentelemetry_logs_otlp_resource_schema_url_identity() +{ + int ret; + int result; + char *input_a; + char *input_b; + char *expected; + flb_sds_t actual_json; + flb_sds_t actual_proto; + flb_sds_t normalized_expected; + struct flb_log_event_encoder encoder; + struct flb_opentelemetry_otlp_logs_options options; + Opentelemetry__Proto__Collector__Logs__V1__ExportLogsServiceRequest *decoded; + + input_a = + "{\"resourceLogs\":[{\"schemaUrl\":\"schema-a\",\"resource\":{\"attributes\":[{" + "\"key\":\"service.name\",\"value\":{\"stringValue\":\"svc\"}}]},\"scopeLogs\":[{" + "\"scope\":{},\"logRecords\":[{\"timeUnixNano\":\"1640995200000000000\"," + "\"body\":{\"stringValue\":\"event-a\"}}]}]}]}"; + + input_b = + "{\"resourceLogs\":[{\"schemaUrl\":\"schema-b\",\"resource\":{\"attributes\":[{" + "\"key\":\"service.name\",\"value\":{\"stringValue\":\"svc\"}}]},\"scopeLogs\":[{" + "\"scope\":{},\"logRecords\":[{\"timeUnixNano\":\"1640995201000000000\"," + "\"body\":{\"stringValue\":\"event-b\"}}]}]}]}"; + + expected = + "{\"resourceLogs\":[{\"resource\":{\"attributes\":[{\"key\":\"service.name\"," + "\"value\":{\"stringValue\":\"svc\"}}]},\"scopeLogs\":[{\"scope\":{}," + "\"logRecords\":[{\"timeUnixNano\":\"1640995200000000000\"," + "\"body\":{\"stringValue\":\"event-a\"}}]}],\"schemaUrl\":\"schema-a\"},{" + "\"resource\":{\"attributes\":[{\"key\":\"service.name\",\"value\":{\"stringValue\":\"svc\"}}]}," + "\"scopeLogs\":[{\"scope\":{},\"logRecords\":[{\"timeUnixNano\":\"1640995201000000000\"," + "\"body\":{\"stringValue\":\"event-b\"}}]}],\"schemaUrl\":\"schema-b\"}]}"; + + ret = flb_log_event_encoder_init(&encoder, + FLB_LOG_EVENT_FORMAT_DEFAULT); + TEST_CHECK(ret == FLB_EVENT_ENCODER_SUCCESS); + if (ret != FLB_EVENT_ENCODER_SUCCESS) { + return; + } + + ret = flb_opentelemetry_logs_json_to_msgpack(&encoder, + input_a, + strlen(input_a), + "log", + &result); + TEST_CHECK(ret == 0); + TEST_CHECK(result == 0); + + ret = flb_opentelemetry_logs_json_to_msgpack(&encoder, + input_b, + strlen(input_b), + "log", + &result); + TEST_CHECK(ret == 0); + TEST_CHECK(result == 0); + if (ret != 0 || result != 0) { + flb_log_event_encoder_destroy(&encoder); + return; + } + + memset(&options, 0, sizeof(options)); + options.logs_require_otel_metadata = FLB_TRUE; + options.logs_body_key = "log"; + + actual_json = flb_opentelemetry_logs_to_otlp_json(encoder.output_buffer, + encoder.output_length, + &options, + &result); + TEST_CHECK(actual_json != NULL); + TEST_CHECK(result == FLB_OPENTELEMETRY_OTLP_JSON_SUCCESS); + if (actual_json == NULL) { + flb_log_event_encoder_destroy(&encoder); + return; + } + + normalized_expected = test_normalize_json(expected); + TEST_CHECK(normalized_expected != NULL); + if (normalized_expected != NULL) { + TEST_CHECK(strcmp(normalized_expected, actual_json) == 0); + flb_sds_destroy(normalized_expected); + } + flb_sds_destroy(actual_json); + + actual_proto = flb_opentelemetry_logs_to_otlp_proto(encoder.output_buffer, + encoder.output_length, + &options, + &result); + TEST_CHECK(actual_proto != NULL); + TEST_CHECK(result == FLB_OPENTELEMETRY_OTLP_PROTO_SUCCESS); + if (actual_proto == NULL) { + flb_log_event_encoder_destroy(&encoder); + return; + } + + decoded = + opentelemetry__proto__collector__logs__v1__export_logs_service_request__unpack( + NULL, flb_sds_len(actual_proto), (uint8_t *) actual_proto); + TEST_CHECK(decoded != NULL); + if (decoded != NULL) { + TEST_CHECK(decoded->n_resource_logs == 2); + TEST_CHECK(strcmp(decoded->resource_logs[0]->schema_url, "schema-a") == 0); + TEST_CHECK(strcmp(decoded->resource_logs[1]->schema_url, "schema-b") == 0); + opentelemetry__proto__collector__logs__v1__export_logs_service_request__free_unpacked(decoded, + NULL); + } + + flb_sds_destroy(actual_proto); + flb_log_event_encoder_destroy(&encoder); +} + void test_opentelemetry_logs_otlp_json_from_plain_logs() { int ret; @@ -2543,6 +2745,10 @@ TEST_LIST = { { "opentelemetry_cases", test_opentelemetry_cases }, { "opentelemetry_logs_otlp_json_roundtrip", test_opentelemetry_logs_otlp_json_roundtrip }, + { "opentelemetry_logs_otlp_json_preserves_appended_resources", + test_opentelemetry_logs_otlp_json_preserves_appended_resources }, + { "opentelemetry_logs_otlp_resource_schema_url_identity", + test_opentelemetry_logs_otlp_resource_schema_url_identity }, { "opentelemetry_logs_otlp_json_from_plain_logs", test_opentelemetry_logs_otlp_json_from_plain_logs }, { "opentelemetry_logs_otlp_proto_from_plain_logs",