diff --git a/misc/python/materialize/cloudtest/app/materialize_application.py b/misc/python/materialize/cloudtest/app/materialize_application.py index f1eafa8a090ca..9122eb354e77e 100644 --- a/misc/python/materialize/cloudtest/app/materialize_application.py +++ b/misc/python/materialize/cloudtest/app/materialize_application.py @@ -27,6 +27,7 @@ EnvironmentdStatefulSet, ListenersConfigMap, MaterializedAliasService, + MzInstanceIdentity, ) from materialize.cloudtest.k8s.minio import Minio from materialize.cloudtest.k8s.mysql import mysql_resources @@ -52,9 +53,10 @@ def __init__( apply_node_selectors: bool = False, ) -> None: self.tag = tag + self.instance_identity = MzInstanceIdentity.generate() self.secret = EnvironmentdSecret() self.listeners_configmap = ListenersConfigMap() - self.environmentd = EnvironmentdService() + self.environmentd = EnvironmentdService(self.instance_identity) self.materialized_alias = MaterializedAliasService() self.testdrive = TestdrivePod( release_mode=release_mode, @@ -86,6 +88,7 @@ def get_resources(self, log_filter: str | None) -> list[K8sResource]: self.secret, self.listeners_configmap, EnvironmentdStatefulSet( + instance_identity=self.instance_identity, release_mode=self.release_mode, tag=self.tag, log_filter=log_filter, diff --git a/misc/python/materialize/cloudtest/k8s/api/k8s_service.py b/misc/python/materialize/cloudtest/k8s/api/k8s_service.py index 8b9309aad8cca..0c1bcfb759a5f 100644 --- a/misc/python/materialize/cloudtest/k8s/api/k8s_service.py +++ b/misc/python/materialize/cloudtest/k8s/api/k8s_service.py @@ -102,7 +102,7 @@ def sql_query( return cursor.fetchall() def http_get(self, path: str) -> Any: - url = f"http://localhost:{self.node_port('internalhttp')}/{path.lstrip('/')}" + url = f"http://localhost:{self.node_port('internal-http')}/{path.lstrip('/')}" response = requests.get(url) response.raise_for_status() return response.text diff --git a/misc/python/materialize/cloudtest/k8s/environmentd.py b/misc/python/materialize/cloudtest/k8s/environmentd.py index 02de038a8b8ba..a1336e8b9edb0 100644 --- a/misc/python/materialize/cloudtest/k8s/environmentd.py +++ b/misc/python/materialize/cloudtest/k8s/environmentd.py @@ -10,8 +10,12 @@ import json import operator import os +import random +import string import urllib.parse +import uuid from collections.abc import Callable +from dataclasses import dataclass from kubernetes.client import ( V1ConfigMap, @@ -54,6 +58,47 @@ get_default_system_parameters, ) +MZ_ORGANIZATION_NAME_LABEL = "materialize.cloud/organization-name" +MZ_RESOURCE_ID_LABEL = "materialize.cloud/mz-resource-id" + + +@dataclass(frozen=True) +class MzInstanceIdentity: + """Identifies one Materialize instance to self-managed tooling. + + The operator stamps an organization name and a resource ID on every + Kubernetes object it provisions for an instance, and tooling such as + mz-debug discovers an instance's objects through those labels. cloudtest + deploys environmentd directly rather than through the operator, so it has to + stamp the labels itself. + + Every object of an instance must carry the same values, and no two instances + living in the same namespace may share them. + """ + + # Names the instance. mz-debug takes this as its `--mz-instance-name`. + organization_name: str + + # Ties an object to its instance. + resource_id: str + + @classmethod + def generate(cls) -> "MzInstanceIdentity": + # Resource IDs are ten characters of a lowercase alphabet, because the + # operator builds DNS-1035 object names out of them. DNS-1035 names are + # case insensitive, so mixed case would not keep them apart. + resource_id_alphabet = string.ascii_lowercase + string.digits + return cls( + organization_name=str(uuid.uuid4()), + resource_id="".join(random.choices(resource_id_alphabet, k=10)), + ) + + def labels(self) -> dict[str, str]: + return { + MZ_ORGANIZATION_NAME_LABEL: self.organization_name, + MZ_RESOURCE_ID_LABEL: self.resource_id, + } + class EnvironmentdSecret(K8sSecret): def __init__(self, namespace: str = DEFAULT_K8S_NAMESPACE) -> None: @@ -82,16 +127,23 @@ def __init__(self, namespace: str = DEFAULT_K8S_NAMESPACE) -> None: class EnvironmentdService(K8sService): - def __init__(self, namespace: str = DEFAULT_K8S_NAMESPACE) -> None: + def __init__( + self, + instance_identity: MzInstanceIdentity, + namespace: str = DEFAULT_K8S_NAMESPACE, + ) -> None: super().__init__(namespace) service_port = V1ServicePort(name="sql", port=6875) http_port = V1ServicePort(name="http", port=6876) internal_port = V1ServicePort(name="internal", port=6877) - internal_http_port = V1ServicePort(name="internalhttp", port=6878) + internal_http_port = V1ServicePort(name="internal-http", port=6878) self.service = V1Service( api_version="v1", kind="Service", - metadata=V1ObjectMeta(name="environmentd", labels={"app": "environmentd"}), + metadata=V1ObjectMeta( + name="environmentd", + labels={"app": "environmentd", **instance_identity.labels()}, + ), spec=V1ServiceSpec( type="NodePort", ports=[service_port, internal_port, http_port, internal_http_port], @@ -119,6 +171,7 @@ def __init__(self, namespace: str = DEFAULT_K8S_NAMESPACE) -> None: class EnvironmentdStatefulSet(K8sStatefulSet): def __init__( self, + instance_identity: MzInstanceIdentity, tag: str | None = None, release_mode: bool = True, coverage_mode: bool = False, @@ -129,6 +182,7 @@ def __init__( cockroach_namespace: str = DEFAULT_K8S_NAMESPACE, apply_node_selectors: bool = False, ) -> None: + self.instance_identity = instance_identity self.tag = tag self.release_mode = release_mode self.coverage_mode = coverage_mode @@ -142,7 +196,10 @@ def __init__( super().__init__(namespace) def generate_stateful_set(self) -> V1StatefulSet: - metadata = V1ObjectMeta(name="environmentd", labels={"app": "environmentd"}) + metadata = V1ObjectMeta( + name="environmentd", + labels={"app": "environmentd", **self.instance_identity.labels()}, + ) label_selector = V1LabelSelector(match_labels={"app": "environmentd"}) ports = [V1ContainerPort(container_port=5432, name="sql")] diff --git a/src/mz-debug/src/internal_http_dumper.rs b/src/mz-debug/src/internal_http_dumper.rs index 040748dead3c6..79ff3c83f8337 100644 --- a/src/mz-debug/src/internal_http_dumper.rs +++ b/src/mz-debug/src/internal_http_dumper.rs @@ -24,7 +24,8 @@ use tracing::{info, warn}; use url::Url; use crate::kubectl_port_forwarder::{ - KubectlPortForwarder, ServiceInfo, find_cluster_services, find_environmentd_service, + KubectlPortForwarder, PortForwardConnection, PortForwardTarget, ServiceInfo, + find_cluster_services, find_environmentd_service, find_service_pods, }; use crate::{AuthMode, Context, EmulatorContext, PasswordAuthCredentials, SelfManagedContext}; @@ -582,6 +583,35 @@ pub async fn dump_emulator_http_resources( Ok(()) } +/// One pod to dump HTTP resources from, expanded from the service that fronts +/// it. A scaled replica's service fronts more than one such pod. +struct PodDumpTarget { + pod_name: String, + namespace: String, + /// Ports are taken from the fronting service. We assume the service port + /// equals the pod's container port for these HTTP endpoints, which holds + /// for clusterd and environmentd. + service_ports: Vec, + service_type: ServiceType, +} + +/// Spawns a `kubectl port-forward` against a specific pod (not its service), +/// so a scaled replica's individual processes can each be reached. +async fn spawn_pod_port_forward( + self_managed_context: &SelfManagedContext, + pod_target: &PodDumpTarget, + target_port: i32, +) -> Result { + KubectlPortForwarder { + context: self_managed_context.k8s_context.clone(), + namespace: pod_target.namespace.clone(), + target: PortForwardTarget::Pod(pod_target.pod_name.clone()), + target_port, + } + .spawn_port_forward() + .await +} + pub async fn dump_self_managed_http_resources( context: &Context, self_managed_context: &SelfManagedContext, @@ -618,80 +648,103 @@ pub async fn dump_self_managed_http_resources( ))) .collect(); - // Scrape each service for heap profiles and prometheus metrics. + // A replica with `scale > 1` is a single service containing multiple clusterd + // pods. Expand every service into the pods it contains and dump each pod + // individually so no process is silently skipped. + let mut pod_targets: Vec = Vec::new(); for &(service_info, service_type) in &services { - let profiling_endpoint = get_profile_endpoint(&service_type); - let heap_profile_port_label = get_port_labels( - &self_managed_context.http_connection_auth_mode, - &service_type, + let pod_names = match find_service_pods( + &self_managed_context.k8s_client, + &self_managed_context.k8s_namespace, + &service_info.selector, ) - .heap_profile_port_label; + .await + { + Ok(pod_names) => pod_names, + Err(e) => { + warn!( + "Failed to list pods for service {}: {:#}", + service_info.service_name, e + ); + continue; + } + }; + if pod_names.is_empty() { + warn!( + "Found no pods for service {}, skipping", + service_info.service_name + ); + continue; + } + for pod_name in pod_names { + pod_targets.push(PodDumpTarget { + pod_name, + namespace: service_info.namespace.clone(), + service_ports: service_info.service_ports.clone(), + service_type, + }); + } + } - let prom_metrics_port_label = get_port_labels( + // Scrape each pod for heap profiles and prometheus metrics. A failure on + // one pod is logged and skipped rather than aborting the whole dump, so one + // unreachable pod does not cost us every other pod's data. + for pod_target in &pod_targets { + let service_type = pod_target.service_type; + let profiling_endpoint = get_profile_endpoint(&service_type); + let HttpPortLabels { + heap_profile_port_label, + prom_metrics_port_label, + } = get_port_labels( &self_managed_context.http_connection_auth_mode, &service_type, - ) - .prom_metrics_port_label; - - let (heap_profile_http_connection, prom_metrics_http_connection) = { - let maybe_heap_profile_port = service_info - .service_ports - .iter() - .find_map(|port_info| find_http_port_by_label(port_info, heap_profile_port_label)); - let maybe_prom_metrics_port = service_info - .service_ports - .iter() - .find_map(|port_info| find_http_port_by_label(port_info, prom_metrics_port_label)); - if let (Some(heap_profile_port), Some(prom_metrics_port)) = - (maybe_heap_profile_port, maybe_prom_metrics_port) - { - let heap_profile_port_forwarder = KubectlPortForwarder { - context: self_managed_context.k8s_context.clone(), - namespace: service_info.namespace.clone(), - service_name: service_info.service_name.clone(), - target_port: heap_profile_port.port, - }; - let heap_profile_http_connection = Arc::new( - heap_profile_port_forwarder - .spawn_port_forward() - .await - .with_context(|| { - format!( - "Failed to spawn port forwarder for service {}", - service_info.service_name - ) - })?, - ); - let prom_metrics_http_connection = if heap_profile_port == prom_metrics_port { - Arc::clone(&heap_profile_http_connection) - } else { - let prom_metrics_port_forwarder = KubectlPortForwarder { - context: self_managed_context.k8s_context.clone(), - namespace: service_info.namespace.clone(), - service_name: service_info.service_name.clone(), - target_port: prom_metrics_port.port, - }; - Arc::new( - prom_metrics_port_forwarder - .spawn_port_forward() - .await - .with_context(|| { - format!( - "Failed to spawn port forwarder for service {}", - service_info.service_name - ) - })?, - ) - }; + ); - (heap_profile_http_connection, prom_metrics_http_connection) - } else { - return Err(anyhow::anyhow!( - "Failed to find HTTP port for service {}, heap_profile_port_label={}, prom_metrics_port_label={}", - service_info.service_name, - heap_profile_port_label, - prom_metrics_port_label - )); + let maybe_heap_profile_port = pod_target + .service_ports + .iter() + .find_map(|port_info| find_http_port_by_label(port_info, heap_profile_port_label)); + let maybe_prom_metrics_port = pod_target + .service_ports + .iter() + .find_map(|port_info| find_http_port_by_label(port_info, prom_metrics_port_label)); + let (Some(heap_profile_port), Some(prom_metrics_port)) = + (maybe_heap_profile_port, maybe_prom_metrics_port) + else { + warn!( + "Failed to find HTTP port for pod {}, heap_profile_port_label={}, prom_metrics_port_label={}", + pod_target.pod_name, heap_profile_port_label, prom_metrics_port_label + ); + continue; + }; + + let heap_profile_http_connection = + match spawn_pod_port_forward(self_managed_context, pod_target, heap_profile_port.port) + .await + { + Ok(connection) => Arc::new(connection), + Err(e) => { + warn!( + "Failed to spawn port forwarder for pod {}: {:#}", + pod_target.pod_name, e + ); + continue; + } + }; + let prom_metrics_http_connection = if heap_profile_port == prom_metrics_port { + Arc::clone(&heap_profile_http_connection) + } else { + match spawn_pod_port_forward(self_managed_context, pod_target, prom_metrics_port.port) + .await + { + Ok(connection) => Arc::new(connection), + Err(e) => { + warn!( + "Failed to spawn port forwarder for pod {}: {:#}", + pod_target.pod_name, e + ); + continue; + } } }; @@ -703,17 +756,14 @@ pub async fn dump_self_managed_http_resources( profiling_endpoint ); - info!( - "Dumping heap profile for service {}", - service_info.service_name - ); + info!("Dumping heap profile for pod {}", pod_target.pod_name); if let Err(e) = dump_task - .dump_heap_profile(&profiling_endpoint, &service_info.service_name) + .dump_heap_profile(&profiling_endpoint, &pod_target.pod_name) .await { warn!( - "Failed to dump heap profile for service {}: {:#}", - service_info.service_name, e + "Failed to dump heap profile for pod {}: {:#}", + pod_target.pod_name, e ); } } @@ -725,32 +775,30 @@ pub async fn dump_self_managed_http_resources( prom_metrics_http_connection.local_port, PROM_METRICS_ENDPOINT ); - info!( - "Dumping prometheus metrics for service {}", - service_info.service_name - ); + info!("Dumping prometheus metrics for pod {}", pod_target.pod_name); if let Err(e) = dump_task - .dump_prometheus_metrics(&prom_metrics_endpoint, &service_info.service_name) + .dump_prometheus_metrics(&prom_metrics_endpoint, &pod_target.pod_name) .await { warn!( - "Failed to dump prometheus metrics for service {}: {:#}", - service_info.service_name, e + "Failed to dump prometheus metrics for pod {}: {:#}", + pod_target.pod_name, e ); } } } // Capture CPU profiles after memory profiling, since each capture - // temporarily disables memory profiling on its service. The captures run - // in parallel, and a failure on one service does not abort the others. + // temporarily disables memory profiling on its pod. The captures run in + // parallel, and a failure on one pod does not abort the others. if context.dump_cpu_profiles { info!( - "Capturing CPU profiles for {} seconds. Memory profiling is temporarily disabled on each service during its capture and restored afterwards.", + "Capturing CPU profiles for {} seconds. Memory profiling is temporarily disabled on each pod during its capture and restored afterwards.", context.cpu_profile_duration_secs ); - let cpu_profile_futures = services.iter().map(|&(service_info, service_type)| { + let cpu_profile_futures = pod_targets.iter().map(|pod_target| { + let service_type = pod_target.service_type; // The CPU and mode endpoints are served on the same port as the heap // profile endpoint. let port_label = get_port_labels( @@ -758,39 +806,34 @@ pub async fn dump_self_managed_http_resources( &service_type, ) .heap_profile_port_label; - let k8s_context = self_managed_context.k8s_context.clone(); let dump_task = &dump_task; let duration_secs = context.cpu_profile_duration_secs; async move { - let Some(port) = service_info + let Some(port) = pod_target .service_ports .iter() .find_map(|port_info| find_http_port_by_label(port_info, port_label)) else { warn!( - "Failed to find HTTP port `{}` for CPU profiling of service {}", - port_label, service_info.service_name + "Failed to find HTTP port `{}` for CPU profiling of pod {}", + port_label, pod_target.pod_name ); return; }; - let port_forwarder = KubectlPortForwarder { - context: k8s_context, - namespace: service_info.namespace.clone(), - service_name: service_info.service_name.clone(), - target_port: port.port, - }; - let connection = match port_forwarder.spawn_port_forward().await { - Ok(connection) => connection, - Err(e) => { - warn!( - "Failed to spawn port forwarder for CPU profiling of service {}: {:#}", - service_info.service_name, e - ); - return; - } - }; + let connection = + match spawn_pod_port_forward(self_managed_context, pod_target, port.port).await + { + Ok(connection) => connection, + Err(e) => { + warn!( + "Failed to spawn port forwarder for CPU profiling of pod {}: {:#}", + pod_target.pod_name, e + ); + return; + } + }; let cpu_endpoint = format!( "{}:{}/{}", @@ -809,7 +852,7 @@ pub async fn dump_self_managed_http_resources( dump_task, &cpu_endpoint, &mode_endpoint, - &service_info.service_name, + &pod_target.pod_name, duration_secs, ) .await; diff --git a/src/mz-debug/src/kubectl_port_forwarder.rs b/src/mz-debug/src/kubectl_port_forwarder.rs index 33689be4588a2..3b9a938aa25b2 100644 --- a/src/mz-debug/src/kubectl_port_forwarder.rs +++ b/src/mz-debug/src/kubectl_port_forwarder.rs @@ -15,19 +15,51 @@ //! Port forwards k8s service via Kubectl +use std::collections::BTreeMap; + use anyhow::{Context, Result}; use k8s_openapi::api::apps::v1::StatefulSet; -use k8s_openapi::api::core::v1::{Service, ServicePort}; +use k8s_openapi::api::core::v1::{Pod, Service, ServicePort}; use kube::api::ListParams; use kube::{Api, Client}; use tokio::io::AsyncBufReadExt; use tracing::info; +/// A Kubernetes resource that `kubectl port-forward` can target. +/// +/// Forwarding a `Service` lets Kubernetes pick one arbitrary backing pod, which +/// is what you want when any pod will do (for example the environmentd SQL +/// listener, where the service routes to the active leader). Forwarding a `Pod` +/// reaches one specific process, which is what profiling a multi-pod (scaled) +#[derive(Debug, Clone)] +pub enum PortForwardTarget { + Service(String), + Pod(String), +} + +impl PortForwardTarget { + /// The `kubectl` resource argument, for example `services/foo` or + /// `pods/foo`. + fn kubectl_arg(&self) -> String { + match self { + PortForwardTarget::Service(name) => format!("services/{name}"), + PortForwardTarget::Pod(name) => format!("pods/{name}"), + } + } + + /// The bare resource name, for logging and error messages. + pub fn name(&self) -> &str { + match self { + PortForwardTarget::Service(name) | PortForwardTarget::Pod(name) => name, + } + } +} + #[derive(Debug)] pub struct KubectlPortForwarder { pub namespace: String, - pub service_name: String, + pub target: PortForwardTarget, pub target_port: i32, pub context: Option, } @@ -48,10 +80,10 @@ impl KubectlPortForwarder { /// the port forward is established. pub async fn spawn_port_forward(&self) -> Result { let port_arg_str = format!(":{}", &self.target_port); - let service_name_arg_str = format!("services/{}", &self.service_name); + let target_arg_str = self.target.kubectl_arg(); let mut args = vec![ "port-forward", - &service_name_arg_str, + &target_arg_str, &port_arg_str, "-n", &self.namespace, @@ -103,7 +135,10 @@ impl KubectlPortForwarder { if let (Some(local_address), Some(local_port)) = (local_address, local_port) { info!( "Port forwarding established for {} from ports {}:{} -> {}", - &self.service_name, local_address, local_port, &self.target_port + self.target.name(), + local_address, + local_port, + &self.target_port ); return Ok(PortForwardConnection { _lines: lines, @@ -127,6 +162,9 @@ pub struct ServiceInfo { pub service_name: String, pub service_ports: Vec, pub namespace: String, + /// The service's pod selector, used to enumerate the pods it fronts. A + /// scaled replica's service selects more than one pod. + pub selector: BTreeMap, } /// Returns ServiceInfo for balancerd @@ -164,6 +202,7 @@ pub async fn find_environmentd_service( service_name: service_name.clone(), service_ports: ports.clone(), namespace: k8s_namespace.clone(), + selector: spec.selector.clone().unwrap_or_default(), }) } else { None @@ -237,6 +276,7 @@ pub async fn find_cluster_services( service_name: name, service_ports: ports, namespace: k8s_namespace.clone(), + selector, }) }) .collect(); @@ -273,7 +313,7 @@ pub async fn create_pg_wire_port_forwarder( Ok(KubectlPortForwarder { context: k8s_context.clone(), namespace: service_info.namespace, - service_name: service_info.service_name, + target: PortForwardTarget::Service(service_info.service_name), target_port: external_sql_port.port, }) } else { @@ -282,3 +322,38 @@ pub async fn create_pg_wire_port_forwarder( )) } } + +/// Lists the names of the pods a service contains, matched by its `selector`. +/// +/// A service with `scale > 1` contains multiple pods, 1 per process. +/// The names of the pods are returned sorted so output is deterministic across runs. +pub async fn find_service_pods( + client: &Client, + k8s_namespace: &str, + selector: &BTreeMap, +) -> Result> { + // An empty selector would match every pod in the namespace, which is never + // what a caller means. Treat it as "no pods". + if selector.is_empty() { + return Ok(Vec::new()); + } + + let label_filter = selector + .iter() + .map(|(k, v)| format!("{k}={v}")) + .collect::>() + .join(","); + + let pods_api: Api = Api::namespaced(client.clone(), k8s_namespace); + let pods = pods_api + .list(&ListParams::default().labels(&label_filter)) + .await + .with_context(|| format!("Failed to list pods in namespace {}", k8s_namespace))?; + + let mut pod_names: Vec = pods + .iter() + .filter_map(|pod| pod.metadata.name.clone()) + .collect(); + pod_names.sort(); + Ok(pod_names) +} diff --git a/test/cloudtest/test_mz_debug_tool.py b/test/cloudtest/test_mz_debug_tool.py index 03d8d0960c5bc..21e4a6850fdf7 100644 --- a/test/cloudtest/test_mz_debug_tool.py +++ b/test/cloudtest/test_mz_debug_tool.py @@ -8,13 +8,22 @@ # by the Apache License, Version 2.0. import glob +import os import subprocess +import time + +import pytest from materialize import MZ_ROOT, spawn from materialize.cloudtest import DEFAULT_K8S_CONTEXT_NAME, DEFAULT_K8S_NAMESPACE from materialize.cloudtest.app.materialize_application import MaterializeApplication +from materialize.cloudtest.util.cluster import cluster_pod_name from materialize.cloudtest.util.wait import wait +# environmentd runs as a single-replica StatefulSet, so mz-debug profiles it as +# the pod `environmentd-0`. +ENVIRONMENTD_POD = "environmentd-0" + def test_successful_zip_creation(mz: MaterializeApplication) -> None: # Wait until the Materialize instance is ready @@ -50,7 +59,7 @@ def test_successful_zip_creation(mz: MaterializeApplication) -> None: "--k8s-namespace", DEFAULT_K8S_NAMESPACE, "--mz-instance-name", - "12345678-1234-1234-1234-123456789012", + mz.instance_identity.organization_name, "--mz-connection-url", "postgresql://mz_system@localhost:6877/materialize", ], @@ -60,3 +69,142 @@ def test_successful_zip_creation(mz: MaterializeApplication) -> None: print("-- Looking for mz-debug zip files") zip_files = glob.glob(str(MZ_ROOT / "mz_debug*.zip")) assert len(zip_files) > 0, "No mz-debug zip file was created" + + +def _newest_dump_dir() -> str: + """The most recently written `mz_debug_` directory in MZ_ROOT, + where mz-debug writes its output. Tests run sequentially against the shared + (session-scoped) instance, so the newest directory belongs to the mz-debug + run this test just made.""" + dump_dirs = [d for d in glob.glob(str(MZ_ROOT / "mz_debug_*")) if os.path.isdir(d)] + assert dump_dirs, "mz-debug did not create an mz_debug_* output directory" + return max(dump_dirs, key=os.path.getmtime) + + +def _profile_names(profiles_dir: str, kind: str, written_after: float) -> list[str]: + """Basenames of the `..pprof.gz` profiles written after + `written_after`, a `time.time()` timestamp taken before the mz-debug run + under test. `kind` is `cpuprof` or `memprof`. + + mz-debug names its output directory after the current minute, so runs a few + seconds apart share one and a run overwrites what an earlier run captured for + the same pod. Ignoring profiles older than the run keeps a failure to capture + from being masked by an earlier run's leftovers.""" + return sorted( + os.path.basename(p) + for p in glob.glob(os.path.join(profiles_dir, f"*.{kind}.pprof.gz")) + if os.path.getmtime(p) >= written_after + ) + + +@pytest.mark.parametrize( + "scale,replication_factor", + [ + # A single service fronting one pod per process. + (2, 1), + # One service per replica, each fronting a single pod. + (1, 2), + ], +) +def test_self_managed_profiles( + mz: MaterializeApplication, scale: int, replication_factor: int +) -> None: + """ + mz-debug must capture both a CPU and a heap profile from environmentd and + from every clusterd pod of a cluster. + + A `scale=N` replica is a single Kubernetes service with N processes, but + contains one pod per process. A cluster of replication factor N is N such + services. Both dimensions have to be walked to reach every pod. + """ + cluster_name = f"dbg_scale{scale}_rf{replication_factor}" + + # Wait until the default cluster is ready, so environmentd is serving SQL. + wait( + condition="condition=Ready", + resource="pod", + label="cluster.environmentd.materialize.cloud/cluster-id=u1", + ) + + mz.environmentd.sql( + f"CREATE CLUSTER {cluster_name} SIZE 'scale={scale},workers=1', " + f"REPLICATION FACTOR {replication_factor}" + ) + rows = mz.environmentd.sql_query( + "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}'" + ) + assert ( + len(rows) == replication_factor + ), f"expected {replication_factor} replica(s), got {rows}" + cluster_id = rows[0][0] + replica_ids = [replica_id for _, replica_id in rows] + + # Each replica is served by one clusterd pod per process, ordinals 0..scale, + # all behind a single service. `cluster_pod_name` returns the `pod/...` + # resource string `kubectl wait` expects. + pod_resources = [ + cluster_pod_name(cluster_id, replica_id, process) + for replica_id in replica_ids + for process in range(scale) + ] + for pod_resource in pod_resources: + wait(condition="condition=Ready", resource=pod_resource) + + print("-- Running mz-debug (CPU and heap profiles)") + # Filesystems can store modification times at a coarser resolution than + # `time.time()` reports, so leave a second of slack for a profile written + # right after the run starts. + run_started = time.time() - 1 + # Capture only profiles to keep the run focused. + spawn.runv( + [ + "cargo", + "run", + "--bin", + "mz-debug", + "--", + "self-managed", + "--k8s-context", + DEFAULT_K8S_CONTEXT_NAME, + "--k8s-namespace", + DEFAULT_K8S_NAMESPACE, + "--mz-instance-name", + mz.instance_identity.organization_name, + "--mz-connection-url", + "postgresql://mz_system@localhost:6877/materialize", + "--dump-k8s=false", + "--dump-system-catalog=false", + "--dump-prometheus-metrics=false", + "--dump-heap-profiles=true", + "--dump-cpu-profiles=true", + "--cpu-profile-duration-seconds=1", + ], + cwd=MZ_ROOT, + ) + + # mz-debug writes `.cpuprof.pprof.gz` and `.memprof.pprof.gz` under + # the run's `profiles/` directory. Both kinds must be there for environmentd + # and for every clusterd pod of every replica, each named after the pod it + # came from. + profiles_dir = os.path.join(_newest_dump_dir(), "profiles") + expected_pods = [ENVIRONMENTD_POD] + [ + pod_resource.removeprefix("pod/") for pod_resource in pod_resources + ] + + for kind in ("cpuprof", "memprof"): + names = _profile_names(profiles_dir, kind, run_started) + print(f"{kind} profiles: {names}") + + missing = [ + pod for pod in expected_pods if f"{pod}.{kind}.pprof.gz" not in names + ] + assert not missing, ( + f"mz-debug captured no {kind} profile for {missing}. Every pod of " + f"every replica must be profiled, under a name that identifies the " + f"pod. {kind} profiles: {names}" + ) + + mz.environmentd.sql(f"DROP CLUSTER {cluster_name} CASCADE") diff --git a/test/mz-debug/mzcompose.py b/test/mz-debug/mzcompose.py index b605a59f788c3..db2287bb740ba 100644 --- a/test/mz-debug/mzcompose.py +++ b/test/mz-debug/mzcompose.py @@ -12,6 +12,7 @@ """ import urllib.request +from pathlib import Path from materialize import spawn from materialize.mzcompose.composition import ( @@ -136,6 +137,55 @@ def _assert_cpu_capture_preserves_heap_profile( ) +def _newest_dump_dir() -> Path: + """Returns the most recently written `mz_debug_` directory in the + working directory, where `mz-debug` writes its output.""" + dump_dirs = [p for p in Path.cwd().glob("mz_debug_*") if p.is_dir()] + assert dump_dirs, "mz-debug did not create an mz_debug_* output directory" + return max(dump_dirs, key=lambda p: p.stat().st_mtime) + + +def _assert_default_dump_files(dump_dir: Path, container_id: str) -> None: + """Asserts that a default `mz-debug emulator` run wrote every artifact it is + meant to. + + A default run enables every collector, so `dump_dir` must contain the docker + dumps, the heap and CPU profiles, the prometheus metrics, the tool's own log, + and a non-empty system catalog dump. A sibling `.zip` archive of the whole + directory must also exist. + """ + expected_files = [ + dump_dir / "tracing.log", + dump_dir / "profiles" / "environmentd.memprof.pprof.gz", + dump_dir / "profiles" / "environmentd.cpuprof.pprof.gz", + dump_dir / "prom_metrics" / "environmentd.metrics.txt", + dump_dir / "docker" / container_id / "logs-stdout.txt", + dump_dir / "docker" / container_id / "logs-stderr.txt", + dump_dir / "docker" / container_id / "inspect.txt", + dump_dir / "docker" / container_id / "stats.txt", + dump_dir / "docker" / container_id / "top.txt", + ] + missing = [str(p) for p in expected_files if not p.is_file()] + + # The system catalog is dumped as one CSV per relation. The exact set is + # large and partly depends on live replicas, so require at least one CSV + # rather than enumerating relations. + catalog_dir = dump_dir / "system_catalog" + if not any(catalog_dir.rglob("*.csv")): + missing.append(f"{catalog_dir}/**/*.csv (system catalog dump is empty)") + + # The whole directory is also archived as a sibling zip. + zip_path = dump_dir.with_name(f"{dump_dir.name}.zip") + if not zip_path.is_file(): + missing.append(str(zip_path)) + + assert ( + not missing + ), "mz-debug default run did not produce all expected files:\n" + "\n".join( + f" - {m}" for m in missing + ) + + 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", ".") @@ -147,7 +197,7 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None: _assert_cpu_capture_preserves_heap_profile(c, container_id) # Smoke test: a full `mz-debug` run against the emulator completes without - # error. + # error and produces the complete set of default output files. spawn.runv( [ "./mz-debug", @@ -155,6 +205,7 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None: "--docker-container-id", container_id, "--mz-connection-url", - "postgres://mz_system@localhost:6877/materialize", + "postgres://mz_system@127.0.0.1:6877/materialize", ] ) + _assert_default_dump_files(_newest_dump_dir(), container_id)