Conservative with checks we make while dual uploads in dev

Author: gyuheon0h

src/native/data_pipeline/mod.rs

@@ -148,11 +148,14 @@ impl TraceExporterBuilderPy {
         heartbeat_ms: u64,
         runtime_id: String,
     ) -> PyResult<Py<Self>> {
-        slf.try_as_mut()?.enable_telemetry(Some(TelemetryConfig {
-            heartbeat: heartbeat_ms,
-            runtime_id: Some(runtime_id),
-            debug_enabled: true,
-        }));
+        slf.try_as_mut()?.enable_telemetry(
+            Some(TelemetryConfig {
+                heartbeat: heartbeat_ms,
+                runtime_id: Some(runtime_id),
+                debug_enabled: true,
+            })
+            .unwrap(),
+        );
         Ok(slf.into())
     }
 

src/native/library_config.rs

@@ -118,6 +118,8 @@ pub fn store_metadata(data: &PyTracerMetadata) -> PyResult<PyAnonymousFileHandle
         service_name: data.service_name.clone(),
         service_env: data.service_env.clone(),
         service_version: data.service_version.clone(),
+        container_id: None,
+        process_tags: None,
     };
 
     let res = store_tracer_metadata(&metadata);

tests/internal/crashtracker/test_crashtracker.py

@@ -735,19 +735,15 @@ def test_crashtracker_no_zombies():
 @pytest.mark.skipif(not sys.platform.startswith("linux"), reason="Linux only")
 @pytest.mark.subprocess()
 def test_crashtracker_upload_errors_intake():
-    """Test specifically for errors intake message structure and content."""
     import ctypes
-    import json
     import os
-    import sys
-    import base64
 
     import tests.internal.crashtracker.utils as utils
 
     with utils.with_test_agent() as client:
         pid = os.fork()
         if pid == 0:
-            ct = utils.CrashtrackerWrapper(base_name="errors_verification")
+            ct = utils.CrashtrackerWrapper(base_name="tags_required")
             assert ct.start()
             stdout_msg, stderr_msg = ct.logs()
             assert not stdout_msg
@@ -756,14 +752,11 @@ def test_crashtracker_upload_errors_intake():
             ctypes.string_at(0)
             sys.exit(-1)
 
-        all_requests = client.requests()
-        errors_intake_messages = utils.get_all_errors_intake_messages(client)
-        assert len(errors_intake_messages) == 2, "Should have 2 errors intake messages"
-        assert len(all_requests) == 4, "Should have 4 total requests (2 telemetry, 2 errors intake)"
+        # 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"
 
-        errors_intake_report = utils.get_errors_intake_report(client)
-        errors_intake_ping = utils.get_errors_intake_ping(client)
-        assert errors_intake_report is not None, "Should have an errors intake report"
-        assert errors_intake_ping is not None, "Should have an errors intake ping"
-
-        
+        all_requests = client.requests()
+        assert len(all_requests) == 4, f"Should have 4 total requests (2 telemetry, 2 errors intake) {all_requests}"

tests/internal/crashtracker/utils.py

@@ -141,26 +141,6 @@ def get_all_crash_messages(test_agent_client: TestAgentClient) -> List[TestAgent
     return crash_messages
 
 
-def get_all_errors_intake_messages(test_agent_client: TestAgentClient) -> List[TestAgentRequest]:
-    """Wait for all errors intake messages from the crashtracker listener socket."""
-    seen_errors_intake_ids = set()
-    errors_intake_messages = []
-    for _ in range(5):
-        incoming_messages = test_agent_client.errors_intake_messages()
-        for message in incoming_messages:
-            body = message.get("body", b"")
-            if isinstance(body, str):
-                body = body.encode("utf-8")
-            errors_intake_id = (hash(body), frozenset(message.get("headers", {}).items()))
-            if errors_intake_id not in seen_errors_intake_ids:
-                seen_errors_intake_ids.add(errors_intake_id)
-                errors_intake_messages.append(message)
-            if len(errors_intake_messages) >= 2:
-                return errors_intake_messages
-        time.sleep(0.2)
-    return errors_intake_messages
-
-
 def get_crash_report(test_agent_client: TestAgentClient) -> TestAgentRequest:
     """Wait for a crash report from the crashtracker listener socket."""
     crash_messages = get_all_crash_messages(test_agent_client)
@@ -194,32 +174,6 @@ def get_crash_ping(test_agent_client: TestAgentClient) -> TestAgentRequest:
     return crash_ping
 
 
-def get_errors_intake_report(test_agent_client: TestAgentClient) -> TestAgentRequest:
-    """Wait for an errors intake message from the crashtracker listener socket."""
-    errors_intake_messages = get_all_errors_intake_messages(test_agent_client)
-    assert len(errors_intake_messages) == 2, f"Expected 2 errors intake messages; got {len(errors_intake_messages)}"
-    errors_intake_report = None
-    for message in errors_intake_messages:
-        if b"is_crash:true" in message["body"]:
-            errors_intake_report = message
-            break
-    assert errors_intake_report is not None, "Could not find errors intake report with 'is_crash:true' tag"
-    return errors_intake_report
-
-
-def get_errors_intake_ping(test_agent_client: TestAgentClient) -> TestAgentRequest:
-    """Wait for an errors intake ping from the crashtracker listener socket."""
-    errors_intake_messages = get_all_errors_intake_messages(test_agent_client)
-    assert len(errors_intake_messages) == 2, f"Expected 2 errors intake messages; got {len(errors_intake_messages)}"
-    errors_intake_ping = None
-    for message in errors_intake_messages:
-        if b"is_crash_ping:true" in message["body"]:
-            errors_intake_ping = message
-            break
-    assert errors_intake_ping is not None, "Could not find errors intake ping with 'is_crash_ping:true' tag"
-    return errors_intake_ping
-
-
 @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/utils.py

@@ -1180,36 +1180,6 @@ def crash_messages(self) -> List[TestAgentRequest]:
             reqs.append(req)
         return reqs
 
-    def errors_intake_messages(self) -> List[TestAgentRequest]:
-        reqs = []
-        for req in self.requests():
-            url = req.get("url", "")
-            if not (url.endswith("/errorsintake") or "/api/v2/errorsintake" in url or "/evp_proxy/v4/api/v2/errorsintake" in url):
-                continue
-
-            # Check for errors intake specific headers
-            headers = req.get("headers", {})
-            content_type = headers.get("content-type", "")
-            if "application/json" not in content_type:
-                continue
-
-            try:
-                if isinstance(req["body"], str):
-                    data = json.loads(req["body"])
-                else:
-                    data = json.loads(base64.b64decode(req["body"]))
-
-                if data.get("ddsource") != "crashtracker":
-                    continue
-
-                if not isinstance(req["body"], str):
-                    req["body"] = base64.b64decode(req["body"])
-
-                reqs.append(req)
-            except Exception:
-                continue
-        return reqs
-
     def clear(self) -> None:
         status, body = self._request("GET", self._url("/test/session/clear"))
         assert status == 200, (