mz-debug CPU profiling

[SangJunBak]

Extend profiling APIs with CPU profiling support and wire it into mz-debug

Extend the profiling HTTP APIs with /cpu and /mode routes.

• /cpu turns on CPU profiling. Before, users could only turn it on via a webpage. CPU and memory profiling cannot be turned on at the same time, because doing so can cause deadlocks (see Don't allow simultaneous memory and CPU profiling #3922 for more context). Before starting CPU profiling, we turn off memory profiling, then create a drop guard for the remainder of the CPU profiling execution that ensures memory profiling is turned back to its previous state after a drop.
• /mode returns what profiling is enabled and allows users to turn memory profiling on/off.

Add a test asserting that CPU profiling disables memory profiling.

Implement CPU profiling in mz-debug: using the new APIs, scrape CPU profiles from each service, also logging if memory profiling is still disabled afterwards. This shouldn't be the case usually, but can happen if for whatever reason, activate on jemalloc fails.

Fix an underflow bug found while testing that caused a panic.

Update the regression test.

Reuse endpoint functions for the emulator: make the code more consistent by reusing what we already use to branch clusterd and environmentd services for the emulator.

Motivation

For many self managed customers, we've had to manually tell users to go to some webpage to scrape CPU profiles which is too much TOIL. This feature now also gets CPU profiles along with memory profiles, default to 10 seconds.

Verification

Created some unit tests and manually tested via my personal kind environment.

[sjwiesman]

Whenever this is out of draft, happy to review.

[SangJunBak]

@sjwiesman Would appreciate a review on the mz-debug: ... commit! Asked someone on the database team to review the other CPU profiling code however~ https://materializeinc.slack.com/archives/C08ACQNGSQK/p1784644907627489

[antiguru]

Nice change, and the MemProfilingSuspendGuard restoring memory profiling on Drop is a good fix for the pre-existing leak where time_prof deactivated jemalloc profiling and never restored it. A couple of things to consider before merging.

1. send_with_fallback: HTTPS→HTTP fallback narrowed to is_connect() (internal_http_dumper.rs)

if response.as_ref().is_err_and(reqwest::Error::is_connect) {
    let _ = url.set_scheme("http");
    response = build_with_auth(&url).send().await;
}

The old code fell back on any error. This is correct and deliberate for the side-effectful /cpu POST, but it's shared code, so the existing heap-profile GET now also only retries on connect errors. If an HTTPS attempt against a plaintext-HTTP server surfaces as something other than Kind::Connect (e.g. a TLS-handshake error classified elsewhere), the fallback silently stops and heap profiling breaks in HTTP-only self-managed deployments. Can you confirm reqwest classifies a TLS-handshake-to-plaintext failure as is_connect()? If not, consider gating the narrowing to the POST path only and keeping the broader fallback on the idempotent GET.

2. Emulator CPU branch bypasses the endpoint helpers (internal_http_dumper.rs, minor)

The self-managed path routes through get_cpu_profile_endpoint/get_prof_mode_endpoint, but the emulator branch hardcodes ENVD_CPU_PROFILE_ENDPOINT/ENVD_PROF_MODE_ENDPOINT. It works (emulator is always environmentd) but is a second construction site that can drift from the helpers. Consider get_cpu_profile_endpoint(&ServiceType::Environmentd) for consistency.

Everything else I probed held up: guard-acquire-before-jemalloc-lock ordering plus the cpu_profiling_active() fast-fail correctly prevents mid-capture re-activation, the null-IP filter in time.rs fixes the real addr - 1 underflow, client/server CpuProfileRequest serde shapes match, and the mzcompose 500→400 update matches the new validation boundary.

---

Unrelated to the code: could you copy the description from the commit messages into the PR body? This PR looks set up for squash merge, which will collapse the individual commit messages, so the "See commit messages for details" content would be lost from the squashed commit. Putting it in the PR body preserves it.

[def-]

With this diff:

diff --git a/ci/nightly/pipeline.template.yml b/ci/nightly/pipeline.template.yml
index 655c0ae37f..9adf19d8ee 100644
--- a/ci/nightly/pipeline.template.yml
+++ b/ci/nightly/pipeline.template.yml
@@ -2687,6 +2687,21 @@ steps:
         agents:
           queue: hetzner-aarch64-16cpu-32gb

+      - id: orchestratord-mz-debug-scaled-replica
+        topics: [self-managed]
+        label: Orchestratord mz-debug profiles every pod of a scaled replica
+        artifact_paths: ["mz_debug_*.zip"]
+        depends_on: devel-docker-tags
+        timeout_in_minutes: 60
+        plugins:
+          - ./ci/plugins/mzcompose:
+              composition: orchestratord
+              run: mz-debug-scaled-replica
+              args: [--recreate-cluster]
+              ci-builder: stable
+        agents:
+          queue: hetzner-aarch64-16cpu-32gb
+
       - id: orchestratord-v1-opt-in
         topics: [self-managed]
         label: "Orchestratord v1 opt-in tests"
diff --git a/test/mz-debug/mzcompose.py b/test/mz-debug/mzcompose.py
index 94c63f6650..53846dc384 100644
--- a/test/mz-debug/mzcompose.py
+++ b/test/mz-debug/mzcompose.py
@@ -8,9 +8,12 @@
 # by the Apache License, Version 2.0.

 """
-Basic test for mz-debug
+Basic test for mz-debug, plus a regression test that a CPU profile capture does
+not destroy the accumulated heap profile (PR #37283, Finding 1).
 """

+import urllib.request
+
 from materialize import spawn
 from materialize.mzcompose.composition import (
     Composition,
@@ -20,17 +23,133 @@ from materialize.mzcompose.composition import (
 from materialize.mzcompose.services.materialized import Materialized
 from materialize.mzcompose.services.mz_debug import MzDebug

+# The internal HTTP port of `materialized`, which serves the unauthenticated
+# `/prof` profiling endpoints (heap, cpu, mode).
+INTERNAL_HTTP_PORT = 6878
+
 SERVICES = [
     Materialized(
         ports=[
             "6875:6875",
             "6877:6877",
+            f"{INTERNAL_HTTP_PORT}:{INTERNAL_HTTP_PORT}",
         ]
     ),
     MzDebug(),
 ]


+def _heap_profile_inuse(port: int) -> tuple[int, int]:
+    """Dumps the jemalloc heap profile and returns
+    `(total in-use sampled bytes, number of sampled stacks)`.
+
+    The `dump_jeheap` action returns the raw jemalloc profile in `jeprof` text
+    format. Each sampled stack is a line starting with `@` followed by a
+    `t*: <objs>: <bytes> [...]` line whose `<bytes>` is the in-use bytes charged
+    to that stack. The leading global `t*:` summary line has no preceding `@`
+    and is therefore ignored, so stacks are not double counted.
+    """
+    request = urllib.request.Request(
+        f"http://localhost:{port}/prof/",
+        data=b"action=dump_jeheap",
+        headers={
+            "Content-Type": "application/x-www-form-urlencoded",
+            # The internal listener runs with no authenticator, but the
+            # profiling route group still requires an authenticated identity.
+            # `mz_system` is accepted on the internal listener without a
+            # password.
+            "x-materialize-user": "mz_system",
+        },
+        method="POST",
+    )
+    with urllib.request.urlopen(request, timeout=60) as response:
+        text = response.read().decode("utf-8", "replace")
+
+    total_bytes = 0
+    num_stacks = 0
+    pending_stack = False
+    for line in text.splitlines():
+        line = line.strip()
+        if line.startswith("@"):
+            pending_stack = True
+            num_stacks += 1
+        elif pending_stack and line.startswith("t*:"):
+            # Format: `t*: <objs>: <bytes> [<cum objs>: <cum bytes>]`.
+            total_bytes += int(line.split()[2])
+            pending_stack = False
+    return total_bytes, num_stacks
+
+
+def _assert_cpu_capture_preserves_heap_profile(
+    c: Composition, container_id: str
+) -> None:
+    """Regression test for PR #37283, Finding 1.
+
+    `mz-debug` captures a CPU profile by default. The capture suspends memory
+    profiling for its duration, but the suspend path calls jemalloc's
+    `prof.reset`, which discards the continuously accumulated heap-profile
+    statistics rather than just pausing sampling. A normal `mz-debug` run
+    therefore silently wipes the heap profile of every service it touches, so
+    later heap dumps no longer include long-lived allocations, exactly the
+    evidence a memory investigation needs.
+
+    We plant several large, long-lived allocations (view definitions held in the
+    catalog), snapshot the heap profile, run a CPU-profile-only `mz-debug`
+    capture, and snapshot again. A correct implementation preserves the planted
+    allocations across the capture. The buggy implementation resets them away.
+    """
+    # Plant long-lived allocations in environmentd's heap. Each view definition
+    # embeds a ~512 KiB literal that the catalog holds verbatim. We stay under
+    # the 1 MiB statement-batch limit and well above jemalloc's 512 KiB average
+    # sampling interval so that the allocations are reliably sampled.
+    ballast = "x" * (512 * 1024)
+    for i in range(16):
+        c.sql(
+            f"CREATE VIEW heap_ballast_{i} AS SELECT '{ballast}'::text AS c",
+            print_statement=False,
+        )
+
+    before_bytes, before_stacks = _heap_profile_inuse(INTERNAL_HTTP_PORT)
+    print(
+        f"heap profile before CPU capture: {before_bytes} bytes across {before_stacks} stacks"
+    )
+    # Sanity check: profiling must have captured the planted ballast, otherwise
+    # the comparison below is meaningless and could pass spuriously.
+    assert before_bytes >= 1_000_000, (
+        f"expected the planted allocations to show up in the heap profile, "
+        f"only saw {before_bytes} bytes across {before_stacks} stacks"
+    )
+
+    # A default `mz-debug` run captures a CPU profile, which is what triggers the
+    # heap-profile reset. Capture only the CPU profile (and only for a second) to
+    # keep the trigger isolated and fast.
+    spawn.runv(
+        [
+            "./mz-debug",
+            "emulator",
+            "--docker-container-id",
+            container_id,
+            "--dump-cpu-profiles=true",
+            "--dump-heap-profiles=false",
+            "--dump-prometheus-metrics=false",
+            "--dump-system-catalog=false",
+            "--dump-docker=false",
+            "--cpu-profile-duration-seconds=1",
+        ]
+    )
+
+    after_bytes, after_stacks = _heap_profile_inuse(INTERNAL_HTTP_PORT)
+    print(
+        f"heap profile after CPU capture: {after_bytes} bytes across {after_stacks} stacks"
+    )
+    assert after_bytes >= before_bytes // 2, (
+        "CPU profile capture reset the accumulated heap profile: "
+        f"{before_bytes} bytes / {before_stacks} stacks before, "
+        f"{after_bytes} bytes / {after_stacks} stacks after. "
+        "The capture must suspend memory profiling without calling prof.reset."
+    )
+
+
 def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
     c.up("materialized", Service("mz-debug", idle=True))
     c.invoke("cp", "mz-debug:/usr/local/bin/mz-debug", ".")
@@ -38,6 +157,12 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
     if container_id is None:
         raise ValueError("Failed to get materialized container ID")

+    # Regression test for PR #37283, Finding 1. Runs first so that it observes a
+    # pristine heap profile that no earlier CPU capture has reset.
+    _assert_cpu_capture_preserves_heap_profile(c, container_id)
+
+    # Smoke test: a full `mz-debug` run against the emulator completes without
+    # error.
     spawn.runv(
         [
             "./mz-debug",
diff --git a/test/orchestratord/mzcompose.py b/test/orchestratord/mzcompose.py
index dd2d9c1a95..6035c51cad 100644
--- a/test/orchestratord/mzcompose.py
+++ b/test/orchestratord/mzcompose.py
@@ -5359,3 +5359,195 @@ def workflow_clusterd_generation_scheduling(
         f"{old_gen} and {new_gen} live and that new-gen selectors filter "
         f"to generation {new_gen}"
     )
+
+
+def workflow_mz_debug_scaled_replica(
+    c: Composition, parser: WorkflowArgumentParser
+) -> None:
+    """Regression test for PR #37283, Finding 2.
+
+    mz-debug's self-managed CPU-profile pass port-forwards each clusterd
+    *service* and captures a single profile per service. A managed replica with
+    `scale > 1` is one headless service in front of N clusterd pods, and
+    `kubectl port-forward service/...` selects one arbitrary pod. mz-debug
+    therefore captures a CPU profile for only one of the N processes, writes it
+    under the service-level filename (no pod ordinal), and silently omits the
+    other N-1 pods. A correct implementation profiles every pod behind the
+    service.
+
+    We stand up a managed cluster whose one replica has scale=2 (two clusterd
+    pods behind one service), run mz-debug, and assert one CPU profile per pod.
+    The buggy implementation produces exactly one.
+    """
+    parser.add_argument(
+        "--recreate-cluster",
+        action=argparse.BooleanOptionalAction,
+        help="Recreate cluster if it exists already",
+    )
+    parser.add_argument("--tag", type=str, help="Custom version tag to use")
+    parser.add_argument(
+        "--orchestratord-override",
+        default=True,
+        action=argparse.BooleanOptionalAction,
+        help="Override orchestratord tag",
+    )
+    args = parser.parse_args()
+
+    SCALE = 2
+    CLUSTER_NAME = "scaled_dbg"
+    # The name mz-debug matches k8s resources against, taken from the testing
+    # Materialize CR (`misc/helm-charts/testing/materialize.yaml`).
+    MZ_INSTANCE_NAME = "12345678-1234-1234-1234-123456789012"
+
+    definition = setup(c, args)
+
+    # The only scale>1 size the operator ships (`6400cc`) demands 62 CPUs and
+    # ~470 GiB per pod, which will never schedule on kind. Register a tiny
+    # scale=2 size for the test instead.
+    definition["operator"]["clusters"]["sizes"]["mz_debug_scale2"] = {
+        "workers": 1,
+        "scale": SCALE,
+        "cpu_exclusive": False,
+        "cpu_limit": 0.1,
+        "credits_per_hour": "0.00",
+        "disk_limit": "1552MiB",
+        "memory_limit": "776MiB",
+    }
+
+    init(definition)
+    run(definition, expect_fail=False)
+
+    # Create a managed cluster whose single replica has scale=2. The internal
+    # SQL listener at 6877 has no authenticator, so a plain mz_system connection
+    # works.
+    with port_forward_environmentd(6877) as port:
+        with (
+            psycopg.connect(
+                f"host=localhost port={port} user=mz_system sslmode=disable",
+                autocommit=True,
+            ) as conn,
+            conn.cursor() as cur,
+        ):
+            # `REPLICATION FACTOR 1` pins the cluster to exactly one replica, so
+            # the scale=2 size yields exactly two clusterd pods behind one
+            # service.
+            cur.execute(
+                f"CREATE CLUSTER {CLUSTER_NAME} SIZE 'mz_debug_scale2', REPLICATION FACTOR 1"
+            )
+            cur.execute(
+                "SELECT c.id, r.id "
+                "FROM mz_cluster_replicas r "
+                "JOIN mz_clusters c ON r.cluster_id = c.id "
+                f"WHERE c.name = '{CLUSTER_NAME}'"
+            )
+            rows = cur.fetchall()
+            assert len(rows) == 1, f"expected exactly one replica, got {rows}"
+            cluster_id, replica_id = rows[0]
+
+    cluster_id_selector = (
+        f"cluster.environmentd.materialize.cloud/cluster-id={cluster_id}"
+    )
+
+    def clusterd_pod_names() -> list[str]:
+        pods = json.loads(
+            spawn.capture(
+                [
+                    "kubectl",
+                    "get",
+                    "pods",
+                    "-l",
+                    cluster_id_selector,
+                    "-n",
+                    "materialize-environment",
+                    "-o",
+                    "json",
+                ]
+            )
+        )["items"]
+        return sorted(p["metadata"]["name"] for p in pods)
+
+    # The clusterd pods are created asynchronously after the catalog commit, so
+    # wait for them to appear before waiting on readiness (`kubectl wait` errors
+    # when no pods match its selector).
+    for _ in range(300):
+        if len(clusterd_pod_names()) >= SCALE:
+            break
+        time.sleep(1)
+    else:
+        raise RuntimeError(
+            f"scaled replica never brought up {SCALE} clusterd pods: "
+            f"{clusterd_pod_names()}"
+        )
+
+    spawn.runv(
+        [
+            "kubectl",
+            "wait",
+            "--for=condition=Ready",
+            "pod",
+            "-l",
+            cluster_id_selector,
+            "-n",
+            "materialize-environment",
+            "--timeout=300s",
+        ]
+    )
+
+    pod_names = clusterd_pod_names()
+    assert (
+        len(pod_names) == SCALE
+    ), f"expected {SCALE} clusterd pods for the scaled replica, got {pod_names}"
+
+    # Run mz-debug, capturing only CPU profiles to keep the run fast and the
+    # trigger isolated. `--dump-cpu-profiles` is on by default; we disable the
+    # other collectors. The connection URL is only parsed (system-catalog dump
+    # is off), so no live SQL connection is needed.
+    spawn.runv(
+        [
+            "./mz-debug",
+            "self-managed",
+            "--k8s-namespace",
+            "materialize-environment",
+            "--mz-instance-name",
+            MZ_INSTANCE_NAME,
+            "--mz-connection-url",
+            "postgresql://mz_system@localhost:6877/materialize",
+            "--dump-k8s=false",
+            "--dump-system-catalog=false",
+            "--dump-heap-profiles=false",
+            "--dump-prometheus-metrics=false",
+            "--dump-cpu-profiles=true",
+            "--cpu-profile-duration-seconds=1",
+        ]
+    )
+
+    # mz-debug writes `mz_debug_<timestamp>/profiles/<service>.cpuprof.pprof.gz`
+    # to its working directory. Match the scaled replica's files by its unique
+    # cluster/replica id, which appears in the clusterd service name regardless
+    # of the operator's name prefix or deploy generation.
+    replica_marker = f"cluster-{cluster_id}-replica-{replica_id}"
+    all_cpu_profiles = sorted(MZ_ROOT.glob("mz_debug_*/profiles/*.cpuprof.pprof.gz"))
+    matching = [p for p in all_cpu_profiles if replica_marker in p.name]
+    print(
+        f"CPU profiles for replica {cluster_id}/{replica_id}: "
+        f"{[p.name for p in matching]} (all: {[p.name for p in all_cpu_profiles]})"
+    )
+
+    assert matching, (
+        "mz-debug produced no CPU profile for the scaled replica's clusterd "
+        "service, so cluster discovery or CPU capture failed and Finding 2 "
+        f"cannot be assessed. All CPU profiles: {[p.name for p in all_cpu_profiles]}"
+    )
+    assert len(matching) == len(pod_names), (
+        f"Finding 2: mz-debug captured {len(matching)} CPU profile(s) "
+        f"({[p.name for p in matching]}) for the scale-{SCALE} replica, but it "
+        f"has {len(pod_names)} clusterd pods ({pod_names}). "
+        "`kubectl port-forward service/...` profiled only one arbitrary pod; "
+        "every pod behind the service must be profiled, with the pod ordinal in "
+        "the filename."
+    )
+
+    print(
+        f"verified mz-debug captured a CPU profile for all {len(pod_names)} "
+        f"pods of the scale-{SCALE} replica {cluster_id}/{replica_id}"
+    )

Running bin/mzcompose --find mz-debug run default fails because the heap-profiling stats are reset:

Traceback (most recent call last):
  File "/home/deen/git/materialize4/misc/python/materialize/mzcompose/composition.py", line 753, in test_case
    yield
  File "/home/deen/git/materialize4/misc/python/materialize/cli/mzcompose.py", line 868, in handle_composition
    composition.workflow(
    ~~~~~~~~~~~~~~~~~~~~^
        workflow_name, *args.unknown_subargs[1:], *extra_args
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    )
    ^
  File "/home/deen/git/materialize4/misc/python/materialize/mzcompose/composition.py", line 635, in workflow
    func(self, parser)
    ~~~~^^^^^^^^^^^^^^
  File "/home/deen/git/materialize4/test/mz-debug/mzcompose.py", line 162, in workflow_default
    _assert_cpu_capture_preserves_heap_profile(c, container_id)
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^
  File "/home/deen/git/materialize4/test/mz-debug/mzcompose.py", line 145, in _assert_cpu_capture_preserves_heap_profile
    assert after_bytes >= before_bytes // 2, (
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AssertionError: CPU profile capture reset the accumulated heap profile: 103487710 bytes / 111 stacks before, 16 bytes / 1 stacks after. The capture must suspend memory profiling without calling prof.reset.
mzcompose: error: at least one test case failed

QA LLM review found this: https://github.com/MaterializeInc/qa-llm-review/blob/master/commit-bugs/done/analysis-pr-37283.md

[SangJunBak]

@def- Fixed the heap allocation bug! Nice catch! As for the multi-process pods, I'm splitting that out into a new PR #37816

[def-]

No further complaints from QA side