feat(profiling): surface native-heap partition arming via dogstatsd gauge + warning

The native-heap ownership-partition arming decision was only observable via a
one-shot stdlib INFO log, which real deploys drop (structured JSON loggers
filter ddtrace stdlib records) and pods/exec is RBAC-blocked, so armed=true|false
could not be confirmed at runtime.

Surface the decision two ways at service start (gated on native-heap being
enabled, so the default majority of processes stay silent):
- emit a dogstatsd gauge profiling.native_heap.partition_armed (1=armed, 0=not)
  via the canonical internal client; fail-safe if the client is unavailable
- upgrade the existing INFO arming line to WARNING (same fields)

Author: vlad-scherbich

ddtrace/profiling/profiler.py

@@ -34,6 +34,16 @@
 
 LOG = logging.getLogger(__name__)
 
+# pymalloc's small-request threshold (see CPython Objects/obmalloc.c). Requests
+# larger than this are delegated to glibc malloc, where the native-heap gotter
+# owns them, so the in-process sampler skips them when the partition is armed.
+_NATIVE_HEAP_SIZE_THRESHOLD_BYTES = 512
+# Dogstatsd gauge surfacing the native-heap ownership-partition arming decision
+# (1=armed, 0=not). Structured JSON loggers in real deploys drop ddtrace stdlib
+# records and pods/exec is RBAC-blocked, so the log line alone is not always
+# observable; this gauge gives an out-of-band, queryable signal at service start.
+_NATIVE_HEAP_PARTITION_ARMED_METRIC = "profiling.native_heap.partition_armed"
+
 
 class Profiler(object):
     """Run profiling while code is executed.
@@ -406,6 +416,30 @@ def _arm_native_heap(self) -> bool:
             LOG.debug("Failed to arm native heap profiling", exc_info=True)
         return False
 
+    def _emit_native_heap_partition_armed_gauge(self, armed: bool) -> None:
+        """Emit a one-shot dogstatsd gauge reporting the arming decision.
+
+        Value is 1 when the native-heap ownership partition is armed and 0
+        otherwise. Uses the same internal dogstatsd client and agent URL the
+        rest of ddtrace uses. Fail-safe: a missing or broken dogstatsd client
+        must never break profiler startup, so any error is swallowed.
+        """
+        try:
+            from ddtrace.internal.dogstatsd import get_dogstatsd_client
+            from ddtrace.internal.settings._agent import config as agent_config
+
+            client = get_dogstatsd_client(agent_config.dogstatsd_url)
+            client.gauge(
+                _NATIVE_HEAP_PARTITION_ARMED_METRIC,
+                1 if armed else 0,
+                tags=[
+                    "domains:OBJ_MEM",
+                    "size_threshold_bytes:%d" % _NATIVE_HEAP_SIZE_THRESHOLD_BYTES,
+                ],
+            )
+        except Exception:
+            LOG.debug("Failed to emit native heap partition arming gauge", exc_info=True)
+
     def _start_service(self) -> None:
         """Start the profiler."""
         native_heap_armed: bool = self._arm_native_heap()
@@ -440,17 +474,25 @@ def _start_service(self) -> None:
         # gotter did not arm, the partition is off and the in-process sampler keeps
         # sampling all sizes (no behavior change).
         memalloc.set_native_heap_partition(native_heap_armed)
-        # One-shot, unambiguous startup line reporting the arming/partition
-        # decision so it can be verified from pod logs without DEBUG. Emitted
-        # exactly once as the profiler starts (never per-sample). When
-        # armed=false the size split is inactive and the in-process sampler
-        # keeps sampling all sizes; the threshold is still reported as the
-        # configured value for clarity.
-        LOG.info(
-            "native heap ownership partition: armed=%s size_threshold_bytes=%d domains=OBJ|MEM",
-            native_heap_armed,
-            512,
-        )
+        # Surface the arming/partition decision so Phase 2 A/B validation can
+        # confirm armed=true|false at runtime. Gated on the feature being
+        # enabled so we never spam the WARNING/metric for the (default) majority
+        # of processes that don't use native-heap profiling. Emitted exactly
+        # once as the profiler starts (never per-sample). When armed=false the
+        # size split is inactive and the in-process sampler keeps sampling all
+        # sizes; the threshold is still reported as the configured value for
+        # clarity.
+        if profiling_config.native_heap.enabled:
+            # WARNING (not INFO) so it survives deploys whose structured JSON
+            # loggers filter out ddtrace INFO records.
+            LOG.warning(
+                "native heap ownership partition: armed=%s size_threshold_bytes=%d domains=OBJ|MEM",
+                native_heap_armed,
+                _NATIVE_HEAP_SIZE_THRESHOLD_BYTES,
+            )
+            # Also surface the decision as a dogstatsd gauge (1=armed, 0=not) for
+            # the deploys where even WARNING logs are unobservable.
+            self._emit_native_heap_partition_armed_gauge(native_heap_armed)
         if native_heap_armed:
             LOG.debug(
                 "Native heap profiling armed; in-process managed-heap (pymalloc OBJ/MEM) sampling "

tests/profiling/test_native_heap_gotter.py

@@ -41,6 +41,18 @@
     _GOTTER_TEST_HOOK_AVAILABLE = False
 
 
+def _only_arming_warning(err: str) -> bool:
+    """Allow the one-shot native-heap arming WARNING on stderr, nothing else.
+
+    Once native-heap profiling is enabled the arming decision is intentionally
+    logged at WARNING (so it survives prod log filtering), which the subprocess
+    harness would otherwise reject as unexpected stderr. Runs in the parent test
+    process against the subprocess's decoded stderr.
+    """
+    lines = [line for line in err.splitlines() if line.strip()]
+    return all("native heap ownership partition:" in line for line in lines)
+
+
 @pytest.mark.skipif(sys.platform != "linux", reason="native heap gotter is Linux-only")
 @pytest.mark.subprocess
 def test_native_heap_gotter_smoke() -> None:
@@ -72,7 +84,7 @@ def test_native_heap_gotter_smoke() -> None:
         assert len(blobs) == 200
 
 
-@pytest.mark.subprocess(env=dict(DD_PROFILING_ENABLED="true"))
+@pytest.mark.subprocess(env=dict(DD_PROFILING_ENABLED="true"), err=_only_arming_warning)
 def test_profiler_start_arms_native_heap_when_enabled() -> None:
     """Starting the profiler with native heap enabled invokes the activator.
 
@@ -124,7 +136,7 @@ def test_profiler_start_skips_native_heap_when_disabled() -> None:
             prof.stop(flush=False)
 
 
-@pytest.mark.subprocess(env=dict(DD_PROFILING_ENABLED="true"))
+@pytest.mark.subprocess(env=dict(DD_PROFILING_ENABLED="true"), err=_only_arming_warning)
 def test_profiler_start_survives_native_heap_install_error() -> None:
     """A failure while arming native heap profiling must not break the profiler.
 
@@ -150,7 +162,7 @@ def test_profiler_start_survives_native_heap_install_error() -> None:
             prof.stop(flush=False)
 
 
-@pytest.mark.subprocess(env=dict(DD_PROFILING_ENABLED="true"))
+@pytest.mark.subprocess(env=dict(DD_PROFILING_ENABLED="true"), err=_only_arming_warning)
 def test_profiler_keeps_managed_heap_when_native_heap_armed() -> None:
     """Ownership partition (Phase 2 de-dup): the partition is by allocator domain.
 
@@ -192,7 +204,7 @@ def test_profiler_keeps_managed_heap_when_native_heap_armed() -> None:
                     prof.stop(flush=False)
 
 
-@pytest.mark.subprocess(env=dict(DD_PROFILING_ENABLED="true"))
+@pytest.mark.subprocess(env=dict(DD_PROFILING_ENABLED="true"), err=_only_arming_warning)
 def test_profiler_keeps_managed_heap_when_gotter_not_installed() -> None:
     """Fail-safe: if native heap is enabled but the gotter did NOT install
     (install() returned False), the in-process sampler must remain active so
@@ -221,6 +233,103 @@ def test_profiler_keeps_managed_heap_when_gotter_not_installed() -> None:
                 prof.stop(flush=False)
 
 
+@pytest.mark.subprocess(env=dict(DD_PROFILING_ENABLED="true"))
+def test_profiler_start_emits_partition_armed_gauge_when_armed() -> None:
+    """Arming observability: when the gotter installs (armed=True), the profiler
+    emits the ``profiling.native_heap.partition_armed`` gauge with value 1 and
+    logs the arming decision at WARNING level.
+    """
+    from unittest import mock
+
+    from ddtrace.internal.datadog.profiling import heap_gotter
+    import ddtrace.internal.dogstatsd
+    from ddtrace.internal.settings.profiling import config as profiling_config
+    import ddtrace.profiling.profiler as profiler_mod
+
+    profiling_config.native_heap.enabled = True  # pyright: ignore[reportAttributeAccessIssue]
+
+    client = mock.Mock()
+    with mock.patch.object(heap_gotter, "install", return_value=True):
+        with mock.patch.object(heap_gotter, "live_heap_enabled", return_value=False):
+            with mock.patch.object(ddtrace.internal.dogstatsd, "get_dogstatsd_client", return_value=client):
+                with mock.patch.object(profiler_mod.LOG, "warning") as warning:
+                    prof = profiler_mod.Profiler()
+                    prof.start()
+                    try:
+                        assert client.gauge.call_count == 1
+                        args, kwargs = client.gauge.call_args
+                        assert args[0] == "profiling.native_heap.partition_armed"
+                        assert args[1] == 1, "gauge value must be 1 when armed"
+                        tags = kwargs["tags"]
+                        assert "domains:OBJ_MEM" in tags
+                        assert "size_threshold_bytes:512" in tags
+
+                        assert warning.called, "arming decision must be logged at WARNING"
+                        msg = warning.call_args[0][0]
+                        assert "native heap ownership partition" in msg
+                        assert warning.call_args[0][1] is True, "WARNING must report armed=True"
+                    finally:
+                        prof.stop(flush=False)
+
+
+@pytest.mark.subprocess(env=dict(DD_PROFILING_ENABLED="true"))
+def test_profiler_start_emits_partition_armed_gauge_zero_when_not_armed() -> None:
+    """Arming observability: when native heap is enabled but the gotter did NOT
+    install (armed=False), the gauge is still emitted, with value 0.
+    """
+    from unittest import mock
+
+    from ddtrace.internal.datadog.profiling import heap_gotter
+    import ddtrace.internal.dogstatsd
+    from ddtrace.internal.settings.profiling import config as profiling_config
+    import ddtrace.profiling.profiler as profiler_mod
+
+    profiling_config.native_heap.enabled = True  # pyright: ignore[reportAttributeAccessIssue]
+
+    client = mock.Mock()
+    with mock.patch.object(heap_gotter, "install", return_value=False):
+        with mock.patch.object(ddtrace.internal.dogstatsd, "get_dogstatsd_client", return_value=client):
+            with mock.patch.object(profiler_mod.LOG, "warning") as warning:
+                prof = profiler_mod.Profiler()
+                prof.start()
+                try:
+                    assert client.gauge.call_count == 1
+                    args, _ = client.gauge.call_args
+                    assert args[0] == "profiling.native_heap.partition_armed"
+                    assert args[1] == 0, "gauge value must be 0 when not armed"
+                    assert warning.called
+                    assert warning.call_args[0][1] is False, "WARNING must report armed=False"
+                finally:
+                    prof.stop(flush=False)
+
+
+@pytest.mark.subprocess(env=dict(DD_PROFILING_ENABLED="true"), err=_only_arming_warning)
+def test_profiler_start_survives_partition_armed_gauge_error() -> None:
+    """Fail-safe: a broken/unavailable dogstatsd client must never break arming
+    or profiler startup.
+    """
+    from unittest import mock
+
+    from ddtrace.internal.datadog.profiling import heap_gotter
+    import ddtrace.internal.dogstatsd
+    from ddtrace.internal.settings.profiling import config as profiling_config
+    import ddtrace.profiling.profiler as profiler_mod
+
+    profiling_config.native_heap.enabled = True  # pyright: ignore[reportAttributeAccessIssue]
+
+    with mock.patch.object(heap_gotter, "install", return_value=True):
+        with mock.patch.object(heap_gotter, "live_heap_enabled", return_value=False):
+            with mock.patch.object(
+                ddtrace.internal.dogstatsd, "get_dogstatsd_client", side_effect=RuntimeError("no agent")
+            ):
+                prof = profiler_mod.Profiler()
+                prof.start()  # must not raise
+                try:
+                    assert prof.status.value == "running"
+                finally:
+                    prof.stop(flush=False)
+
+
 @pytest.mark.subprocess(env=dict(DD_PROFILING_ENABLED="true"))
 def test_profiler_keeps_managed_heap_when_native_heap_disabled() -> None:
     """With native heap disabled, the in-process memory collector runs unchanged