tests: integration: cover otlp resource schema url grouping

Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>

Author: edsiper

tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py

@@ -252,7 +252,7 @@ def maybe_read_prometheus_metric_value(metrics_text, metric_name, input_name):
         return None
 
 
-def build_resource_collision_logs_payload(user_id, body):
+def build_resource_collision_logs_payload(user_id, body, schema_url=None):
     payload = {
         "resourceLogs": [
             {
@@ -283,9 +283,49 @@ def build_resource_collision_logs_payload(user_id, 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
@@ -709,6 +749,49 @@ def test_in_opentelemetry_stdout_otlp_json_logs_preserve_resources_across_reques
     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()

tests/integration/scenarios/out_kafka/tests/test_out_kafka_001.py

@@ -218,8 +218,8 @@ def _build_multi_resource_payload(service, signal_type, json_file):
     return payload
 
 
-def _build_resource_collision_payload(user_id, body):
-    return {
+def _build_resource_collision_payload(user_id, body, schema_url=None):
+    payload = {
         "resource_logs": [
             {
                 "resource": {
@@ -249,6 +249,11 @@ def _build_resource_collision_payload(user_id, 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":
@@ -600,3 +605,43 @@ def test_out_kafka_otlp_logs_preserve_resources_across_requests_in_same_chunk(
     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