From 750e8f1f59e73fef9841d7db9babe7d9bc2746d2 Mon Sep 17 00:00:00 2001 From: Benjamin Knofe-Vider Date: Mon, 6 Jul 2026 16:33:45 +0200 Subject: [PATCH] k8s workers: bound pre-session memory (DUCKGRES_MEMORY_LIMIT, GOMEMLIMIT, THREADS at spawn) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Worker pods were spawned with no memory configuration, so the worker process sized its base DuckDB off the NODE's /proc/meminfo (sysinfo.AutoMemoryLimit = 75% of node RAM): all pre-session work — DuckLake ATTACH, tenant activation, warmup, the controlDB side connection — ran with a memory_limit potentially far above the pod's cgroup limit, and the Go runtime had no ceiling at all. Combined with memory the DuckDB buffer manager does not track (C++ catalog objects for large DuckLake catalogs, postgres_scanner/libpq full-result buffering of metadata reads), this OOM-killed workers that were designed to run slow but never die (observed twice on 64Gi dev workers during pg_catalog metadata queries over a ~3k-table catalog, and reproducibly during CREATE TABLE streams on 16Gi default workers). spawnWorker now stamps three env vars derived from the pod's resolved resource requests, via workerMemoryHygieneEnv: - DUCKGRES_MEMORY_LIMIT: the same 75%-of-pod value the per-session SET uses (shared helper duckdbMemoryLimitForPodMemory extracted from workerDuckDBLimits so the two cannot disagree), bounding the base DB from process start; - GOMEMLIMIT: 1/8 of the pod, a soft ceiling so Go GC pushes back before the cgroup kill; - DUCKGRES_THREADS: the pod's CPU request (fractional cores round up). The worker consumes the first and third via the existing configresolve env surface; GOMEMLIMIT is read by the Go runtime. Tests: TestWorkerMemoryHygieneEnv (shapes incl. sub-GB formatting and the no-requests fallback), extended TestK8sPool_SpawnWorkerCreatesCorrectPod, and new assert_worker_pod checks in tests/e2e-mw-dev/harness.sh pinned to the e2e pool default (1536Mi -> 1GB / 192MiB / 1). --- controlplane/control.go | 32 ++++++++++++------ controlplane/k8s_pool_spawn.go | 41 ++++++++++++++++++++++- controlplane/k8s_pool_test.go | 61 ++++++++++++++++++++++++++++++++++ tests/e2e-mw-dev/harness.sh | 13 ++++++++ 4 files changed, 135 insertions(+), 12 deletions(-) diff --git a/controlplane/control.go b/controlplane/control.go index 439cc992..855680b5 100644 --- a/controlplane/control.go +++ b/controlplane/control.go @@ -1403,17 +1403,7 @@ func (cp *ControlPlane) workerDuckDBLimits(profile *WorkerProfile) (memLimit str } if memReq != "" { - memBytes := parseK8sMemory(memReq) - if memBytes > 0 { - duckdbBytes := memBytes * 3 / 4 // 75% of worker memory for DuckDB - const gb = 1024 * 1024 * 1024 - const mb = 1024 * 1024 - if duckdbBytes >= gb { - memLimit = fmt.Sprintf("%dGB", duckdbBytes/gb) - } else { - memLimit = fmt.Sprintf("%dMB", duckdbBytes/mb) - } - } + memLimit = duckdbMemoryLimitForPodMemory(parseK8sMemory(memReq)) } if cpuReq != "" { @@ -1423,6 +1413,26 @@ func (cp *ControlPlane) workerDuckDBLimits(profile *WorkerProfile) (memLimit str return memLimit, threads } +// duckdbMemoryLimitForPodMemory formats the DuckDB memory_limit for a worker +// pod of the given memory size: 75% of the pod, leaving the remainder as +// headroom for everything memory_limit does NOT govern (DuckDB C++ catalog +// objects, postgres_scanner/libpq result buffers, the Go runtime, page cache). +// Returns "" when memBytes is 0/unparseable (DuckDB then auto-detects). +// Shared between session sizing (workerDuckDBLimits) and the spawn-time +// DUCKGRES_MEMORY_LIMIT env so the two can never disagree. +func duckdbMemoryLimitForPodMemory(memBytes uint64) string { + if memBytes == 0 { + return "" + } + duckdbBytes := memBytes * 3 / 4 // 75% of worker memory for DuckDB + const gb = 1024 * 1024 * 1024 + const mb = 1024 * 1024 + if duckdbBytes >= gb { + return fmt.Sprintf("%dGB", duckdbBytes/gb) + } + return fmt.Sprintf("%dMB", duckdbBytes/mb) +} + // workerBillingSize returns the provisioned worker pod size for compute-usage // billing, in milli-units (millicores, MiB). It mirrors workerDuckDBLimits's // source-of-truth selection: a non-default profile sizes from the profile's pod diff --git a/controlplane/k8s_pool_spawn.go b/controlplane/k8s_pool_spawn.go index e883e539..bb0585c8 100644 --- a/controlplane/k8s_pool_spawn.go +++ b/controlplane/k8s_pool_spawn.go @@ -74,6 +74,10 @@ func (p *K8sWorkerPool) spawnWorker(ctx context.Context, id int, image string, p podLabels["duckgres/org"] = p.orgID } + // Resolve the pod's resource shape once: the container Resources AND the + // memory-hygiene env vars below must derive from the same values. + workerResources := p.workerResourcesForProfile(profile) + // Build pod spec pod := &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{ @@ -152,7 +156,7 @@ func (p *K8sWorkerPool) spawnWorker(ctx context.Context, id int, image string, p SecurityContext: &corev1.SecurityContext{ AllowPrivilegeEscalation: boolPtr(false), }, - Resources: p.workerResourcesForProfile(profile), + Resources: workerResources, }, }, }, @@ -162,6 +166,18 @@ func (p *K8sWorkerPool) spawnWorker(ctx context.Context, id int, image string, p Value: "true", }) + // Pre-session memory hygiene. Without an explicit DUCKGRES_MEMORY_LIMIT the + // worker's ConfigureMainDB falls back to sysinfo.AutoMemoryLimit(), which + // reads the NODE's /proc/meminfo — so all pre-session work (DuckLake + // ATTACH, activation, warmup, controlDB) runs with a memory_limit sized to + // the node, not the pod cgroup. Pass the pod-derived limit (same 75% the + // per-session SET uses, via duckdbMemoryLimitForPodMemory) and the pod's + // CPU count so the base DB is correctly bounded from process start. + // GOMEMLIMIT (read by the Go runtime directly) gives the Go side a soft + // ceiling at 1/8 of the pod so GC pushes back before the cgroup OOM-kills; + // DuckDB's buffer manager is unaffected (C allocations are untracked by Go). + pod.Spec.Containers[0].Env = append(pod.Spec.Containers[0].Env, workerMemoryHygieneEnv(workerResources)...) + // Stamp every log line with pod and node identifiers via the Downward API. pod.Spec.Containers[0].Env = append(pod.Spec.Containers[0].Env, corev1.EnvVar{ @@ -621,6 +637,29 @@ func (p *K8sWorkerPool) workerResources() corev1.ResourceRequirements { return p.workerResourcesForProfile(WorkerProfile{}) } +// workerMemoryHygieneEnv derives the memory/thread env vars for a worker pod +// from its resolved resource requests. See the call site comment in +// spawnWorker for why these must be set before any session exists. +func workerMemoryHygieneEnv(res corev1.ResourceRequirements) []corev1.EnvVar { + var env []corev1.EnvVar + if mem, ok := res.Requests[corev1.ResourceMemory]; ok { + memBytes := uint64(mem.Value()) + if lim := duckdbMemoryLimitForPodMemory(memBytes); lim != "" { + env = append(env, corev1.EnvVar{Name: "DUCKGRES_MEMORY_LIMIT", Value: lim}) + } + if goLimitMiB := memBytes / 8 / (1 << 20); goLimitMiB > 0 { + env = append(env, corev1.EnvVar{Name: "GOMEMLIMIT", Value: fmt.Sprintf("%dMiB", goLimitMiB)}) + } + } + if cpu, ok := res.Requests[corev1.ResourceCPU]; ok { + // Value() rounds fractional cores up, so a "500m" worker gets 1 thread. + if threads := cpu.Value(); threads > 0 { + env = append(env, corev1.EnvVar{Name: "DUCKGRES_THREADS", Value: strconv.FormatInt(threads, 10)}) + } + } + return env +} + // workerResourcesForProfile builds the pod resource requirements for a worker of // the given profile: the profile's CPU/Memory when set, otherwise the pool-global // request (today's behavior). Requests are mirrored to Limits so the pod is diff --git a/controlplane/k8s_pool_test.go b/controlplane/k8s_pool_test.go index bbf37838..abe50f9a 100644 --- a/controlplane/k8s_pool_test.go +++ b/controlplane/k8s_pool_test.go @@ -20,6 +20,7 @@ import ( "github.com/posthog/duckgres/server" corev1 "k8s.io/api/core/v1" k8serrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" @@ -3001,7 +3002,9 @@ func assertSpawnedWorkerPod(t *testing.T, pod *corev1.Pod) { foundTLSCertEnv := false foundTLSKeyEnv := false foundMaxSessionsEnv := false + envByName := map[string]string{} for _, env := range c.Env { + envByName[env.Name] = env.Value if env.Name == "DUCKGRES_DUCKDB_TOKEN" && env.ValueFrom != nil && env.ValueFrom.SecretKeyRef != nil && env.ValueFrom.SecretKeyRef.Name == "test-secret-duckgres-worker-test-cp-0" { @@ -3033,6 +3036,22 @@ func assertSpawnedWorkerPod(t *testing.T, pod *corev1.Pod) { t.Fatal("expected DUCKGRES_DUCKDB_MAX_SESSIONS=1 (one query session per worker)") } + // Pre-session memory hygiene: the worker process must know the POD's + // memory shape from the start. Without these, ConfigureMainDB sizes the + // base DB off the NODE's /proc/meminfo (sysinfo.AutoMemoryLimit) and all + // pre-session work (DuckLake ATTACH, activation) runs effectively + // unbounded, and the Go runtime has no ceiling at all. + // Default pool shape is 16Gi/8cpu -> 75% = 12GB, GOMEMLIMIT = 1/8 pod. + if got := envByName["DUCKGRES_MEMORY_LIMIT"]; got != "12GB" { + t.Fatalf("expected DUCKGRES_MEMORY_LIMIT=12GB (75%% of 16Gi pod), got %q", got) + } + if got := envByName["DUCKGRES_THREADS"]; got != "8" { + t.Fatalf("expected DUCKGRES_THREADS=8 (pod CPU request), got %q", got) + } + if got := envByName["GOMEMLIMIT"]; got != "2048MiB" { + t.Fatalf("expected GOMEMLIMIT=2048MiB (1/8 of 16Gi pod), got %q", got) + } + if len(pod.Spec.Volumes) == 0 { t.Fatal("expected configmap volume") } @@ -4614,3 +4633,45 @@ func TestWorkerDoNotDisruptLifecycle(t *testing.T) { t.Fatalf("do-not-disrupt annotation = %q,%v after protect, want \"true\"", v, ok) } } + +func TestWorkerMemoryHygieneEnv(t *testing.T) { + mk := func(cpu, mem string) corev1.ResourceRequirements { + req := corev1.ResourceList{} + if cpu != "" { + req[corev1.ResourceCPU] = resource.MustParse(cpu) + } + if mem != "" { + req[corev1.ResourceMemory] = resource.MustParse(mem) + } + return corev1.ResourceRequirements{Requests: req} + } + cases := []struct { + name string + res corev1.ResourceRequirements + expected map[string]string + }{ + {"prod-like 15/120Gi", mk("15", "120Gi"), map[string]string{ + "DUCKGRES_MEMORY_LIMIT": "90GB", "GOMEMLIMIT": "15360MiB", "DUCKGRES_THREADS": "15"}}, + {"dev 15/64Gi", mk("15", "64Gi"), map[string]string{ + "DUCKGRES_MEMORY_LIMIT": "48GB", "GOMEMLIMIT": "8192MiB", "DUCKGRES_THREADS": "15"}}, + {"sub-GB pod formats MB", mk("500m", "512Mi"), map[string]string{ + "DUCKGRES_MEMORY_LIMIT": "384MB", "GOMEMLIMIT": "64MiB", "DUCKGRES_THREADS": "1"}}, + {"no requests -> no env", corev1.ResourceRequirements{}, map[string]string{}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := map[string]string{} + for _, e := range workerMemoryHygieneEnv(tc.res) { + got[e.Name] = e.Value + } + if len(got) != len(tc.expected) { + t.Fatalf("env mismatch: got %v want %v", got, tc.expected) + } + for k, v := range tc.expected { + if got[k] != v { + t.Fatalf("%s = %q, want %q (all: %v)", k, got[k], v, got) + } + } + }) + } +} diff --git a/tests/e2e-mw-dev/harness.sh b/tests/e2e-mw-dev/harness.sh index 41dc2b60..d0780711 100755 --- a/tests/e2e-mw-dev/harness.sh +++ b/tests/e2e-mw-dev/harness.sh @@ -852,6 +852,19 @@ assert_worker_pod() { # org dmem="$(k get pod "$pod" -o jsonpath="${WORKER_C}.resources.requests.memory}")" [ "$dcpu" = "750m" ] || fail "default worker $pod requests.cpu='$dcpu' want '750m' (pool default; BestEffort regression?)" [ "$dmem" = "1536Mi" ] || fail "default worker $pod requests.memory='$dmem' want '1536Mi'" + + # Pre-session memory hygiene env (workerMemoryHygieneEnv): the CP must stamp + # a pod-derived DuckDB memory_limit, a Go soft memory ceiling, and the thread + # count at spawn — otherwise the worker sizes its base DB off the NODE's + # /proc/meminfo and all pre-session work (DuckLake ATTACH, activation) runs + # effectively unbounded. Values derive from the 750m/1536Mi pool default: + # 75% of 1536Mi floors to 1GB; GOMEMLIMIT is 1/8 pod = 192MiB; 750m -> 1. + dml="$(k get pod "$pod" -o jsonpath="${WORKER_C}.env[?(@.name==\"DUCKGRES_MEMORY_LIMIT\")].value}")" + [ "$dml" = "1GB" ] || fail "default worker $pod DUCKGRES_MEMORY_LIMIT='$dml' want '1GB' (75% of 1536Mi, GB-floored)" + gml="$(k get pod "$pod" -o jsonpath="${WORKER_C}.env[?(@.name==\"GOMEMLIMIT\")].value}")" + [ "$gml" = "192MiB" ] || fail "default worker $pod GOMEMLIMIT='$gml' want '192MiB' (1/8 of 1536Mi)" + thr="$(k get pod "$pod" -o jsonpath="${WORKER_C}.env[?(@.name==\"DUCKGRES_THREADS\")].value}")" + [ "$thr" = "1" ] || fail "default worker $pod DUCKGRES_THREADS='$thr' want '1' (750m rounds up)" } # ---- worker sizing (TTL-pool model) ---------------------------------------