From 2558df56a8f2f6c954eaaa20c0372f2b21b17b77 Mon Sep 17 00:00:00 2001 From: Bill Guowei Yang Date: Thu, 30 Jul 2026 05:08:16 -0400 Subject: [PATCH 1/2] feat(metrics): record terminal PostgreSQL session starts --- README.md | 1 + controlplane/control.go | 108 ++++++++++++++++++++++------ controlplane/control_cancel_test.go | 91 +++++++++++++++++------ docs/metrics.md | 14 +++- server/observe/metrics.go | 52 +++++++++++++- server/observe/metrics_test.go | 89 ++++++++++++++++++++++- 6 files changed, 304 insertions(+), 51 deletions(-) diff --git a/README.md b/README.md index aac942e5..c50e6dba 100644 --- a/README.md +++ b/README.md @@ -93,6 +93,7 @@ labels, aggregation rules, PromQL examples, and admission metric migration. | `duckgres_session_admission_reclaim_reservation_capacity` | Gauge | Cleanup-ownership slot capacity for this control-plane process (4096 per reclaimer by default) | | `duckgres_session_admission_reclaim_reservation_rejections_total{reason}` | Counter | Reservations rejected because capacity was `full`, the reclaimer was `closed`, or the exact reference was a `duplicate` | | `duckgres_session_start_duration_seconds{org,protocol,outcome}` | Histogram | Authenticated PostgreSQL session bootstrap through flushed `ReadyForQuery` | +| `duckgres_postgres_session_start_total{org,outcome,failure_class}` | Counter | Exactly one terminal result per authenticated PostgreSQL session start after server retries; `outcome` is `success\|failure` and failure classes distinguish operator-actionable failures from client/lifecycle noise | | `duckgres_flight_rpc_duration_seconds{method}` | Histogram | Flight ingress RPC duration by method | | `duckgres_flight_ingress_sessions_total{outcome}` | Counter | Flight ingress session outcomes (`created|reused|auth_failed|rate_limited|create_failed|token_invalid`) | | `duckgres_flight_sessions_reaped_total{trigger}` | Counter | Number of Flight auth sessions reaped (`trigger=periodic|forced`) | diff --git a/controlplane/control.go b/controlplane/control.go index 25b67bd4..de21deab 100644 --- a/controlplane/control.go +++ b/controlplane/control.go @@ -1144,7 +1144,7 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { server.RecordSuccessfulAuthAttempt(cp.rateLimiter, remoteAddr) clog.Info("User authenticated.") sessionStart := observe.BeginSessionStart(orgID, "postgres") - defer sessionStart.Finish("error") + defer sessionStart.Finish("error", observe.SessionStartFailureUnknown) // Resolve the requested worker shape from the connection-string startup // options (duckgres.worker_cpu / worker_memory / worker_ttl), layered on @@ -1159,6 +1159,7 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { } workerProfile, profileWarns, orgProfileApplied, profileErr := cp.resolveWorkerProfile(startupOptions, orgProfileDefaults) if profileErr != nil { + sessionStart.Finish("error", observe.SessionStartFailureClient) clog.Warn("Rejected worker profile.", "error", profileErr) _ = server.WriteErrorResponse(writer, "FATAL", "22023", profileErr.Error()) _ = writer.Flush() @@ -1181,6 +1182,7 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { rawS3Cache, s3CacheRequested := startupOptions[server.S3CacheGUCName] if s3CacheRequested { if err := server.ValidateS3CacheOption(rawS3Cache); err != nil { + sessionStart.Finish("error", observe.SessionStartFailureClient) clog.Warn("Rejected s3_cache startup option.", "error", err) _ = server.WriteErrorResponse(writer, "FATAL", "22023", err.Error()) _ = writer.Flush() @@ -1197,6 +1199,7 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { // hitting a partially-migrated catalog. The client gets a clear error // and can retry after the migration completes. if cp.orgRouter.IsMigratingForOrg(orgID) { + sessionStart.Finish("error", observe.SessionStartFailureLifecycle) clog.Info("Connection rejected during DuckLake migration.") _ = server.WriteErrorResponse(writer, "FATAL", "57P03", "DuckLake catalog upgrade in progress for your organization, please retry in a few moments") @@ -1212,6 +1215,7 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { // connection advisory lock.) if whState, ok := cp.configStore.OrgWarehouseStatus(orgID); ok && whState == string(configstore.ManagedWarehouseStateResharding) { + sessionStart.Finish("error", observe.SessionStartFailureLifecycle) clog.Info("Connection rejected during metadata-store reshard.") _ = server.WriteErrorResponse(writer, "FATAL", "57P03", "metadata-store reshard in progress for your organization, please retry shortly") @@ -1226,6 +1230,7 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { // the stack absence is expected and the client should be told to // retry rather than receive a misleading auth-style error. whState, orgExists := cp.configStore.OrgWarehouseStatus(orgID) + sessionStart.Finish("error", missingOrgStackFailureClass(whState, orgExists)) switch { case !orgExists: _ = server.WriteErrorResponse(writer, "FATAL", "28000", "no org configured for user") @@ -1262,14 +1267,14 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { rebalancer = cp.rebalancer } if cp.isDraining() { - sessionStart.Finish("draining") + sessionStart.Finish("draining", observe.SessionStartFailureLifecycle) _ = server.WriteErrorResponse(writer, "FATAL", "57P03", "control plane is draining, retry shortly") _ = writer.Flush() return } preReadyCtx, finishPreReady, err := cp.beginPreReadyHandshake(context.Background()) if err != nil { - sessionStart.Finish("draining") + sessionStart.Finish("draining", observe.SessionStartFailureLifecycle) _ = server.WriteErrorResponse(writer, "FATAL", "57P03", "control plane is draining, retry shortly") _ = writer.Flush() return @@ -1286,6 +1291,7 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { defer server.CancelClientConn(tmpCC) server.SendInitialParams(tmpCC) if err := writer.Flush(); err != nil { + sessionStart.Finish("error", observe.SessionStartFailureTransport) clog.Error("Failed to flush initial params.", "error", err) return } @@ -1330,7 +1336,8 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { sessions.DestroySession(createdPID) executor = nil } - sessionStart.Finish(controlPlaneSessionStartOutcome(err)) + outcome, failureClass := controlPlaneSessionStartResult(err) + sessionStart.Finish(outcome, failureClass) clog.Error("Failed to create session.", "error", err) code, message := sessionCreationErrorResponse(err) _ = server.WriteErrorResponse(writer, "FATAL", code, message) @@ -1360,7 +1367,13 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { attachContextErr := attachCtx.Err() attachCancel() if probeErr != nil { - sessionStart.Finish(controlPlaneSessionStartOperationOutcome(probeErr, attachContextErr, cp.isDraining())) + outcome, failureClass := controlPlaneSessionStartOperationResult( + probeErr, + attachContextErr, + cp.isDraining(), + observe.SessionStartFailureMetadataStore, + ) + sessionStart.Finish(outcome, failureClass) clog.Error("Failed to detect attached catalogs.", "error", probeErr) _ = server.WriteErrorResponse(writer, "FATAL", "XX000", "failed to detect attached catalogs") _ = writer.Flush() @@ -1371,6 +1384,7 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { var ok bool effectiveCatalog, ok = resolveEffectiveCatalog(requestedCatalog, duckLakeAttached) if !ok { + sessionStart.Finish("error", observe.SessionStartFailureMetadataStore) clog.Warn("Postgres connection rejected: requested catalog is not available for this connection.", "requested", requestedCatalog, "ducklake_attached", duckLakeAttached) msg := "no catalog is available for this connection" @@ -1413,7 +1427,13 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { if err := sessionmeta.InitSessionDatabaseMetadataWithAccess(initCtx, executor, effectiveCatalog, metadataAccess); err != nil { initContextErr := initCtx.Err() initCancel() - sessionStart.Finish(controlPlaneSessionStartOperationOutcome(err, initContextErr, cp.isDraining())) + outcome, failureClass := controlPlaneSessionStartOperationResult( + err, + initContextErr, + cp.isDraining(), + observe.SessionStartFailureMetadataStore, + ) + sessionStart.Finish(outcome, failureClass) clog.Error("Failed to initialize session database metadata.", "database", database, "error", err) _ = server.WriteErrorResponse(writer, "FATAL", "XX000", "failed to initialize session database metadata") _ = writer.Flush() @@ -1433,7 +1453,13 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { spCancel() if err != nil { if source == sessionDefaultSourceConfiguredCatalog { - sessionStart.Finish(controlPlaneSessionStartOperationOutcome(err, spContextErr, cp.isDraining())) + outcome, failureClass := controlPlaneSessionStartOperationResult( + err, + spContextErr, + cp.isDraining(), + observe.SessionStartFailureMetadataStore, + ) + sessionStart.Finish(outcome, failureClass) clog.Error("Failed to apply session default catalog.", "catalog", effectiveCatalog, "error", err) _ = server.WriteErrorResponse(writer, "FATAL", "XX000", "failed to apply default catalog") _ = writer.Flush() @@ -1456,7 +1482,13 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { initContextErr := initCtx.Err() initCancel() if err != nil { - sessionStart.Finish(controlPlaneSessionStartOperationOutcome(err, initContextErr, cp.isDraining())) + outcome, failureClass := controlPlaneSessionStartOperationResult( + err, + initContextErr, + cp.isDraining(), + observe.SessionStartFailureMetadataStore, + ) + sessionStart.Finish(outcome, failureClass) clog.Error("Failed to apply passthrough session default catalog.", "command", cmd, "error", err) _ = server.WriteErrorResponse(writer, "FATAL", "XX000", "failed to apply default catalog") _ = writer.Flush() @@ -1521,6 +1553,7 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { // s3_cache=off must never silently run through the cache. if s3CacheRequested { if err := server.ApplyConnectionS3CacheOption(cc, rawS3Cache); err != nil { + sessionStart.Finish("error", observe.SessionStartFailureWorker) clog.Error("Failed to apply s3_cache startup option.", "error", err) _ = server.WriteErrorResponse(writer, "FATAL", "XX000", err.Error()) _ = writer.Flush() @@ -1534,24 +1567,26 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { disconnect := disconnectWatcher.Stop() drained := finishPreReady() if disconnect.ClientCanceled { - sessionStart.Finish("canceled") + sessionStart.Finish("canceled", observe.SessionStartFailureCanceled) return } if drained { - sessionStart.Finish("draining") + sessionStart.Finish("draining", observe.SessionStartFailureLifecycle) return } // Send ReadyForQuery to signal that the handshake is complete if err := server.WriteReadyForQuery(writer, 'I'); err != nil { + sessionStart.Finish("error", observe.SessionStartFailureTransport) clog.Error("Failed to send ReadyForQuery.", "error", err) return } if err := writer.Flush(); err != nil { + sessionStart.Finish("error", observe.SessionStartFailureTransport) clog.Error("Failed to flush writer.", "error", err) return } - sessionStart.Finish("success") + sessionStart.Finish("success", observe.SessionStartFailureNone) if orgID != "" { observeOrgPgSessionAccepted(orgID, passthroughUser) } @@ -1564,33 +1599,60 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { } } -func controlPlaneSessionStartOutcome(err error) string { +func controlPlaneSessionStartResult(err error) (outcome, failureClass string) { var capacityErr *WorkerCapacityExhaustedError var rejection *configstore.OrgConnectionAdmissionRejectedError switch { case err == nil: - return "success" - case errors.As(err, &capacityErr), errors.As(err, &rejection): - return "capacity" + return "success", observe.SessionStartFailureNone + case errors.As(err, &capacityErr): + return "capacity", observe.SessionStartFailureCapacity + case errors.As(err, &rejection): + return "capacity", observe.SessionStartFailureClient case errors.Is(err, context.Canceled): - return "canceled" + return "canceled", observe.SessionStartFailureCanceled case errors.Is(err, context.DeadlineExceeded), errors.Is(err, ErrTooManyConnections): - return "timeout" + return "timeout", observe.SessionStartFailureCapacity case errors.Is(err, ErrSessionManagerDraining): - return "draining" + return "draining", observe.SessionStartFailureLifecycle default: - return "error" + return "error", observe.SessionStartFailureWorker } } -func controlPlaneSessionStartOperationOutcome(err, contextErr error, draining bool) string { +func controlPlaneSessionStartOperationResult( + err, contextErr error, + draining bool, + failureClass string, +) (outcome, classifiedFailure string) { if draining { - return "draining" + return "draining", observe.SessionStartFailureLifecycle } + terminalErr := err if contextErr != nil { - return controlPlaneSessionStartOutcome(contextErr) + terminalErr = contextErr + } + outcome, _ = controlPlaneSessionStartResult(terminalErr) + switch outcome { + case "success": + return outcome, observe.SessionStartFailureNone + case "canceled": + return outcome, observe.SessionStartFailureCanceled + case "draining": + return outcome, observe.SessionStartFailureLifecycle + default: + return outcome, failureClass + } +} + +func missingOrgStackFailureClass(warehouseState string, orgExists bool) string { + if !orgExists { + return observe.SessionStartFailureClient + } + if warehouseState == "" || warehouseState == string(configstore.ManagedWarehouseStateReady) { + return observe.SessionStartFailureControlPlane } - return controlPlaneSessionStartOutcome(err) + return observe.SessionStartFailureLifecycle } func authorizedClientSearchPath(searchPath string, policy *server.QueryAccessPolicy) string { diff --git a/controlplane/control_cancel_test.go b/controlplane/control_cancel_test.go index 77f87f5e..fa8ae761 100644 --- a/controlplane/control_cancel_test.go +++ b/controlplane/control_cancel_test.go @@ -11,6 +11,7 @@ import ( "github.com/posthog/duckgres/controlplane/configstore" "github.com/posthog/duckgres/server" "github.com/posthog/duckgres/server/flightclient" + "github.com/posthog/duckgres/server/observe" ) func TestCreateSessionWithRegisteredCancel_CancelQueryCancelsWait(t *testing.T) { @@ -196,18 +197,19 @@ func TestSessionCreationErrorResponse(t *testing.T) { }) } -func TestControlPlaneSessionStartOutcome(t *testing.T) { +func TestControlPlaneSessionStartResult(t *testing.T) { tests := []struct { - name string - err error - want string + name string + err error + wantOutcome string + wantFailureClass string }{ - {name: "success", want: "success"}, - {name: "canceled", err: context.Canceled, want: "canceled"}, - {name: "deadline", err: context.DeadlineExceeded, want: "timeout"}, - {name: "queue timeout", err: ErrTooManyConnections, want: "timeout"}, - {name: "draining", err: ErrSessionManagerDraining, want: "draining"}, - {name: "worker capacity", err: NewWorkerCapacityExhaustedError(time.Second), want: "capacity"}, + {name: "success", wantOutcome: "success", wantFailureClass: observe.SessionStartFailureNone}, + {name: "canceled", err: context.Canceled, wantOutcome: "canceled", wantFailureClass: observe.SessionStartFailureCanceled}, + {name: "deadline", err: context.DeadlineExceeded, wantOutcome: "timeout", wantFailureClass: observe.SessionStartFailureCapacity}, + {name: "queue timeout", err: ErrTooManyConnections, wantOutcome: "timeout", wantFailureClass: observe.SessionStartFailureCapacity}, + {name: "draining", err: ErrSessionManagerDraining, wantOutcome: "draining", wantFailureClass: observe.SessionStartFailureLifecycle}, + {name: "worker capacity", err: NewWorkerCapacityExhaustedError(time.Second), wantOutcome: "capacity", wantFailureClass: observe.SessionStartFailureCapacity}, { name: "admission hard rejection", err: &configstore.OrgConnectionAdmissionRejectedError{ @@ -215,33 +217,76 @@ func TestControlPlaneSessionStartOutcome(t *testing.T) { RequestedVCPUs: 4, MaximumVCPUs: 2, }, - want: "capacity", + wantOutcome: "capacity", + wantFailureClass: observe.SessionStartFailureClient, }, - {name: "generic error", err: errors.New("bootstrap failed"), want: "error"}, + {name: "generic error", err: errors.New("bootstrap failed"), wantOutcome: "error", wantFailureClass: observe.SessionStartFailureWorker}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - if got := controlPlaneSessionStartOutcome(tt.err); got != tt.want { - t.Fatalf("controlPlaneSessionStartOutcome(%v) = %q, want %q", tt.err, got, tt.want) + gotOutcome, gotFailureClass := controlPlaneSessionStartResult(tt.err) + if gotOutcome != tt.wantOutcome || gotFailureClass != tt.wantFailureClass { + t.Fatalf("controlPlaneSessionStartResult(%v) = (%q, %q), want (%q, %q)", + tt.err, gotOutcome, gotFailureClass, tt.wantOutcome, tt.wantFailureClass) } }) } } -func TestControlPlaneSessionStartOperationOutcomePrefersContext(t *testing.T) { +func TestControlPlaneSessionStartOperationResultPrefersContext(t *testing.T) { operationErr := errors.New("metadata initialization failed") - if got := controlPlaneSessionStartOperationOutcome(operationErr, context.Canceled, false); got != "canceled" { - t.Fatalf("canceled operation outcome = %q, want canceled", got) + tests := []struct { + name string + contextErr error + draining bool + wantOutcome string + wantFailureClass string + }{ + {name: "client canceled", contextErr: context.Canceled, wantOutcome: "canceled", wantFailureClass: observe.SessionStartFailureCanceled}, + {name: "operation timed out", contextErr: context.DeadlineExceeded, wantOutcome: "timeout", wantFailureClass: observe.SessionStartFailureMetadataStore}, + {name: "ordinary metadata error", wantOutcome: "error", wantFailureClass: observe.SessionStartFailureMetadataStore}, + {name: "draining wins", contextErr: context.Canceled, draining: true, wantOutcome: "draining", wantFailureClass: observe.SessionStartFailureLifecycle}, } - if got := controlPlaneSessionStartOperationOutcome(operationErr, context.DeadlineExceeded, false); got != "timeout" { - t.Fatalf("timed-out operation outcome = %q, want timeout", got) + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotOutcome, gotFailureClass := controlPlaneSessionStartOperationResult( + operationErr, + tt.contextErr, + tt.draining, + observe.SessionStartFailureMetadataStore, + ) + if gotOutcome != tt.wantOutcome || gotFailureClass != tt.wantFailureClass { + t.Fatalf("operation result = (%q, %q), want (%q, %q)", + gotOutcome, gotFailureClass, tt.wantOutcome, tt.wantFailureClass) + } + }) } - if got := controlPlaneSessionStartOperationOutcome(operationErr, nil, false); got != "error" { - t.Fatalf("ordinary operation outcome = %q, want error", got) +} + +func TestMissingOrgStackFailureClass(t *testing.T) { + tests := []struct { + name string + state string + orgExists bool + want string + }{ + {name: "missing org", orgExists: false, want: observe.SessionStartFailureClient}, + {name: "legacy ready state", orgExists: true, state: "", want: observe.SessionStartFailureControlPlane}, + {name: "ready warehouse", orgExists: true, state: string(configstore.ManagedWarehouseStateReady), want: observe.SessionStartFailureControlPlane}, + {name: "failed provisioning", orgExists: true, state: string(configstore.ManagedWarehouseStateFailed), want: observe.SessionStartFailureLifecycle}, + {name: "deleting", orgExists: true, state: string(configstore.ManagedWarehouseStateDeleting), want: observe.SessionStartFailureLifecycle}, + {name: "resharding", orgExists: true, state: string(configstore.ManagedWarehouseStateResharding), want: observe.SessionStartFailureLifecycle}, + {name: "pending", orgExists: true, state: string(configstore.ManagedWarehouseStatePending), want: observe.SessionStartFailureLifecycle}, } - if got := controlPlaneSessionStartOperationOutcome(operationErr, context.Canceled, true); got != "draining" { - t.Fatalf("drain-canceled operation outcome = %q, want draining", got) + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := missingOrgStackFailureClass(tt.state, tt.orgExists); got != tt.want { + t.Fatalf("missingOrgStackFailureClass(%q, %t) = %q, want %q", tt.state, tt.orgExists, got, tt.want) + } + }) } } diff --git a/docs/metrics.md b/docs/metrics.md index ca2620b3..b2385abd 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -33,7 +33,7 @@ Histograms expose the usual `_bucket`, `_count`, and `_sum` series. | Admission queue | `duckgres_session_admission_wait_seconds`, `duckgres_session_admission_requests_total` | After a request is successfully enqueued until grant, hard rejection, timeout, cancellation, or evaluation error. Enqueue failures are excluded. | | Admission state | `duckgres_session_admission_queue_depth`, `duckgres_session_admission_active_vcpus`, `duckgres_session_admission_limit_vcpus` | Local waiting callers, live local lease handles, and the effective org cap reconciled for active org stacks from each control-plane process's current config snapshot. | | Worker acquisition | `duckgres_worker_acquire_*` | After admission grants until an existing, hot-idle, or newly spawned worker is allocated. | -| Session start | `duckgres_session_start_duration_seconds` | After successful PostgreSQL authentication until `ReadyForQuery` is flushed or session bootstrap terminates. | +| Session start | `duckgres_session_start_duration_seconds`, `duckgres_postgres_session_start_total` | After successful PostgreSQL authentication until `ReadyForQuery` is flushed or session bootstrap terminates. The counter records exactly one terminal result after server-side retries. | | Query | `duckgres_query_total`, `duckgres_query_duration_seconds` | One non-empty query attempt and its execution duration. | Session start includes profile resolution, admission, worker @@ -52,6 +52,7 @@ session defaults, and the final ready flush. It excludes failed authentication. | `duckgres_session_admission_active_vcpus` | Gauge | `org` | Requested vCPUs held by live local lease handles. It is admitted capacity, not measured CPU usage or the exact durable lease-row total. | | `duckgres_session_admission_limit_vcpus` | Gauge | `org` | Effective org cap for an active org stack, reconciled from this process's current config snapshot. `0` means unlimited. | | `duckgres_session_start_duration_seconds` | Histogram | `org`, `protocol`, `outcome` | Authenticated PostgreSQL create-to-ready latency. | +| `duckgres_postgres_session_start_total` | Counter | `org`, `outcome`, `failure_class` | Exactly one terminal authenticated PostgreSQL session-start result after server-side retries. | Admission request outcomes are `granted`, `rejected`, `timeout`, `canceled`, and `error`. `rejected` means the requested worker shape can never fit its hard @@ -59,6 +60,17 @@ organization or user vCPU ceiling. Session-start outcomes are `success`, `timeout`, `canceled`, `capacity`, `draining`, and `error`. +The PostgreSQL terminal counter collapses those outcomes to `success` or +`failure`. Success always has `failure_class="none"`. Failure classes are +`capacity`, `worker`, `metadata_store`, `control_plane`, `client`, `lifecycle`, +`canceled`, `transport`, and `unknown`. The first four represent failures an +operator can usually alleviate. The remaining classes let alerts exclude bad +client input, planned lifecycle transitions, client disconnects, wire errors, +and newly added paths that have not yet been classified. Flight SQL does not +emit this counter. `capacity` covers runtime worker exhaustion and admission +timeouts; requests that exceed a configured hard org or user vCPU limit are +classified as `client`. + Evaluation decisions are `granted_current`, `already_granted`, `rejected`, `blocked`, `waiting`, `inactive`, `missing`, `canceled`, `timeout`, and `error`. Each evaluation describes only the polling request. Evaluation reasons are diff --git a/server/observe/metrics.go b/server/observe/metrics.go index 8207c917..8cc75556 100644 --- a/server/observe/metrics.go +++ b/server/observe/metrics.go @@ -41,6 +41,27 @@ func ObserveConnectionDuration(org string, seconds float64) { connectionDurationHistogram.WithLabelValues(org).Observe(seconds) } +// PostgreSQL session-start failure classes are intentionally bounded. Alerting +// may allowlist operator-actionable classes, while client, lifecycle, transport, +// and unknown failures remain available for diagnosis without paging. +const ( + SessionStartFailureNone = "none" + SessionStartFailureCapacity = "capacity" + SessionStartFailureWorker = "worker" + SessionStartFailureMetadataStore = "metadata_store" + SessionStartFailureControlPlane = "control_plane" + SessionStartFailureClient = "client" + SessionStartFailureLifecycle = "lifecycle" + SessionStartFailureCanceled = "canceled" + SessionStartFailureTransport = "transport" + SessionStartFailureUnknown = "unknown" +) + +var postgresSessionStartCounter = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "duckgres_postgres_session_start_total", + Help: "Terminal outcomes for authenticated PostgreSQL session starts after all server-side retries.", +}, []string{"org", "outcome", "failure_class"}) + var sessionStartDurationHistogram = promauto.NewHistogramVec(prometheus.HistogramOpts{ Name: "duckgres_session_start_duration_seconds", Help: "Authenticated session bootstrap time through protocol readiness, partitioned by org, protocol, and terminal outcome.", @@ -48,7 +69,7 @@ var sessionStartDurationHistogram = promauto.NewHistogramVec(prometheus.Histogra }, []string{"org", "protocol", "outcome"}) // SessionStartScope records one logical protocol bootstrap. Finish is -// idempotent so a default-error defer can coexist with an explicit success. +// idempotent so a default-error defer can coexist with an explicit result. type SessionStartScope struct { started time.Time org string @@ -65,7 +86,7 @@ func BeginSessionStart(org, protocol string) *SessionStartScope { return &SessionStartScope{started: time.Now(), org: org, protocol: protocol} } -func (s *SessionStartScope) Finish(outcome string) { +func (s *SessionStartScope) Finish(outcome, failureClass string) { if s == nil { return } @@ -74,15 +95,42 @@ func (s *SessionStartScope) Finish(outcome string) { default: outcome = "error" } + terminalOutcome, failureClass := normalizePostgresSessionStartResult(outcome, failureClass) s.once.Do(func() { duration := time.Since(s.started) if duration < 0 { duration = 0 } sessionStartDurationHistogram.WithLabelValues(s.org, s.protocol, outcome).Observe(duration.Seconds()) + if s.protocol == "postgres" { + postgresSessionStartCounter.WithLabelValues(s.org, terminalOutcome, failureClass).Inc() + } }) } +func normalizePostgresSessionStartResult(outcome, failureClass string) (string, string) { + if outcome == "success" { + return "success", SessionStartFailureNone + } + + switch failureClass { + case SessionStartFailureCapacity, + SessionStartFailureWorker, + SessionStartFailureMetadataStore, + SessionStartFailureControlPlane, + SessionStartFailureClient, + SessionStartFailureLifecycle, + SessionStartFailureCanceled, + SessionStartFailureTransport, + SessionStartFailureUnknown: + return "failure", failureClass + default: + // A failed session must never be reported with class=none, and newly + // added unclassified paths must stay outside paging allowlists. + return "failure", SessionStartFailureUnknown + } +} + // S3BytesReadTotal counts bytes read from S3 by DuckDB, labeled by org. // Bumped from EnrichSpanWithProfiling when DuckDB reports total_bytes_read // in its profiling output. diff --git a/server/observe/metrics_test.go b/server/observe/metrics_test.go index 09707dea..d6008d5d 100644 --- a/server/observe/metrics_test.go +++ b/server/observe/metrics_test.go @@ -1,6 +1,7 @@ package observe import ( + "fmt" "testing" "time" @@ -85,10 +86,12 @@ func TestObserveConnectionDurationRecordsPerOrgSample(t *testing.T) { func TestSessionStartScopeRecordsExactlyOnce(t *testing.T) { const org = "session-start-test-org" before := sessionStartSampleCount(t, org, "postgres", "success") + terminalBefore := postgresSessionStartCount(t, org, "success", SessionStartFailureNone) + failureBefore := postgresSessionStartCount(t, org, "failure", SessionStartFailureControlPlane) scope := BeginSessionStart(org, "postgres") - scope.Finish("success") - scope.Finish("error") + scope.Finish("success", SessionStartFailureControlPlane) + scope.Finish("error", SessionStartFailureControlPlane) if got := sessionStartSampleCount(t, org, "postgres", "success"); got != before+1 { t.Fatalf("success sample count = %d, want %d", got, before+1) @@ -96,6 +99,88 @@ func TestSessionStartScopeRecordsExactlyOnce(t *testing.T) { if got := sessionStartSampleCount(t, org, "postgres", "error"); got != 0 { t.Fatalf("error sample count = %d, want 0", got) } + if got := postgresSessionStartCount(t, org, "success", SessionStartFailureNone); got != terminalBefore+1 { + t.Fatalf("terminal success count = %v, want %v", got, terminalBefore+1) + } + if got := postgresSessionStartCount(t, org, "failure", SessionStartFailureControlPlane); got != failureBefore { + t.Fatalf("second Finish changed terminal classification: count = %v, want %v", got, failureBefore) + } +} + +func TestSessionStartScopeRecordsPostgresTerminalFailure(t *testing.T) { + const org = "session-start-terminal-failure-org" + before := postgresSessionStartCount(t, org, "failure", SessionStartFailureMetadataStore) + histogramBefore := sessionStartSampleCount(t, org, "postgres", "timeout") + + BeginSessionStart(org, "postgres").Finish("timeout", SessionStartFailureMetadataStore) + + if got := postgresSessionStartCount(t, org, "failure", SessionStartFailureMetadataStore); got != before+1 { + t.Fatalf("terminal failure count = %v, want %v", got, before+1) + } + if got := sessionStartSampleCount(t, org, "postgres", "timeout"); got != histogramBefore+1 { + t.Fatalf("timeout histogram count = %d, want %d", got, histogramBefore+1) + } +} + +func TestSessionStartScopeNormalizesUnsafeFailureClasses(t *testing.T) { + tests := []struct { + name string + outcome string + failureClass string + wantOutcome string + wantClass string + }{ + { + name: "success always uses none", + outcome: "success", + failureClass: SessionStartFailureWorker, + wantOutcome: "success", + wantClass: SessionStartFailureNone, + }, + { + name: "failure cannot use none", + outcome: "error", + failureClass: SessionStartFailureNone, + wantOutcome: "failure", + wantClass: SessionStartFailureUnknown, + }, + { + name: "unknown class stays non-pageable", + outcome: "error", + failureClass: "new-unclassified-path", + wantOutcome: "failure", + wantClass: SessionStartFailureUnknown, + }, + } + + for i, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + org := fmt.Sprintf("session-start-normalization-%d", i) + before := postgresSessionStartCount(t, org, tt.wantOutcome, tt.wantClass) + + BeginSessionStart(org, "postgres").Finish(tt.outcome, tt.failureClass) + + if got := postgresSessionStartCount(t, org, tt.wantOutcome, tt.wantClass); got != before+1 { + t.Fatalf("normalized terminal count = %v, want %v", got, before+1) + } + }) + } +} + +func TestSessionStartScopeDoesNotExportFlightTerminalCounter(t *testing.T) { + const org = "session-start-obsolete-flight-org" + before := postgresSessionStartCount(t, org, "failure", SessionStartFailureMetadataStore) + + BeginSessionStart(org, "flight").Finish("error", SessionStartFailureMetadataStore) + + if got := postgresSessionStartCount(t, org, "failure", SessionStartFailureMetadataStore); got != before { + t.Fatalf("obsolete Flight path changed PostgreSQL terminal count: got %v, want %v", got, before) + } +} + +func postgresSessionStartCount(t *testing.T, org, outcome, failureClass string) float64 { + t.Helper() + return counterVecValue(t, postgresSessionStartCounter, org, outcome, failureClass) } func sessionStartSampleCount(t *testing.T, org, protocol, outcome string) uint64 { From 9ef9cefffb6dd06b5b7e5d77655d0bb5723aaab6 Mon Sep 17 00:00:00 2001 From: Bill Guowei Yang Date: Thu, 30 Jul 2026 05:28:44 -0400 Subject: [PATCH 2/2] refactor(metrics): use reason for session-start causes --- README.md | 2 +- controlplane/control.go | 98 ++++++++++++++--------------- controlplane/control_cancel_test.go | 78 +++++++++++------------ docs/metrics.md | 8 +-- server/observe/metrics.go | 62 +++++++++--------- server/observe/metrics_test.go | 87 ++++++++++++++----------- 6 files changed, 173 insertions(+), 162 deletions(-) diff --git a/README.md b/README.md index c50e6dba..e99f70f5 100644 --- a/README.md +++ b/README.md @@ -93,7 +93,7 @@ labels, aggregation rules, PromQL examples, and admission metric migration. | `duckgres_session_admission_reclaim_reservation_capacity` | Gauge | Cleanup-ownership slot capacity for this control-plane process (4096 per reclaimer by default) | | `duckgres_session_admission_reclaim_reservation_rejections_total{reason}` | Counter | Reservations rejected because capacity was `full`, the reclaimer was `closed`, or the exact reference was a `duplicate` | | `duckgres_session_start_duration_seconds{org,protocol,outcome}` | Histogram | Authenticated PostgreSQL session bootstrap through flushed `ReadyForQuery` | -| `duckgres_postgres_session_start_total{org,outcome,failure_class}` | Counter | Exactly one terminal result per authenticated PostgreSQL session start after server retries; `outcome` is `success\|failure` and failure classes distinguish operator-actionable failures from client/lifecycle noise | +| `duckgres_postgres_session_start_total{org,outcome,reason}` | Counter | Exactly one terminal result per authenticated PostgreSQL session start after server retries; `outcome` is `success\|failure` and bounded reasons distinguish operator-actionable failures from client/lifecycle noise | | `duckgres_flight_rpc_duration_seconds{method}` | Histogram | Flight ingress RPC duration by method | | `duckgres_flight_ingress_sessions_total{outcome}` | Counter | Flight ingress session outcomes (`created|reused|auth_failed|rate_limited|create_failed|token_invalid`) | | `duckgres_flight_sessions_reaped_total{trigger}` | Counter | Number of Flight auth sessions reaped (`trigger=periodic|forced`) | diff --git a/controlplane/control.go b/controlplane/control.go index de21deab..8a655edf 100644 --- a/controlplane/control.go +++ b/controlplane/control.go @@ -1144,7 +1144,7 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { server.RecordSuccessfulAuthAttempt(cp.rateLimiter, remoteAddr) clog.Info("User authenticated.") sessionStart := observe.BeginSessionStart(orgID, "postgres") - defer sessionStart.Finish("error", observe.SessionStartFailureUnknown) + defer sessionStart.Finish("error", observe.SessionStartReasonUnknown) // Resolve the requested worker shape from the connection-string startup // options (duckgres.worker_cpu / worker_memory / worker_ttl), layered on @@ -1159,7 +1159,7 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { } workerProfile, profileWarns, orgProfileApplied, profileErr := cp.resolveWorkerProfile(startupOptions, orgProfileDefaults) if profileErr != nil { - sessionStart.Finish("error", observe.SessionStartFailureClient) + sessionStart.Finish("error", observe.SessionStartReasonClient) clog.Warn("Rejected worker profile.", "error", profileErr) _ = server.WriteErrorResponse(writer, "FATAL", "22023", profileErr.Error()) _ = writer.Flush() @@ -1182,7 +1182,7 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { rawS3Cache, s3CacheRequested := startupOptions[server.S3CacheGUCName] if s3CacheRequested { if err := server.ValidateS3CacheOption(rawS3Cache); err != nil { - sessionStart.Finish("error", observe.SessionStartFailureClient) + sessionStart.Finish("error", observe.SessionStartReasonClient) clog.Warn("Rejected s3_cache startup option.", "error", err) _ = server.WriteErrorResponse(writer, "FATAL", "22023", err.Error()) _ = writer.Flush() @@ -1199,7 +1199,7 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { // hitting a partially-migrated catalog. The client gets a clear error // and can retry after the migration completes. if cp.orgRouter.IsMigratingForOrg(orgID) { - sessionStart.Finish("error", observe.SessionStartFailureLifecycle) + sessionStart.Finish("error", observe.SessionStartReasonLifecycle) clog.Info("Connection rejected during DuckLake migration.") _ = server.WriteErrorResponse(writer, "FATAL", "57P03", "DuckLake catalog upgrade in progress for your organization, please retry in a few moments") @@ -1215,7 +1215,7 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { // connection advisory lock.) if whState, ok := cp.configStore.OrgWarehouseStatus(orgID); ok && whState == string(configstore.ManagedWarehouseStateResharding) { - sessionStart.Finish("error", observe.SessionStartFailureLifecycle) + sessionStart.Finish("error", observe.SessionStartReasonLifecycle) clog.Info("Connection rejected during metadata-store reshard.") _ = server.WriteErrorResponse(writer, "FATAL", "57P03", "metadata-store reshard in progress for your organization, please retry shortly") @@ -1230,7 +1230,7 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { // the stack absence is expected and the client should be told to // retry rather than receive a misleading auth-style error. whState, orgExists := cp.configStore.OrgWarehouseStatus(orgID) - sessionStart.Finish("error", missingOrgStackFailureClass(whState, orgExists)) + sessionStart.Finish("error", missingOrgStackReason(whState, orgExists)) switch { case !orgExists: _ = server.WriteErrorResponse(writer, "FATAL", "28000", "no org configured for user") @@ -1267,14 +1267,14 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { rebalancer = cp.rebalancer } if cp.isDraining() { - sessionStart.Finish("draining", observe.SessionStartFailureLifecycle) + sessionStart.Finish("draining", observe.SessionStartReasonLifecycle) _ = server.WriteErrorResponse(writer, "FATAL", "57P03", "control plane is draining, retry shortly") _ = writer.Flush() return } preReadyCtx, finishPreReady, err := cp.beginPreReadyHandshake(context.Background()) if err != nil { - sessionStart.Finish("draining", observe.SessionStartFailureLifecycle) + sessionStart.Finish("draining", observe.SessionStartReasonLifecycle) _ = server.WriteErrorResponse(writer, "FATAL", "57P03", "control plane is draining, retry shortly") _ = writer.Flush() return @@ -1291,7 +1291,7 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { defer server.CancelClientConn(tmpCC) server.SendInitialParams(tmpCC) if err := writer.Flush(); err != nil { - sessionStart.Finish("error", observe.SessionStartFailureTransport) + sessionStart.Finish("error", observe.SessionStartReasonTransport) clog.Error("Failed to flush initial params.", "error", err) return } @@ -1336,8 +1336,8 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { sessions.DestroySession(createdPID) executor = nil } - outcome, failureClass := controlPlaneSessionStartResult(err) - sessionStart.Finish(outcome, failureClass) + outcome, reason := controlPlaneSessionStartResult(err) + sessionStart.Finish(outcome, reason) clog.Error("Failed to create session.", "error", err) code, message := sessionCreationErrorResponse(err) _ = server.WriteErrorResponse(writer, "FATAL", code, message) @@ -1367,13 +1367,13 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { attachContextErr := attachCtx.Err() attachCancel() if probeErr != nil { - outcome, failureClass := controlPlaneSessionStartOperationResult( + outcome, reason := controlPlaneSessionStartOperationResult( probeErr, attachContextErr, cp.isDraining(), - observe.SessionStartFailureMetadataStore, + observe.SessionStartReasonMetadataStore, ) - sessionStart.Finish(outcome, failureClass) + sessionStart.Finish(outcome, reason) clog.Error("Failed to detect attached catalogs.", "error", probeErr) _ = server.WriteErrorResponse(writer, "FATAL", "XX000", "failed to detect attached catalogs") _ = writer.Flush() @@ -1384,7 +1384,7 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { var ok bool effectiveCatalog, ok = resolveEffectiveCatalog(requestedCatalog, duckLakeAttached) if !ok { - sessionStart.Finish("error", observe.SessionStartFailureMetadataStore) + sessionStart.Finish("error", observe.SessionStartReasonMetadataStore) clog.Warn("Postgres connection rejected: requested catalog is not available for this connection.", "requested", requestedCatalog, "ducklake_attached", duckLakeAttached) msg := "no catalog is available for this connection" @@ -1427,13 +1427,13 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { if err := sessionmeta.InitSessionDatabaseMetadataWithAccess(initCtx, executor, effectiveCatalog, metadataAccess); err != nil { initContextErr := initCtx.Err() initCancel() - outcome, failureClass := controlPlaneSessionStartOperationResult( + outcome, reason := controlPlaneSessionStartOperationResult( err, initContextErr, cp.isDraining(), - observe.SessionStartFailureMetadataStore, + observe.SessionStartReasonMetadataStore, ) - sessionStart.Finish(outcome, failureClass) + sessionStart.Finish(outcome, reason) clog.Error("Failed to initialize session database metadata.", "database", database, "error", err) _ = server.WriteErrorResponse(writer, "FATAL", "XX000", "failed to initialize session database metadata") _ = writer.Flush() @@ -1453,13 +1453,13 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { spCancel() if err != nil { if source == sessionDefaultSourceConfiguredCatalog { - outcome, failureClass := controlPlaneSessionStartOperationResult( + outcome, reason := controlPlaneSessionStartOperationResult( err, spContextErr, cp.isDraining(), - observe.SessionStartFailureMetadataStore, + observe.SessionStartReasonMetadataStore, ) - sessionStart.Finish(outcome, failureClass) + sessionStart.Finish(outcome, reason) clog.Error("Failed to apply session default catalog.", "catalog", effectiveCatalog, "error", err) _ = server.WriteErrorResponse(writer, "FATAL", "XX000", "failed to apply default catalog") _ = writer.Flush() @@ -1482,13 +1482,13 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { initContextErr := initCtx.Err() initCancel() if err != nil { - outcome, failureClass := controlPlaneSessionStartOperationResult( + outcome, reason := controlPlaneSessionStartOperationResult( err, initContextErr, cp.isDraining(), - observe.SessionStartFailureMetadataStore, + observe.SessionStartReasonMetadataStore, ) - sessionStart.Finish(outcome, failureClass) + sessionStart.Finish(outcome, reason) clog.Error("Failed to apply passthrough session default catalog.", "command", cmd, "error", err) _ = server.WriteErrorResponse(writer, "FATAL", "XX000", "failed to apply default catalog") _ = writer.Flush() @@ -1553,7 +1553,7 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { // s3_cache=off must never silently run through the cache. if s3CacheRequested { if err := server.ApplyConnectionS3CacheOption(cc, rawS3Cache); err != nil { - sessionStart.Finish("error", observe.SessionStartFailureWorker) + sessionStart.Finish("error", observe.SessionStartReasonWorker) clog.Error("Failed to apply s3_cache startup option.", "error", err) _ = server.WriteErrorResponse(writer, "FATAL", "XX000", err.Error()) _ = writer.Flush() @@ -1567,26 +1567,26 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { disconnect := disconnectWatcher.Stop() drained := finishPreReady() if disconnect.ClientCanceled { - sessionStart.Finish("canceled", observe.SessionStartFailureCanceled) + sessionStart.Finish("canceled", observe.SessionStartReasonCanceled) return } if drained { - sessionStart.Finish("draining", observe.SessionStartFailureLifecycle) + sessionStart.Finish("draining", observe.SessionStartReasonLifecycle) return } // Send ReadyForQuery to signal that the handshake is complete if err := server.WriteReadyForQuery(writer, 'I'); err != nil { - sessionStart.Finish("error", observe.SessionStartFailureTransport) + sessionStart.Finish("error", observe.SessionStartReasonTransport) clog.Error("Failed to send ReadyForQuery.", "error", err) return } if err := writer.Flush(); err != nil { - sessionStart.Finish("error", observe.SessionStartFailureTransport) + sessionStart.Finish("error", observe.SessionStartReasonTransport) clog.Error("Failed to flush writer.", "error", err) return } - sessionStart.Finish("success", observe.SessionStartFailureNone) + sessionStart.Finish("success", observe.SessionStartReasonNone) if orgID != "" { observeOrgPgSessionAccepted(orgID, passthroughUser) } @@ -1599,34 +1599,34 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) { } } -func controlPlaneSessionStartResult(err error) (outcome, failureClass string) { +func controlPlaneSessionStartResult(err error) (outcome, reason string) { var capacityErr *WorkerCapacityExhaustedError var rejection *configstore.OrgConnectionAdmissionRejectedError switch { case err == nil: - return "success", observe.SessionStartFailureNone + return "success", observe.SessionStartReasonNone case errors.As(err, &capacityErr): - return "capacity", observe.SessionStartFailureCapacity + return "capacity", observe.SessionStartReasonCapacity case errors.As(err, &rejection): - return "capacity", observe.SessionStartFailureClient + return "capacity", observe.SessionStartReasonClient case errors.Is(err, context.Canceled): - return "canceled", observe.SessionStartFailureCanceled + return "canceled", observe.SessionStartReasonCanceled case errors.Is(err, context.DeadlineExceeded), errors.Is(err, ErrTooManyConnections): - return "timeout", observe.SessionStartFailureCapacity + return "timeout", observe.SessionStartReasonCapacity case errors.Is(err, ErrSessionManagerDraining): - return "draining", observe.SessionStartFailureLifecycle + return "draining", observe.SessionStartReasonLifecycle default: - return "error", observe.SessionStartFailureWorker + return "error", observe.SessionStartReasonWorker } } func controlPlaneSessionStartOperationResult( err, contextErr error, draining bool, - failureClass string, -) (outcome, classifiedFailure string) { + reason string, +) (outcome, classifiedReason string) { if draining { - return "draining", observe.SessionStartFailureLifecycle + return "draining", observe.SessionStartReasonLifecycle } terminalErr := err if contextErr != nil { @@ -1635,24 +1635,24 @@ func controlPlaneSessionStartOperationResult( outcome, _ = controlPlaneSessionStartResult(terminalErr) switch outcome { case "success": - return outcome, observe.SessionStartFailureNone + return outcome, observe.SessionStartReasonNone case "canceled": - return outcome, observe.SessionStartFailureCanceled + return outcome, observe.SessionStartReasonCanceled case "draining": - return outcome, observe.SessionStartFailureLifecycle + return outcome, observe.SessionStartReasonLifecycle default: - return outcome, failureClass + return outcome, reason } } -func missingOrgStackFailureClass(warehouseState string, orgExists bool) string { +func missingOrgStackReason(warehouseState string, orgExists bool) string { if !orgExists { - return observe.SessionStartFailureClient + return observe.SessionStartReasonClient } if warehouseState == "" || warehouseState == string(configstore.ManagedWarehouseStateReady) { - return observe.SessionStartFailureControlPlane + return observe.SessionStartReasonControlPlane } - return observe.SessionStartFailureLifecycle + return observe.SessionStartReasonLifecycle } func authorizedClientSearchPath(searchPath string, policy *server.QueryAccessPolicy) string { diff --git a/controlplane/control_cancel_test.go b/controlplane/control_cancel_test.go index fa8ae761..55309fcb 100644 --- a/controlplane/control_cancel_test.go +++ b/controlplane/control_cancel_test.go @@ -199,17 +199,17 @@ func TestSessionCreationErrorResponse(t *testing.T) { func TestControlPlaneSessionStartResult(t *testing.T) { tests := []struct { - name string - err error - wantOutcome string - wantFailureClass string + name string + err error + wantOutcome string + wantReason string }{ - {name: "success", wantOutcome: "success", wantFailureClass: observe.SessionStartFailureNone}, - {name: "canceled", err: context.Canceled, wantOutcome: "canceled", wantFailureClass: observe.SessionStartFailureCanceled}, - {name: "deadline", err: context.DeadlineExceeded, wantOutcome: "timeout", wantFailureClass: observe.SessionStartFailureCapacity}, - {name: "queue timeout", err: ErrTooManyConnections, wantOutcome: "timeout", wantFailureClass: observe.SessionStartFailureCapacity}, - {name: "draining", err: ErrSessionManagerDraining, wantOutcome: "draining", wantFailureClass: observe.SessionStartFailureLifecycle}, - {name: "worker capacity", err: NewWorkerCapacityExhaustedError(time.Second), wantOutcome: "capacity", wantFailureClass: observe.SessionStartFailureCapacity}, + {name: "success", wantOutcome: "success", wantReason: observe.SessionStartReasonNone}, + {name: "canceled", err: context.Canceled, wantOutcome: "canceled", wantReason: observe.SessionStartReasonCanceled}, + {name: "deadline", err: context.DeadlineExceeded, wantOutcome: "timeout", wantReason: observe.SessionStartReasonCapacity}, + {name: "queue timeout", err: ErrTooManyConnections, wantOutcome: "timeout", wantReason: observe.SessionStartReasonCapacity}, + {name: "draining", err: ErrSessionManagerDraining, wantOutcome: "draining", wantReason: observe.SessionStartReasonLifecycle}, + {name: "worker capacity", err: NewWorkerCapacityExhaustedError(time.Second), wantOutcome: "capacity", wantReason: observe.SessionStartReasonCapacity}, { name: "admission hard rejection", err: &configstore.OrgConnectionAdmissionRejectedError{ @@ -217,18 +217,18 @@ func TestControlPlaneSessionStartResult(t *testing.T) { RequestedVCPUs: 4, MaximumVCPUs: 2, }, - wantOutcome: "capacity", - wantFailureClass: observe.SessionStartFailureClient, + wantOutcome: "capacity", + wantReason: observe.SessionStartReasonClient, }, - {name: "generic error", err: errors.New("bootstrap failed"), wantOutcome: "error", wantFailureClass: observe.SessionStartFailureWorker}, + {name: "generic error", err: errors.New("bootstrap failed"), wantOutcome: "error", wantReason: observe.SessionStartReasonWorker}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - gotOutcome, gotFailureClass := controlPlaneSessionStartResult(tt.err) - if gotOutcome != tt.wantOutcome || gotFailureClass != tt.wantFailureClass { + gotOutcome, gotReason := controlPlaneSessionStartResult(tt.err) + if gotOutcome != tt.wantOutcome || gotReason != tt.wantReason { t.Fatalf("controlPlaneSessionStartResult(%v) = (%q, %q), want (%q, %q)", - tt.err, gotOutcome, gotFailureClass, tt.wantOutcome, tt.wantFailureClass) + tt.err, gotOutcome, gotReason, tt.wantOutcome, tt.wantReason) } }) } @@ -237,54 +237,54 @@ func TestControlPlaneSessionStartResult(t *testing.T) { func TestControlPlaneSessionStartOperationResultPrefersContext(t *testing.T) { operationErr := errors.New("metadata initialization failed") tests := []struct { - name string - contextErr error - draining bool - wantOutcome string - wantFailureClass string + name string + contextErr error + draining bool + wantOutcome string + wantReason string }{ - {name: "client canceled", contextErr: context.Canceled, wantOutcome: "canceled", wantFailureClass: observe.SessionStartFailureCanceled}, - {name: "operation timed out", contextErr: context.DeadlineExceeded, wantOutcome: "timeout", wantFailureClass: observe.SessionStartFailureMetadataStore}, - {name: "ordinary metadata error", wantOutcome: "error", wantFailureClass: observe.SessionStartFailureMetadataStore}, - {name: "draining wins", contextErr: context.Canceled, draining: true, wantOutcome: "draining", wantFailureClass: observe.SessionStartFailureLifecycle}, + {name: "client canceled", contextErr: context.Canceled, wantOutcome: "canceled", wantReason: observe.SessionStartReasonCanceled}, + {name: "operation timed out", contextErr: context.DeadlineExceeded, wantOutcome: "timeout", wantReason: observe.SessionStartReasonMetadataStore}, + {name: "ordinary metadata error", wantOutcome: "error", wantReason: observe.SessionStartReasonMetadataStore}, + {name: "draining wins", contextErr: context.Canceled, draining: true, wantOutcome: "draining", wantReason: observe.SessionStartReasonLifecycle}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - gotOutcome, gotFailureClass := controlPlaneSessionStartOperationResult( + gotOutcome, gotReason := controlPlaneSessionStartOperationResult( operationErr, tt.contextErr, tt.draining, - observe.SessionStartFailureMetadataStore, + observe.SessionStartReasonMetadataStore, ) - if gotOutcome != tt.wantOutcome || gotFailureClass != tt.wantFailureClass { + if gotOutcome != tt.wantOutcome || gotReason != tt.wantReason { t.Fatalf("operation result = (%q, %q), want (%q, %q)", - gotOutcome, gotFailureClass, tt.wantOutcome, tt.wantFailureClass) + gotOutcome, gotReason, tt.wantOutcome, tt.wantReason) } }) } } -func TestMissingOrgStackFailureClass(t *testing.T) { +func TestMissingOrgStackReason(t *testing.T) { tests := []struct { name string state string orgExists bool want string }{ - {name: "missing org", orgExists: false, want: observe.SessionStartFailureClient}, - {name: "legacy ready state", orgExists: true, state: "", want: observe.SessionStartFailureControlPlane}, - {name: "ready warehouse", orgExists: true, state: string(configstore.ManagedWarehouseStateReady), want: observe.SessionStartFailureControlPlane}, - {name: "failed provisioning", orgExists: true, state: string(configstore.ManagedWarehouseStateFailed), want: observe.SessionStartFailureLifecycle}, - {name: "deleting", orgExists: true, state: string(configstore.ManagedWarehouseStateDeleting), want: observe.SessionStartFailureLifecycle}, - {name: "resharding", orgExists: true, state: string(configstore.ManagedWarehouseStateResharding), want: observe.SessionStartFailureLifecycle}, - {name: "pending", orgExists: true, state: string(configstore.ManagedWarehouseStatePending), want: observe.SessionStartFailureLifecycle}, + {name: "missing org", orgExists: false, want: observe.SessionStartReasonClient}, + {name: "legacy ready state", orgExists: true, state: "", want: observe.SessionStartReasonControlPlane}, + {name: "ready warehouse", orgExists: true, state: string(configstore.ManagedWarehouseStateReady), want: observe.SessionStartReasonControlPlane}, + {name: "failed provisioning", orgExists: true, state: string(configstore.ManagedWarehouseStateFailed), want: observe.SessionStartReasonLifecycle}, + {name: "deleting", orgExists: true, state: string(configstore.ManagedWarehouseStateDeleting), want: observe.SessionStartReasonLifecycle}, + {name: "resharding", orgExists: true, state: string(configstore.ManagedWarehouseStateResharding), want: observe.SessionStartReasonLifecycle}, + {name: "pending", orgExists: true, state: string(configstore.ManagedWarehouseStatePending), want: observe.SessionStartReasonLifecycle}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - if got := missingOrgStackFailureClass(tt.state, tt.orgExists); got != tt.want { - t.Fatalf("missingOrgStackFailureClass(%q, %t) = %q, want %q", tt.state, tt.orgExists, got, tt.want) + if got := missingOrgStackReason(tt.state, tt.orgExists); got != tt.want { + t.Fatalf("missingOrgStackReason(%q, %t) = %q, want %q", tt.state, tt.orgExists, got, tt.want) } }) } diff --git a/docs/metrics.md b/docs/metrics.md index b2385abd..b6b30148 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -52,7 +52,7 @@ session defaults, and the final ready flush. It excludes failed authentication. | `duckgres_session_admission_active_vcpus` | Gauge | `org` | Requested vCPUs held by live local lease handles. It is admitted capacity, not measured CPU usage or the exact durable lease-row total. | | `duckgres_session_admission_limit_vcpus` | Gauge | `org` | Effective org cap for an active org stack, reconciled from this process's current config snapshot. `0` means unlimited. | | `duckgres_session_start_duration_seconds` | Histogram | `org`, `protocol`, `outcome` | Authenticated PostgreSQL create-to-ready latency. | -| `duckgres_postgres_session_start_total` | Counter | `org`, `outcome`, `failure_class` | Exactly one terminal authenticated PostgreSQL session-start result after server-side retries. | +| `duckgres_postgres_session_start_total` | Counter | `org`, `outcome`, `reason` | Exactly one terminal authenticated PostgreSQL session-start result after server-side retries. | Admission request outcomes are `granted`, `rejected`, `timeout`, `canceled`, and `error`. `rejected` means the requested worker shape can never fit its hard @@ -61,15 +61,15 @@ Session-start outcomes are `success`, `timeout`, `canceled`, `capacity`, `draining`, and `error`. The PostgreSQL terminal counter collapses those outcomes to `success` or -`failure`. Success always has `failure_class="none"`. Failure classes are +`failure`. Success always has `reason="none"`. Failure reasons are `capacity`, `worker`, `metadata_store`, `control_plane`, `client`, `lifecycle`, `canceled`, `transport`, and `unknown`. The first four represent failures an -operator can usually alleviate. The remaining classes let alerts exclude bad +operator can usually alleviate. The remaining reasons let alerts exclude bad client input, planned lifecycle transitions, client disconnects, wire errors, and newly added paths that have not yet been classified. Flight SQL does not emit this counter. `capacity` covers runtime worker exhaustion and admission timeouts; requests that exceed a configured hard org or user vCPU limit are -classified as `client`. +reported with reason `client`. Evaluation decisions are `granted_current`, `already_granted`, `rejected`, `blocked`, `waiting`, `inactive`, `missing`, `canceled`, `timeout`, and `error`. diff --git a/server/observe/metrics.go b/server/observe/metrics.go index 8cc75556..afe9a46b 100644 --- a/server/observe/metrics.go +++ b/server/observe/metrics.go @@ -41,26 +41,26 @@ func ObserveConnectionDuration(org string, seconds float64) { connectionDurationHistogram.WithLabelValues(org).Observe(seconds) } -// PostgreSQL session-start failure classes are intentionally bounded. Alerting -// may allowlist operator-actionable classes, while client, lifecycle, transport, +// PostgreSQL session-start reasons are intentionally bounded. Alerting may +// allowlist operator-actionable reasons, while client, lifecycle, transport, // and unknown failures remain available for diagnosis without paging. const ( - SessionStartFailureNone = "none" - SessionStartFailureCapacity = "capacity" - SessionStartFailureWorker = "worker" - SessionStartFailureMetadataStore = "metadata_store" - SessionStartFailureControlPlane = "control_plane" - SessionStartFailureClient = "client" - SessionStartFailureLifecycle = "lifecycle" - SessionStartFailureCanceled = "canceled" - SessionStartFailureTransport = "transport" - SessionStartFailureUnknown = "unknown" + SessionStartReasonNone = "none" + SessionStartReasonCapacity = "capacity" + SessionStartReasonWorker = "worker" + SessionStartReasonMetadataStore = "metadata_store" + SessionStartReasonControlPlane = "control_plane" + SessionStartReasonClient = "client" + SessionStartReasonLifecycle = "lifecycle" + SessionStartReasonCanceled = "canceled" + SessionStartReasonTransport = "transport" + SessionStartReasonUnknown = "unknown" ) var postgresSessionStartCounter = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "duckgres_postgres_session_start_total", Help: "Terminal outcomes for authenticated PostgreSQL session starts after all server-side retries.", -}, []string{"org", "outcome", "failure_class"}) +}, []string{"org", "outcome", "reason"}) var sessionStartDurationHistogram = promauto.NewHistogramVec(prometheus.HistogramOpts{ Name: "duckgres_session_start_duration_seconds", @@ -86,7 +86,7 @@ func BeginSessionStart(org, protocol string) *SessionStartScope { return &SessionStartScope{started: time.Now(), org: org, protocol: protocol} } -func (s *SessionStartScope) Finish(outcome, failureClass string) { +func (s *SessionStartScope) Finish(outcome, reason string) { if s == nil { return } @@ -95,7 +95,7 @@ func (s *SessionStartScope) Finish(outcome, failureClass string) { default: outcome = "error" } - terminalOutcome, failureClass := normalizePostgresSessionStartResult(outcome, failureClass) + terminalOutcome, reason := normalizePostgresSessionStartResult(outcome, reason) s.once.Do(func() { duration := time.Since(s.started) if duration < 0 { @@ -103,31 +103,31 @@ func (s *SessionStartScope) Finish(outcome, failureClass string) { } sessionStartDurationHistogram.WithLabelValues(s.org, s.protocol, outcome).Observe(duration.Seconds()) if s.protocol == "postgres" { - postgresSessionStartCounter.WithLabelValues(s.org, terminalOutcome, failureClass).Inc() + postgresSessionStartCounter.WithLabelValues(s.org, terminalOutcome, reason).Inc() } }) } -func normalizePostgresSessionStartResult(outcome, failureClass string) (string, string) { +func normalizePostgresSessionStartResult(outcome, reason string) (string, string) { if outcome == "success" { - return "success", SessionStartFailureNone + return "success", SessionStartReasonNone } - switch failureClass { - case SessionStartFailureCapacity, - SessionStartFailureWorker, - SessionStartFailureMetadataStore, - SessionStartFailureControlPlane, - SessionStartFailureClient, - SessionStartFailureLifecycle, - SessionStartFailureCanceled, - SessionStartFailureTransport, - SessionStartFailureUnknown: - return "failure", failureClass + switch reason { + case SessionStartReasonCapacity, + SessionStartReasonWorker, + SessionStartReasonMetadataStore, + SessionStartReasonControlPlane, + SessionStartReasonClient, + SessionStartReasonLifecycle, + SessionStartReasonCanceled, + SessionStartReasonTransport, + SessionStartReasonUnknown: + return "failure", reason default: - // A failed session must never be reported with class=none, and newly + // A failed session must never be reported with reason=none, and newly // added unclassified paths must stay outside paging allowlists. - return "failure", SessionStartFailureUnknown + return "failure", SessionStartReasonUnknown } } diff --git a/server/observe/metrics_test.go b/server/observe/metrics_test.go index d6008d5d..9b6449a8 100644 --- a/server/observe/metrics_test.go +++ b/server/observe/metrics_test.go @@ -2,6 +2,7 @@ package observe import ( "fmt" + "strings" "testing" "time" @@ -86,12 +87,12 @@ func TestObserveConnectionDurationRecordsPerOrgSample(t *testing.T) { func TestSessionStartScopeRecordsExactlyOnce(t *testing.T) { const org = "session-start-test-org" before := sessionStartSampleCount(t, org, "postgres", "success") - terminalBefore := postgresSessionStartCount(t, org, "success", SessionStartFailureNone) - failureBefore := postgresSessionStartCount(t, org, "failure", SessionStartFailureControlPlane) + terminalBefore := postgresSessionStartCount(t, org, "success", SessionStartReasonNone) + failureBefore := postgresSessionStartCount(t, org, "failure", SessionStartReasonControlPlane) scope := BeginSessionStart(org, "postgres") - scope.Finish("success", SessionStartFailureControlPlane) - scope.Finish("error", SessionStartFailureControlPlane) + scope.Finish("success", SessionStartReasonControlPlane) + scope.Finish("error", SessionStartReasonControlPlane) if got := sessionStartSampleCount(t, org, "postgres", "success"); got != before+1 { t.Fatalf("success sample count = %d, want %d", got, before+1) @@ -99,22 +100,32 @@ func TestSessionStartScopeRecordsExactlyOnce(t *testing.T) { if got := sessionStartSampleCount(t, org, "postgres", "error"); got != 0 { t.Fatalf("error sample count = %d, want 0", got) } - if got := postgresSessionStartCount(t, org, "success", SessionStartFailureNone); got != terminalBefore+1 { + if got := postgresSessionStartCount(t, org, "success", SessionStartReasonNone); got != terminalBefore+1 { t.Fatalf("terminal success count = %v, want %v", got, terminalBefore+1) } - if got := postgresSessionStartCount(t, org, "failure", SessionStartFailureControlPlane); got != failureBefore { + if got := postgresSessionStartCount(t, org, "failure", SessionStartReasonControlPlane); got != failureBefore { t.Fatalf("second Finish changed terminal classification: count = %v, want %v", got, failureBefore) } } +func TestPostgresSessionStartCounterUsesCanonicalReasonLabel(t *testing.T) { + descriptions := make(chan *prometheus.Desc, 1) + postgresSessionStartCounter.Describe(descriptions) + description := <-descriptions + + if got := description.String(); !strings.Contains(got, "variableLabels: {org,outcome,reason}") { + t.Fatalf("postgres session start labels = %q, want org, outcome, reason", got) + } +} + func TestSessionStartScopeRecordsPostgresTerminalFailure(t *testing.T) { const org = "session-start-terminal-failure-org" - before := postgresSessionStartCount(t, org, "failure", SessionStartFailureMetadataStore) + before := postgresSessionStartCount(t, org, "failure", SessionStartReasonMetadataStore) histogramBefore := sessionStartSampleCount(t, org, "postgres", "timeout") - BeginSessionStart(org, "postgres").Finish("timeout", SessionStartFailureMetadataStore) + BeginSessionStart(org, "postgres").Finish("timeout", SessionStartReasonMetadataStore) - if got := postgresSessionStartCount(t, org, "failure", SessionStartFailureMetadataStore); got != before+1 { + if got := postgresSessionStartCount(t, org, "failure", SessionStartReasonMetadataStore); got != before+1 { t.Fatalf("terminal failure count = %v, want %v", got, before+1) } if got := sessionStartSampleCount(t, org, "postgres", "timeout"); got != histogramBefore+1 { @@ -122,45 +133,45 @@ func TestSessionStartScopeRecordsPostgresTerminalFailure(t *testing.T) { } } -func TestSessionStartScopeNormalizesUnsafeFailureClasses(t *testing.T) { +func TestSessionStartScopeNormalizesUnsafeReasons(t *testing.T) { tests := []struct { - name string - outcome string - failureClass string - wantOutcome string - wantClass string + name string + outcome string + reason string + wantOutcome string + wantReason string }{ { - name: "success always uses none", - outcome: "success", - failureClass: SessionStartFailureWorker, - wantOutcome: "success", - wantClass: SessionStartFailureNone, + name: "success always uses none", + outcome: "success", + reason: SessionStartReasonWorker, + wantOutcome: "success", + wantReason: SessionStartReasonNone, }, { - name: "failure cannot use none", - outcome: "error", - failureClass: SessionStartFailureNone, - wantOutcome: "failure", - wantClass: SessionStartFailureUnknown, + name: "failure cannot use none", + outcome: "error", + reason: SessionStartReasonNone, + wantOutcome: "failure", + wantReason: SessionStartReasonUnknown, }, { - name: "unknown class stays non-pageable", - outcome: "error", - failureClass: "new-unclassified-path", - wantOutcome: "failure", - wantClass: SessionStartFailureUnknown, + name: "unknown reason stays non-pageable", + outcome: "error", + reason: "new-unclassified-path", + wantOutcome: "failure", + wantReason: SessionStartReasonUnknown, }, } for i, tt := range tests { t.Run(tt.name, func(t *testing.T) { org := fmt.Sprintf("session-start-normalization-%d", i) - before := postgresSessionStartCount(t, org, tt.wantOutcome, tt.wantClass) + before := postgresSessionStartCount(t, org, tt.wantOutcome, tt.wantReason) - BeginSessionStart(org, "postgres").Finish(tt.outcome, tt.failureClass) + BeginSessionStart(org, "postgres").Finish(tt.outcome, tt.reason) - if got := postgresSessionStartCount(t, org, tt.wantOutcome, tt.wantClass); got != before+1 { + if got := postgresSessionStartCount(t, org, tt.wantOutcome, tt.wantReason); got != before+1 { t.Fatalf("normalized terminal count = %v, want %v", got, before+1) } }) @@ -169,18 +180,18 @@ func TestSessionStartScopeNormalizesUnsafeFailureClasses(t *testing.T) { func TestSessionStartScopeDoesNotExportFlightTerminalCounter(t *testing.T) { const org = "session-start-obsolete-flight-org" - before := postgresSessionStartCount(t, org, "failure", SessionStartFailureMetadataStore) + before := postgresSessionStartCount(t, org, "failure", SessionStartReasonMetadataStore) - BeginSessionStart(org, "flight").Finish("error", SessionStartFailureMetadataStore) + BeginSessionStart(org, "flight").Finish("error", SessionStartReasonMetadataStore) - if got := postgresSessionStartCount(t, org, "failure", SessionStartFailureMetadataStore); got != before { + if got := postgresSessionStartCount(t, org, "failure", SessionStartReasonMetadataStore); got != before { t.Fatalf("obsolete Flight path changed PostgreSQL terminal count: got %v, want %v", got, before) } } -func postgresSessionStartCount(t *testing.T, org, outcome, failureClass string) float64 { +func postgresSessionStartCount(t *testing.T, org, outcome, reason string) float64 { t.Helper() - return counterVecValue(t, postgresSessionStartCounter, org, outcome, failureClass) + return counterVecValue(t, postgresSessionStartCounter, org, outcome, reason) } func sessionStartSampleCount(t *testing.T, org, protocol, outcome string) uint64 {