diff --git a/docs/runbooks/scenario-runner.md b/docs/runbooks/scenario-runner.md index 792f305b..0b05f733 100644 --- a/docs/runbooks/scenario-runner.md +++ b/docs/runbooks/scenario-runner.md @@ -48,14 +48,24 @@ Frozen dataset scenarios additionally require: export DUCKGRES_SCENARIO_FROZEN_S3_URI="s3:///frozen_v1/" ``` -The full suite, fast suite, and targeted frozen perf scenarios exercise PGWire -only. Frozen perf records per-query success and failure rows in -`query_results.csv`. +The targeted frozen perf scenario also requires the standard Duckgres worker +shape deployed with the scenario stack: + +```bash +export DUCKGRES_K8S_WORKER_CPU_REQUEST="3" +export DUCKGRES_K8S_WORKER_MEMORY_REQUEST="12Gi" +``` + +The full and fast suites exercise PGWire only. The targeted frozen perf +scenario compares PGWire and Trino. It records per-query success and failure +rows in `query_results.csv`. Measured query errors fail the perf DAG step after its artifacts are written; independent sibling steps continue to run. A `perf_queries` step can set `with.targets` to a non-empty subset of the -catalog's targets. Duckgres perf catalogs and scenarios are pgwire-only. +catalog's targets. Optional `with.worker_cpu` and `with.worker_memory` values +are sent as PGWire startup options. Both default to empty, which leaves worker +selection to the server; set both for resource-controlled comparisons. Do not commit concrete dev endpoints, secrets, org IDs, or private bucket names. @@ -149,7 +159,7 @@ The frozen dbt scenario uses: dbt artifacts are written under `artifacts/scenario//dbt/`, including per-command stdout/stderr logs, `target/` artifacts, and dbt logs. Install `dbt-postgres` locally or set `DUCKGRES_SCENARIO_DBT_BIN` to the dbt executable to use. -The frozen dbt workload requests a 2 CPU, 4Gi worker through the dbt connection's `duckgres.worker_cpu` and `duckgres.worker_memory` startup options. It also sets `with.connect_timeout: 360`, long enough for the control plane's five-minute worker queue to provision a cold Karpenter node. Other scenario workloads use the isolated control plane's default worker size; `scenario-dev` sets that default to 2 CPU and 8Gi to add process headroom for repeated frozen pgwire aggregates. A `dbt_run` step can opt into a different size or connection window with `with.worker_cpu`, `with.worker_memory`, and `with.connect_timeout`. +The frozen dbt workload requests a 2 CPU, 4Gi worker through the dbt connection's `duckgres.worker_cpu` and `duckgres.worker_memory` startup options. It also sets `with.connect_timeout: 360`, long enough for the control plane's five-minute worker queue to provision a cold Karpenter node. Frozen perf requests the scenario stack's standard worker shape through the same startup GUCs; `scenario-dev` currently configures 3 CPU and 12Gi, matching the aggregate execution resources of its three Trino workers. This explicit sizing bypasses the exploratory worker tier so every measured PGWire query uses the comparison resource budget. A `dbt_run` or `perf_queries` step can opt into a different worker shape with `with.worker_cpu` and `with.worker_memory`. `perf_queries` defaults `with.fail_on_query_errors` to `true`. A measured query error therefore marks that DAG step failed and appears in the scenario result, diff --git a/tests/mw-dev/README.md b/tests/mw-dev/README.md index 6474d0cc..8fed41ed 100644 --- a/tests/mw-dev/README.md +++ b/tests/mw-dev/README.md @@ -97,6 +97,8 @@ verified HTTPS and never disables certificate verification. Optional perf-step settings are: +- `worker_cpu` (default empty, leaving PGWire worker selection to the server) +- `worker_memory` (default empty, leaving PGWire worker selection to the server) - `trino_schema` (default `posthog`) - `trino_ca_cert_file` (default empty, using system roots) - `trino_startup_timeout` (default `2m`) @@ -110,7 +112,10 @@ before warmup or measured statements run. For the isolated mw-dev cell, use scenario Job mounts the per-run CA from `duckgres-trino-tls` and passes that path through `DUCKGRES_SCENARIO_TRINO_CA_CERT`; the perf adapter verifies the coordinator certificate and retries its first authenticated query for the -bounded Secret-projection window. The paired catalog remains the single SQL +bounded Secret-projection window. It also passes the deployed +`DUCKGRES_K8S_WORKER_CPU_REQUEST` and `DUCKGRES_K8S_WORKER_MEMORY_REQUEST` into +the perf step, which requests that exact shape through PGWire startup options +and therefore bypasses the exploratory worker tier. The paired catalog remains the single SQL source: direct-Parquet `raw_view` members run only through PGWire, while each production-shaped `ducklake_table` member runs through both PGWire and Trino. To reproduce the scheduled run, deploy and test with `E2E_SUITE=trino` and the diff --git a/tests/mw-dev/run.sh b/tests/mw-dev/run.sh index dd0e22d4..02902cd7 100755 --- a/tests/mw-dev/run.sh +++ b/tests/mw-dev/run.sh @@ -536,6 +536,8 @@ spec: - { name: DUCKGRES_SCENARIO_FROZEN_S3_URI, value: "$FROZEN_S3_URI" } - { name: DUCKGRES_SCENARIO_TRINO_CA_CERT, value: "/trino-ca/ca.crt" } - { name: DUCKGRES_SCENARIO_DBT_BIN, value: "dbt" } + - { name: DUCKGRES_K8S_WORKER_CPU_REQUEST, value: "$DUCKGRES_K8S_WORKER_CPU_REQUEST" } + - { name: DUCKGRES_K8S_WORKER_MEMORY_REQUEST, value: "$DUCKGRES_K8S_WORKER_MEMORY_REQUEST" } # The Crossplane composition grants this isolated service account # exact-name access to only the matching CNPG credential Secret. - { name: DUCKGRES_SCENARIO_ORG_ID, value: "ci-pr-${PR_NUMBER}-cnpg" } diff --git a/tests/mw-dev/run_sh_test.go b/tests/mw-dev/run_sh_test.go index b8e48030..b683ca0e 100644 --- a/tests/mw-dev/run_sh_test.go +++ b/tests/mw-dev/run_sh_test.go @@ -1156,6 +1156,13 @@ func TestScenarioPodIsProtectedFromKarpenterDisruption(t *testing.T) { if got := annotations["karpenter.sh/do-not-disrupt"]; got != "true" { t.Fatalf("scenario Pod karpenter.sh/do-not-disrupt = %v, want true", got) } + env := deploymentContainerEnv(manifest, "scenario") + if got := env["DUCKGRES_K8S_WORKER_CPU_REQUEST"]; got != "$DUCKGRES_K8S_WORKER_CPU_REQUEST" { + t.Fatalf("scenario worker CPU env = %q, want deployed standard worker CPU", got) + } + if got := env["DUCKGRES_K8S_WORKER_MEMORY_REQUEST"]; got != "$DUCKGRES_K8S_WORKER_MEMORY_REQUEST" { + t.Fatalf("scenario worker memory env = %q, want deployed standard worker memory", got) + } } func TestControlPlaneServiceDoesNotExposeFlight(t *testing.T) { diff --git a/tests/mw-dev/scenario/perf/adapter_test.go b/tests/mw-dev/scenario/perf/adapter_test.go index c893458e..65e6acb8 100644 --- a/tests/mw-dev/scenario/perf/adapter_test.go +++ b/tests/mw-dev/scenario/perf/adapter_test.go @@ -50,6 +50,8 @@ func TestExecutorRunsPerfStepAndWritesArtifacts(t *testing.T) { "catalog_file": catalogPath, "run_id": "scenario-run-1", "dataset_version": "posthog-file-views-v1", + "worker_cpu": "3", + "worker_memory": "12Gi", }, }) if err != nil { @@ -70,6 +72,9 @@ func TestExecutorRunsPerfStepAndWritesArtifacts(t *testing.T) { if strings.Contains(pgwireDSN, "hostaddr=") { t.Fatalf("pgwire dsn = %q, should not use unsupported lib/pq hostaddr", pgwireDSN) } + if !strings.Contains(pgwireDSN, "options='-c duckgres.worker_cpu=3 -c duckgres.worker_memory=12Gi'") { + t.Fatalf("pgwire dsn = %q, want explicit standard worker sizing options", pgwireDSN) + } if factory.pgwireConnection.DialAddress != "10.0.0.10:5432" { t.Fatalf("pgwire direct address = %q, want 10.0.0.10:5432", factory.pgwireConnection.DialAddress) } diff --git a/tests/mw-dev/scenario/perf/steps.go b/tests/mw-dev/scenario/perf/steps.go index 14172491..4bfe679b 100644 --- a/tests/mw-dev/scenario/perf/steps.go +++ b/tests/mw-dev/scenario/perf/steps.go @@ -5,6 +5,7 @@ import ( "fmt" "path/filepath" "strconv" + "strings" "sync" "time" @@ -64,6 +65,8 @@ type stepSpec struct { OutputSubdir string ReadOnly bool FailOnQueryErrors bool + WorkerCPU string + WorkerMemory string TrinoSchema string TrinoCACertFile string TrinoStartup trinodriver.StartupOptions @@ -255,6 +258,8 @@ func (e *Executor) parseStep(step core.Step) (stepSpec, error) { OutputSubdir: stringFromWith(step, "output_subdir", "perf"), ReadOnly: boolFromWith(step, "read_only", true), FailOnQueryErrors: boolFromWith(step, "fail_on_query_errors", true), + WorkerCPU: stringFromWith(step, "worker_cpu", ""), + WorkerMemory: stringFromWith(step, "worker_memory", ""), TrinoSchema: stringFromWith(step, "trino_schema", "posthog"), TrinoCACertFile: stringFromWith(step, "trino_ca_cert_file", ""), TrinoStartup: trinodriver.StartupOptions{ @@ -392,6 +397,19 @@ func (e *Executor) pgwireConnection(spec stepSpec) (scenariosql.PGWireConnection cfg.Database = spec.Database cfg.Username = spec.Username cfg.Password = spec.Password + workerOptions := make([]string, 0, 4) + if spec.WorkerCPU != "" { + workerOptions = append(workerOptions, "-c", "duckgres.worker_cpu="+spec.WorkerCPU) + } + if spec.WorkerMemory != "" { + workerOptions = append(workerOptions, "-c", "duckgres.worker_memory="+spec.WorkerMemory) + } + if len(workerOptions) > 0 { + if cfg.StartupOptions != "" { + cfg.StartupOptions += " " + } + cfg.StartupOptions += strings.Join(workerOptions, " ") + } connection, err := cfg.PGWire() if err != nil { return scenariosql.PGWireConnection{}, classified(ErrorClassConfig, err) diff --git a/tests/mw-dev/scenario/runner_test.go b/tests/mw-dev/scenario/runner_test.go index e94ac3b6..b4e45ee2 100644 --- a/tests/mw-dev/scenario/runner_test.go +++ b/tests/mw-dev/scenario/runner_test.go @@ -164,6 +164,8 @@ func TestProvisionSmokeScenarioUsesIsolatedStackWarehouseIdentityAndSupportedSte func TestFrozenSuccessScenariosUseIsolatedStackWarehouseIdentity(t *testing.T) { const scenarioOrgID = "ci-pr-123-cnpg" t.Setenv("DUCKGRES_SCENARIO_TRINO_CA_CERT", "/tmp/test-trino-ca.crt") + t.Setenv("DUCKGRES_K8S_WORKER_CPU_REQUEST", "3") + t.Setenv("DUCKGRES_K8S_WORKER_MEMORY_REQUEST", "12Gi") for _, scenarioFile := range []string{ "posthog_frozen_metadata.yaml", @@ -450,6 +452,8 @@ func TestFrozenPerfScenarioUsesSupportedStepsAndRelativeCatalog(t *testing.T) { t.Setenv("DUCKGRES_SCENARIO_FROZEN_S3_URI", "s3://example-frozen/frozen_v1/") t.Setenv("DUCKGRES_SCENARIO_ORG_ID", "ci-pr-123-cnpg") t.Setenv("DUCKGRES_SCENARIO_TRINO_CA_CERT", "/tmp/test-trino-ca.crt") + t.Setenv("DUCKGRES_K8S_WORKER_CPU_REQUEST", "3") + t.Setenv("DUCKGRES_K8S_WORKER_MEMORY_REQUEST", "12Gi") scenario, _, err := loadScenarioForRun(filepath.Join("scenarios", "posthog_frozen_perf.yaml")) if err != nil { @@ -490,6 +494,12 @@ func TestFrozenPerfScenarioUsesSupportedStepsAndRelativeCatalog(t *testing.T) { if got, _ := step.With["trino_ca_cert_file"].(string); got != "/tmp/test-trino-ca.crt" { t.Fatalf("perf Trino CA file = %q, want resolved environment path", got) } + if got, _ := step.With["worker_cpu"].(string); got != "3" { + t.Fatalf("perf worker_cpu = %q, want standard worker CPU", got) + } + if got, _ := step.With["worker_memory"].(string); got != "12Gi" { + t.Fatalf("perf worker_memory = %q, want standard worker memory", got) + } } if !foundPerf { t.Fatal("expected frozen perf scenario to include a perf_queries step") @@ -500,6 +510,8 @@ func TestFrozenPerfScenarioBuildsAndValidatesPostHogTablesBeforePerf(t *testing. t.Setenv("DUCKGRES_SCENARIO_FROZEN_S3_URI", "s3://example-frozen/frozen_v1/") t.Setenv("DUCKGRES_SCENARIO_ORG_ID", "ci-pr-123-cnpg") t.Setenv("DUCKGRES_SCENARIO_TRINO_CA_CERT", "/tmp/test-trino-ca.crt") + t.Setenv("DUCKGRES_K8S_WORKER_CPU_REQUEST", "3") + t.Setenv("DUCKGRES_K8S_WORKER_MEMORY_REQUEST", "12Gi") scenario, _, err := loadScenarioForRun(filepath.Join("scenarios", "posthog_frozen_perf.yaml")) if err != nil { diff --git a/tests/mw-dev/scenario/scenarios/posthog_frozen_perf.yaml b/tests/mw-dev/scenario/scenarios/posthog_frozen_perf.yaml index ba6bc8c1..7f7add96 100644 --- a/tests/mw-dev/scenario/scenarios/posthog_frozen_perf.yaml +++ b/tests/mw-dev/scenario/scenarios/posthog_frozen_perf.yaml @@ -4,6 +4,8 @@ required_env: - DUCKGRES_SCENARIO_ORG_ID - DUCKGRES_SCENARIO_FROZEN_S3_URI - DUCKGRES_SCENARIO_TRINO_CA_CERT + - DUCKGRES_K8S_WORKER_CPU_REQUEST + - DUCKGRES_K8S_WORKER_MEMORY_REQUEST steps: - id: provision type: provision_warehouse @@ -81,6 +83,8 @@ steps: run_id: ${run_id} dataset_version: posthog-file-views-v1 fail_on_query_errors: true + worker_cpu: ${env:DUCKGRES_K8S_WORKER_CPU_REQUEST} + worker_memory: ${env:DUCKGRES_K8S_WORKER_MEMORY_REQUEST} - id: deprovision type: deprovision_warehouse diff --git a/tests/mw-dev/scenario/script_test.go b/tests/mw-dev/scenario/script_test.go index 7047ce0f..72cfde35 100644 --- a/tests/mw-dev/scenario/script_test.go +++ b/tests/mw-dev/scenario/script_test.go @@ -51,6 +51,8 @@ func TestScenarioRunScriptCheckEnvIncludesScenarioRequiredEnv(t *testing.T) { "DUCKGRES_SCENARIO_ORG_ID", "DUCKGRES_SCENARIO_FROZEN_S3_URI", "DUCKGRES_SCENARIO_TRINO_CA_CERT", + "DUCKGRES_K8S_WORKER_CPU_REQUEST", + "DUCKGRES_K8S_WORKER_MEMORY_REQUEST", } { if !strings.Contains(text, name) { t.Fatalf("script output %q missing %s", text, name) diff --git a/tests/mw-dev/scenario/sql/connection.go b/tests/mw-dev/scenario/sql/connection.go index 8fea1c58..300002ff 100644 --- a/tests/mw-dev/scenario/sql/connection.go +++ b/tests/mw-dev/scenario/sql/connection.go @@ -23,6 +23,7 @@ type ConnectionConfig struct { SSLMode string ConnectTimeout int ApplicationName string + StartupOptions string } // PGWireConnection separates the hostname used for managed-warehouse TLS @@ -75,6 +76,9 @@ func (c ConnectionConfig) PGWire() (PGWireConnection, error) { if c.ApplicationName != "" { values = append(values, [2]string{"application_name", c.ApplicationName}) } + if c.StartupOptions != "" { + values = append(values, [2]string{"options", c.StartupOptions}) + } parts := make([]string, 0, len(values)) for _, kv := range values {