diff --git a/README.md b/README.md index aac942e5..e99f70f5 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,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 25b67bd4..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") + 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,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.SessionStartReasonClient) 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.SessionStartReasonClient) 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.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") @@ -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.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") @@ -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", missingOrgStackReason(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.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") + sessionStart.Finish("draining", observe.SessionStartReasonLifecycle) _ = 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.SessionStartReasonTransport) 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, 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) @@ -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, reason := controlPlaneSessionStartOperationResult( + probeErr, + attachContextErr, + cp.isDraining(), + observe.SessionStartReasonMetadataStore, + ) + 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() @@ -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.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" @@ -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, reason := controlPlaneSessionStartOperationResult( + err, + initContextErr, + cp.isDraining(), + observe.SessionStartReasonMetadataStore, + ) + 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() @@ -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, reason := controlPlaneSessionStartOperationResult( + err, + spContextErr, + cp.isDraining(), + observe.SessionStartReasonMetadataStore, + ) + 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() @@ -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, reason := controlPlaneSessionStartOperationResult( + err, + initContextErr, + cp.isDraining(), + observe.SessionStartReasonMetadataStore, + ) + 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() @@ -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.SessionStartReasonWorker) 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.SessionStartReasonCanceled) return } if drained { - sessionStart.Finish("draining") + 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.SessionStartReasonTransport) clog.Error("Failed to send ReadyForQuery.", "error", err) return } if err := writer.Flush(); err != nil { + sessionStart.Finish("error", observe.SessionStartReasonTransport) clog.Error("Failed to flush writer.", "error", err) return } - sessionStart.Finish("success") + sessionStart.Finish("success", observe.SessionStartReasonNone) 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, reason 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.SessionStartReasonNone + case errors.As(err, &capacityErr): + return "capacity", observe.SessionStartReasonCapacity + case errors.As(err, &rejection): + return "capacity", observe.SessionStartReasonClient case errors.Is(err, context.Canceled): - return "canceled" + return "canceled", observe.SessionStartReasonCanceled case errors.Is(err, context.DeadlineExceeded), errors.Is(err, ErrTooManyConnections): - return "timeout" + return "timeout", observe.SessionStartReasonCapacity case errors.Is(err, ErrSessionManagerDraining): - return "draining" + return "draining", observe.SessionStartReasonLifecycle default: - return "error" + return "error", observe.SessionStartReasonWorker } } -func controlPlaneSessionStartOperationOutcome(err, contextErr error, draining bool) string { +func controlPlaneSessionStartOperationResult( + err, contextErr error, + draining bool, + reason string, +) (outcome, classifiedReason string) { if draining { - return "draining" + return "draining", observe.SessionStartReasonLifecycle } + terminalErr := err if contextErr != nil { - return controlPlaneSessionStartOutcome(contextErr) + terminalErr = contextErr + } + outcome, _ = controlPlaneSessionStartResult(terminalErr) + switch outcome { + case "success": + return outcome, observe.SessionStartReasonNone + case "canceled": + return outcome, observe.SessionStartReasonCanceled + case "draining": + return outcome, observe.SessionStartReasonLifecycle + default: + return outcome, reason + } +} + +func missingOrgStackReason(warehouseState string, orgExists bool) string { + if !orgExists { + return observe.SessionStartReasonClient + } + if warehouseState == "" || warehouseState == string(configstore.ManagedWarehouseStateReady) { + return observe.SessionStartReasonControlPlane } - return controlPlaneSessionStartOutcome(err) + 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 77f87f5e..55309fcb 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 + wantReason 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", 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{ @@ -215,33 +217,76 @@ func TestControlPlaneSessionStartOutcome(t *testing.T) { RequestedVCPUs: 4, MaximumVCPUs: 2, }, - want: "capacity", + wantOutcome: "capacity", + wantReason: observe.SessionStartReasonClient, }, - {name: "generic error", err: errors.New("bootstrap failed"), want: "error"}, + {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) { - if got := controlPlaneSessionStartOutcome(tt.err); got != tt.want { - t.Fatalf("controlPlaneSessionStartOutcome(%v) = %q, want %q", tt.err, got, tt.want) + gotOutcome, gotReason := controlPlaneSessionStartResult(tt.err) + if gotOutcome != tt.wantOutcome || gotReason != tt.wantReason { + t.Fatalf("controlPlaneSessionStartResult(%v) = (%q, %q), want (%q, %q)", + tt.err, gotOutcome, gotReason, tt.wantOutcome, tt.wantReason) } }) } } -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 + wantReason string + }{ + {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}, } - 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, gotReason := controlPlaneSessionStartOperationResult( + operationErr, + tt.contextErr, + tt.draining, + observe.SessionStartReasonMetadataStore, + ) + if gotOutcome != tt.wantOutcome || gotReason != tt.wantReason { + t.Fatalf("operation result = (%q, %q), want (%q, %q)", + gotOutcome, gotReason, tt.wantOutcome, tt.wantReason) + } + }) } - if got := controlPlaneSessionStartOperationOutcome(operationErr, nil, false); got != "error" { - t.Fatalf("ordinary operation outcome = %q, want error", got) +} + +func TestMissingOrgStackReason(t *testing.T) { + tests := []struct { + name string + state string + orgExists bool + want string + }{ + {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}, } - 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 := 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 ca2620b3..b6b30148 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`, `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 @@ -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 `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 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 +reported with reason `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..afe9a46b 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 reasons are intentionally bounded. Alerting may +// allowlist operator-actionable reasons, while client, lifecycle, transport, +// and unknown failures remain available for diagnosis without paging. +const ( + 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", "reason"}) + 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, reason string) { if s == nil { return } @@ -74,15 +95,42 @@ func (s *SessionStartScope) Finish(outcome string) { default: outcome = "error" } + terminalOutcome, reason := normalizePostgresSessionStartResult(outcome, reason) 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, reason).Inc() + } }) } +func normalizePostgresSessionStartResult(outcome, reason string) (string, string) { + if outcome == "success" { + return "success", SessionStartReasonNone + } + + switch reason { + case SessionStartReasonCapacity, + SessionStartReasonWorker, + SessionStartReasonMetadataStore, + SessionStartReasonControlPlane, + SessionStartReasonClient, + SessionStartReasonLifecycle, + SessionStartReasonCanceled, + SessionStartReasonTransport, + SessionStartReasonUnknown: + return "failure", reason + default: + // A failed session must never be reported with reason=none, and newly + // added unclassified paths must stay outside paging allowlists. + return "failure", SessionStartReasonUnknown + } +} + // 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..9b6449a8 100644 --- a/server/observe/metrics_test.go +++ b/server/observe/metrics_test.go @@ -1,6 +1,8 @@ package observe import ( + "fmt" + "strings" "testing" "time" @@ -85,10 +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", SessionStartReasonNone) + failureBefore := postgresSessionStartCount(t, org, "failure", SessionStartReasonControlPlane) scope := BeginSessionStart(org, "postgres") - scope.Finish("success") - scope.Finish("error") + 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) @@ -96,6 +100,98 @@ 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", SessionStartReasonNone); got != terminalBefore+1 { + t.Fatalf("terminal success count = %v, want %v", got, terminalBefore+1) + } + 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", SessionStartReasonMetadataStore) + histogramBefore := sessionStartSampleCount(t, org, "postgres", "timeout") + + BeginSessionStart(org, "postgres").Finish("timeout", SessionStartReasonMetadataStore) + + 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 { + t.Fatalf("timeout histogram count = %d, want %d", got, histogramBefore+1) + } +} + +func TestSessionStartScopeNormalizesUnsafeReasons(t *testing.T) { + tests := []struct { + name string + outcome string + reason string + wantOutcome string + wantReason string + }{ + { + name: "success always uses none", + outcome: "success", + reason: SessionStartReasonWorker, + wantOutcome: "success", + wantReason: SessionStartReasonNone, + }, + { + name: "failure cannot use none", + outcome: "error", + reason: SessionStartReasonNone, + wantOutcome: "failure", + wantReason: SessionStartReasonUnknown, + }, + { + 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.wantReason) + + BeginSessionStart(org, "postgres").Finish(tt.outcome, tt.reason) + + if got := postgresSessionStartCount(t, org, tt.wantOutcome, tt.wantReason); 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", SessionStartReasonMetadataStore) + + BeginSessionStart(org, "flight").Finish("error", SessionStartReasonMetadataStore) + + 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, reason string) float64 { + t.Helper() + return counterVecValue(t, postgresSessionStartCounter, org, outcome, reason) } func sessionStartSampleCount(t *testing.T, org, protocol, outcome string) uint64 {