tests: integration: Add an integration test for reamining metadata on upstream

Signed-off-by: Hiroshi Hatake <hiroshi@chronosphere.io>

Author: cosmo0920

tests/integration/scenarios/in_forward/config/in_opentelemetry_to_forward_receiver_retain_metadata.yaml

@@ -0,0 +1,22 @@
+service:
+  flush: 1
+  log_level: info
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+
+pipeline:
+  inputs:
+    - name: opentelemetry
+      port: ${FLUENT_BIT_TEST_LISTENER_PORT}
+      http2: off
+      tls: off
+      profiles_support: true
+
+  outputs:
+    - name: forward
+      match: '*'
+      host: 127.0.0.1
+      port: ${FORWARD_RECEIVER_PORT}
+      send_options: true
+      retain_metadata_in_forward_mode: true
+      require_ack_response: true

tests/integration/scenarios/in_forward/tests/test_in_forward_001.py

@@ -924,6 +924,50 @@ def test_out_forward_logs_signal_e2e_with_forward_receiver():
     assert "This is another example log message." in log_bodies
 
 
+def test_out_forward_logs_signal_e2e_with_forward_receiver_retain_metadata():
+    service = ForwardReceiverService(
+        "in_opentelemetry_to_forward_receiver_retain_metadata.yaml"
+    )
+    service.start()
+
+    try:
+        service.send_json_as_otel_protobuf("test_logs_001.in.json", "logs")
+        messages = service.wait_for_forward_messages(1)
+    finally:
+        service.stop()
+
+    message = messages[0]
+    _assert_forward_signal_message(message, expected_tag="v1_logs", expected_signal=0)
+    assert len(message["records"]) > 0
+
+    schema_metadata_count = 0
+    otlp_metadata_count = 0
+    empty_metadata_count = 0
+
+    for record in message["records"]:
+        metadata = record["metadata"]
+
+        assert isinstance(metadata, dict)
+
+        if "schema" in metadata:
+            assert metadata["schema"] == "otlp"
+            assert metadata["resource_id"] in (0, 1)
+            assert metadata["scope_id"] in (0, 1)
+            schema_metadata_count += 1
+        elif "otlp" in metadata:
+            assert metadata["otlp"]["severity_number"] == 9
+            assert metadata["otlp"]["severity_text"] == "INFO"
+            assert metadata["otlp"]["attributes"]["example_key"] == "example_value"
+            otlp_metadata_count += 1
+        else:
+            assert metadata == {}
+            empty_metadata_count += 1
+
+    assert schema_metadata_count > 0
+    assert otlp_metadata_count > 0
+    assert empty_metadata_count > 0
+
+
 def test_out_forward_metrics_signal_e2e_with_forward_receiver():
     service = ForwardReceiverService("in_opentelemetry_to_forward_receiver.yaml")
     service.start()