diff --git a/apiproxy/proxy.go b/apiproxy/proxy.go index 63c2561..24ac179 100644 --- a/apiproxy/proxy.go +++ b/apiproxy/proxy.go @@ -39,7 +39,7 @@ var ( } ) -func Init(mux *http.ServeMux, db *db.Database) { +func Init(mux *http.ServeMux, db *db.Database) *requestHealthRecorder { // Setup Azure Vars and Connection String azconf := &AzureConfig{ DeploymentName: os.Getenv("DEPLOYMENT_NAME"), @@ -60,8 +60,9 @@ func Init(mux *http.ServeMux, db *db.Database) { rc: rc, rerankClient: &http.Client{Timeout: azconf.RerankTimeout}, } + h.health = newRequestHealthRecorder(db) mux.Handle("/api/", h) - + return h.health } type baseHandle struct { @@ -71,6 +72,7 @@ type baseHandle struct { rerankClient *http.Client clientTokenValidator func(http.ResponseWriter, *http.Request) bool rerankEndpointOverride string + health *requestHealthRecorder } type ProxyDB interface { @@ -103,11 +105,13 @@ func (h *baseHandle) ServeHTTP(w http.ResponseWriter, r *http.Request) { func (h *baseHandle) HandleAzure(w http.ResponseWriter, r *http.Request, backend string) { azureToken := h.ValidateToken(w, r) if azureToken == "" { - http.Error(w, "Error Processing Request", http.StatusUnauthorized) + // ValidateToken writes the authentication error response. return } r.Header.Set("Content-Type", "application/json") r.Header.Set("Api-Key", azureToken) + originalEndpoint := r.URL.Path + requestID := r.Header.Get("X-Request-ID") remoteUrl := h.SetAzureUrl(r) if remoteUrl == nil { @@ -116,6 +120,9 @@ func (h *baseHandle) HandleAzure(w http.ResponseWriter, r *http.Request, backend } proxy := httputil.NewSingleHostReverseProxy(remoteUrl) + if h.health != nil { + proxy.Transport = &requestHealthTransport{base: http.DefaultTransport, recorder: h.health} + } r.Host = remoteUrl.Host // Remove proxy/ingress headers that should not be forwarded to Azure. // These headers are added by our ingress and clients and may be rejected @@ -145,6 +152,13 @@ func (h *baseHandle) HandleAzure(w http.ResponseWriter, r *http.Request, backend // `/openai` base will produce `/openai/v1/responses` as desired. r.URL.Path = strings.TrimPrefix(r.URL.Path, "/api") ensureStreamUsageForChatCompletions(r) + model, streaming := captureRequestHealthRequest(r) + r = withRequestHealthMetadata(r, requestHealthMetadata{ + Endpoint: originalEndpoint, + Model: model, + RequestID: requestID, + Streaming: streaming, + }) // Before proxying, log the intended complete URL. actualURL := *remoteUrl // Make a copy of the URL struct diff --git a/apiproxy/request_health.go b/apiproxy/request_health.go new file mode 100644 index 0000000..7f42f98 --- /dev/null +++ b/apiproxy/request_health.go @@ -0,0 +1,439 @@ +package apiproxy + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "net" + "net/http" + "strings" + "sync" + "sync/atomic" + "time" + + db "openai-api-proxy/db" +) + +const ( + requestHealthQueueSize = 1024 + requestHealthWorkers = 2 + requestHealthRetries = 5 + requestHealthRetryWait = 250 * time.Millisecond + requestHealthWriteTimeout = 5 * time.Second + requestHealthShutdownTimeout = 30 * time.Second +) + +type requestHealthStore interface { + WriteRequestHealthAttemptContext(context.Context, *db.RequestHealthAttempt) error + RequestHealthRetention() (time.Duration, error) + DeleteExpiredRequestHealthAttempts(time.Time) (int64, error) +} + +type requestHealthRecorder struct { + store requestHealthStore + queue chan *db.RequestHealthAttempt + stop chan struct{} + wg sync.WaitGroup + mu sync.Mutex + open bool + once sync.Once + closed chan struct{} + workerCtx context.Context + workerCancel context.CancelFunc + dropped atomic.Uint64 +} + +func newRequestHealthRecorder(store requestHealthStore) *requestHealthRecorder { + if store == nil { + return nil + } + r := &requestHealthRecorder{ + store: store, + queue: make(chan *db.RequestHealthAttempt, requestHealthQueueSize), + stop: make(chan struct{}), + closed: make(chan struct{}), + open: true, + } + r.workerCtx, r.workerCancel = context.WithCancel(context.Background()) + for i := 0; i < requestHealthWorkers; i++ { + r.wg.Add(1) + go r.runWorker() + } + r.wg.Add(1) + go r.runCleanup() + return r +} + +func (r *requestHealthRecorder) record(attempt *db.RequestHealthAttempt) { + if r == nil || attempt == nil { + return + } + r.mu.Lock() + defer r.mu.Unlock() + if !r.open { + return + } + select { + case r.queue <- attempt: + default: + r.logDrop("queue is full") + } +} + +func (r *requestHealthRecorder) logDrop(reason string) { + total := r.dropped.Add(1) + log.Printf("request health telemetry dropped: %s (total=%d)", reason, total) +} + +func (r *requestHealthRecorder) runWorker() { + defer r.wg.Done() + for { + select { + case attempt, ok := <-r.queue: + if !ok { + return + } + r.writeAttempt(attempt) + case <-r.workerCtx.Done(): + r.discardQueued() + return + } + } +} + +func (r *requestHealthRecorder) writeAttempt(attempt *db.RequestHealthAttempt) { + var err error + for retry := 0; retry <= requestHealthRetries; retry++ { + if r.workerCtx.Err() != nil { + r.logDrop("shutdown canceled queued write") + return + } + writeCtx, cancel := context.WithTimeout(r.workerCtx, requestHealthWriteTimeout) + err = r.store.WriteRequestHealthAttemptContext(writeCtx, attempt) + cancel() + if err == nil { + break + } + if retry < requestHealthRetries { + wait := time.NewTimer(requestHealthRetryWait << retry) + select { + case <-wait.C: + case <-r.workerCtx.Done(): + if !wait.Stop() { + <-wait.C + } + r.logDrop("shutdown canceled retry") + return + } + } + } + if err != nil { + r.logDrop(fmt.Sprintf("after retries: %v", err)) + } +} + +func (r *requestHealthRecorder) runCleanup() { + defer r.wg.Done() + cleanup := func() { + retention, err := r.store.RequestHealthRetention() + if err != nil { + log.Printf("request health retention lookup failed: %v", err) + return + } + if _, err := r.store.DeleteExpiredRequestHealthAttempts(time.Now().Add(-retention)); err != nil { + log.Printf("request health retention cleanup failed: %v", err) + } + } + cleanup() + ticker := time.NewTicker(time.Hour) + defer ticker.Stop() + for { + select { + case <-ticker.C: + cleanup() + case <-r.stop: + return + } + } +} + +func (r *requestHealthRecorder) Close() { + ctx, cancel := context.WithTimeout(context.Background(), requestHealthShutdownTimeout) + defer cancel() + r.CloseContext(ctx) +} + +func (r *requestHealthRecorder) CloseContext(ctx context.Context) { + if r == nil { + return + } + if ctx == nil { + ctx = context.Background() + } + r.once.Do(func() { + r.mu.Lock() + r.open = false + close(r.queue) + close(r.stop) + r.mu.Unlock() + go func() { + select { + case <-ctx.Done(): + r.workerCancel() + case <-r.closed: + } + }() + go func() { + r.wg.Wait() + close(r.closed) + }() + }) + select { + case <-r.closed: + log.Printf("request health telemetry shutdown: dropped=%d", r.dropped.Load()) + case <-ctx.Done(): + r.workerCancel() + r.discardQueued() + log.Printf("request health telemetry shutdown timed out: dropped=%d", r.dropped.Load()) + } +} + +func (r *requestHealthRecorder) discardQueued() { + discarded := 0 + for { + select { + case _, ok := <-r.queue: + if !ok { + if discarded > 0 { + r.dropped.Add(uint64(discarded)) + log.Printf("request health telemetry discarded queued records: count=%d", discarded) + } + return + } + discarded++ + default: + if discarded > 0 { + r.dropped.Add(uint64(discarded)) + log.Printf("request health telemetry discarded queued records: count=%d", discarded) + } + return + } + } +} + +func (r *requestHealthRecorder) close() { + r.Close() +} + +type requestHealthTransport struct { + base http.RoundTripper + recorder *requestHealthRecorder +} + +type requestHealthMetadataKey struct{} + +type requestHealthMetadata struct { + Endpoint string + Model string + RequestID string + Streaming bool +} + +func withRequestHealthMetadata(req *http.Request, metadata requestHealthMetadata) *http.Request { + return req.WithContext(context.WithValue(req.Context(), requestHealthMetadataKey{}, metadata)) +} + +func requestHealthMetadataFromRequest(req *http.Request) requestHealthMetadata { + if req == nil { + return requestHealthMetadata{} + } + metadata, _ := req.Context().Value(requestHealthMetadataKey{}).(requestHealthMetadata) + return metadata +} + +func (t *requestHealthTransport) RoundTrip(req *http.Request) (*http.Response, error) { + started := time.Now() + metadata := requestHealthMetadataFromRequest(req) + endpoint := metadata.Endpoint + if endpoint == "" { + endpoint = req.URL.Path + } + requestID := metadata.RequestID + if requestID == "" { + requestID = req.Header.Get("X-Request-ID") + } + attempt := &db.RequestHealthAttempt{ + AttemptID: newRequestHealthID(), + RequestID: requestID, + Endpoint: endpoint, + Upstream: req.URL.Host, + Model: metadata.Model, + Method: req.Method, + StartedAt: started, + Streaming: metadata.Streaming || requestAcceptsEventStream(req), + } + base := t.base + if base == nil { + base = http.DefaultTransport + } + response, err := base.RoundTrip(req) + if err != nil { + attempt.Duration = time.Since(started) + attempt.CompletedAt = time.Now() + attempt.Outcome = requestHealthErrorOutcome(err, req.Context()) + attempt.ClientCancelled = attempt.Outcome == "caller_canceled" + t.recorder.record(attempt) + return nil, err + } + attempt.StatusCode = response.StatusCode + attempt.Streaming = attempt.Streaming || strings.Contains(strings.ToLower(response.Header.Get("Content-Type")), "text/event") + response.Body = &requestHealthBody{ + ReadCloser: response.Body, + attempt: attempt, + recorder: t.recorder, + started: started, + ctx: req.Context(), + } + return response, nil +} + +const requestHealthModelCaptureLimit = 1024 * 1024 + +func captureRequestHealthModel(req *http.Request) string { + model, _ := captureRequestHealthRequest(req) + return model +} + +func captureRequestHealthRequest(req *http.Request) (string, bool) { + if req == nil || req.Body == nil || (req.ContentLength >= 0 && req.ContentLength > requestHealthModelCaptureLimit) { + return "", false + } + body, err := io.ReadAll(io.LimitReader(req.Body, requestHealthModelCaptureLimit+1)) + if err != nil || int64(len(body)) > requestHealthModelCaptureLimit { + req.Body = io.NopCloser(io.MultiReader(bytes.NewReader(body), req.Body)) + return "", false + } + req.Body = io.NopCloser(bytes.NewReader(body)) + var payload struct { + Model string `json:"model"` + Stream *bool `json:"stream"` + } + if err := json.Unmarshal(body, &payload); err != nil { + return "", false + } + return strings.TrimSpace(payload.Model), payload.Stream != nil && *payload.Stream +} + +type requestHealthBody struct { + io.ReadCloser + attempt *db.RequestHealthAttempt + recorder *requestHealthRecorder + started time.Time + ctx context.Context + mu sync.Mutex + finalized bool + firstByte time.Time + readErr error +} + +func (b *requestHealthBody) Read(p []byte) (int, error) { + n, err := b.ReadCloser.Read(p) + b.mu.Lock() + if !b.finalized { + if n > 0 && b.firstByte.IsZero() { + b.firstByte = time.Now() + d := b.firstByte.Sub(b.started) + b.attempt.FirstByteDuration = &d + } + if err != nil && err != io.EOF { + b.readErr = err + } + } + b.mu.Unlock() + return n, err +} + +func (b *requestHealthBody) Close() error { + err := b.ReadCloser.Close() + b.mu.Lock() + if b.finalized { + b.mu.Unlock() + return err + } + b.finalized = true + b.attempt.Duration = time.Since(b.started) + b.attempt.CompletedAt = time.Now() + if b.ctx.Err() != nil { + b.attempt.ClientCancelled = true + if b.attempt.StatusCode > 0 { + b.attempt.Outcome = requestHealthStatusOutcome(b.attempt.StatusCode) + } else { + b.attempt.Outcome = "caller_canceled" + } + } else if b.readErr != nil { + b.attempt.Outcome = requestHealthErrorOutcome(b.readErr, b.ctx) + } else if err != nil { + b.attempt.Outcome = requestHealthErrorOutcome(err, b.ctx) + } else { + b.attempt.Outcome = requestHealthStatusOutcome(b.attempt.StatusCode) + } + attempt := b.attempt + b.mu.Unlock() + // Finalization precedes queueing, so a concurrent Read cannot mutate the + // attempt after the worker can observe it. + b.recorder.record(attempt) + return err +} + +func requestHealthStatusOutcome(status int) string { + switch { + case status >= http.StatusOK && status < http.StatusMultipleChoices: + return "success_2xx" + case status >= http.StatusMultipleChoices && status < http.StatusBadRequest: + return "redirect_3xx" + case status >= http.StatusBadRequest && status < http.StatusInternalServerError: + return "client_error_4xx" + case status >= http.StatusInternalServerError && status <= 599: + return "server_error_5xx" + default: + return "transport_error" + } +} + +func requestHealthErrorOutcome(err error, ctx context.Context) string { + if ctx != nil && ctx.Err() == context.Canceled { + return "caller_canceled" + } + if errors.Is(err, context.DeadlineExceeded) { + return "timeout" + } + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + return "timeout" + } + if errors.Is(err, context.Canceled) { + return "upstream_canceled" + } + return "transport_error" +} + +func requestAcceptsEventStream(req *http.Request) bool { + return req != nil && strings.Contains(strings.ToLower(req.Header.Get("Accept")), "text/event") +} + +func newRequestHealthID() string { + var raw [16]byte + if _, err := rand.Read(raw[:]); err != nil { + return fmt.Sprintf("attempt-%d", time.Now().UnixNano()) + } + raw[6] = (raw[6] & 0x0f) | 0x40 + raw[8] = (raw[8] & 0x3f) | 0x80 + encoded := hex.EncodeToString(raw[:]) + return encoded[:8] + "-" + encoded[8:12] + "-" + encoded[12:16] + "-" + encoded[16:20] + "-" + encoded[20:] +} diff --git a/apiproxy/request_health_test.go b/apiproxy/request_health_test.go new file mode 100644 index 0000000..fc08b4d --- /dev/null +++ b/apiproxy/request_health_test.go @@ -0,0 +1,289 @@ +package apiproxy + +import ( + "context" + "errors" + "io" + "net/http" + "strings" + "sync" + "testing" + "time" + + db "openai-api-proxy/db" +) + +type healthStoreTest struct { + mu sync.Mutex + attempts []*db.RequestHealthAttempt + blockWrites bool + writeStarted chan struct{} + writeStartedOnce sync.Once +} + +func (s *healthStoreTest) RequestHealthRetention() (time.Duration, error) { + return 30 * 24 * time.Hour, nil +} + +func (s *healthStoreTest) DeleteExpiredRequestHealthAttempts(time.Time) (int64, error) { + return 0, nil +} + +func (s *healthStoreTest) WriteRequestHealthAttempt(a *db.RequestHealthAttempt) error { + s.mu.Lock() + defer s.mu.Unlock() + s.attempts = append(s.attempts, a) + return nil +} + +func (s *healthStoreTest) WriteRequestHealthAttemptContext(ctx context.Context, a *db.RequestHealthAttempt) error { + if s.blockWrites { + s.writeStartedOnce.Do(func() { close(s.writeStarted) }) + <-ctx.Done() + return ctx.Err() + } + return s.WriteRequestHealthAttempt(a) +} + +func waitForHealthAttempt(t *testing.T, store *healthStoreTest) *db.RequestHealthAttempt { + t.Helper() + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + store.mu.Lock() + if len(store.attempts) > 0 { + attempt := store.attempts[0] + store.mu.Unlock() + return attempt + } + store.mu.Unlock() + time.Sleep(time.Millisecond) + } + t.Fatal("expected one attempt") + return nil +} + +func newTestRecorder(store *healthStoreTest) *requestHealthRecorder { + return newRequestHealthRecorder(store) +} + +func TestRequestHealthBodyRecordsStatusAndFirstByte(t *testing.T) { + store := &healthStoreTest{} + recorder := newTestRecorder(store) + defer recorder.Close() + started := time.Now().Add(-time.Millisecond) + attempt := &db.RequestHealthAttempt{AttemptID: "attempt-1", StatusCode: http.StatusOK, StartedAt: started} + body := &requestHealthBody{ + ReadCloser: io.NopCloser(strings.NewReader("ok")), + attempt: attempt, + recorder: recorder, + started: started, + ctx: context.Background(), + } + if _, err := io.ReadAll(body); err != nil { + t.Fatal(err) + } + if err := body.Close(); err != nil { + t.Fatal(err) + } + recorded := waitForHealthAttempt(t, store) + if recorded.Outcome != "success_2xx" || recorded.FirstByteDuration == nil || recorded.CompletedAt.IsZero() { + t.Fatalf("unexpected attempt: %+v", recorded) + } +} + +func TestRequestHealthStatusOutcomes(t *testing.T) { + tests := []struct { + status int + want string + }{ + {http.StatusOK, "success_2xx"}, + {http.StatusMultipleChoices, "redirect_3xx"}, + {http.StatusBadRequest, "client_error_4xx"}, + {http.StatusInternalServerError, "server_error_5xx"}, + } + for _, test := range tests { + if got := requestHealthStatusOutcome(test.status); got != test.want { + t.Errorf("status %d: got %q, want %q", test.status, got, test.want) + } + } +} + +func TestRequestHealthErrorOutcomes(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if got := requestHealthErrorOutcome(context.Canceled, ctx); got != "caller_canceled" { + t.Fatalf("caller cancellation classified as %q", got) + } + if got := requestHealthErrorOutcome(context.DeadlineExceeded, context.Background()); got != "timeout" { + t.Fatalf("timeout classified as %q", got) + } + if got := requestHealthErrorOutcome(errors.New("connection reset"), context.Background()); got != "transport_error" { + t.Fatalf("transport error classified as %q", got) + } +} + +func TestRequestHealthBodyRecordsCallerCancellation(t *testing.T) { + store := &healthStoreTest{} + recorder := newTestRecorder(store) + defer recorder.Close() + ctx, cancel := context.WithCancel(context.Background()) + attempt := &db.RequestHealthAttempt{AttemptID: "attempt-2", StatusCode: http.StatusOK} + body := &requestHealthBody{ + ReadCloser: io.NopCloser(strings.NewReader("ok")), + attempt: attempt, + recorder: recorder, + started: time.Now(), + ctx: ctx, + } + cancel() + _, _ = body.Read(make([]byte, 2)) + _ = body.Close() + recorded := waitForHealthAttempt(t, store) + if recorded.Outcome != "success_2xx" || !recorded.ClientCancelled { + t.Fatalf("unexpected cancellation record: %+v", recorded) + } +} + +func TestRequestHealthRecorderCloseDrainsAndIsIdempotent(t *testing.T) { + store := &healthStoreTest{} + recorder := newTestRecorder(store) + for i := 0; i < 10; i++ { + recorder.record(&db.RequestHealthAttempt{AttemptID: string(rune('a' + i))}) + } + + recorder.Close() + recorder.Close() + + store.mu.Lock() + defer store.mu.Unlock() + if len(store.attempts) != 10 { + t.Fatalf("got %d persisted attempts, want 10", len(store.attempts)) + } +} + +func TestRequestHealthRecorderConcurrentRecordAndClose(t *testing.T) { + store := &healthStoreTest{} + recorder := newTestRecorder(store) + var records sync.WaitGroup + for i := 0; i < 100; i++ { + records.Add(1) + go func(i int) { + defer records.Done() + recorder.record(&db.RequestHealthAttempt{AttemptID: string(rune(i))}) + }(i) + } + recorder.Close() + records.Wait() + recorder.Close() +} + +func TestRequestHealthRecorderCloseContextIsBounded(t *testing.T) { + store := &healthStoreTest{blockWrites: true, writeStarted: make(chan struct{})} + recorder := newTestRecorder(store) + recorder.record(&db.RequestHealthAttempt{AttemptID: "blocked"}) + select { + case <-store.writeStarted: + case <-time.After(time.Second): + t.Fatal("worker did not start the blocked write") + } + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + started := time.Now() + recorder.CloseContext(ctx) + if elapsed := time.Since(started); elapsed > 500*time.Millisecond { + t.Fatalf("bounded close took %s", elapsed) + } + recorder.Close() +} + +type closeUnblocksReader struct { + readStarted chan struct{} + closed chan struct{} + closeOnce sync.Once +} + +func (r *closeUnblocksReader) Read([]byte) (int, error) { + close(r.readStarted) + <-r.closed + return 1, nil +} + +func (r *closeUnblocksReader) Close() error { + r.closeOnce.Do(func() { close(r.closed) }) + return nil +} + +func TestRequestHealthBodyConcurrentReadAndCloseDoesNotMutateAfterFinalize(t *testing.T) { + store := &healthStoreTest{} + recorder := newTestRecorder(store) + defer recorder.Close() + reader := &closeUnblocksReader{readStarted: make(chan struct{}), closed: make(chan struct{})} + attempt := &db.RequestHealthAttempt{AttemptID: "concurrent", StatusCode: http.StatusOK} + body := &requestHealthBody{ + ReadCloser: reader, + attempt: attempt, + recorder: recorder, + started: time.Now(), + ctx: context.Background(), + } + readDone := make(chan struct{}) + go func() { + _, _ = body.Read(make([]byte, 1)) + close(readDone) + }() + <-reader.readStarted + if err := body.Close(); err != nil { + t.Fatal(err) + } + select { + case <-readDone: + case <-time.After(time.Second): + t.Fatal("concurrent read did not finish") + } + recorded := waitForHealthAttempt(t, store) + if recorded.FirstByteDuration != nil || recorded.Outcome != "success_2xx" { + t.Fatalf("read mutated finalized attempt: %+v", recorded) + } +} + +func TestCaptureRequestHealthModelPreservesOversizedBody(t *testing.T) { + payload := strings.Repeat("x", requestHealthModelCaptureLimit+1) + req, err := http.NewRequest(http.MethodPost, "https://example.test", strings.NewReader(payload)) + if err != nil { + t.Fatal(err) + } + + if model := captureRequestHealthModel(req); model != "" { + t.Fatalf("expected no model for oversized body, got %q", model) + } + restored, err := io.ReadAll(req.Body) + if err != nil { + t.Fatal(err) + } + if string(restored) != payload { + t.Fatalf("request body was not preserved") + } +} + +func TestCaptureRequestHealthRequestUsesStreamField(t *testing.T) { + for _, test := range []struct { + name string + body string + streaming bool + }{ + {name: "non-streaming", body: `{"model":"gpt-5-mini","stream":false}`, streaming: false}, + {name: "streaming", body: `{"model":"gpt-5-mini","stream":true}`, streaming: true}, + } { + t.Run(test.name, func(t *testing.T) { + req, err := http.NewRequest(http.MethodPost, "https://example.test", strings.NewReader(test.body)) + if err != nil { + t.Fatal(err) + } + model, streaming := captureRequestHealthRequest(req) + if model != "gpt-5-mini" || streaming != test.streaming { + t.Fatalf("got model=%q streaming=%t", model, streaming) + } + }) + } +} diff --git a/apiproxy/rerank.go b/apiproxy/rerank.go index 07d514b..e38902c 100644 --- a/apiproxy/rerank.go +++ b/apiproxy/rerank.go @@ -191,11 +191,20 @@ func (h *baseHandle) HandleRerank(w http.ResponseWriter, r *http.Request) { upstreamRequest.Header.Set("Content-Type", "application/json") upstreamRequest.Header.Set("Authorization", "Bearer "+h.az.ApiKey) upstreamRequest.Header.Set("X-Request-ID", requestID) + upstreamRequest = withRequestHealthMetadata(upstreamRequest, requestHealthMetadata{ + Endpoint: r.URL.Path, + Model: request.Model, + }) client := h.rerankClient if client == nil { client = http.DefaultClient } + if h.health != nil { + clientCopy := *client + clientCopy.Transport = &requestHealthTransport{base: client.Transport, recorder: h.health} + client = &clientCopy + } response, err := client.Do(upstreamRequest) if err != nil { writeRerankError(w, http.StatusBadGateway, "rerank upstream request failed", "upstream_error") diff --git a/app/RELEASE_NOTES.md b/app/RELEASE_NOTES.md index 32bc599..e23e927 100644 --- a/app/RELEASE_NOTES.md +++ b/app/RELEASE_NOTES.md @@ -1,6 +1,7 @@ # Release Notes ## Unreleased +- Added an authenticated service-health dashboard with local range controls and admin-configurable retention with immediate purge. - Added lazy, exact hourly request-statistics caching for faster administrator reports, with pricing invalidation and manual UTC-range rebuilds. - Added separate chat-token and rerank-search reporting and dashboard charts. - Added request-type-aware rerank pricing management and filters. diff --git a/app/src/app/admin/service-health/page.tsx b/app/src/app/admin/service-health/page.tsx new file mode 100644 index 0000000..2782326 --- /dev/null +++ b/app/src/app/admin/service-health/page.tsx @@ -0,0 +1,11 @@ +import { redirect } from "next/navigation"; + +import { ServiceHealthSettings } from "~/components/service-health-settings"; +import { auth } from "~/server/auth"; +import { api } from "~/trpc/server"; + +export default async function ServiceHealthSettingsPage() { + if (!(await auth())?.user?.isAdmin) redirect("/"); + const settings = await api.serviceHealth.getSettings(); + return ; +} diff --git a/app/src/app/service-health/page.tsx b/app/src/app/service-health/page.tsx new file mode 100644 index 0000000..3eb0091 --- /dev/null +++ b/app/src/app/service-health/page.tsx @@ -0,0 +1,33 @@ +import { redirect } from "next/navigation"; + +import { ServiceHealthDashboard } from "~/components/service-health-dashboard"; +import { auth } from "~/server/auth"; +import { api } from "~/trpc/server"; + +function defaultRange() { + const end = new Date(); + const start = new Date(end.getTime() - 24 * 60 * 60 * 1000); + return { start: start.toISOString(), end: end.toISOString() }; +} + +function validDate(value: string | undefined) { + if (!value) return null; + const date = new Date(value); + return Number.isFinite(date.getTime()) ? date.toISOString() : null; +} + +export default async function ServiceHealthPage({ + searchParams, +}: { + searchParams?: Promise<{ start?: string; end?: string }>; +}) { + if (!(await auth())?.user) redirect("/"); + const params = await searchParams; + const defaults = defaultRange(); + const start = validDate(params?.start) ?? defaults.start; + const end = validDate(params?.end) ?? defaults.end; + const dashboard = await api.serviceHealth.getDashboard({ start, end }); + return ( + + ); +} diff --git a/app/src/components/service-health-dashboard.tsx b/app/src/components/service-health-dashboard.tsx new file mode 100644 index 0000000..c72b75c --- /dev/null +++ b/app/src/components/service-health-dashboard.tsx @@ -0,0 +1,609 @@ +"use client"; + +import { + Activity, + AlertTriangle, + CheckCircle2, + Clock3, + XCircle, +} from "lucide-react"; +import { useRouter } from "next/navigation"; +import { useEffect, useState } from "react"; +import { + Bar, + BarChart, + CartesianGrid, + Line, + LineChart, + XAxis, + YAxis, +} from "recharts"; + +import { Button } from "~/components/ui/button"; +import { Card, CardContent, CardHeader, CardTitle } from "~/components/ui/card"; +import { + ChartContainer, + ChartTooltip, + ChartTooltipContent, +} from "~/components/ui/chart"; +import { Input } from "~/components/ui/input"; + +type Latency = { + averageMs: number | null; + p50Ms: number | null; + p95Ms: number | null; + maximumMs: number | null; +}; + +type Dashboard = { + range: { start: string; end: string; bucketSeconds: number }; + totals: { + total: number; + successful: number; + redirects: number; + clientErrors: number; + serverErrors: number; + timeouts: number; + upstreamCancellations: number; + transportErrors: number; + callerCancellations: number; + unsuccessful: number; + failureRate: number; + }; + latency: { duration: Latency; firstByte: Latency }; + statusCodes: { + statusCode: number | null; + outcome: string; + count: number; + }[]; + providers: { + provider: string; + count: number; + unsuccessful: number; + avgDurationMs: number; + }[]; + endpoints: { endpoint: string; count: number; unsuccessful: number }[]; + models: { model: string; count: number; unsuccessful: number }[]; + timeline: { + bucket: string; + total: number; + successful: number; + unsuccessful: number; + callerCancellations: number; + avgDurationMs: number | null; + p95DurationMs: number | null; + }[]; +}; + +function localInputValue(iso: string, timeZone: string) { + const date = new Date(iso); + const parts = new Intl.DateTimeFormat("en-US", { + timeZone, + year: "numeric", + month: "2-digit", + day: "2-digit", + hour: "2-digit", + minute: "2-digit", + hourCycle: "h23", + }).formatToParts(date); + const value = (type: Intl.DateTimeFormatPartTypes) => + parts.find((part) => part.type === type)?.value ?? "00"; + return `${value("year")}-${value("month")}-${value("day")}T${value("hour")}:${value("minute")}`; +} + +function toISO(value: string) { + const date = new Date(value); + return Number.isFinite(date.getTime()) ? date.toISOString() : null; +} + +function formatBucket(seconds: number) { + if (seconds < 3600) return `${Math.round(seconds / 60)} min`; + if (seconds < 86400) return `${Math.round(seconds / 3600)} Std.`; + return `${Math.round(seconds / 3600)} Std.`; +} + +function formatLatency(value: number | null, number: Intl.NumberFormat) { + return value === null ? "No data" : `${number.format(Math.round(value))} ms`; +} + +export function ServiceHealthDashboard({ + dashboard, + start, + end, +}: { + dashboard: Dashboard; + start: string; + end: string; +}) { + const router = useRouter(); + const [displayLocale, setDisplayLocale] = useState("en-US"); + const [displayTimeZone, setDisplayTimeZone] = useState("UTC"); + const [localStart, setLocalStart] = useState(localInputValue(start, "UTC")); + const [localEnd, setLocalEnd] = useState(localInputValue(end, "UTC")); + const [customRangeError, setCustomRangeError] = useState(null); + useEffect(() => { + const resolved = Intl.DateTimeFormat().resolvedOptions(); + const locale = resolved.locale || "en-US"; + const timeZone = resolved.timeZone || "UTC"; + setDisplayLocale(locale); + setDisplayTimeZone(timeZone); + setLocalStart(localInputValue(start, timeZone)); + setLocalEnd(localInputValue(end, timeZone)); + }, [start, end]); + const number = new Intl.NumberFormat(displayLocale); + const date = new Intl.DateTimeFormat(displayLocale, { + dateStyle: "medium", + timeStyle: "short", + timeZone: displayTimeZone, + }); + + const navigate = (nextStart: Date, nextEnd: Date) => { + router.push( + `/service-health?start=${encodeURIComponent(nextStart.toISOString())}&end=${encodeURIComponent(nextEnd.toISOString())}`, + ); + }; + const applyCustomRange = () => { + const nextStart = toISO(localStart); + const nextEnd = toISO(localEnd); + if (!nextStart || !nextEnd) { + setCustomRangeError("Bitte gib einen gültigen Zeitraum ein."); + return; + } + if (nextStart >= nextEnd) { + setCustomRangeError("Das Ende muss nach dem Beginn liegen."); + return; + } + setCustomRangeError(null); + navigate(new Date(nextStart), new Date(nextEnd)); + }; + const preset = (minutes: number) => { + const nextEnd = new Date(); + navigate(new Date(nextEnd.getTime() - minutes * 60 * 1000), nextEnd); + }; + + return ( +
+
+
+

Service Health

+

+ Aggregierte Upstream-Metriken. Zeitangaben werden lokal angezeigt. +

+
+
+ + + + + + + +
+
+ + + + + + {customRangeError ? ( + + {customRangeError} + + ) : null} + + Buckets: {formatBucket(dashboard.range.bucketSeconds)} + + + + +
+ } + title="Requests" + value={number.format(dashboard.totals.total)} + /> + } + title="2xx erfolgreich" + value={number.format(dashboard.totals.successful)} + /> + } + title="Unsuccessful" + value={number.format(dashboard.totals.unsuccessful)} + /> + } + title="Fehlerrate (reine Caller-Abbrüche separat)" + value={`${dashboard.totals.failureRate.toFixed(1)}%`} + /> + } + title="p95 Dauer" + value={formatLatency(dashboard.latency.duration.p95Ms, number)} + /> +
+ +
+ + + + + date.format(new Date(value))} + /> + + } /> + + + + + + + + + + + + date.format(new Date(value))} + /> + + } /> + + + + + + + + + ({ + ...row, + label: + row.statusCode === null + ? row.outcome + : String(row.statusCode), + }))} + > + + + + } /> + + + + + + +
+ + ({ + label: row.provider, + count: row.count, + unsuccessful: row.unsuccessful, + suffix: `Ø ${Math.round(row.avgDurationMs)} ms`, + }))} + title="Provider" + /> + ({ + label: row.endpoint, + count: row.count, + unsuccessful: row.unsuccessful, + }))} + title="Endpoints" + /> + ({ + label: row.model, + count: row.count, + unsuccessful: row.unsuccessful, + }))} + title="Modelle" + /> +
+ ); +} + +function Metric({ + title, + value, + icon, +}: { + title: string; + value: string | number; + icon: React.ReactNode; +}) { + return ( + + +
+

{title}

+

{value}

+
+ {icon} + + + ); +} + +function ChartCard({ + title, + children, +}: { + title: string; + children: React.ReactNode; +}) { + return ( + + + {title} + + {children} + + ); +} + +function LatencyCard({ + latency, + number, +}: { + latency: { duration: Latency; firstByte: Latency }; + number: Intl.NumberFormat; +}) { + return ( + + + Latency-Zusammenfassung + + + + + + + ); +} + +function TimelineTable({ timeline }: { timeline: Dashboard["timeline"] }) { + return ( + + + + + + + + + + + + {timeline.map((row) => ( + + + + + + + ))} + +
Requests und Fehler pro Zeitintervall
ZeitRequestsUnsuccessfulCaller cancel
{new Date(row.bucket).toISOString()}{row.total}{row.unsuccessful}{row.callerCancellations}
+ ); +} + +function LatencyTimelineTable({ + timeline, +}: { + timeline: Dashboard["timeline"]; +}) { + return ( + + + + + + + + + + + {timeline.map((row) => ( + + + + + + ))} + +
Durchschnittliche und p95-Latenz pro Zeitintervall
ZeitDurchschnitt in Millisekundenp95 in Millisekunden
{new Date(row.bucket).toISOString()}{row.avgDurationMs ?? "Keine Daten"}{row.p95DurationMs ?? "Keine Daten"}
+ ); +} + +function StatusCodeTable({ rows }: { rows: Dashboard["statusCodes"] }) { + return ( + + + + + + + + + + {rows.map((row) => ( + + + + + ))} + +
Requests nach HTTP-Statuscode
Statuscode oder ErgebnisRequests
{row.statusCode ?? row.outcome}{row.count}
+ ); +} + +function LatencyGroup({ + label, + latency, + number, +}: { + label: string; + latency: Latency; + number: Intl.NumberFormat; +}) { + const format = (value: number | null) => formatLatency(value, number); + return ( +
+

{label}

+
+
Ø
+
{format(latency.averageMs)}
+
p50
+
{format(latency.p50Ms)}
+
p95
+
{format(latency.p95Ms)}
+
Max
+
{format(latency.maximumMs)}
+
+
+ ); +} + +function BreakdownCard({ + title, + rows, +}: { + title: string; + rows: { + label: string; + count: number; + unsuccessful: number; + suffix?: string; + }[]; +}) { + return ( + + + {title} + + +
+ {rows.length ? ( + rows.map((row) => ( +
+ {row.label} + + {row.count} Requests · {row.unsuccessful} unsuccessful + {row.suffix ? ` · ${row.suffix}` : ""} + +
+ )) + ) : ( +

+ Keine Daten im gewählten Zeitraum. +

+ )} +
+ + + ); +} diff --git a/app/src/components/service-health-settings.tsx b/app/src/components/service-health-settings.tsx new file mode 100644 index 0000000..038e27f --- /dev/null +++ b/app/src/components/service-health-settings.tsx @@ -0,0 +1,63 @@ +"use client"; + +import { useState } from "react"; +import { toast } from "sonner"; + +import { Button } from "~/components/ui/button"; +import { Card, CardContent, CardHeader, CardTitle } from "~/components/ui/card"; +import { Input } from "~/components/ui/input"; +import { api } from "~/trpc/react"; + +export function ServiceHealthSettings({ value }: { value: string }) { + const [retention, setRetention] = useState(value); + const utils = api.useUtils(); + const update = api.serviceHealth.updateSettings.useMutation({ + onSuccess: async (result) => { + setRetention(result.value); + await utils.serviceHealth.getSettings.invalidate(); + toast.success(`${result.deleted} alte Health-Metriken gelöscht.`); + }, + }); + + return ( +
+

Health-Aufbewahrung

+

+ Lege fest, wie lange Request-Health-Metriken gespeichert werden. +

+ + + Aufbewahrungsdauer + + +
+ + +
+

+ Erlaubt sind positive Werte mit m (Minuten),{" "} + h (Stunden) oder d (Tage), zum Beispiel{" "} + 30d; maximal 10 Jahre. +

+ {update.error ? ( +

{update.error.message}

+ ) : null} + + +
+ ); +} diff --git a/app/src/components/sidebar-nav.tsx b/app/src/components/sidebar-nav.tsx index 61a5d2a..be89848 100644 --- a/app/src/components/sidebar-nav.tsx +++ b/app/src/components/sidebar-nav.tsx @@ -5,6 +5,7 @@ import { Boxes, Coins, FileText, + HeartPulse, KeyRound, Users, } from "lucide-react"; @@ -26,6 +27,11 @@ const userItems = [ label: "Meine Nutzung", icon: BarChart3, }, + { + href: "/service-health", + label: "Service Health", + icon: HeartPulse, + }, { href: "/my-api-keys", label: "Meine API-Schlüssel", @@ -54,6 +60,11 @@ const adminItems = [ label: "Modelle", icon: Boxes, }, + { + href: "/admin/service-health", + label: "Health-Aufbewahrung", + icon: HeartPulse, + }, ]; const reportingItems = [ diff --git a/app/src/server/api/root.ts b/app/src/server/api/root.ts index 59fcadd..771600f 100644 --- a/app/src/server/api/root.ts +++ b/app/src/server/api/root.ts @@ -2,6 +2,7 @@ import { adminRouter } from "~/server/api/routers/admin"; import { apiKeyRouter } from "~/server/api/routers/api-key"; import { postRouter } from "~/server/api/routers/post"; import { reportingRouter } from "~/server/api/routers/reporting"; +import { serviceHealthRouter } from "~/server/api/routers/service-health"; import { createCallerFactory, createTRPCRouter } from "~/server/api/trpc"; /** @@ -14,6 +15,7 @@ export const appRouter = createTRPCRouter({ post: postRouter, apiKey: apiKeyRouter, reporting: reportingRouter, + serviceHealth: serviceHealthRouter, }); // export type definition of API diff --git a/app/src/server/api/routers/service-health.ts b/app/src/server/api/routers/service-health.ts new file mode 100644 index 0000000..e703b2f --- /dev/null +++ b/app/src/server/api/routers/service-health.ts @@ -0,0 +1,422 @@ +import { TRPCError } from "@trpc/server"; +import { and, asc, desc, gte, lt, sql } from "drizzle-orm"; +import { z } from "zod"; + +import { createTRPCRouter, protectedProcedure } from "~/server/api/trpc"; +import { + requestHealthAttempts, + requestHealthSettings, +} from "~/server/db/schema"; + +const adminProcedure = protectedProcedure.use(({ ctx, next }) => { + if (!ctx.session.user.isAdmin) throw new TRPCError({ code: "FORBIDDEN" }); + return next({ ctx: { session: ctx.session } }); +}); + +const dashboardInput = z.object({ + start: z.string().datetime({ offset: true }), + end: z.string().datetime({ offset: true }), +}); + +const maxRequestHealthRetentionSeconds = 10 * 365 * 24 * 60 * 60; +const maxTimelineBuckets = 2000; +const maxBreakdownRows = 100; +const requestHealthBucketOriginMs = Date.UTC(2000, 0, 1); + +type LatencySummary = { + averageMs: number | null; + p50Ms: number | null; + p95Ms: number | null; + maximumMs: number | null; +}; + +function bucketSecondsForRange(durationMs: number) { + const baseBucketSeconds = + durationMs <= 15 * 60 * 1000 + ? 60 + : durationMs <= 60 * 60 * 1000 + ? 5 * 60 + : durationMs <= 24 * 60 * 60 * 1000 + ? 15 * 60 + : durationMs <= 7 * 24 * 60 * 60 * 1000 + ? 60 * 60 + : 6 * 60 * 60; + return Math.max( + baseBucketSeconds, + Math.ceil(durationMs / 1000 / maxTimelineBuckets), + ); +} + +export function requestHealthBucketStartMs( + timestampMs: number, + bucketSeconds: number, +) { + const bucketMs = bucketSeconds * 1000; + return ( + requestHealthBucketOriginMs + + Math.floor((timestampMs - requestHealthBucketOriginMs) / bucketMs) * + bucketMs + ); +} + +function parseRange(input: z.infer) { + const start = new Date(input.start); + const end = new Date(input.end); + if (!Number.isFinite(start.getTime()) || !Number.isFinite(end.getTime())) { + throw new TRPCError({ code: "BAD_REQUEST", message: "Invalid date range" }); + } + if (end <= start) { + throw new TRPCError({ + code: "BAD_REQUEST", + message: "The end of the range must be after its start", + }); + } + if ( + end.getTime() - start.getTime() > + maxRequestHealthRetentionSeconds * 1000 + ) { + throw new TRPCError({ + code: "BAD_REQUEST", + message: "The date range must not exceed 10 years", + }); + } + return { + start, + end, + bucketSeconds: bucketSecondsForRange(end.getTime() - start.getTime()), + }; +} + +function latencySummary(row: { + averageMs: number | null; + p50Ms: number | null; + p95Ms: number | null; + maximumMs: number | null; +}): LatencySummary { + return { + averageMs: row.averageMs === null ? null : Number(row.averageMs), + p50Ms: row.p50Ms === null ? null : Number(row.p50Ms), + p95Ms: row.p95Ms === null ? null : Number(row.p95Ms), + maximumMs: row.maximumMs === null ? null : Number(row.maximumMs), + }; +} + +function durationExpression( + column: + | typeof requestHealthAttempts.durationMs + | typeof requestHealthAttempts.firstByteDurationMs, +) { + return { + averageMs: sql`avg(${column})`, + p50Ms: sql`percentile_cont(0.50) within group (order by ${column})`, + p95Ms: sql`percentile_cont(0.95) within group (order by ${column})`, + maximumMs: sql`max(${column})`, + }; +} + +function retentionSeconds(value: string) { + const match = /^(?[1-9][0-9]*)(?[mhd])$/.exec(value.trim()); + if (!match?.groups) { + throw new TRPCError({ + code: "BAD_REQUEST", + message: "Use a positive duration such as 30d, 720h, or 90m", + }); + } + const amount = Number(match.groups.amount); + const multiplier = + match.groups.unit === "d" ? 86400 : match.groups.unit === "h" ? 3600 : 60; + const seconds = amount * multiplier; + if ( + !Number.isSafeInteger(seconds) || + seconds <= 0 || + seconds > maxRequestHealthRetentionSeconds + ) { + throw new TRPCError({ + code: "BAD_REQUEST", + message: "Retention duration must not exceed 10 years", + }); + } + return seconds; +} + +function formatRetention(seconds: number) { + if (seconds % 86400 === 0) return `${seconds / 86400}d`; + if (seconds % 3600 === 0) return `${seconds / 3600}h`; + return `${Math.floor(seconds / 60)}m`; +} + +export const serviceHealthRouter = createTRPCRouter({ + getDashboard: protectedProcedure + .input(dashboardInput) + .query(async ({ ctx, input }) => { + const { start, end, bucketSeconds } = parseRange(input); + const filter = and( + gte(requestHealthAttempts.startedAt, start.toISOString()), + lt(requestHealthAttempts.startedAt, end.toISOString()), + ); + const successfulCondition = sql`( + ${requestHealthAttempts.outcome} = 'success_2xx' + OR ( + ${requestHealthAttempts.clientCancelled} = true + AND ${requestHealthAttempts.statusCode} >= 200 + AND ${requestHealthAttempts.statusCode} < 300 + ) + )`; + const redirectCondition = sql`( + ${requestHealthAttempts.outcome} = 'redirect_3xx' + OR ( + ${requestHealthAttempts.clientCancelled} = true + AND ${requestHealthAttempts.statusCode} >= 300 + AND ${requestHealthAttempts.statusCode} < 400 + ) + )`; + const clientErrorCondition = sql`( + ${requestHealthAttempts.outcome} = 'client_error_4xx' + OR ( + ${requestHealthAttempts.clientCancelled} = true + AND ${requestHealthAttempts.statusCode} >= 400 + AND ${requestHealthAttempts.statusCode} < 500 + ) + )`; + const serverErrorCondition = sql`( + ${requestHealthAttempts.outcome} = 'server_error_5xx' + OR ( + ${requestHealthAttempts.clientCancelled} = true + AND ${requestHealthAttempts.statusCode} >= 500 + AND ${requestHealthAttempts.statusCode} <= 599 + ) + )`; + const unsuccessfulCondition = sql`( + ${requestHealthAttempts.outcome} IN ( + 'client_error_4xx', 'server_error_5xx', 'timeout', + 'upstream_canceled', 'transport_error' + ) + OR ( + ${requestHealthAttempts.clientCancelled} = true + AND ${requestHealthAttempts.statusCode} >= 400 + ) + )`; + const duration = durationExpression(requestHealthAttempts.durationMs); + const firstByte = durationExpression( + requestHealthAttempts.firstByteDurationMs, + ); + + const [ + summaryRow, + statusRows, + providerRows, + endpointRows, + modelRows, + timelineRows, + ] = await Promise.all([ + ctx.db + .select({ + total: sql`count(*)::int`, + successful: sql`count(*) filter (where ${successfulCondition})::int`, + redirects: sql`count(*) filter (where ${redirectCondition})::int`, + clientErrors: sql`count(*) filter (where ${clientErrorCondition})::int`, + serverErrors: sql`count(*) filter (where ${serverErrorCondition})::int`, + timeouts: sql`count(*) filter (where ${requestHealthAttempts.outcome} = 'timeout')::int`, + upstreamCancellations: sql`count(*) filter (where ${requestHealthAttempts.outcome} = 'upstream_canceled')::int`, + transportErrors: sql`count(*) filter (where ${requestHealthAttempts.outcome} = 'transport_error')::int`, + callerCancellations: sql`count(*) filter (where ${requestHealthAttempts.clientCancelled} = true)::int`, + unsuccessful: sql`count(*) filter (where ${unsuccessfulCondition})::int`, + duration, + firstByte, + }) + .from(requestHealthAttempts) + .where(filter), + ctx.db + .select({ + statusCode: requestHealthAttempts.statusCode, + outcome: sql`case when ${requestHealthAttempts.statusCode} is null then ${requestHealthAttempts.outcome} else '' end`, + count: sql`count(*)::int`, + }) + .from(requestHealthAttempts) + .where(filter) + .groupBy( + requestHealthAttempts.statusCode, + sql`case when ${requestHealthAttempts.statusCode} is null then ${requestHealthAttempts.outcome} else '' end`, + ) + .orderBy(desc(sql`count(*)`)), + ctx.db + .select({ + provider: requestHealthAttempts.upstream, + count: sql`count(*)::int`, + unsuccessful: sql`count(*) filter (where ${unsuccessfulCondition})::int`, + avgDurationMs: sql`coalesce(round(avg(${requestHealthAttempts.durationMs})), 0)::int`, + }) + .from(requestHealthAttempts) + .where(filter) + .groupBy(requestHealthAttempts.upstream) + .orderBy(desc(sql`count(*)`)) + .limit(maxBreakdownRows), + ctx.db + .select({ + endpoint: requestHealthAttempts.endpoint, + count: sql`count(*)::int`, + unsuccessful: sql`count(*) filter (where ${unsuccessfulCondition})::int`, + }) + .from(requestHealthAttempts) + .where(filter) + .groupBy(requestHealthAttempts.endpoint) + .orderBy(desc(sql`count(*)`)) + .limit(maxBreakdownRows), + ctx.db + .select({ + model: sql`coalesce(${requestHealthAttempts.model}, 'Unknown')`, + count: sql`count(*)::int`, + unsuccessful: sql`count(*) filter (where ${unsuccessfulCondition})::int`, + }) + .from(requestHealthAttempts) + .where(filter) + .groupBy(sql`coalesce(${requestHealthAttempts.model}, 'Unknown')`) + .orderBy(desc(sql`count(*)`)) + .limit(maxBreakdownRows), + (() => { + const bucket = sql`date_bin(${sql.raw(`interval '${bucketSeconds} seconds'`)}, ${requestHealthAttempts.startedAt}, timestamptz '2000-01-01 00:00:00+00')`; + return ctx.db + .select({ + bucket, + total: sql`count(*)::int`, + successful: sql`count(*) filter (where ${successfulCondition})::int`, + unsuccessful: sql`count(*) filter (where ${unsuccessfulCondition})::int`, + callerCancellations: sql`count(*) filter (where ${requestHealthAttempts.clientCancelled} = true)::int`, + avgDurationMs: sql`coalesce(round(avg(${requestHealthAttempts.durationMs})), 0)::int`, + p95DurationMs: sql`coalesce(percentile_cont(0.95) within group (order by ${requestHealthAttempts.durationMs}), 0)::int`, + }) + .from(requestHealthAttempts) + .where(filter) + .groupBy(bucket) + .orderBy(asc(bucket)); + })(), + ]); + + const summary = summaryRow[0]; + const total = Number(summary?.total ?? 0); + const unsuccessful = Number(summary?.unsuccessful ?? 0); + const callerCancellations = Number(summary?.callerCancellations ?? 0); + const bucketMs = bucketSeconds * 1000; + const firstBucket = requestHealthBucketStartMs( + start.getTime(), + bucketSeconds, + ); + const timelineByBucket = new Map( + timelineRows.map((row) => [new Date(row.bucket).getTime(), row]), + ); + const timeline = []; + for ( + let cursor = firstBucket; + cursor < end.getTime(); + cursor += bucketMs + ) { + const row = timelineByBucket.get(cursor); + timeline.push({ + bucket: new Date(cursor).toISOString(), + total: Number(row?.total ?? 0), + successful: Number(row?.successful ?? 0), + unsuccessful: Number(row?.unsuccessful ?? 0), + callerCancellations: Number(row?.callerCancellations ?? 0), + avgDurationMs: row ? Number(row.avgDurationMs) : null, + p95DurationMs: row ? Number(row.p95DurationMs) : null, + }); + } + + return { + range: { + start: start.toISOString(), + end: end.toISOString(), + bucketSeconds, + }, + totals: { + total, + successful: Number(summary?.successful ?? 0), + redirects: Number(summary?.redirects ?? 0), + clientErrors: Number(summary?.clientErrors ?? 0), + serverErrors: Number(summary?.serverErrors ?? 0), + timeouts: Number(summary?.timeouts ?? 0), + upstreamCancellations: Number(summary?.upstreamCancellations ?? 0), + transportErrors: Number(summary?.transportErrors ?? 0), + callerCancellations, + unsuccessful, + failureRate: total ? (unsuccessful / total) * 100 : 0, + }, + latency: { + duration: latencySummary( + summary?.duration ?? { + averageMs: null, + p50Ms: null, + p95Ms: null, + maximumMs: null, + }, + ), + firstByte: latencySummary( + summary?.firstByte ?? { + averageMs: null, + p50Ms: null, + p95Ms: null, + maximumMs: null, + }, + ), + }, + statusCodes: statusRows.map((row) => ({ + statusCode: row.statusCode, + outcome: row.outcome, + count: Number(row.count), + })), + providers: providerRows.map((row) => ({ + provider: row.provider, + count: Number(row.count), + unsuccessful: Number(row.unsuccessful), + avgDurationMs: Number(row.avgDurationMs), + })), + endpoints: endpointRows.map((row) => ({ + endpoint: row.endpoint, + count: Number(row.count), + unsuccessful: Number(row.unsuccessful), + })), + models: modelRows.map((row) => ({ + model: row.model, + count: Number(row.count), + unsuccessful: Number(row.unsuccessful), + })), + timeline, + }; + }), + getSettings: adminProcedure.query(async ({ ctx }) => { + const [settings] = await ctx.db + .select() + .from(requestHealthSettings) + .limit(1); + const seconds = Number(settings?.retentionSeconds ?? 2592000n); + return { value: formatRetention(seconds) }; + }), + updateSettings: adminProcedure + .input(z.object({ value: z.string().trim().min(1).max(32) })) + .mutation(async ({ ctx, input }) => { + const seconds = retentionSeconds(input.value); + const cutoff = new Date(Date.now() - seconds * 1000).toISOString(); + const deleted = await ctx.db.transaction(async (tx) => { + await tx + .insert(requestHealthSettings) + .values({ + id: 1, + retentionSeconds: BigInt(seconds), + updatedAt: new Date().toISOString(), + }) + .onConflictDoUpdate({ + target: requestHealthSettings.id, + set: { + retentionSeconds: BigInt(seconds), + updatedAt: new Date().toISOString(), + }, + }); + const rows = await tx + .delete(requestHealthAttempts) + .where(lt(requestHealthAttempts.startedAt, cutoff)) + .returning({ attemptId: requestHealthAttempts.attemptId }); + return rows.length; + }); + return { value: formatRetention(seconds), deleted }; + }), +}); diff --git a/app/src/server/db/schema.ts b/app/src/server/db/schema.ts index c3de608..acd2518 100644 --- a/app/src/server/db/schema.ts +++ b/app/src/server/db/schema.ts @@ -11,6 +11,7 @@ import { pgEnum, pgTable, primaryKey, + smallint, text, timestamp, uniqueIndex, @@ -415,3 +416,58 @@ export const azurePricingAudits = pgTable("azure_pricing_audits", { rows: jsonb().notNull().default(sql`'[]'::jsonb`), error: text(), }); + +export const requestHealthAttempts = pgTable( + "request_health_attempts", + { + attemptId: varchar("attempt_id", { length: 36 }).primaryKey().notNull(), + requestId: varchar("request_id", { length: 255 }), + endpoint: varchar({ length: 255 }).notNull(), + upstream: varchar({ length: 255 }).notNull(), + model: varchar({ length: 255 }), + method: varchar({ length: 16 }).notNull(), + statusCode: smallint("status_code"), + outcome: varchar({ length: 32 }).notNull(), + durationMs: bigint("duration_ms", { mode: "number" }).notNull(), + firstByteDurationMs: bigint("first_byte_duration_ms", { mode: "number" }), + streaming: boolean().notNull().default(false), + clientCancelled: boolean("client_cancelled").notNull().default(false), + startedAt: timestamp("started_at", { + withTimezone: true, + mode: "string", + }) + .defaultNow() + .notNull(), + completedAt: timestamp("completed_at", { + withTimezone: true, + mode: "string", + }) + .defaultNow() + .notNull(), + createdAt: timestamp("created_at", { withTimezone: true, mode: "string" }) + .defaultNow() + .notNull(), + }, + (table) => [ + index("request_health_attempts_started_at_idx").on(table.startedAt), + index("request_health_attempts_outcome_started_at_idx").on( + table.outcome, + table.startedAt, + ), + index("request_health_attempts_status_started_at_idx").on( + table.statusCode, + table.startedAt, + ), + ], +); + +export const requestHealthSettings = pgTable("request_health_settings", { + id: smallint().primaryKey().default(1), + retentionSeconds: bigint("retention_seconds", { mode: "bigint" }) + .notNull() + .default(2592000n), + updatedAt: timestamp("updated_at", { withTimezone: true, mode: "string" }) + .defaultNow() + .notNull(), + updatedBy: varchar("updated_by", { length: 255 }), +}); diff --git a/cmd/main.go b/cmd/main.go index 3c91fd8..1e56f12 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -1,6 +1,7 @@ package main import ( + "context" "log" "net/http" api "openai-api-proxy/api" @@ -11,6 +12,7 @@ import ( "os" "os/signal" "syscall" + "time" "github.com/joho/godotenv" ) @@ -24,33 +26,31 @@ func main() { db := db.DatabaseInit() defer db.Close() - osExit(db) - defer log.Println("Closing DB Clients :)") - mux := http.NewServeMux() a := auth.Init(mux, db) - // - proxy.Init(mux, db) // Start AI Proxy - web.Init(mux, a) // Start Web UI - api.ApiInit(mux, a, db) // Start Backend API + recorder := proxy.Init(mux, db) // Start AI Proxy + web.Init(mux, a) // Start Web UI + api.ApiInit(mux, a, db) // Start Backend API log.Printf("Serving on http://localhost:%d", 8082) - log.Fatal(http.ListenAndServe(":8082", mux)) - -} - -// Close DB on Program Exit -func osExit(db *db.Database) { + server := &http.Server{Addr: ":8082", Handler: mux} sigc := make(chan os.Signal, 1) - signal.Notify(sigc, - syscall.SIGHUP, - syscall.SIGINT, - syscall.SIGTERM, - syscall.SIGQUIT) + signal.Notify(sigc, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT) go func() { s := <-sigc - log.Printf("Exit: %s Closing DB Clients :)", s) + log.Printf("Exit: %s", s) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := server.Shutdown(ctx); err != nil { + log.Printf("HTTP server shutdown failed: %v", err) + } + recorder.CloseContext(ctx) db.Close() - os.Exit(1) }() + if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Printf("HTTP server stopped unexpectedly: %v", err) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + recorder.CloseContext(ctx) + cancel() + } } diff --git a/db/migrations/20260831100000_request_health.sql b/db/migrations/20260831100000_request_health.sql new file mode 100644 index 0000000..428e538 --- /dev/null +++ b/db/migrations/20260831100000_request_health.sql @@ -0,0 +1,44 @@ +CREATE TABLE IF NOT EXISTS request_health_settings ( + id smallint PRIMARY KEY DEFAULT 1 CHECK (id = 1), + retention_seconds bigint NOT NULL DEFAULT 2592000 CHECK (retention_seconds > 0), + updated_at timestamptz NOT NULL DEFAULT now(), + updated_by varchar(255) REFERENCES users(id) ON DELETE SET NULL +); + +INSERT INTO request_health_settings (id) +VALUES (1) +ON CONFLICT (id) DO NOTHING; + +CREATE TABLE IF NOT EXISTS request_health_attempts ( + attempt_id varchar(36) PRIMARY KEY, + request_id varchar(255), + endpoint varchar(255) NOT NULL, + upstream varchar(255) NOT NULL, + model varchar(255), + method varchar(16) NOT NULL, + status_code smallint, + outcome varchar(32) NOT NULL, + duration_ms bigint NOT NULL CHECK (duration_ms >= 0), + first_byte_duration_ms bigint CHECK (first_byte_duration_ms >= 0), + streaming boolean NOT NULL DEFAULT false, + client_cancelled boolean NOT NULL DEFAULT false, + started_at timestamptz NOT NULL DEFAULT now(), + completed_at timestamptz NOT NULL DEFAULT now(), + created_at timestamptz NOT NULL DEFAULT now(), + CONSTRAINT request_health_attempts_status_check CHECK ( + status_code IS NULL OR (status_code >= 100 AND status_code <= 599) + ), + CONSTRAINT request_health_attempts_outcome_check CHECK ( + outcome IN ( + 'success_2xx', 'redirect_3xx', 'client_error_4xx', 'server_error_5xx', + 'timeout', 'upstream_canceled', 'transport_error', 'caller_canceled' + ) + ) +); + +CREATE INDEX IF NOT EXISTS request_health_attempts_started_at_idx + ON request_health_attempts (started_at); +CREATE INDEX IF NOT EXISTS request_health_attempts_outcome_started_at_idx + ON request_health_attempts (outcome, started_at); +CREATE INDEX IF NOT EXISTS request_health_attempts_status_started_at_idx + ON request_health_attempts (status_code, started_at); diff --git a/db/migrations/atlas.sum b/db/migrations/atlas.sum index 6c09ab1..8f12178 100644 --- a/db/migrations/atlas.sum +++ b/db/migrations/atlas.sum @@ -1,4 +1,4 @@ -h1:M+IJgytlEV67h1svjx6prA48NmxiXduDienWfaST2fI= +h1:Q2yvd0OOFoDBQismvZyHqkxB3zxF9EASgvamsTzbjlE= 20240924151008_init.sql h1:rTjsfTqruGXjIUITaR6Pqt0n4lm+4EcdvESuYXgmyP4= 20240924154252_request_monitoring.sql h1:VWH6kcc2DWS7L9tBQ8lgS858/9f5HP7PMYZo0fZlCX8= 20241001124048_id_datatype.sql h1:Nb1OpAzcVshPzgEqQ3w/sCFjorO4h/GTOBeuISanu4k= @@ -28,3 +28,4 @@ h1:M+IJgytlEV67h1svjx6prA48NmxiXduDienWfaST2fI= 20260825120000_reporting_request_indexes.sql h1:fHCPTleO4cX5OAcwoJd4n5UJD8vWty/r4sNlItuXyVw= 20260826100000_azure_retail_pricing.sql h1:Xd8OtUZg4VZlsYlSjLmYyxOwJVNHPXgK/Mdxft5lURc= 20260826130000_request_statistics_cache.sql h1:GKodt4zvUbq+KMFdKM6pPPI6v5Q37d9nFjP9j1CLcmc= +20260831100000_request_health.sql h1:Wp16grY0k+BPaLVRZ9UnNCqoUzPyv9Dyr5MCNXMDORk= diff --git a/db/request_health.go b/db/request_health.go new file mode 100644 index 0000000..eae0031 --- /dev/null +++ b/db/request_health.go @@ -0,0 +1,93 @@ +package database + +import ( + "context" + "time" +) + +// RequestHealthAttempt is an append-only, payload-free record of one upstream +// attempt. It is intentionally separate from usage accounting because failed +// requests do not have usage records. +type RequestHealthAttempt struct { + AttemptID string + RequestID string + Endpoint string + Upstream string + Model string + Method string + StatusCode int + Outcome string + Duration time.Duration + FirstByteDuration *time.Duration + Streaming bool + ClientCancelled bool + StartedAt time.Time + CompletedAt time.Time +} + +const defaultRequestHealthRetention = 30 * 24 * time.Hour + +func (d *Database) WriteRequestHealthAttempt(a *RequestHealthAttempt) error { + return d.WriteRequestHealthAttemptContext(context.Background(), a) +} + +func (d *Database) WriteRequestHealthAttemptContext(ctx context.Context, a *RequestHealthAttempt) error { + _, err := d.db.ExecContext(ctx, ` + INSERT INTO request_health_attempts + (attempt_id, request_id, endpoint, upstream, model, method, status_code, + outcome, duration_ms, first_byte_duration_ms, streaming, client_cancelled, started_at, completed_at) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,COALESCE($13, now()),COALESCE($14, now())) + ON CONFLICT (attempt_id) DO NOTHING`, + a.AttemptID, nullOrString(a.RequestID), a.Endpoint, a.Upstream, + nullOrString(a.Model), a.Method, nullableStatus(a.StatusCode), a.Outcome, + a.Duration.Milliseconds(), nullableDurationMillis(a.FirstByteDuration), + a.Streaming, a.ClientCancelled, nullableTime(a.StartedAt), nullableTime(a.CompletedAt)) + return err +} + +func (d *Database) RequestHealthRetention() (time.Duration, error) { + var seconds int64 + err := d.db.QueryRow(` + SELECT retention_seconds + FROM request_health_settings + WHERE id = 1`).Scan(&seconds) + if err != nil { + return defaultRequestHealthRetention, err + } + if seconds <= 0 { + return defaultRequestHealthRetention, nil + } + return time.Duration(seconds) * time.Second, nil +} + +func (d *Database) DeleteExpiredRequestHealthAttempts(before time.Time) (int64, error) { + result, err := d.db.Exec( + "DELETE FROM request_health_attempts WHERE started_at < $1", + before.UTC(), + ) + if err != nil { + return 0, err + } + return result.RowsAffected() +} + +func nullableStatus(status int) any { + if status == 0 { + return nil + } + return status +} + +func nullableDurationMillis(duration *time.Duration) any { + if duration == nil { + return nil + } + return duration.Milliseconds() +} + +func nullableTime(value time.Time) any { + if value.IsZero() { + return nil + } + return value +} diff --git a/db/schema.sql b/db/schema.sql index ad7fda1..c5cc5e0 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -153,6 +153,44 @@ CREATE TABLE IF NOT EXISTS request_statistics_cache ( CONSTRAINT request_statistics_cache_pkey PRIMARY KEY (bucket_start, api_key_id, model, request_type) ); +CREATE TABLE IF NOT EXISTS request_health_settings ( + id smallint PRIMARY KEY DEFAULT 1 CHECK (id = 1), + retention_seconds bigint NOT NULL DEFAULT 2592000 CHECK (retention_seconds > 0), + updated_at timestamptz NOT NULL DEFAULT now(), + updated_by varchar(255) REFERENCES users(id) ON DELETE SET NULL +); +INSERT INTO request_health_settings (id) VALUES (1) ON CONFLICT (id) DO NOTHING; + +CREATE TABLE IF NOT EXISTS request_health_attempts ( + attempt_id varchar(36) PRIMARY KEY, + request_id varchar(255), + endpoint varchar(255) NOT NULL, + upstream varchar(255) NOT NULL, + model varchar(255), + method varchar(16) NOT NULL, + status_code smallint, + outcome varchar(32) NOT NULL, + duration_ms bigint NOT NULL CHECK (duration_ms >= 0), + first_byte_duration_ms bigint CHECK (first_byte_duration_ms >= 0), + streaming boolean NOT NULL DEFAULT false, + client_cancelled boolean NOT NULL DEFAULT false, + started_at timestamptz NOT NULL DEFAULT now(), + completed_at timestamptz NOT NULL DEFAULT now(), + created_at timestamptz NOT NULL DEFAULT now(), + CONSTRAINT request_health_attempts_status_check CHECK ( + status_code IS NULL OR (status_code >= 100 AND status_code <= 599) + ), + CONSTRAINT request_health_attempts_outcome_check CHECK ( + outcome IN ( + 'success_2xx', 'redirect_3xx', 'client_error_4xx', 'server_error_5xx', + 'timeout', 'upstream_canceled', 'transport_error', 'caller_canceled' + ) + ) +); +CREATE INDEX IF NOT EXISTS request_health_attempts_started_at_idx ON request_health_attempts (started_at); +CREATE INDEX IF NOT EXISTS request_health_attempts_outcome_started_at_idx ON request_health_attempts (outcome, started_at); +CREATE INDEX IF NOT EXISTS request_health_attempts_status_started_at_idx ON request_health_attempts (status_code, started_at); + CREATE INDEX IF NOT EXISTS request_statistics_cache_bucket_idx ON request_statistics_cache (bucket_start);