From cd461b95f88d08bf6939cf1f631841240e13896d Mon Sep 17 00:00:00 2001 From: Mathias Date: Fri, 12 Jun 2026 08:49:51 +0200 Subject: [PATCH] feat(observability): instrument AI + HTTP paths, serve /metrics on a side port (ADR-030, #15) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire the metrics package into the live paths and serve it: - summarizer: per-endpoint latency by model/outcome(success|error|parse_error)/fallback + slog. - youtube.FetchTranscript: latency by outcome (captions|none|rate_limited) + slog. - chat: answer latency by model + slog. - llm usage hook → token counts (prompt|completion) per model, wired in buildSummarizer/buildChat. - oidc callback: login counter. - cmdServe: wrap Router in metrics.HTTPMiddleware (request count + latency by bounded route pattern) and serve /metrics on TAPIR_METRICS_ADDR (default :9090), a SEPARATE port — never on the public app mux. BDD: observability.feature scenarios un-pended + mapped. TDD: summarizer wiring tested black-box via the /metrics scrape; metrics-not-on-public-mux asserted. Co-Authored-By: Claude Opus 4.8 (1M context) --- cmd/tapir/main.go | 21 ++++++++++++- cmd/tapir/processor.go | 5 +-- docs/use-cases/observability.feature | 10 +----- internal/adapters/chat/chat.go | 8 +++++ internal/adapters/summarizer/summarizer.go | 11 ++++++- .../adapters/summarizer/summarizer_test.go | 23 ++++++++++++++ internal/adapters/youtube/captions.go | 31 +++++++++++++++++++ internal/config/config.go | 7 +++++ internal/web/handlers_test.go | 9 ++++++ internal/web/oidc/oidc.go | 2 ++ test/acceptance/scenario_coverage_test.go | 10 ++++++ 11 files changed, 124 insertions(+), 13 deletions(-) diff --git a/cmd/tapir/main.go b/cmd/tapir/main.go index 2ae7fd0..aad0746 100644 --- a/cmd/tapir/main.go +++ b/cmd/tapir/main.go @@ -28,6 +28,7 @@ import ( "gitea.d-ma.be/mathias/tapir/internal/adapters/youtube" "gitea.d-ma.be/mathias/tapir/internal/auth" "gitea.d-ma.be/mathias/tapir/internal/config" + "gitea.d-ma.be/mathias/tapir/internal/metrics" "gitea.d-ma.be/mathias/tapir/internal/runner" "gitea.d-ma.be/mathias/tapir/internal/web" "gitea.d-ma.be/mathias/tapir/internal/web/oidc" @@ -310,16 +311,34 @@ func cmdServe(ctx context.Context, log *slog.Logger) error { srv := &http.Server{ Addr: cfg.HTTPAddr, - Handler: app.Router(), + Handler: metrics.HTTPMiddleware(app.Router()), ReadHeaderTimeout: 10 * time.Second, } + // Prometheus /metrics on a SEPARATE port (ADR-030) — never on the public app + // mux, so a scrape is in-cluster only. Empty TAPIR_METRICS_ADDR disables it. + var metricsSrv *http.Server + if cfg.MetricsAddr != "" { + mmux := http.NewServeMux() + mmux.Handle("GET /metrics", metrics.Handler()) + metricsSrv = &http.Server{Addr: cfg.MetricsAddr, Handler: mmux, ReadHeaderTimeout: 10 * time.Second} + go func() { + log.Info("serving metrics", "addr", cfg.MetricsAddr) + if err := metricsSrv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + log.Error("metrics server", "err", err) + } + }() + } + // Graceful shutdown on signal: stop accepting, drain in-flight requests. go func() { <-ctx.Done() shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() _ = srv.Shutdown(shutdownCtx) + if metricsSrv != nil { + _ = metricsSrv.Shutdown(shutdownCtx) + } }() log.Info("serving web ui", "addr", cfg.HTTPAddr, "user", cfg.UserID) diff --git a/cmd/tapir/processor.go b/cmd/tapir/processor.go index bb15676..ccdb2e3 100644 --- a/cmd/tapir/processor.go +++ b/cmd/tapir/processor.go @@ -13,6 +13,7 @@ import ( "gitea.d-ma.be/mathias/tapir/internal/adapters/youtube" "gitea.d-ma.be/mathias/tapir/internal/config" "gitea.d-ma.be/mathias/tapir/internal/domain" + "gitea.d-ma.be/mathias/tapir/internal/metrics" "gitea.d-ma.be/mathias/tapir/internal/ports" "gitea.d-ma.be/mathias/tapir/internal/usecase" "gitea.d-ma.be/mathias/tapir/internal/web" @@ -69,7 +70,7 @@ func buildSummarizer(cfg config.Config) *summarizer.Summarizer { func summarizerEndpoint(cfg config.Config) func(model string) summarizer.Endpoint { return func(model string) summarizer.Endpoint { return summarizer.Endpoint{ - Client: llm.New(cfg.GatewayURL, cfg.GatewayKey, model, cfg.SummarizerTimeout, llm.WithMaxTokens(cfg.SummaryMaxTokens)), + Client: llm.New(cfg.GatewayURL, cfg.GatewayKey, model, cfg.SummarizerTimeout, llm.WithMaxTokens(cfg.SummaryMaxTokens), llm.WithUsageHook(metrics.RecordTokens)), Provider: providerOf(model), Model: model, } @@ -150,7 +151,7 @@ func buildChat(cfg config.Config) *chat.Service { return nil } newClient := func(model string) chat.Completer { - return llm.New(cfg.GatewayURL, cfg.GatewayKey, model, cfg.SummarizerTimeout, llm.WithMaxTokens(cfg.SummaryMaxTokens)) + return llm.New(cfg.GatewayURL, cfg.GatewayKey, model, cfg.SummarizerTimeout, llm.WithMaxTokens(cfg.SummaryMaxTokens), llm.WithUsageHook(metrics.RecordTokens)) } return chat.New(newClient, models, cfg.MaxTranscriptChars) } diff --git a/docs/use-cases/observability.feature b/docs/use-cases/observability.feature index 3a1a933..b4c6b0e 100644 --- a/docs/use-cases/observability.feature +++ b/docs/use-cases/observability.feature @@ -4,50 +4,42 @@ Feature: Observability — timing and metrics for performance and UX (ADR-030, # So that I can see latency, model behaviour, and usage — and feed the Stage-0 eval gate # AI metrics are the priority (ADR-030 R3). Each scenario maps to a Go test in - # scenarioCoverage once implemented; tagged @pending until the TDD step lands it. + # test/acceptance/scenario_coverage_test.go (the BDD name-coverage gate). - @pending # TestObserveSummarizeRecordsModelOutcomeFallback Scenario: Summarization latency is recorded per endpoint Given the summarizer runs a transcript through its endpoint chain When an endpoint returns a parseable summary Then the summarize latency is recorded with the model, outcome "success", and whether it was a fallback - @pending # TestObserveSummarizeRecordsFailureOutcomes Scenario: A failing summarizer endpoint records its failure outcome Given the summarizer runs a transcript through its endpoint chain When an endpoint errors or returns unparseable output Then the summarize latency is recorded with outcome "error" or "parse_error" before the chain advances - @pending # TestObserveCaptionFetchByOutcome Scenario: Caption fetch latency is recorded by outcome Given a caption fetch is attempted for a video When it resolves to captions, no captions, or a rate limit Then the caption-fetch latency is recorded labelled by that outcome - @pending # TestLLMUsageHookRecordsTokens Scenario: LLM token usage is recorded from the completion Given an LLM completion returns a usage block with prompt and completion tokens When the client finishes the call Then the prompt and completion tokens are recorded for that model - @pending # TestChatAnswerLatencyRecorded Scenario: Q&A answer latency is recorded Given a user asks a question about a video When the answer is produced from the stored transcript Then the chat answer latency is recorded for the answering model - @pending # TestHTTPMiddlewareRecordsByRoutePattern Scenario: HTTP requests are counted by route, method, and status Given the metrics HTTP middleware wraps the app When a request is served against a registered route Then it is counted and timed under the bounded route pattern, not the raw path - @pending # TestLoginCounted Scenario: A successful login is counted Given a user completes the OIDC callback and a session is established Then the login counter is incremented - @pending # TestMetricsNotOnPublicMux Scenario: The metrics endpoint is not on the public app port Given the service is running When the public app mux is inspected diff --git a/internal/adapters/chat/chat.go b/internal/adapters/chat/chat.go index f37d19f..3fabc10 100644 --- a/internal/adapters/chat/chat.go +++ b/internal/adapters/chat/chat.go @@ -13,8 +13,12 @@ package chat import ( "context" "fmt" + "log/slog" "strings" + "time" "unicode/utf8" + + "gitea.d-ma.be/mathias/tapir/internal/metrics" ) // Completer is the minimal LLM chat surface the Service needs. *llm.Client @@ -113,10 +117,14 @@ func (s *Service) Answer(ctx context.Context, req Request) (Reply, error) { system := buildSystem(transcript, truncated) user := buildUser(req.History, req.Question) + start := time.Now() out, err := s.newClient(model).Complete(ctx, system, user) if err != nil { return Reply{}, fmt.Errorf("chat: %s: %w", model, err) } + dur := time.Since(start) + metrics.ObserveChat(model, dur) + slog.Default().Info("chat answer", "model", model, "elapsed_ms", dur.Milliseconds()) answer := strings.TrimSpace(out) if answer == "" { return Reply{}, fmt.Errorf("chat: %s returned an empty answer", model) diff --git a/internal/adapters/summarizer/summarizer.go b/internal/adapters/summarizer/summarizer.go index 0312ee7..c18b4ce 100644 --- a/internal/adapters/summarizer/summarizer.go +++ b/internal/adapters/summarizer/summarizer.go @@ -13,11 +13,13 @@ import ( "encoding/json" "errors" "fmt" + "log/slog" "strings" "time" "unicode/utf8" "gitea.d-ma.be/mathias/tapir/internal/domain" + "gitea.d-ma.be/mathias/tapir/internal/metrics" ) // Completer is the minimal LLM chat surface the Summarizer needs. @@ -95,16 +97,23 @@ func (s *Summarizer) Summarize(ctx context.Context, v domain.Video, t domain.Tra var errs []error for i, ep := range s.endpoints { + fallback := i > 0 + start := time.Now() out, err := ep.Client.Complete(ctx, systemPrompt, user) + dur := time.Since(start) if err != nil { + metrics.ObserveSummarize(ep.Model, "error", fallback, dur) errs = append(errs, fmt.Errorf("%s/%s call: %w", ep.Provider, ep.Model, err)) continue } - sum, perr := s.build(v, ep, i > 0, out) + sum, perr := s.build(v, ep, fallback, out) if perr != nil { + metrics.ObserveSummarize(ep.Model, "parse_error", fallback, dur) errs = append(errs, fmt.Errorf("%s/%s output: %w", ep.Provider, ep.Model, perr)) continue } + metrics.ObserveSummarize(ep.Model, "success", fallback, dur) + slog.Default().Info("summarized", "model", ep.Model, "fallback", fallback, "elapsed_ms", dur.Milliseconds()) return sum, nil } return domain.Summary{}, fmt.Errorf("summarize: all %d endpoint(s) failed: %w", len(s.endpoints), errors.Join(errs...)) diff --git a/internal/adapters/summarizer/summarizer_test.go b/internal/adapters/summarizer/summarizer_test.go index e74dec5..c066ec8 100644 --- a/internal/adapters/summarizer/summarizer_test.go +++ b/internal/adapters/summarizer/summarizer_test.go @@ -7,13 +7,36 @@ package summarizer import ( "context" "errors" + "net/http" + "net/http/httptest" "strings" "testing" "gitea.d-ma.be/mathias/tapir/internal/domain" + "gitea.d-ma.be/mathias/tapir/internal/metrics" "gitea.d-ma.be/mathias/tapir/internal/ports" ) +// TestSummarizerRecordsMetric verifies the summarizer→metrics wiring (ADR-030) +// black-box: after a successful summarize, the public /metrics scrape shows a +// success observation for that endpoint's model. +func TestSummarizerRecordsMetric(t *testing.T) { + const model = "metrics-test-model" + s := New(Endpoint{Client: &fakeClient{reply: goodReply}, Provider: "local", Model: model}, nil) + if _, err := s.Summarize(context.Background(), testVideo(), testTranscript()); err != nil { + t.Fatalf("Summarize: %v", err) + } + + rec := httptest.NewRecorder() + metrics.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + body := rec.Body.String() + if !strings.Contains(body, `tapir_summarize_duration_seconds`) || + !strings.Contains(body, `model="`+model+`"`) || + !strings.Contains(body, `outcome="success"`) { + t.Errorf("metrics scrape missing summarize success for %s", model) + } +} + // compile-time check: Summarizer satisfies the port. var _ ports.Summarizer = (*Summarizer)(nil) diff --git a/internal/adapters/youtube/captions.go b/internal/adapters/youtube/captions.go index f0b0762..a441749 100644 --- a/internal/adapters/youtube/captions.go +++ b/internal/adapters/youtube/captions.go @@ -7,10 +7,13 @@ import ( "encoding/xml" "fmt" "io" + "log/slog" "net/http" "strings" + "time" "gitea.d-ma.be/mathias/tapir/internal/domain" + "gitea.d-ma.be/mathias/tapir/internal/metrics" ) // defaultPlayerBaseURL is the InnerTube / watch-page host. Overridable via @@ -43,7 +46,35 @@ const maxCaptionBytes = 16 << 20 // 16 MiB // fetch, or an unparseable body all yield SourceNone rather than an error. Only // genuine transport (network) faults return an error. Audio download and // speech-to-text remain absent (ADR-007). +// FetchTranscript times the caption fetch and records its latency by outcome +// (ADR-030) before returning. Transport errors are surfaced to the caller and not +// recorded as an outcome (logged upstream); the three resolved outcomes +// captions|none|rate_limited are the ones that consume the scarce fetch budget. func (a *Adapter) FetchTranscript(ctx context.Context, v domain.Video) (domain.Transcript, error) { + start := time.Now() + tr, err := a.fetchTranscript(ctx, v) + if err == nil { + dur := time.Since(start) + outcome := captionOutcome(tr.Source) + metrics.ObserveCaptionFetch(outcome, dur) + slog.Default().Info("caption fetch", "video", v.ProviderVideoID, "outcome", outcome, "elapsed_ms", dur.Milliseconds()) + } + return tr, err +} + +// captionOutcome maps a transcript source to the metric outcome label. +func captionOutcome(s domain.TranscriptSource) string { + switch s { + case domain.SourceCaptions: + return "captions" + case domain.SourceRateLimited: + return "rate_limited" + default: + return "none" + } +} + +func (a *Adapter) fetchTranscript(ctx context.Context, v domain.Video) (domain.Transcript, error) { client := a.plainClient() tracks, err := a.captionTracks(ctx, client, v.ProviderVideoID) diff --git a/internal/config/config.go b/internal/config/config.go index ed01146..01f01f5 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -143,6 +143,11 @@ type Config struct { // HTTPAddr is the listen address for `tapir serve` (the Stage-0 web UI). HTTPAddr string + // MetricsAddr is the listen address for the Prometheus /metrics endpoint + // (ADR-030). A SEPARATE port from HTTPAddr so /metrics is never exposed on the + // public app — only scraped in-cluster. Empty disables the metrics server. + MetricsAddr string + // PublicURL is the externally-reachable base URL of the deployed service, // e.g. "https://tapir.d-ma.be". Used to build absolute links handed to humans // (the `tapir invite` URL). No trailing slash is assumed — callers trim it. @@ -178,6 +183,7 @@ const ( defaultYTConnectRedirectURL = "https://tapir.d-ma.be/oauth/youtube/callback" defaultOAuthRedirectAddr = "localhost:8080" defaultHTTPAddr = ":8080" + defaultMetricsAddr = ":9090" defaultFetchBackoff = time.Hour defaultFetchRate = 2 * time.Second defaultPublicURL = "https://tapir.d-ma.be" @@ -209,6 +215,7 @@ func Load() (Config, error) { SecretsFile: envOr("TAPIR_SECRETS_FILE", defaultSecretsFile()), OAuthRedirectAddr: envOr("TAPIR_OAUTH_REDIRECT_ADDR", defaultOAuthRedirectAddr), HTTPAddr: envOr("TAPIR_HTTP_ADDR", defaultHTTPAddr), + MetricsAddr: lookupOr("TAPIR_METRICS_ADDR", defaultMetricsAddr), PublicURL: envOr("TAPIR_PUBLIC_URL", defaultPublicURL), OIDCIssuer: os.Getenv("TAPIR_OIDC_ISSUER"), DexClientID: os.Getenv("TAPIR_DEX_CLIENT_ID"), diff --git a/internal/web/handlers_test.go b/internal/web/handlers_test.go index d36c183..608c79f 100644 --- a/internal/web/handlers_test.go +++ b/internal/web/handlers_test.go @@ -524,3 +524,12 @@ func TestListAutoModeBannerCopy(t *testing.T) { require.Contains(t, html, "land gradually") require.NotContains(t, html, "are not summarized automatically") } + +// TestMetricsNotOnPublicMux: the public app router exposes no /metrics route — +// Prometheus is served on the dedicated metrics port only (ADR-030, security R6). +func TestMetricsNotOnPublicMux(t *testing.T) { + app := newApp(t) + resetDB(t, rawPool(t)) + rec := do(t, app, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + require.Equal(t, http.StatusNotFound, rec.Code, "/metrics must not be on the public mux") +} diff --git a/internal/web/oidc/oidc.go b/internal/web/oidc/oidc.go index 64d20a4..cb8c91d 100644 --- a/internal/web/oidc/oidc.go +++ b/internal/web/oidc/oidc.go @@ -27,6 +27,7 @@ import ( "github.com/coreos/go-oidc/v3/oidc" "golang.org/x/oauth2" + "gitea.d-ma.be/mathias/tapir/internal/metrics" "gitea.d-ma.be/mathias/tapir/internal/web" ) @@ -253,6 +254,7 @@ func (d *DexAuth) handleCallback(w http.ResponseWriter, r *http.Request) { user := web.User{Subject: idToken.Subject, Email: claims.Email} d.setSessionCookie(w, d.encodeSession(user, d.now().Add(d.sessionTTL))) + metrics.IncLogin() http.Redirect(w, r, "/", http.StatusFound) } diff --git a/test/acceptance/scenario_coverage_test.go b/test/acceptance/scenario_coverage_test.go index 502b945..9ced566 100644 --- a/test/acceptance/scenario_coverage_test.go +++ b/test/acceptance/scenario_coverage_test.go @@ -25,6 +25,16 @@ import ( // fails if a scenario is unmapped, a mapped test is missing, or an entry no // longer matches a real non-pending scenario. var scenarioCoverage = map[string]string{ + // observability.feature (ADR-030, #15) + "Summarization latency is recorded per endpoint": "TestSummarizerRecordsMetric", + "A failing summarizer endpoint records its failure outcome": "TestObserveSummarizeRecordsFailureOutcomes", + "Caption fetch latency is recorded by outcome": "TestObserveCaptionFetchByOutcome", + "LLM token usage is recorded from the completion": "TestClient_UsageHookRecordsTokens", + "Q&A answer latency is recorded": "TestChatAnswerLatencyRecorded", + "HTTP requests are counted by route, method, and status": "TestHTTPMiddlewareRecordsByRoutePattern", + "A successful login is counted": "TestLoginCounted", + "The metrics endpoint is not on the public app port": "TestMetricsNotOnPublicMux", + // ai_routing.feature "Local AI produces the summary": "TestSummarize_LocalSucceeds", "Local AI fails and the user has a BYO provider configured": "TestSummarize_FallsBackToBYO",