Handle v1 payload output from trace-agent

tests/apm_tracing_e2e/test_process_tags.py

@@ -18,10 +18,10 @@ def setup_tracing_process_tags(self):
     def test_tracing_process_tags(self):
         # Get all the spans from the agent
         found = False
-        for data, _ in interfaces.agent.get_spans(self.req):
+        for data, _ in interfaces.agent.get_chunks_v1(self.req):
             # Check that the agent managed to extract the process tags from the first chunk
-            for payload in data["request"]["content"]["tracerPayloads"]:
-                process_tags = payload["tags"]["_dd.tags.process"]
+            for payload in data["request"]["content"]["idxTracerPayloads"]:
+                process_tags = payload["attributes"]["_dd.tags.process"]
                 validate_process_tags(process_tags)
                 found = True
         assert found, "Process tags are missing"

tests/integrations/test_open_telemetry.py

@@ -20,21 +20,21 @@ def test_properties(self):
             assert span is not None, f"Span is not found for {db_operation}"
 
             # DEPRECATED!! Now it is db.instance. The name of the database being connected to. Database instance name.
-            assert span["meta"]["db.name"] == db_container.db_instance
+            assert span["attributes"]["db.name"] == db_container.db_instance
 
             # Describes the relationship between the Span, its parents, and its children in a Trace.
-            assert span["meta"]["span.kind"] == "client"
+            assert span["kind"] == "SPAN_KIND_CLIENT"
 
             # An identifier for the database management system (DBMS) product being used. Formerly db.type
             # Must be one of the available values:
             # https://datadoghq.atlassian.net/wiki/spaces/APM/pages/2357395856/Span+attributes#db.system
-            assert span["meta"]["db.system"] == self.db_service
+            assert span["attributes"]["db.system"] == self.db_service
 
             # Username for accessing the database.
-            assert span["meta"]["db.user"].casefold() == db_container.db_user.casefold()
+            assert span["attributes"]["db.user"].casefold() == db_container.db_user.casefold()
 
             # The database password should not show in the traces
-            for key in span["meta"]:
+            for key in span["attributes"]:
                 if key not in [
                     "peer.hostname",
                     "db.user",
@@ -46,20 +46,20 @@ def test_properties(self):
                     "net.peer.name",
                     "server.address",
                 ]:  # These fields hostname, user... are the same as password
-                    assert span["meta"][key] != db_container.db_password, f"Test is failing for {db_operation}"
+                    assert span["attributes"][key] != db_container.db_password, f"Test is failing for {db_operation}"
 
     def test_resource(self):
         """Usually the query"""
         for db_operation, request in self.get_requests(excluded_operations=["procedure", "select_error"]):
             span = self.get_span_from_agent(request)
-            assert db_operation in span["resource"].lower()
+            assert db_operation in span["resourceRef"].lower()
 
     @missing_feature(library="python_otel", reason="Open telemetry doesn't send this span for python")
     def test_db_connection_string(self):
         """The connection string used to connect to the database."""
         for db_operation, request in self.get_requests():
             span = self.get_span_from_agent(request)
-            assert span["meta"]["db.connection_string"].strip(), f"Test is failing for {db_operation}"
+            assert span["attributes"]["db.connection_string"].strip(), f"Test is failing for {db_operation}"
 
     @missing_feature(library="python_otel", reason="Open Telemetry doesn't send this span for python but it should do")
     @missing_feature(library="nodejs_otel", reason="Open Telemetry doesn't send this span for nodejs but it should do")
@@ -69,11 +69,11 @@ def test_db_operation(self):
             span = self.get_span_from_agent(request)
 
             if db_operation == "procedure":
-                assert any(substring in span["meta"]["db.operation"].lower() for substring in ["call", "exec"]), (
+                assert any(substring in span["attributes"]["db.operation"].lower() for substring in ["call", "exec"]), (
                     "db.operation span not found for procedure operation"
                 )
             else:
-                assert db_operation.lower() in span["meta"]["db.operation"].lower(), (
+                assert db_operation.lower() in span["attributes"]["db.operation"].lower(), (
                     f"Test is failing for {db_operation}"
                 )
 
@@ -85,22 +85,22 @@ def test_db_sql_table(self):
         """The name of the primary table that the operation is acting upon, including the database name (if applicable)."""
         for db_operation, request in self.get_requests(excluded_operations=["procedure"]):
             span = self.get_span_from_agent(request)
-            assert span["meta"]["db.sql.table"].strip(), f"Test is failing for {db_operation}"
+            assert span["attributes"]["db.sql.table"].strip(), f"Test is failing for {db_operation}"
 
     def test_error_message(self):
         """A string representing the error message."""
         span = self.get_span_from_agent(self.requests[self.db_service]["select_error"])
-        assert len(span["meta"]["error.msg"].strip()) != 0
+        assert len(span["attributes"]["error.msg"].strip()) != 0
 
     @missing_feature(library="nodejs_otel", reason="Open telemetry with nodejs is not generating this information.")
     def test_error_type_and_stack(self):
         span = self.get_span_from_agent(self.requests[self.db_service]["select_error"])
 
         # A string representing the type of the error
-        assert span["meta"]["error.type"].strip()
+        assert span["attributes"]["error.type"].strip()
 
         # A human readable version of the stack trace
-        assert span["meta"]["error.stack"].strip()
+        assert span["attributes"]["error.stack"].strip()
 
     @bug(library="python_otel", reason="OTEL-940")
     @bug(library="nodejs_otel", reason="OTEL-940")
@@ -110,11 +110,11 @@ def test_obfuscate_query(self):
         for db_operation, request in self.get_requests():
             span = self.get_span_from_agent(request)
             if db_operation in ["update", "delete", "procedure", "select_error", "select"]:
-                assert span["meta"]["db.statement"].count("?") == 2, (
+                assert span["attributes"]["db.statement"].count("?") == 2, (
                     f"The query is not properly obfuscated for operation {db_operation}"
                 )
             else:
-                assert span["meta"]["db.statement"].count("?") == 3, (
+                assert span["attributes"]["db.statement"].count("?") == 3, (
                     f"The query is not properly obfuscated for operation {db_operation}"
                 )
 
@@ -128,8 +128,8 @@ def test_db_statement_query(self):
         """Usually the query"""
         for db_operation, request in self.get_requests(excluded_operations=["procedure", "select_error"]):
             span = self.get_span_from_agent(request)
-            assert db_operation in span["meta"]["db.statement"].lower(), (
-                f"{db_operation}  not found in {span['meta']['db.statement']}"
+            assert db_operation in span["attributes"]["db.statement"].lower(), (
+                f"{db_operation}  not found in {span['attributes']['db.statement']}"
             )
 
 

tests/integrations/utils.py

@@ -13,7 +13,7 @@ class BaseDbIntegrationsTestClass:
 
     def _setup(self):
         """Make request to weblog for each operation: select, update...
-        those requests will be permored only one time for the entire test run
+        those requests will be performed only one time for the entire test run
         """
 
         assert self.db_service is not None, "db_service must be defined"
@@ -74,6 +74,7 @@ def _setup(self):
     def get_requests(
         self, excluded_operations: Sequence[str] = (), operations: Sequence[str] | None = None
     ) -> Generator[tuple[str, HttpResponse], None, None]:
+        logger.debug(f"Getting requests for {self.db_service} from {self.requests.keys()}")
         for db_operation, request in self.requests[self.db_service].items():
             if operations is not None and db_operation not in operations:
                 continue
@@ -108,40 +109,37 @@ def get_span_from_tracer(weblog_request: HttpResponse) -> dict:
 
     @staticmethod
     def get_span_from_agent(weblog_request: HttpResponse) -> dict:
-        for data, span in interfaces.agent.get_spans(weblog_request):
-            logger.debug(f"Span found: trace id={span['traceID']}; span id={span['spanID']} ({data['log_filename']})")
+        for data, chunk in interfaces.agent.get_chunks_v1(weblog_request):
+            logger.debug(f"Chunk found: trace id={chunk['traceID']}; ({data['log_filename']})")
 
             # iterate over everything to be sure to miss nothing
-            for _, span_child in interfaces.agent.get_spans():
-                if span_child["traceID"] != span["traceID"]:
-                    continue
-
-                logger.debug(f"Checking if span {span_child['spanID']} could match")
+            for span in chunk["spans"]:
+                logger.debug(f"Checking if span {span['spanID']} could match")
 
-                if span_child.get("type") not in ("sql", "db"):
-                    logger.debug(f"Wrong type:{span_child.get('type')}, continue...")
+                if span.get("typeRef") not in ("sql", "db"):
+                    logger.debug(f"Wrong type:{span.get('typeRef')}, continue...")
                     # no way it's the span we're looking for
                     continue
 
                 # workaround to avoid conflicts on connection check on mssql
                 # workaround to avoid conflicts on connection check on mssql + nodejs + opentelemetry (there is a bug in the sql obfuscation)
-                if span_child["resource"] in ("SELECT ?", "SELECT 1;"):
-                    logger.debug(f"Wrong resource:{span_child.get('resource')}, continue...")
+                if span.get("resourceRef") in ("SELECT ?", "SELECT 1;"):
+                    logger.debug(f"Wrong resource:{span.get('resourceRef')}, continue...")
                     continue
 
                 # workaround to avoid conflicts on postgres + nodejs + opentelemetry
-                if span_child["name"] == "pg.connect":
-                    logger.debug(f"Wrong name:{span_child.get('name')}, continue...")
+                if span.get("nameRef") == "pg.connect":
+                    logger.debug(f"Wrong name:{span.get('nameRef')}, continue...")
                     continue
 
                 # workaround to avoid conflicts on mssql + nodejs + opentelemetry
-                if span_child["meta"].get("db.statement") == "SELECT 1;":
-                    logger.debug(f"Wrong db.statement:{span_child.get('meta', {}).get('db.statement')}, continue...")
+                if span.get("attributes", {}).get("db.statement") == "SELECT 1;":
+                    logger.debug(f"Wrong db.statement:{span.get('attributes', {}).get('db.statement')}, continue...")
                     continue
 
-                logger.info(f"Span type==sql found: spanId={span_child['spanID']}")
+                logger.info(f"Span type==sql found: spanId={span['spanID']}")
 
-                return span_child
+                return span
 
         raise ValueError(f"Span is not found for {weblog_request.request.url}")
 

tests/test_config_consistency.py

@@ -43,11 +43,14 @@ def test_status_code_400(self):
         assert self.r.status_code == 400
 
         interfaces.library.assert_trace_exists(self.r)
-        spans = interfaces.agent.get_spans_list(self.r)
+        chunks = interfaces.agent.get_chunks_v1(self.r)
+        chunks = [chunk for _, chunk in chunks]
+        assert len(chunks) == 1, "Agent received the incorrect amount of chunks"
+        spans = chunks[0]["spans"]
         assert len(spans) == 1, "Agent received the incorrect amount of spans"
 
-        assert spans[0]["type"] == "web"
-        assert spans[0]["meta"]["http.status_code"] == "400"
+        assert spans[0]["typeRef"] == "web"
+        assert spans[0]["attributes"]["http.status_code"] == "400"
         assert "error" not in spans[0] or spans[0]["error"] == 0
 
     def setup_status_code_500(self):
@@ -57,11 +60,14 @@ def test_status_code_500(self):
         assert self.r.status_code == 500
 
         interfaces.library.assert_trace_exists(self.r)
-        spans = interfaces.agent.get_spans_list(self.r)
+        chunks = interfaces.agent.get_chunks_v1(self.r)
+        chunks = [chunk for _, chunk in chunks]
+        assert len(chunks) == 1, "Agent received the incorrect amount of chunks"
+        spans = chunks[0]["spans"]
         assert len(spans) == 1, "Agent received the incorrect amount of spans"
 
-        assert spans[0]["meta"]["http.status_code"] == "500"
-        assert spans[0]["error"] == 1
+        assert spans[0]["attributes"]["http.status_code"] == "500"
+        assert spans[0]["error"]
 
 
 @scenarios.tracing_config_nondefault
@@ -76,12 +82,15 @@ def test_status_code_200(self):
         assert self.r.status_code == 200
 
         interfaces.library.assert_trace_exists(self.r)
-        spans = interfaces.agent.get_spans_list(self.r)
+        chunks = interfaces.agent.get_chunks_v1(self.r)
+        chunks = [chunk for _, chunk in chunks]
+        assert len(chunks) == 1, "Agent received the incorrect amount of chunks"
+        spans = chunks[0]["spans"]
         assert len(spans) == 1, "Agent received the incorrect amount of spans"
 
-        assert spans[0]["type"] == "web"
-        assert spans[0]["meta"]["http.status_code"] == "200"
-        assert spans[0]["error"] == 1
+        assert spans[0]["typeRef"] == "web"
+        assert spans[0]["attributes"]["http.status_code"] == "200"
+        assert spans[0]["error"]
 
     def setup_status_code_202(self):
         self.r = weblog.get("/status?code=202")
@@ -90,12 +99,15 @@ def test_status_code_202(self):
         assert self.r.status_code == 202
 
         interfaces.library.assert_trace_exists(self.r)
-        spans = interfaces.agent.get_spans_list(self.r)
+        chunks = interfaces.agent.get_chunks_v1(self.r)
+        chunks = [chunk for _, chunk in chunks]
+        assert len(chunks) == 1, "Agent received the incorrect amount of chunks"
+        spans = chunks[0]["spans"]
         assert len(spans) == 1, "Agent received the incorrect amount of spans"
 
-        assert spans[0]["type"] == "web"
-        assert spans[0]["meta"]["http.status_code"] == "202"
-        assert spans[0]["error"] == 1
+        assert spans[0]["typeRef"] == "web"
+        assert spans[0]["attributes"]["http.status_code"] == "202"
+        assert spans[0]["error"]
 
 
 # Tests for verifying default query string obfuscation behavior can be found in the Test_StandardTagsUrl test class
@@ -421,9 +433,12 @@ def setup_specified_service_name(self):
     )
     def test_specified_service_name(self):
         interfaces.library.assert_trace_exists(self.r)
-        spans = interfaces.agent.get_spans_list(self.r)
+        chunks = interfaces.agent.get_chunks_v1(self.r)
+        chunks = [chunk for _, chunk in chunks]
+        assert len(chunks) == 1, "Agent received the incorrect amount of chunks"
+        spans = chunks[0]["spans"]
         assert len(spans) == 1, "Agent received the incorrect amount of spans"
-        assert spans[0]["service"] == "service_test"
+        assert spans[0]["serviceRef"] == "service_test"
 
 
 @scenarios.default
@@ -436,10 +451,13 @@ def setup_default_service_name(self):
 
     def test_default_service_name(self):
         interfaces.library.assert_trace_exists(self.r)
-        spans = interfaces.agent.get_spans_list(self.r)
+        chunks = interfaces.agent.get_chunks_v1(self.r)
+        chunks = [chunk for _, chunk in chunks]
+        assert len(chunks) == 1, "Agent received the incorrect amount of chunks"
+        spans = chunks[0]["spans"]
         assert len(spans) == 1, "Agent received the incorrect amount of spans"
         assert (
-            spans[0]["service"] != "service_test"
+            spans[0]["serviceRef"] != "service_test"
         )  # in default scenario, DD_SERVICE is set to "weblog" in the dockerfile; this is a temp fix to test that it is not the value we manually set in the specific scenario
 
 

tests/test_data_integrity.py

@@ -246,13 +246,17 @@ def test_agent_do_not_drop_traces(self):
 
         # get list of trace ids reported by the agent
         trace_ids_reported_by_agent = set()
-        for _, span in interfaces.agent.get_spans():
-            trace_ids_reported_by_agent.add(int(span["traceID"]))
+        for _, chunk in interfaces.agent.get_chunks_v1():
+            # the chunk TraceID is a hex encoded string like "0x69274AA50000000068F1C3D5F2D1A9B0"
+            # We need to convert it to an integer taking only the lower 64 bits
+            # Note that this ignores the upper 64 bits, but this is fine for just verifying that the trace is reported for our test
+            trace_id = int(chunk["traceID"], 16) & 0xFFFFFFFFFFFFFFFF
+            trace_ids_reported_by_agent.add(trace_id)
 
         def get_span_with_sampling_data(trace: list):
             # The root span is not necessarily the span wherein the sampling priority can be found.
             # If present, the root will take precedence, and otherwise the first span with the
-            # sampling priority tag will be returned. This isthe same logic found on the trace-agent.
+            # sampling priority tag will be returned. This is the same logic found on the trace-agent.
             span_with_sampling_data = None
             for span in trace:
                 if span.get("metrics", {}).get("_sampling_priority_v1", None) is not None:

tests/test_telemetry.py

@@ -692,12 +692,12 @@ def test_traces_contain_install_id(self):
         """Assert that at least one trace carries APM onboarding info"""
 
         def validate_at_least_one_span_with_tag(tag: str):
-            for _, span in interfaces.agent.get_spans():
-                meta = span.get("meta", {})
-                if tag in meta:
-                    break
-            else:
-                raise Exception(f"Did not find tag {tag} in any spans")
+            for _, chunk in interfaces.agent.get_chunks_v1():
+                for span in chunk.get("spans", []):
+                    attributes = span.get("attributes", {})
+                    if tag in attributes:
+                        return
+            raise Exception(f"Did not find tag {tag} in any spans")
 
         validate_at_least_one_span_with_tag("_dd.install.id")
         validate_at_least_one_span_with_tag("_dd.install.time")

utils/_context/_scenarios/open_telemetry.py

@@ -134,7 +134,24 @@ def _wait_for_app_readiness(self):
             logger.debug("Open telemetry ready")
 
     def post_setup(self, session: pytest.Session):  # noqa: ARG002
-        if self.use_proxy:
+        if self.replay:
+            logger.terminal.write(
+                "\nReplay mode is not fully functional for this scenario, you may encounter errors\n",
+                bold=True,
+                red=True,
+            )
+            logger.terminal.write_sep("-", "Load all data from logs")
+            logger.terminal.flush()
+
+            interfaces.open_telemetry.load_data_from_logs()
+            interfaces.open_telemetry.check_deserialization_errors()
+
+            if self.include_agent:
+                interfaces.agent.load_data_from_logs()
+                interfaces.agent.check_deserialization_errors()
+
+            interfaces.backend.load_data_from_logs()
+        elif self.use_proxy:
             self._wait_interface(interfaces.open_telemetry, 5)
             self._wait_interface(interfaces.backend, self.backend_interface_timeout)