From 304e44fc171567f4514e97a818cb13b2e05dcf58 Mon Sep 17 00:00:00 2001 From: Bhagya Amarasinghe Date: Tue, 4 Aug 2026 16:04:31 +0530 Subject: [PATCH 1/3] feat: add taxonomy lifecycle observability --- .env.example | 1 + cmd/api/app.go | 46 +++++- cmd/api/main.go | 29 +--- cmd/worker/main.go | 3 + .../api/handlers/taxonomy_internal_handler.go | 5 +- internal/config/config.go | 5 + internal/models/taxonomy.go | 22 ++- internal/observability/aggregate.go | 7 + internal/observability/logging.go | 26 ++++ internal/observability/logging_test.go | 38 +++++ internal/observability/names.go | 10 +- internal/observability/taxonomy.go | 134 ++++++++++++++++++ internal/observability/taxonomy_test.go | 48 +++++++ internal/repository/taxonomy_repository.go | 48 ++++--- internal/service/taxonomy_service.go | 106 +++++++++++++- internal/service/taxonomy_service_test.go | 46 +++++- tests/taxonomy_persistence_test.go | 5 +- 17 files changed, 522 insertions(+), 57 deletions(-) create mode 100644 internal/observability/logging_test.go create mode 100644 internal/observability/taxonomy.go create mode 100644 internal/observability/taxonomy_test.go diff --git a/.env.example b/.env.example index dad59610..9e103e1d 100644 --- a/.env.example +++ b/.env.example @@ -42,6 +42,7 @@ PORT=8080 # Default: info # Valid values: debug, info, warn, error LOG_LEVEL=info +# LOG_FORMAT=json # Taxonomy service integration (optional; beta) # TAXONOMY_SERVICE_URL is the internal URL Hub uses to call the standalone taxonomy service. diff --git a/cmd/api/app.go b/cmd/api/app.go index 9202c3f7..0c5f9630 100644 --- a/cmd/api/app.go +++ b/cmd/api/app.go @@ -550,11 +550,17 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) { taxonomyStarter = taxonomyClient } + var taxonomyMetrics observability.TaxonomyMetrics + if metrics != nil { + taxonomyMetrics = metrics.Taxonomy + } + taxonomyService := service.NewTaxonomyService(service.NewTaxonomyServiceParams{ Repo: taxonomyRepo, Starter: taxonomyStarter, EmbeddingModel: taxonomyEmbeddingModel, MinimumEmbeddingCount: cfg.Taxonomy.MinimumEmbeddedRecords, + Metrics: taxonomyMetrics, }) taxonomyHandler := handlers.NewTaxonomyHandler(taxonomyService) feedbackRecordsHandler := handlers.NewFeedbackRecordsHandler(feedbackRecordsService) @@ -721,7 +727,7 @@ func (a *App) Run(ctx context.Context) error { // Reap taxonomy runs orphaned in a non-terminal state, but only when the taxonomy service is wired // (no runs exist otherwise, so the sweep would be pointless). if a.taxonomyRepo != nil && (a.cfg.Taxonomy.ServiceURL != "" || a.cfg.Taxonomy.ServiceToken != "") { - go runTaxonomyRunReaper(ctx, a.taxonomyRepo, + go runTaxonomyRunReaper(ctx, a.taxonomyRepo, taxonomyMetricsFromAggregate(a.metrics), a.cfg.Taxonomy.StuckRunTimeout.Duration(), a.cfg.Taxonomy.ReaperInterval.Duration()) } @@ -928,7 +934,8 @@ const stuckTaxonomyRunMessage = "taxonomy run timed out without completing" // forever in the UI and generation can be retried. Idempotent: the repository's status filter skips // runs that finished on their own between sweeps. func runTaxonomyRunReaper( - ctx context.Context, repo *repository.TaxonomyRepository, timeout, interval time.Duration, + ctx context.Context, repo *repository.TaxonomyRepository, metrics observability.TaxonomyMetrics, + timeout, interval time.Duration, ) { // A non-positive interval panics time.NewTicker, and a non-positive timeout would reap active // runs (cutoff would be now or in the future). Either is misconfiguration — disable the reaper. @@ -943,10 +950,31 @@ func runTaxonomyRunReaper( defer ticker.Stop() reap := func() { - failed, err := repo.FailStuckRuns(ctx, timeout, stuckTaxonomyRunMessage, + reaped, err := repo.FailStuckRuns(ctx, timeout, stuckTaxonomyRunMessage, models.TaxonomyRunFailureCodeInternalError) - if failed > 0 { - slog.InfoContext(ctx, "taxonomy stuck-run reaper failed stalled runs", "count", failed) + for _, run := range reaped { + slog.ErrorContext(ctx, "taxonomy stuck-run reaper failed stalled run", + "event", "hub.taxonomy.run.reaped", "run_id", run.ID, "tenant_id", run.TenantID, + "scope_type", run.ScopeType, "source_type", run.SourceType, + "source_id", run.SourceID, "field_id", run.FieldID, + "failure_code", models.TaxonomyRunFailureCodeInternalError) + } + + if len(reaped) > 0 && metrics != nil { + metrics.RecordRunsReaped(ctx, int64(len(reaped))) + + for _, run := range reaped { + metrics.RecordRunOutcome(ctx, string(models.TaxonomyRunStatusFailed), + string(models.TaxonomyRunFailureCodeInternalError), string(run.ScopeType)) + + started := run.CreatedAt + if run.StartedAt != nil { + started = *run.StartedAt + } + + metrics.RecordRunDuration(ctx, time.Since(started), string(models.TaxonomyRunStatusFailed), + string(run.ScopeType)) + } } if err != nil { @@ -966,6 +994,14 @@ func runTaxonomyRunReaper( } } +func taxonomyMetricsFromAggregate(metrics *observability.Metrics) observability.TaxonomyMetrics { + if metrics == nil { + return nil + } + + return metrics.Taxonomy +} + // shutdownObservability shuts down tracer and meter providers. Logs secondary errors, returns the first. func shutdownObservability(ctx context.Context, tracer *sdktrace.TracerProvider, meter *sdkmetric.MeterProvider) error { var first error diff --git a/cmd/api/main.go b/cmd/api/main.go index 436f9b1f..b3419ec0 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -6,12 +6,12 @@ import ( "log/slog" "os" "os/signal" - "strings" "syscall" pgxvec "github.com/pgvector/pgvector-go/pgx" "github.com/formbricks/hub/internal/config" + "github.com/formbricks/hub/internal/observability" "github.com/formbricks/hub/pkg/database" ) @@ -27,20 +27,20 @@ func main() { func run() int { cfg, err := config.Load() if err != nil { - setupLogging("info") + setupLogging("info", "text") slog.Error("Failed to load configuration", "error", err) return exitFailure } if cfg.Server.HubAPIKey == "" { - setupLogging(cfg.Server.LogLevel) + setupLogging(cfg.Server.LogLevel, cfg.Server.LogFormat) slog.Error("API_KEY is required for hub-api") return exitFailure } - setupLogging(cfg.Server.LogLevel) + setupLogging(cfg.Server.LogLevel, cfg.Server.LogFormat) ctx := context.Background() @@ -92,23 +92,6 @@ func run() int { return exitSuccess } -func setupLogging(level string) { - var logLevel slog.Level - - switch strings.ToLower(level) { - case "debug": - logLevel = slog.LevelDebug - case "info": - logLevel = slog.LevelInfo - case "warn": - logLevel = slog.LevelWarn - case "error": - logLevel = slog.LevelError - default: - logLevel = slog.LevelInfo - } - - opts := &slog.HandlerOptions{Level: logLevel} - handler := slog.NewTextHandler(os.Stdout, opts) - slog.SetDefault(slog.New(handler)) +func setupLogging(level, format string) { + slog.SetDefault(slog.New(observability.NewLogHandler(os.Stdout, level, format))) } diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 88b7eb8c..18bc32fb 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -12,6 +12,7 @@ import ( pgxvec "github.com/pgvector/pgvector-go/pgx" "github.com/formbricks/hub/internal/config" + "github.com/formbricks/hub/internal/observability" "github.com/formbricks/hub/pkg/database" ) @@ -32,6 +33,8 @@ func run() int { return exitFailure } + slog.SetDefault(slog.New(observability.NewLogHandler(os.Stdout, cfg.Server.LogLevel, cfg.Server.LogFormat))) + if cfg.Database.URL == "" || cfg.Database.URL == config.DefaultDatabaseURL { slog.Error("DATABASE_URL must be set explicitly for hub-worker (do not use the default test URL)") diff --git a/internal/api/handlers/taxonomy_internal_handler.go b/internal/api/handlers/taxonomy_internal_handler.go index 07049c5d..85411d30 100644 --- a/internal/api/handlers/taxonomy_internal_handler.go +++ b/internal/api/handlers/taxonomy_internal_handler.go @@ -17,8 +17,7 @@ type TaxonomyInternalService interface { FailRun( ctx context.Context, runID uuid.UUID, - message string, - errorCode models.TaxonomyRunFailureCode, + req models.TaxonomyRunFailedRequest, ) (*models.TaxonomyRun, error) Heartbeat(ctx context.Context, runID uuid.UUID) error } @@ -119,7 +118,7 @@ func (h *TaxonomyInternalHandler) FailRun(w http.ResponseWriter, r *http.Request return } - result, err := h.service.FailRun(r.Context(), runID, req.Error, req.ErrorCode) + result, err := h.service.FailRun(r.Context(), runID, req) if err != nil { respondTaxonomyError(w, r, err) diff --git a/internal/config/config.go b/internal/config/config.go index 8c7022fe..19c75d79 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -71,6 +71,7 @@ type ServerConfig struct { HubAPIKey string `env:"API_KEY"` PublicBaseURL string `env:"PUBLIC_BASE_URL"` LogLevel string `env:"LOG_LEVEL" env-default:"info"` + LogFormat string `env:"LOG_FORMAT" env-default:"text"` ShutdownTimeout DurationSec `env:"SHUTDOWN_TIMEOUT_SECONDS" env-default:"30"` } @@ -339,6 +340,10 @@ func applyDefaults(cfg *Config) { cfg.Server.LogLevel = "info" } + if cfg.Server.LogFormat == "" { + cfg.Server.LogFormat = "text" + } + const defaultShutdownSec = 30 if cfg.Server.ShutdownTimeout.Duration() == 0 { cfg.Server.ShutdownTimeout = DurationSec(time.Duration(defaultShutdownSec) * time.Second) diff --git a/internal/models/taxonomy.go b/internal/models/taxonomy.go index 64ee9e6a..b14ed373 100644 --- a/internal/models/taxonomy.go +++ b/internal/models/taxonomy.go @@ -251,8 +251,26 @@ type TaxonomyRunResultRequest struct { // TaxonomyRunFailedRequest records a taxonomy run failure. type TaxonomyRunFailedRequest struct { - Error string `json:"error" validate:"required,no_null_bytes,min=1,max=2000"` - ErrorCode TaxonomyRunFailureCode `json:"error_code,omitempty" validate:"omitempty,oneof=insufficient_data service_unavailable generation_failed invalid_output internal_error"` //nolint:lll // Validator oneof values are space-delimited. + Error string `json:"error" validate:"required,no_null_bytes,min=1,max=2000"` + ErrorCode TaxonomyRunFailureCode `json:"error_code,omitempty" validate:"omitempty,oneof=insufficient_data service_unavailable generation_failed invalid_output internal_error"` //nolint:lll // Validator oneof values are space-delimited. + Diagnostics *TaxonomyRunFailureDiagnostics `json:"diagnostics,omitempty" validate:"omitempty"` +} + +// TaxonomyRunFailureDiagnostics contains bounded, non-sensitive compute diagnostics. It is stored +// inside the existing metrics JSON column, keeping the database and public API schema unchanged. +type TaxonomyRunFailureDiagnostics struct { + Phase string `json:"phase,omitempty" validate:"omitempty,oneof=fetch cluster label tree persist unknown"` //nolint:lll + FailureReason string `json:"failure_reason,omitempty" validate:"omitempty,oneof=provider_authentication provider_rate_limit provider_timeout provider_unavailable provider_response invalid_output validation_failed insufficient_data hub_unavailable internal_error unknown"` //nolint:lll + Provider string `json:"provider,omitempty" validate:"omitempty,oneof=openai bedrock vertex unknown"` + Model string `json:"model,omitempty" validate:"omitempty,no_null_bytes,max=255"` + ProviderAdapter string `json:"provider_adapter,omitempty" validate:"omitempty,no_null_bytes,max=100"` + AdapterVersion string `json:"adapter_version,omitempty" validate:"omitempty,no_null_bytes,max=100"` + ProviderSDKVersion string `json:"provider_sdk_version,omitempty" validate:"omitempty,no_null_bytes,max=100"` + LLMAttempts *int `json:"llm_attempts,omitempty" validate:"omitempty,min=0,max=1000"` + InputTokens *int64 `json:"input_tokens,omitempty" validate:"omitempty,min=0"` + OutputTokens *int64 `json:"output_tokens,omitempty" validate:"omitempty,min=0"` + TotalTokens *int64 `json:"total_tokens,omitempty" validate:"omitempty,min=0"` + PhaseDurations map[string]float64 `json:"phase_durations_seconds,omitempty"` } // RenameTaxonomyNodeRequest renames a generated taxonomy node. diff --git a/internal/observability/aggregate.go b/internal/observability/aggregate.go index 0194bd0b..6a5f45e8 100644 --- a/internal/observability/aggregate.go +++ b/internal/observability/aggregate.go @@ -22,6 +22,7 @@ type Metrics struct { EnrichmentClear EnrichmentClearMetrics // EnrichmentBacklog gauges the aggregate eligible-but-unenriched record count per enrichment. EnrichmentBacklog EnrichmentBacklogMetrics + Taxonomy TaxonomyMetrics } // NewMetrics creates EventMetrics, WebhookMetrics, EmbeddingMetrics, TranslationMetrics, and CacheMetrics from the given meter. @@ -77,6 +78,11 @@ func NewMetrics(meter metric.Meter) (*Metrics, error) { return nil, fmt.Errorf("enrichment backlog metrics: %w", err) } + taxonomy, err := NewTaxonomyMetrics(meter) + if err != nil { + return nil, fmt.Errorf("taxonomy metrics: %w", err) + } + return &Metrics{ Events: events, Webhooks: webhooks, @@ -87,5 +93,6 @@ func NewMetrics(meter metric.Meter) (*Metrics, error) { Cache: cache, EnrichmentClear: enrichmentClear, EnrichmentBacklog: enrichmentBacklog, + Taxonomy: taxonomy, }, nil } diff --git a/internal/observability/logging.go b/internal/observability/logging.go index 94c66a12..51d9c9a7 100644 --- a/internal/observability/logging.go +++ b/internal/observability/logging.go @@ -3,7 +3,9 @@ package observability import ( "context" "fmt" + "io" "log/slog" + "strings" "go.opentelemetry.io/otel/trace" ) @@ -71,3 +73,27 @@ func (h *TraceContextHandler) WithAttrs(attrs []slog.Attr) slog.Handler { func (h *TraceContextHandler) WithGroup(name string) slog.Handler { return &TraceContextHandler{inner: h.inner.WithGroup(name)} } + +// NewLogHandler creates a structured JSON or human-readable text slog handler. +// Text remains the default for local and backwards-compatible deployments. +func NewLogHandler(output io.Writer, level, format string) slog.Handler { + var logLevel slog.Level + + switch strings.ToLower(strings.TrimSpace(level)) { + case "debug": + logLevel = slog.LevelDebug + case "warn": + logLevel = slog.LevelWarn + case "error": + logLevel = slog.LevelError + default: + logLevel = slog.LevelInfo + } + + options := &slog.HandlerOptions{Level: logLevel} + if strings.EqualFold(strings.TrimSpace(format), "json") { + return NewTraceContextHandler(slog.NewJSONHandler(output, options)) + } + + return NewTraceContextHandler(slog.NewTextHandler(output, options)) +} diff --git a/internal/observability/logging_test.go b/internal/observability/logging_test.go new file mode 100644 index 00000000..269edf14 --- /dev/null +++ b/internal/observability/logging_test.go @@ -0,0 +1,38 @@ +package observability + +import ( + "bytes" + "context" + "encoding/json" + "log/slog" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestNewLogHandlerJSONIncludesRequestCorrelation(t *testing.T) { + var output bytes.Buffer + + logger := slog.New(NewLogHandler(&output, "info", "json")) + ctx := context.WithValue(context.Background(), RequestIDKey, "request-1") + + logger.InfoContext(ctx, "taxonomy lifecycle", "event", "hub.taxonomy.run.started") + + var record map[string]any + require.NoError(t, json.Unmarshal(output.Bytes(), &record)) + assert.Equal(t, "request-1", record["request_id"]) + assert.Equal(t, "hub.taxonomy.run.started", record["event"]) +} + +func TestNewLogHandlerDefaultsToText(t *testing.T) { + var output bytes.Buffer + + logger := slog.New(NewLogHandler(&output, "info", "")) + + logger.Info("taxonomy lifecycle", "event", "hub.taxonomy.run.started") + + assert.True(t, strings.HasPrefix(output.String(), "time=")) + assert.Contains(t, output.String(), "event=hub.taxonomy.run.started") +} diff --git a/internal/observability/names.go b/internal/observability/names.go index f78c6c96..a1efea97 100644 --- a/internal/observability/names.go +++ b/internal/observability/names.go @@ -54,6 +54,12 @@ const ( MetricNameCacheHits = "hub_cache_hits_total" MetricNameCacheMisses = "hub_cache_misses_total" + + MetricNameTaxonomyRunsStarted = "hub_taxonomy_runs_started_total" + MetricNameTaxonomyRunOutcomes = "hub_taxonomy_runs_total" + MetricNameTaxonomyRunDuration = "hub_taxonomy_run_duration_seconds" + MetricNameTaxonomyDispatchError = "hub_taxonomy_dispatch_errors_total" + MetricNameTaxonomyRunsReaped = "hub_taxonomy_runs_reaped_total" ) // Attribute keys. @@ -70,7 +76,9 @@ const ( // AttrEnrichment labels the enrichment-backlog gauge; values are the fixed enrichment types // (translation, sentiment, emotions). tenant_id is deliberately NOT a label — the gauge is // aggregated across all tenants to keep cardinality bounded. - AttrEnrichment = "enrichment" + AttrEnrichment = "enrichment" + AttrFailureCode = "failure_code" + AttrScopeType = "scope_type" ) // AllowedEventTypes returns event type strings allowed for metric attributes (bounded cardinality). diff --git a/internal/observability/taxonomy.go b/internal/observability/taxonomy.go new file mode 100644 index 00000000..1d926fc4 --- /dev/null +++ b/internal/observability/taxonomy.go @@ -0,0 +1,134 @@ +package observability + +import ( + "context" + "fmt" + "time" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +// TaxonomyMetrics records bounded-cardinality taxonomy lifecycle metrics. Run, request, +// tenant, source, and field identifiers intentionally belong only in correlated logs. +type TaxonomyMetrics interface { + RecordRunStarted(ctx context.Context, scopeType string) + RecordRunOutcome(ctx context.Context, status, failureCode, scopeType string) + RecordRunDuration(ctx context.Context, duration time.Duration, status, scopeType string) + RecordDispatchError(ctx context.Context, reason string) + RecordRunsReaped(ctx context.Context, count int64) +} + +type taxonomyMetrics struct { + started metric.Int64Counter + outcomes metric.Int64Counter + duration metric.Float64Histogram + dispatchError metric.Int64Counter + reaped metric.Int64Counter +} + +// NewTaxonomyMetrics creates taxonomy lifecycle metrics. +func NewTaxonomyMetrics(meter metric.Meter) (TaxonomyMetrics, error) { + if meter == nil { + return nil, nil //nolint:nilnil // disabled metrics are represented by a nil interface + } + + started, err := meter.Int64Counter(MetricNameTaxonomyRunsStarted, + metric.WithDescription("Taxonomy runs accepted and dispatched by Hub")) + if err != nil { + return nil, fmt.Errorf("create taxonomy started counter: %w", err) + } + + outcomes, err := meter.Int64Counter(MetricNameTaxonomyRunOutcomes, + metric.WithDescription("Terminal taxonomy run outcomes")) + if err != nil { + return nil, fmt.Errorf("create taxonomy outcomes counter: %w", err) + } + + duration, err := meter.Float64Histogram(MetricNameTaxonomyRunDuration, + metric.WithDescription("Taxonomy run wall-clock duration"), metric.WithUnit("s")) + if err != nil { + return nil, fmt.Errorf("create taxonomy duration histogram: %w", err) + } + + dispatchError, err := meter.Int64Counter(MetricNameTaxonomyDispatchError, + metric.WithDescription("Taxonomy dispatch failures")) + if err != nil { + return nil, fmt.Errorf("create taxonomy dispatch error counter: %w", err) + } + + reaped, err := meter.Int64Counter(MetricNameTaxonomyRunsReaped, + metric.WithDescription("Taxonomy runs failed by the stuck-run reaper")) + if err != nil { + return nil, fmt.Errorf("create taxonomy reaped counter: %w", err) + } + + return &taxonomyMetrics{ + started: started, + outcomes: outcomes, + duration: duration, + dispatchError: dispatchError, + reaped: reaped, + }, nil +} + +func (m *taxonomyMetrics) RecordRunStarted(ctx context.Context, scopeType string) { + m.started.Add(ctx, 1, metric.WithAttributes(attribute.String(AttrScopeType, boundedScopeType(scopeType)))) +} + +func (m *taxonomyMetrics) RecordRunOutcome(ctx context.Context, status, failureCode, scopeType string) { + m.outcomes.Add(ctx, 1, metric.WithAttributes( + attribute.String(AttrStatus, boundedTaxonomyStatus(status)), + attribute.String(AttrFailureCode, boundedFailureCode(failureCode)), + attribute.String(AttrScopeType, boundedScopeType(scopeType)), + )) +} + +func (m *taxonomyMetrics) RecordRunDuration( + ctx context.Context, duration time.Duration, status, scopeType string, +) { + m.duration.Record(ctx, duration.Seconds(), metric.WithAttributes( + attribute.String(AttrStatus, boundedTaxonomyStatus(status)), + attribute.String(AttrScopeType, boundedScopeType(scopeType)), + )) +} + +func (m *taxonomyMetrics) RecordDispatchError(ctx context.Context, reason string) { + if reason != "request_failed" && reason != "mark_failed_failed" { + reason = "other" + } + + m.dispatchError.Add(ctx, 1, metric.WithAttributes(attribute.String(AttrReason, reason))) +} + +func (m *taxonomyMetrics) RecordRunsReaped(ctx context.Context, count int64) { + if count > 0 { + m.reaped.Add(ctx, count) + } +} + +func boundedTaxonomyStatus(value string) string { + switch value { + case "succeeded", "failed", "canceled": + return value + default: + return "other" + } +} + +func boundedFailureCode(value string) string { + switch value { + case "none", "insufficient_data", "service_unavailable", "generation_failed", "invalid_output", "internal_error": + return value + default: + return "other" + } +} + +func boundedScopeType(value string) string { + if value == "field" || value == "directory" { + return value + } + + return "other" +} diff --git a/internal/observability/taxonomy_test.go b/internal/observability/taxonomy_test.go new file mode 100644 index 00000000..0dd4a350 --- /dev/null +++ b/internal/observability/taxonomy_test.go @@ -0,0 +1,48 @@ +package observability + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" +) + +func TestTaxonomyMetricsEmitBoundedAttributesWithoutIdentifiers(t *testing.T) { + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + metrics, err := NewTaxonomyMetrics(provider.Meter("test")) + require.NoError(t, err) + require.NotNil(t, metrics) + + ctx := context.Background() + metrics.RecordRunStarted(ctx, "field") + metrics.RecordRunOutcome(ctx, "failed", "provider-secret-value", "tenant-secret-value") + metrics.RecordRunDuration(ctx, 3*time.Second, "failed", "field") + metrics.RecordDispatchError(ctx, "unbounded-error-text") + metrics.RecordRunsReaped(ctx, 2) + + var collected metricdata.ResourceMetrics + require.NoError(t, reader.Collect(ctx, &collected)) + assert.Equal(t, int64(1), counterValue(collected, MetricNameTaxonomyRunsStarted, AttrScopeType, "field")) + assert.Equal(t, int64(1), counterValue(collected, MetricNameTaxonomyRunOutcomes, AttrFailureCode, "other")) + assert.Equal(t, int64(1), counterValue(collected, MetricNameTaxonomyDispatchError, AttrReason, "other")) + assert.Equal(t, int64(2), counterValue(collected, MetricNameTaxonomyRunsReaped, "", "")) + + for _, scope := range collected.ScopeMetrics { + for _, item := range scope.Metrics { + if sum, ok := item.Data.(metricdata.Sum[int64]); ok { + for _, point := range sum.DataPoints { + for _, forbidden := range []string{"run_id", "request_id", "tenant_id", "source_id", "field_id"} { + _, present := point.Attributes.Value(attribute.Key(forbidden)) + assert.False(t, present, "%s must never be a metric attribute", forbidden) + } + } + } + } + } +} diff --git a/internal/repository/taxonomy_repository.go b/internal/repository/taxonomy_repository.go index 502b91b5..9590a791 100644 --- a/internal/repository/taxonomy_repository.go +++ b/internal/repository/taxonomy_repository.go @@ -297,6 +297,7 @@ func (r *TaxonomyRepository) MarkRunFailed( tenantID string, message string, errorCode models.TaxonomyRunFailureCode, + metrics json.RawMessage, ) (*models.TaxonomyRun, error) { var run *models.TaxonomyRun @@ -304,11 +305,12 @@ func (r *TaxonomyRepository) MarkRunFailed( updated, err := queryTaxonomyRun(ctx, dbTx, ` WITH taxonomy_runs AS ( UPDATE taxonomy_runs - SET status = 'failed', error = $2, error_code = $3, finished_at = NOW(), updated_at = NOW() - WHERE id = $1 AND tenant_id = $4 AND status IN ('pending', 'running') + SET status = 'failed', error = $2, error_code = $3, metrics = $4, + finished_at = NOW(), updated_at = NOW() + WHERE id = $1 AND tenant_id = $5 AND status IN ('pending', 'running') RETURNING * )`+taxonomyRunSelect+` FROM taxonomy_runs`, - runID, message, nullableFailureCode(errorCode), tenantID, + runID, message, nullableFailureCode(errorCode), rawOrDefault(metrics, defaultJSONObj), tenantID, ) if err != nil { if errors.Is(err, pgx.ErrNoRows) { @@ -397,6 +399,18 @@ func (r *TaxonomyRepository) FailRunIfStale( return failed, nil } +// ReapedTaxonomyRun identifies a run transitioned by the reaper for correlated operator logs. +type ReapedTaxonomyRun struct { + ID uuid.UUID + TenantID string + ScopeType models.TaxonomyScopeType + SourceType string + SourceID string + FieldID string + StartedAt *time.Time + CreatedAt time.Time +} + // FailStuckRuns marks taxonomy runs stuck in a non-terminal state (pending/running) past olderThan // as failed. Runs are orphaned when the taxonomy service crashes mid-run or its terminal callback is // lost; without this sweep they are polled forever in the UI and block regeneration. @@ -412,40 +426,42 @@ func (r *TaxonomyRepository) FailRunIfStale( // invariant that coordinates tenant-owned writes with tenant-data purges. Stuck runs are rare, so a // per-run loop rather than a batched per-tenant update is sufficient. The final UPDATE re-checks both // status and updated_at under the tenant lock, so a run that heartbeats, reaches a terminal state, or -// is removed between selection and update is skipped. -// Returns the number of runs failed and the first unexpected error, if any. +// is removed between selection and update is skipped. It returns the runs failed and the first +// unexpected error, if any. func (r *TaxonomyRepository) FailStuckRuns( ctx context.Context, olderThan time.Duration, message string, errorCode models.TaxonomyRunFailureCode, -) (int64, error) { +) ([]ReapedTaxonomyRun, error) { cutoff := time.Now().Add(-olderThan) type stuckRun struct { - id uuid.UUID - tenantID string + ReapedTaxonomyRun } rows, err := r.db.Query(ctx, ` - SELECT id, tenant_id + SELECT id, tenant_id, scope_type, source_type, source_id, field_id, started_at, created_at FROM taxonomy_runs WHERE status IN ('pending', 'running') AND updated_at < $1`, cutoff, ) if err != nil { - return 0, fmt.Errorf("select stuck taxonomy runs: %w", err) + return nil, fmt.Errorf("select stuck taxonomy runs: %w", err) } var candidates []stuckRun for rows.Next() { var run stuckRun - if scanErr := rows.Scan(&run.id, &run.tenantID); scanErr != nil { + if scanErr := rows.Scan( + &run.ID, &run.TenantID, &run.ScopeType, &run.SourceType, &run.SourceID, &run.FieldID, + &run.StartedAt, &run.CreatedAt, + ); scanErr != nil { rows.Close() - return 0, fmt.Errorf("scan stuck taxonomy run: %w", scanErr) + return nil, fmt.Errorf("scan stuck taxonomy run: %w", scanErr) } candidates = append(candidates, run) @@ -454,19 +470,19 @@ func (r *TaxonomyRepository) FailStuckRuns( rows.Close() if rowsErr := rows.Err(); rowsErr != nil { - return 0, fmt.Errorf("iterate stuck taxonomy runs: %w", rowsErr) + return nil, fmt.Errorf("iterate stuck taxonomy runs: %w", rowsErr) } var ( - reaped int64 + reaped []ReapedTaxonomyRun firstErr error ) for _, run := range candidates { - failed, markErr := r.FailRunIfStale(ctx, run.id, run.tenantID, cutoff, message, errorCode) + failed, markErr := r.FailRunIfStale(ctx, run.ID, run.TenantID, cutoff, message, errorCode) if markErr == nil { if failed { - reaped++ + reaped = append(reaped, run.ReapedTaxonomyRun) } continue diff --git a/internal/service/taxonomy_service.go b/internal/service/taxonomy_service.go index dadb52e3..5e39d09a 100644 --- a/internal/service/taxonomy_service.go +++ b/internal/service/taxonomy_service.go @@ -5,12 +5,15 @@ import ( "encoding/json" "errors" "fmt" + "log/slog" "strings" + "time" "github.com/google/uuid" "github.com/formbricks/hub/internal/huberrors" "github.com/formbricks/hub/internal/models" + "github.com/formbricks/hub/internal/observability" "github.com/formbricks/hub/internal/repository" ) @@ -39,6 +42,7 @@ type TaxonomyRepository interface { //nolint:interfacebloat // taxonomy service tenantID string, message string, errorCode models.TaxonomyRunFailureCode, + metrics json.RawMessage, ) (*models.TaxonomyRun, error) Heartbeat(ctx context.Context, runID uuid.UUID, tenantID string) error GetRunForInternalService(ctx context.Context, runID uuid.UUID) (*models.TaxonomyRun, error) @@ -75,6 +79,7 @@ type TaxonomyService struct { starter TaxonomyRunStarter embeddingModel string minimumEmbeddingCount int + metrics observability.TaxonomyMetrics } // NewTaxonomyServiceParams configures a TaxonomyService. @@ -83,6 +88,7 @@ type NewTaxonomyServiceParams struct { Starter TaxonomyRunStarter EmbeddingModel string MinimumEmbeddingCount int + Metrics observability.TaxonomyMetrics } // NewTaxonomyService creates a taxonomy application service. @@ -97,6 +103,7 @@ func NewTaxonomyService(params NewTaxonomyServiceParams) *TaxonomyService { starter: params.Starter, embeddingModel: strings.TrimSpace(params.EmbeddingModel), minimumEmbeddingCount: minimumEmbeddingCount, + metrics: params.Metrics, } } @@ -193,18 +200,31 @@ func (s *TaxonomyService) StartManualRun( return nil, fmt.Errorf("mark taxonomy run running: %w", err) } + s.recordStarted(ctx, runningRun) + if err := s.starter.StartRun(ctx, run.ID.String()); err != nil { - _, markErr := s.repo.MarkRunFailed( + failedRun, markErr := s.repo.MarkRunFailed( ctx, run.ID, scope.TenantID, "taxonomy service did not accept the run", models.TaxonomyRunFailureCodeServiceUnavailable, + nil, ) if markErr != nil { + if s.metrics != nil { + s.metrics.RecordDispatchError(ctx, "mark_failed_failed") + } + return nil, fmt.Errorf("mark taxonomy run failed after start error: %w", markErr) } + if s.metrics != nil { + s.metrics.RecordDispatchError(ctx, "request_failed") + } + + s.recordTerminal(ctx, failedRun, models.TaxonomyRunFailureCodeServiceUnavailable) + return nil, fmt.Errorf("%w: %w", ErrTaxonomyServiceStartFailed, err) } @@ -350,6 +370,8 @@ func (s *TaxonomyService) CompleteRun( return nil, fmt.Errorf("complete taxonomy run: %w", err) } + s.recordTerminal(ctx, run, "") + return run, nil } @@ -357,21 +379,27 @@ func (s *TaxonomyService) CompleteRun( func (s *TaxonomyService) FailRun( ctx context.Context, runID uuid.UUID, - message string, - errorCode models.TaxonomyRunFailureCode, + req models.TaxonomyRunFailedRequest, ) (*models.TaxonomyRun, error) { - sanitized, normalizedCode := normalizeRunFailure(message, errorCode) + sanitized, normalizedCode := normalizeRunFailure(req.Error, req.ErrorCode) existingRun, err := s.repo.GetRunForInternalService(ctx, runID) if err != nil { return nil, fmt.Errorf("get taxonomy run: %w", err) } - run, err := s.repo.MarkRunFailed(ctx, runID, existingRun.TenantID, sanitized, normalizedCode) + metrics, err := marshalFailureDiagnostics(req.Diagnostics) + if err != nil { + return nil, fmt.Errorf("marshal taxonomy failure diagnostics: %w", err) + } + + run, err := s.repo.MarkRunFailed(ctx, runID, existingRun.TenantID, sanitized, normalizedCode, metrics) if err != nil { return nil, fmt.Errorf("fail taxonomy run: %w", err) } + s.recordTerminal(ctx, run, normalizedCode) + return run, nil } @@ -494,6 +522,74 @@ func (s *TaxonomyService) ListNodeRecords( return &models.TaxonomyNodeRecordsResponse{Data: records, Limit: limit}, nil } +func (s *TaxonomyService) recordStarted(ctx context.Context, run *models.TaxonomyRun) { + if s.metrics != nil { + s.metrics.RecordRunStarted(ctx, string(run.ScopeType)) + } + + slog.InfoContext(ctx, "taxonomy run started", + "event", "hub.taxonomy.run.started", "run_id", run.ID, "tenant_id", run.TenantID, + "scope_type", run.ScopeType, "source_type", run.SourceType, "source_id", run.SourceID, + "field_id", run.FieldID, "record_count", run.RecordCount, "embedding_count", run.EmbeddingCount) +} + +func (s *TaxonomyService) recordTerminal( + ctx context.Context, run *models.TaxonomyRun, failureCode models.TaxonomyRunFailureCode, +) { + if run == nil { + return + } + + code := "none" + if failureCode != "" { + code = string(failureCode) + } + + duration := taxonomyRunDuration(run) + if s.metrics != nil { + s.metrics.RecordRunOutcome(ctx, string(run.Status), code, string(run.ScopeType)) + s.metrics.RecordRunDuration(ctx, duration, string(run.Status), string(run.ScopeType)) + } + + slog.InfoContext(ctx, "taxonomy run reached terminal state", + "event", "hub.taxonomy.run.terminal", "run_id", run.ID, "tenant_id", run.TenantID, + "scope_type", run.ScopeType, "source_type", run.SourceType, "source_id", run.SourceID, + "field_id", run.FieldID, "status", run.Status, "failure_code", code, + "duration_seconds", duration.Seconds(), "record_count", run.RecordCount, + "embedding_count", run.EmbeddingCount, "cluster_count", run.ClusterCount, "node_count", run.NodeCount) +} + +func marshalFailureDiagnostics(diagnostics *models.TaxonomyRunFailureDiagnostics) (json.RawMessage, error) { + if diagnostics == nil { + return nil, nil + } + + raw, err := json.Marshal(map[string]*models.TaxonomyRunFailureDiagnostics{"failure_diagnostics": diagnostics}) + if err != nil { + return nil, fmt.Errorf("marshal diagnostics: %w", err) + } + + return raw, nil +} + +func taxonomyRunDuration(run *models.TaxonomyRun) time.Duration { + start := run.CreatedAt + if run.StartedAt != nil { + start = *run.StartedAt + } + + end := time.Now() + if run.FinishedAt != nil { + end = *run.FinishedAt + } + + if end.Before(start) { + return 0 + } + + return end.Sub(start) +} + func normalizeTaxonomyScope(scope models.TaxonomyScope) (models.TaxonomyScope, error) { tenantID, err := normalizeRequiredTenantIDValue(scope.TenantID) if err != nil { diff --git a/internal/service/taxonomy_service_test.go b/internal/service/taxonomy_service_test.go index a159fc13..6c790a15 100644 --- a/internal/service/taxonomy_service_test.go +++ b/internal/service/taxonomy_service_test.go @@ -2,6 +2,7 @@ package service import ( "context" + "encoding/json" "errors" "testing" @@ -22,6 +23,7 @@ type mockTaxonomyRepo struct { markRunFailedMessage string markRunFailedCode models.TaxonomyRunFailureCode markRunFailedTenant string + markRunFailedMetrics json.RawMessage heartbeatTenant string countNodeRecords []models.TaxonomyNodeRecordCount @@ -71,10 +73,12 @@ func (m *mockTaxonomyRepo) MarkRunFailed( tenantID string, message string, errorCode models.TaxonomyRunFailureCode, + metrics json.RawMessage, ) (*models.TaxonomyRun, error) { m.markRunFailedTenant = tenantID m.markRunFailedMessage = message m.markRunFailedCode = errorCode + m.markRunFailedMetrics = metrics return &models.TaxonomyRun{ ID: runID, @@ -298,7 +302,9 @@ func TestTaxonomyService_FailRunDefaultsFailureCode(t *testing.T) { repo := &mockTaxonomyRepo{} svc := NewTaxonomyService(NewTaxonomyServiceParams{Repo: repo}) - result, err := svc.FailRun(context.Background(), runID, " generated invalid taxonomy ", "") + result, err := svc.FailRun(context.Background(), runID, models.TaxonomyRunFailedRequest{ + Error: " generated invalid taxonomy ", + }) if err != nil { t.Fatalf("FailRun() error = %v", err) } @@ -320,6 +326,44 @@ func TestTaxonomyService_FailRunDefaultsFailureCode(t *testing.T) { } } +func TestTaxonomyService_FailRunPersistsSafeDiagnosticsInMetrics(t *testing.T) { + runID := uuid.MustParse("018e1234-5678-9abc-def0-222222222223") + attempts := 2 + tokens := int64(42) + repo := &mockTaxonomyRepo{} + svc := NewTaxonomyService(NewTaxonomyServiceParams{Repo: repo}) + + _, err := svc.FailRun(context.Background(), runID, models.TaxonomyRunFailedRequest{ + Error: "provider request failed", + ErrorCode: models.TaxonomyRunFailureCodeServiceUnavailable, + Diagnostics: &models.TaxonomyRunFailureDiagnostics{ + Phase: "tree", + FailureReason: "provider_timeout", + Provider: "vertex", + Model: "gemini-2.5-flash", + LLMAttempts: &attempts, + TotalTokens: &tokens, + }, + }) + if err != nil { + t.Fatalf("FailRun() error = %v", err) + } + + var persisted map[string]models.TaxonomyRunFailureDiagnostics + if err := json.Unmarshal(repo.markRunFailedMetrics, &persisted); err != nil { + t.Fatalf("unmarshal persisted diagnostics: %v", err) + } + + diagnostics := persisted["failure_diagnostics"] + if diagnostics.FailureReason != "provider_timeout" || diagnostics.Provider != "vertex" { + t.Fatalf("persisted diagnostics = %+v", diagnostics) + } + + if diagnostics.TotalTokens == nil || *diagnostics.TotalTokens != 42 { + t.Fatalf("persisted total tokens = %v", diagnostics.TotalTokens) + } +} + func TestTaxonomyService_GetNodeRecordCounts(t *testing.T) { runID := uuid.MustParse("018e1234-5678-9abc-def0-444444444444") nodeID := uuid.MustParse("018e1234-5678-9abc-def0-555555555555") diff --git a/tests/taxonomy_persistence_test.go b/tests/taxonomy_persistence_test.go index 67427d15..312db174 100644 --- a/tests/taxonomy_persistence_test.go +++ b/tests/taxonomy_persistence_test.go @@ -104,6 +104,7 @@ func TestTaxonomyRepository_RunLifecycle(t *testing.T) { failed, err := repo.MarkRunFailed( ctx, run.ID, scope.TenantID, "clustering failed", models.TaxonomyRunFailureCodeInsufficientData, + nil, ) require.NoError(t, err) require.Equal(t, models.TaxonomyRunStatusFailed, failed.Status) @@ -115,6 +116,7 @@ func TestTaxonomyRepository_RunLifecycle(t *testing.T) { _, err = repo.MarkRunFailed( ctx, run.ID, scope.TenantID, "again", models.TaxonomyRunFailureCodeInternalError, + nil, ) require.ErrorIs(t, err, huberrors.ErrConflict, "failed->failed must conflict") }) @@ -165,7 +167,7 @@ func TestTaxonomyRepository_FailStuckRuns(t *testing.T) { failed, err := repo.FailStuckRuns(ctx, time.Hour, "stuck run", models.TaxonomyRunFailureCodeInternalError) require.NoError(t, err) - require.GreaterOrEqual(t, failed, int64(2), "both orphaned runs should be reaped") + require.GreaterOrEqual(t, len(failed), 2, "both orphaned runs should be reaped") for _, id := range []uuid.UUID{stuck.ID, pending.ID} { reaped, err := repo.GetRunForInternalService(ctx, id) @@ -247,6 +249,7 @@ func TestTaxonomyRepository_Heartbeat(t *testing.T) { require.NoError(t, err) failed, err := repo.MarkRunFailed( ctx, run.ID, scope.TenantID, "boom", models.TaxonomyRunFailureCodeInternalError, + nil, ) require.NoError(t, err) From 15f43df7c7344d1f81c922520de0d65cf0aa23e5 Mon Sep 17 00:00:00 2001 From: Bhagya Amarasinghe Date: Tue, 4 Aug 2026 17:12:25 +0530 Subject: [PATCH 2/3] fix: address taxonomy observability review feedback --- cmd/api/app.go | 3 +- internal/models/taxonomy.go | 2 +- internal/repository/taxonomy_repository.go | 32 ++++++++---- tests/taxonomy_api_test.go | 54 +++++++++++++++++++ tests/taxonomy_persistence_test.go | 61 ++++++++++++++++++++-- 5 files changed, 136 insertions(+), 16 deletions(-) diff --git a/cmd/api/app.go b/cmd/api/app.go index 0c5f9630..5e28cc31 100644 --- a/cmd/api/app.go +++ b/cmd/api/app.go @@ -972,7 +972,8 @@ func runTaxonomyRunReaper( started = *run.StartedAt } - metrics.RecordRunDuration(ctx, time.Since(started), string(models.TaxonomyRunStatusFailed), + duration := max(run.FinishedAt.Sub(started), 0) + metrics.RecordRunDuration(ctx, duration, string(models.TaxonomyRunStatusFailed), string(run.ScopeType)) } } diff --git a/internal/models/taxonomy.go b/internal/models/taxonomy.go index b14ed373..1a749949 100644 --- a/internal/models/taxonomy.go +++ b/internal/models/taxonomy.go @@ -270,7 +270,7 @@ type TaxonomyRunFailureDiagnostics struct { InputTokens *int64 `json:"input_tokens,omitempty" validate:"omitempty,min=0"` OutputTokens *int64 `json:"output_tokens,omitempty" validate:"omitempty,min=0"` TotalTokens *int64 `json:"total_tokens,omitempty" validate:"omitempty,min=0"` - PhaseDurations map[string]float64 `json:"phase_durations_seconds,omitempty"` + PhaseDurations map[string]float64 `json:"phase_durations_seconds,omitempty" validate:"omitempty,max=8,dive,keys,oneof=input_fetch input_validation clustering evidence_selection cluster_labeling taxonomy_generation payload_validation persistence,endkeys,gte=0"` //nolint:lll } // RenameTaxonomyNodeRequest renames a generated taxonomy node. diff --git a/internal/repository/taxonomy_repository.go b/internal/repository/taxonomy_repository.go index 9590a791..56eaef85 100644 --- a/internal/repository/taxonomy_repository.go +++ b/internal/repository/taxonomy_repository.go @@ -363,7 +363,8 @@ func (r *TaxonomyRepository) Heartbeat( // FailRunIfStale marks a pending/running taxonomy run as failed only when its liveness timestamp is // still older than cutoff at the moment of the update. The status and freshness checks are part of // the tenant-locked UPDATE so a heartbeat that lands after candidate selection protects the run. -// Returns true when the run was failed and false when it was refreshed, completed, or removed first. +// Returns the database terminal timestamp when the run was failed and nil when it was refreshed, +// completed, or removed first. func (r *TaxonomyRepository) FailRunIfStale( ctx context.Context, runID uuid.UUID, @@ -371,32 +372,39 @@ func (r *TaxonomyRepository) FailRunIfStale( cutoff time.Time, message string, errorCode models.TaxonomyRunFailureCode, -) (bool, error) { - var failed bool +) (*time.Time, error) { + var finishedAt *time.Time err := withTenantWritePoolTx(ctx, r.db, []string{tenantID}, func(dbTx tenantWriteTx) error { - tag, err := dbTx.Exec(ctx, ` + var terminalTime time.Time + + err := dbTx.QueryRow(ctx, ` UPDATE taxonomy_runs SET status = 'failed', error = $3, error_code = $4, finished_at = NOW(), updated_at = NOW() WHERE id = $1 AND tenant_id = $2 AND status IN ('pending', 'running') - AND updated_at < $5`, + AND updated_at < $5 + RETURNING finished_at`, runID, tenantID, message, nullableFailureCode(errorCode), cutoff, - ) + ).Scan(&terminalTime) + if errors.Is(err, pgx.ErrNoRows) { + return nil + } + if err != nil { return fmt.Errorf("fail stale taxonomy run: %w", err) } - failed = tag.RowsAffected() > 0 + finishedAt = &terminalTime return nil }) if err != nil { - return false, err + return nil, err } - return failed, nil + return finishedAt, nil } // ReapedTaxonomyRun identifies a run transitioned by the reaper for correlated operator logs. @@ -409,6 +417,7 @@ type ReapedTaxonomyRun struct { FieldID string StartedAt *time.Time CreatedAt time.Time + FinishedAt time.Time } // FailStuckRuns marks taxonomy runs stuck in a non-terminal state (pending/running) past olderThan @@ -479,9 +488,10 @@ func (r *TaxonomyRepository) FailStuckRuns( ) for _, run := range candidates { - failed, markErr := r.FailRunIfStale(ctx, run.ID, run.TenantID, cutoff, message, errorCode) + finishedAt, markErr := r.FailRunIfStale(ctx, run.ID, run.TenantID, cutoff, message, errorCode) if markErr == nil { - if failed { + if finishedAt != nil { + run.FinishedAt = *finishedAt reaped = append(reaped, run.ReapedTaxonomyRun) } diff --git a/tests/taxonomy_api_test.go b/tests/taxonomy_api_test.go index 306260f8..2eaa4fc5 100644 --- a/tests/taxonomy_api_test.go +++ b/tests/taxonomy_api_test.go @@ -837,6 +837,14 @@ func TestTaxonomyAPI_InternalServiceEndpoints(t *testing.T) { body := models.TaxonomyRunFailedRequest{ Error: "clustering did not converge", ErrorCode: models.TaxonomyRunFailureCodeGenerationFailed, + Diagnostics: &models.TaxonomyRunFailureDiagnostics{ + Phase: "cluster", + PhaseDurations: map[string]float64{ + "input_fetch": 0.5, "input_validation": 0.1, "clustering": 2.5, + "evidence_selection": 0.2, "cluster_labeling": 1.2, "taxonomy_generation": 3.4, + "payload_validation": 0.3, "persistence": 0.7, + }, + }, } failedURL := harness.server.URL + "/internal/v1/taxonomy/runs/" + runID.String() + "/failed" @@ -849,6 +857,12 @@ func TestTaxonomyAPI_InternalServiceEndpoints(t *testing.T) { assert.Equal(t, models.TaxonomyRunStatusFailed, run.Status) require.NotNil(t, run.Error) assert.Equal(t, "clustering did not converge", *run.Error) + assert.JSONEq(t, + `{"failure_diagnostics":{"phase":"cluster","phase_durations_seconds":{`+ + `"input_fetch":0.5,"input_validation":0.1,"clustering":2.5,"evidence_selection":0.2,`+ + `"cluster_labeling":1.2,"taxonomy_generation":3.4,"payload_validation":0.3,"persistence":0.7}}}`, + string(run.Metrics), + ) }) t.Run("heartbeat returns the internal wire contract", func(t *testing.T) { @@ -917,6 +931,46 @@ func TestTaxonomyAPI_InternalErrors(t *testing.T) { assertTaxonomyInvalidParam(t, problem, "error", "required") }) + t.Run("failed diagnostics phase durations are bounded", func(t *testing.T) { + tests := []struct { + name string + durations map[string]float64 + }{ + {name: "disallowed key", durations: map[string]float64{"prompt_rendering": 1}}, + {name: "negative duration", durations: map[string]float64{"clustering": -0.1}}, + { + name: "too many entries", + durations: map[string]float64{ + "input_fetch": 1, "input_validation": 1, "clustering": 1, "evidence_selection": 1, + "cluster_labeling": 1, "taxonomy_generation": 1, "payload_validation": 1, + "persistence": 1, "prompt_rendering": 1, + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + requestTaxonomyProblem( + ctx, + t, + http.MethodPost, + harness.server.URL+"/internal/v1/taxonomy/runs/"+unknownRunID.String()+"/failed", + harness.internalToken, + models.TaxonomyRunFailedRequest{ + Error: "generation failed", + ErrorCode: models.TaxonomyRunFailureCodeGenerationFailed, + Diagnostics: &models.TaxonomyRunFailureDiagnostics{ + PhaseDurations: test.durations, + }, + }, + http.StatusBadRequest, + response.CodeValidation, + response.ProblemTypeValidation, + ) + }) + } + }) + tests := []struct { name string method string diff --git a/tests/taxonomy_persistence_test.go b/tests/taxonomy_persistence_test.go index 312db174..ac206caf 100644 --- a/tests/taxonomy_persistence_test.go +++ b/tests/taxonomy_persistence_test.go @@ -2,6 +2,7 @@ package tests import ( "context" + "encoding/json" "sync" "testing" "time" @@ -113,6 +114,11 @@ func TestTaxonomyRepository_RunLifecycle(t *testing.T) { require.NotNil(t, failed.ErrorCode) require.Equal(t, models.TaxonomyRunFailureCodeInsufficientData, *failed.ErrorCode) require.NotNil(t, failed.FinishedAt) + assert.JSONEq(t, `{}`, string(failed.Metrics)) + + roundTrip, err := repo.GetRunForInternalService(ctx, run.ID) + require.NoError(t, err) + assert.JSONEq(t, `{}`, string(roundTrip.Metrics)) _, err = repo.MarkRunFailed( ctx, run.ID, scope.TenantID, "again", models.TaxonomyRunFailureCodeInternalError, @@ -121,6 +127,26 @@ func TestTaxonomyRepository_RunLifecycle(t *testing.T) { require.ErrorIs(t, err, huberrors.ErrConflict, "failed->failed must conflict") }) + t.Run("mark failed persists metrics JSON", func(t *testing.T) { + scope := uniqueTaxonomyScope("tax-failed-metrics") + cleanupTaxonomyTenant(ctx, t, db, scope.TenantID) + + run, _, err := repo.CreateRunIfAvailable(ctx, repository.CreateTaxonomyRunParams{TaxonomyScope: scope}) + require.NoError(t, err) + + metrics := json.RawMessage(`{"failure_diagnostics":{"phase":"cluster","llm_attempts":2}}`) + failed, err := repo.MarkRunFailed( + ctx, run.ID, scope.TenantID, "clustering failed", models.TaxonomyRunFailureCodeGenerationFailed, + metrics, + ) + require.NoError(t, err) + assert.JSONEq(t, string(metrics), string(failed.Metrics)) + + roundTrip, err := repo.GetRunForInternalService(ctx, run.ID) + require.NoError(t, err) + assert.JSONEq(t, string(metrics), string(roundTrip.Metrics)) + }) + t.Run("unknown run id is not found", func(t *testing.T) { scope := uniqueTaxonomyScope("tax-missing") _, err := repo.MarkRunRunning(ctx, uuid.New(), scope.TenantID) @@ -148,7 +174,7 @@ func TestTaxonomyRepository_FailStuckRuns(t *testing.T) { // A run orphaned in `running` with a stale updated_at (no heartbeat for two hours). stuck, _, err := repo.CreateRunIfAvailable(ctx, repository.CreateTaxonomyRunParams{TaxonomyScope: stuckScope}) require.NoError(t, err) - _, err = repo.MarkRunRunning(ctx, stuck.ID, stuckScope.TenantID) + stuck, err = repo.MarkRunRunning(ctx, stuck.ID, stuckScope.TenantID) require.NoError(t, err) _, err = db.Exec(ctx, `UPDATE taxonomy_runs SET updated_at = NOW() - INTERVAL '2 hours' WHERE id = $1`, stuck.ID) require.NoError(t, err) @@ -169,6 +195,14 @@ func TestTaxonomyRepository_FailStuckRuns(t *testing.T) { require.NoError(t, err) require.GreaterOrEqual(t, len(failed), 2, "both orphaned runs should be reaped") + reapedByID := make(map[uuid.UUID]repository.ReapedTaxonomyRun, len(failed)) + for _, run := range failed { + reapedByID[run.ID] = run + } + + assertReapedMetadata(t, reapedByID[stuck.ID], stuck) + assertReapedMetadata(t, reapedByID[pending.ID], pending) + for _, id := range []uuid.UUID{stuck.ID, pending.ID} { reaped, err := repo.GetRunForInternalService(ctx, id) require.NoError(t, err) @@ -230,11 +264,11 @@ func TestTaxonomyRepository_Heartbeat(t *testing.T) { // The reaper has already selected this run using cutoff, but a heartbeat lands before its // tenant-locked transition. The final UPDATE must re-check freshness and leave the run alive. require.NoError(t, repo.Heartbeat(ctx, run.ID, scope.TenantID)) - failed, err := repo.FailRunIfStale( + finishedAt, err := repo.FailRunIfStale( ctx, run.ID, scope.TenantID, cutoff, "stuck run", models.TaxonomyRunFailureCodeInternalError, ) require.NoError(t, err) - assert.False(t, failed, "a run refreshed after candidate selection must not be failed") + assert.Nil(t, finishedAt, "a run refreshed after candidate selection must not be failed") survivor, err := repo.GetRunForInternalService(ctx, run.ID) require.NoError(t, err) @@ -315,6 +349,27 @@ func TestTaxonomyRepository_FailStuckRunsCoordinatesWithPurge(t *testing.T) { require.ErrorIs(t, err, huberrors.ErrNotFound, "the run must be purged regardless of interleaving") } +func assertReapedMetadata(t *testing.T, actual repository.ReapedTaxonomyRun, expected *models.TaxonomyRun) { + t.Helper() + require.Equal(t, expected.ID, actual.ID) + assert.Equal(t, expected.TenantID, actual.TenantID) + assert.Equal(t, expected.ScopeType, actual.ScopeType) + assert.Equal(t, expected.SourceType, actual.SourceType) + assert.Equal(t, expected.SourceID, actual.SourceID) + assert.Equal(t, expected.FieldID, actual.FieldID) + assert.WithinDuration(t, expected.CreatedAt, actual.CreatedAt, time.Microsecond) + require.False(t, actual.FinishedAt.IsZero()) + + if expected.StartedAt == nil { + assert.Nil(t, actual.StartedAt) + + return + } + + require.NotNil(t, actual.StartedAt) + assert.WithinDuration(t, *expected.StartedAt, *actual.StartedAt, time.Microsecond) +} + // TestTaxonomyRepository_StoreResultAndActivate covers persisting the full artifact graph // (clusters, memberships, nodes), activating the run, replacing a prior active run, and the // conflict when the run is not in the running state. From 1a4ba5ab981b5ed0c5f09a863dca9fb9adcdf3dc Mon Sep 17 00:00:00 2001 From: Bhagya Amarasinghe Date: Tue, 4 Aug 2026 18:37:04 +0530 Subject: [PATCH 3/3] fix: align taxonomy observability contracts --- cmd/api/app.go | 4 ---- internal/models/taxonomy.go | 2 +- internal/observability/logging_test.go | 2 ++ internal/observability/names.go | 2 +- internal/service/taxonomy_service_test.go | 2 +- tests/taxonomy_api_test.go | 4 ++-- 6 files changed, 7 insertions(+), 9 deletions(-) diff --git a/cmd/api/app.go b/cmd/api/app.go index 5e28cc31..3093d07e 100644 --- a/cmd/api/app.go +++ b/cmd/api/app.go @@ -236,10 +236,6 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) { } } - // Install TraceContextHandler unconditionally so request_id (and trace_id/span_id when tracing is on) appear in logs. - defaultHandler := slog.Default().Handler() - slog.SetDefault(slog.New(observability.NewTraceContextHandler(defaultHandler))) - if tracerProvider != nil { otel.SetTracerProvider(tracerProvider) } diff --git a/internal/models/taxonomy.go b/internal/models/taxonomy.go index 1a749949..26b05e15 100644 --- a/internal/models/taxonomy.go +++ b/internal/models/taxonomy.go @@ -259,7 +259,7 @@ type TaxonomyRunFailedRequest struct { // TaxonomyRunFailureDiagnostics contains bounded, non-sensitive compute diagnostics. It is stored // inside the existing metrics JSON column, keeping the database and public API schema unchanged. type TaxonomyRunFailureDiagnostics struct { - Phase string `json:"phase,omitempty" validate:"omitempty,oneof=fetch cluster label tree persist unknown"` //nolint:lll + Phase string `json:"phase,omitempty" validate:"omitempty,oneof=input_fetch input_validation clustering evidence_selection cluster_labeling taxonomy_generation payload_validation persistence unknown"` //nolint:lll FailureReason string `json:"failure_reason,omitempty" validate:"omitempty,oneof=provider_authentication provider_rate_limit provider_timeout provider_unavailable provider_response invalid_output validation_failed insufficient_data hub_unavailable internal_error unknown"` //nolint:lll Provider string `json:"provider,omitempty" validate:"omitempty,oneof=openai bedrock vertex unknown"` Model string `json:"model,omitempty" validate:"omitempty,no_null_bytes,max=255"` diff --git a/internal/observability/logging_test.go b/internal/observability/logging_test.go index 269edf14..6205881e 100644 --- a/internal/observability/logging_test.go +++ b/internal/observability/logging_test.go @@ -20,6 +20,8 @@ func TestNewLogHandlerJSONIncludesRequestCorrelation(t *testing.T) { logger.InfoContext(ctx, "taxonomy lifecycle", "event", "hub.taxonomy.run.started") + assert.Equal(t, 1, strings.Count(output.String(), `"request_id"`)) + var record map[string]any require.NoError(t, json.Unmarshal(output.Bytes(), &record)) assert.Equal(t, "request-1", record["request_id"]) diff --git a/internal/observability/names.go b/internal/observability/names.go index a1efea97..c4dadfa7 100644 --- a/internal/observability/names.go +++ b/internal/observability/names.go @@ -56,7 +56,7 @@ const ( MetricNameCacheMisses = "hub_cache_misses_total" MetricNameTaxonomyRunsStarted = "hub_taxonomy_runs_started_total" - MetricNameTaxonomyRunOutcomes = "hub_taxonomy_runs_total" + MetricNameTaxonomyRunOutcomes = "hub_taxonomy_outcomes_total" MetricNameTaxonomyRunDuration = "hub_taxonomy_run_duration_seconds" MetricNameTaxonomyDispatchError = "hub_taxonomy_dispatch_errors_total" MetricNameTaxonomyRunsReaped = "hub_taxonomy_runs_reaped_total" diff --git a/internal/service/taxonomy_service_test.go b/internal/service/taxonomy_service_test.go index 6c790a15..f97f9f48 100644 --- a/internal/service/taxonomy_service_test.go +++ b/internal/service/taxonomy_service_test.go @@ -337,7 +337,7 @@ func TestTaxonomyService_FailRunPersistsSafeDiagnosticsInMetrics(t *testing.T) { Error: "provider request failed", ErrorCode: models.TaxonomyRunFailureCodeServiceUnavailable, Diagnostics: &models.TaxonomyRunFailureDiagnostics{ - Phase: "tree", + Phase: "taxonomy_generation", FailureReason: "provider_timeout", Provider: "vertex", Model: "gemini-2.5-flash", diff --git a/tests/taxonomy_api_test.go b/tests/taxonomy_api_test.go index 2eaa4fc5..cdb8b8d8 100644 --- a/tests/taxonomy_api_test.go +++ b/tests/taxonomy_api_test.go @@ -838,7 +838,7 @@ func TestTaxonomyAPI_InternalServiceEndpoints(t *testing.T) { Error: "clustering did not converge", ErrorCode: models.TaxonomyRunFailureCodeGenerationFailed, Diagnostics: &models.TaxonomyRunFailureDiagnostics{ - Phase: "cluster", + Phase: "clustering", PhaseDurations: map[string]float64{ "input_fetch": 0.5, "input_validation": 0.1, "clustering": 2.5, "evidence_selection": 0.2, "cluster_labeling": 1.2, "taxonomy_generation": 3.4, @@ -858,7 +858,7 @@ func TestTaxonomyAPI_InternalServiceEndpoints(t *testing.T) { require.NotNil(t, run.Error) assert.Equal(t, "clustering did not converge", *run.Error) assert.JSONEq(t, - `{"failure_diagnostics":{"phase":"cluster","phase_durations_seconds":{`+ + `{"failure_diagnostics":{"phase":"clustering","phase_durations_seconds":{`+ `"input_fetch":0.5,"input_validation":0.1,"clustering":2.5,"evidence_selection":0.2,`+ `"cluster_labeling":1.2,"taxonomy_generation":3.4,"payload_validation":0.3,"persistence":0.7}}}`, string(run.Metrics),