Verify errors intake crash messages

Author: gyuheon0h

src/native/Cargo.lock

@@ -16,21 +16,21 @@ profiling = ["dep:datadog-profiling-ffi"]
 
 [dependencies]
 anyhow = { version = "1.0", optional = true }
-datadog-crashtracker = { git = "https://github.com/DataDog/libdatadog", rev = "f4c0bdfbd620cf4a627d852e8c31eda2b0057cd7", optional = true }
-datadog-ddsketch = { git = "https://github.com/DataDog/libdatadog", rev = "f4c0bdfbd620cf4a627d852e8c31eda2b0057cd7" }
-datadog-library-config = { git = "https://github.com/DataDog/libdatadog", rev = "f4c0bdfbd620cf4a627d852e8c31eda2b0057cd7" }
-datadog-log = { git = "https://github.com/DataDog/libdatadog", rev = "f4c0bdfbd620cf4a627d852e8c31eda2b0057cd7" }
-data-pipeline = { git = "https://github.com/DataDog/libdatadog", rev = "f4c0bdfbd620cf4a627d852e8c31eda2b0057cd7" }
-datadog-profiling-ffi = { git = "https://github.com/DataDog/libdatadog", rev = "f4c0bdfbd620cf4a627d852e8c31eda2b0057cd7", optional = true, features = [
+datadog-crashtracker = { git = "https://github.com/DataDog/libdatadog", rev = "309daec4ef976e3c74c684732aac0e6f18e5e1a4", optional = true }
+datadog-ddsketch = { git = "https://github.com/DataDog/libdatadog", rev = "309daec4ef976e3c74c684732aac0e6f18e5e1a4" }
+datadog-library-config = { git = "https://github.com/DataDog/libdatadog", rev = "309daec4ef976e3c74c684732aac0e6f18e5e1a4" }
+datadog-log = { git = "https://github.com/DataDog/libdatadog", rev = "309daec4ef976e3c74c684732aac0e6f18e5e1a4" }
+data-pipeline = { git = "https://github.com/DataDog/libdatadog", rev = "309daec4ef976e3c74c684732aac0e6f18e5e1a4" }
+datadog-profiling-ffi = { git = "https://github.com/DataDog/libdatadog", rev = "309daec4ef976e3c74c684732aac0e6f18e5e1a4", optional = true, features = [
     "cbindgen",
 ] }
-ddcommon = { git = "https://github.com/DataDog/libdatadog", rev = "f4c0bdfbd620cf4a627d852e8c31eda2b0057cd7" }
+ddcommon = { git = "https://github.com/DataDog/libdatadog", rev = "309daec4ef976e3c74c684732aac0e6f18e5e1a4" }
 pyo3 = { version = "0.25", features = ["extension-module", "anyhow"] }
 tracing = { version = "0.1", default-features = false }
 
 [build-dependencies]
 pyo3-build-config = "0.25"
-build_common = { git = "https://github.com/DataDog/libdatadog", rev = "f4c0bdfbd620cf4a627d852e8c31eda2b0057cd7", features = [
+build_common = { git = "https://github.com/DataDog/libdatadog", rev = "309daec4ef976e3c74c684732aac0e6f18e5e1a4", features = [
     "cbindgen",
 ] }
 

src/native/Cargo.toml

@@ -743,7 +743,7 @@ def test_crashtracker_upload_errors_intake():
     with utils.with_test_agent() as client:
         pid = os.fork()
         if pid == 0:
-            ct = utils.CrashtrackerWrapper(base_name="tags_required")
+            ct = utils.CrashtrackerWrapper(base_name="errors_intake_test")
             assert ct.start()
             stdout_msg, stderr_msg = ct.logs()
             assert not stdout_msg
@@ -752,11 +752,26 @@ def test_crashtracker_upload_errors_intake():
             ctypes.string_at(0)
             sys.exit(-1)
 
-        # Check for crash ping and report
-        _crash_ping = utils.get_crash_ping(client)
-        _crash_report = utils.get_crash_report(client)
-        assert _crash_ping is not None, "Should have a crash ping"
-        assert _crash_report is not None, "Should have a crash report"
-
+        # Check for telemetry crash ping and report
+        _telemetry_crash_ping = utils.get_crash_ping(client)
+        _telemetry_crash_report = utils.get_crash_report(client)
+        assert _telemetry_crash_ping is not None, "Should have a telemetry crash ping"
+        assert _telemetry_crash_report is not None, "Should have a telemetry crash report"
+        # Verify that we have both telemetry and errors intake messages
         all_requests = client.requests()
-        assert len(all_requests) == 4, f"Should have 4 total requests (2 telemetry, 2 errors intake) {all_requests}"
+
+        # Check for errors intake crash ping and report
+        _errors_intake_crash_ping = utils.get_errors_intake_crash_ping(client)
+        _errors_intake_crash_report = utils.get_errors_intake_crash_report(client)
+        assert _errors_intake_crash_ping is not None, "Should have an errors intake crash ping"
+        assert _errors_intake_crash_report is not None, "Should have an errors intake crash report"
+
+        telemetry_requests = client.crash_messages()
+        errors_intake_requests = client.error_intake_messages()
+
+        assert len(
+            telemetry_requests) == 2, f"Should have 2 telemetry crash messages (ping + report), got {len(telemetry_requests)}"
+        assert len(
+            errors_intake_requests) == 2, f"Should have 2 errors intake messages (ping + report), got {len(errors_intake_requests)}"
+        assert len(
+            all_requests) == 4, f"Should have 4 total requests (2 telemetry, 2 errors intake), got {len(all_requests)}"

tests/internal/crashtracker/test_crashtracker.py

@@ -174,6 +174,50 @@ def get_crash_ping(test_agent_client: TestAgentClient) -> TestAgentRequest:
     return crash_ping
 
 
+def get_errors_intake_crash_report(test_agent_client: TestAgentClient) -> TestAgentRequest:
+    """Wait for and retrieve a crash report from the errors intake endpoint."""
+    for _ in range(5):
+        errors_messages = test_agent_client.error_intake_messages()
+        for message in errors_messages:
+            body = message.get("body", b"")
+            if isinstance(body, str):
+                body = body.encode("utf-8")
+            try:
+                if isinstance(message.get("body"), bytes):
+                    data = json.loads(message["body"].decode("utf-8"))
+                else:
+                    data = json.loads(message["body"])
+                if "is_crash:true" in data.get("ddtags", ""):
+                    return message
+            except (json.JSONDecodeError, KeyError, TypeError):
+                continue
+        time.sleep(0.2)
+
+    raise AssertionError("Could not find errors intake crash report with 'is_crash: true'")
+
+
+def get_errors_intake_crash_ping(test_agent_client: TestAgentClient) -> TestAgentRequest:
+    """Wait for and retrieve a crash ping from the errors intake endpoint."""
+    for _ in range(5):
+        errors_messages = test_agent_client.error_intake_messages()
+        for message in errors_messages:
+            body = message.get("body", b"")
+            if isinstance(body, str):
+                body = body.encode("utf-8")
+            try:
+                if isinstance(message.get("body"), bytes):
+                    data = json.loads(message["body"].decode("utf-8"))
+                else:
+                    data = json.loads(message["body"])
+                if "is_crash_ping:true" in data.get("ddtags", ""):
+                    return message
+            except (json.JSONDecodeError, KeyError, TypeError):
+                continue
+        time.sleep(0.2)
+
+    raise AssertionError("Could not find errors intake crash ping with 'is_crash: false' and 'is_crash_ping:true' tag")
+
+
 @contextmanager
 def with_test_agent() -> Generator[TestAgentClient, None, None]:
     base_url = ddtrace.tracer.agent_trace_url or "http://localhost:9126"  # default to local test agent

tests/internal/crashtracker/utils.py

@@ -1180,6 +1180,15 @@ def crash_messages(self) -> List[TestAgentRequest]:
             reqs.append(req)
         return reqs
 
+    def error_intake_messages(self) -> List[TestAgentRequest]:
+        reqs = []
+        for req in self.requests():
+            # Filter requests to the errors intake endpoint
+            if req["url"].endswith("/evp_proxy/v4/api/v2/errorsintake"):
+                req["body"] = base64.b64decode(req["body"])
+                reqs.append(req)
+        return reqs
+
     def clear(self) -> None:
         status, body = self._request("GET", self._url("/test/session/clear"))
         assert status == 200, (