From 2ce3921f7886ec7e048ac9403e3bdd98a0a588cc Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Fri, 22 May 2026 20:47:41 -0400 Subject: [PATCH] =?UTF-8?q?=E2=9A=96=EF=B8=8F=20feat:=20Add=20Operational?= =?UTF-8?q?=20Prometheus=20Metrics=20(#13265)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * add operational prometheus metrics * fix metrics review findings * fix metrics edge cases * gate mongoose metrics instrumentation * gate metrics setup when unconfigured --- api/db/connect.js | 4 +- api/server/index.metrics.spec.js | 4 +- packages/api/src/app/metrics.spec.ts | 338 +++++++++++- packages/api/src/app/metrics.ts | 485 +++++++++++++++++- packages/api/src/auth/openid.spec.ts | 8 + packages/api/src/auth/openid.ts | 125 +++-- .../api/src/stream/GenerationJobManager.ts | 62 ++- 7 files changed, 967 insertions(+), 59 deletions(-) diff --git a/api/db/connect.js b/api/db/connect.js index 3534884b57..a63d3301b6 100644 --- a/api/db/connect.js +++ b/api/db/connect.js @@ -1,10 +1,12 @@ require('dotenv').config(); -const { isEnabled } = require('@librechat/api'); +const { isEnabled, instrumentMongooseQueryMetrics } = require('@librechat/api'); const { logger } = require('@librechat/data-schemas'); const mongoose = require('mongoose'); const MONGO_URI = process.env.MONGO_URI; +instrumentMongooseQueryMetrics(mongoose); + if (!MONGO_URI) { throw new Error('Please define the MONGO_URI environment variable'); } diff --git a/api/server/index.metrics.spec.js b/api/server/index.metrics.spec.js index 3f29347292..c907aca3b6 100644 --- a/api/server/index.metrics.spec.js +++ b/api/server/index.metrics.spec.js @@ -72,16 +72,18 @@ describe('Server metrics route', () => { mongoServer = await MongoMemoryServer.create(); process.env.MONGO_URI = mongoServer.getUri(); process.env.PORT = '0'; + process.env.METRICS_SECRET = 'test-secret'; app = require('~/server'); await healthCheckPoll(app); }); afterEach(() => { - delete process.env.METRICS_SECRET; + process.env.METRICS_SECRET = 'test-secret'; }); afterAll(async () => { + delete process.env.METRICS_SECRET; await mongoServer.stop(); await mongoose.disconnect(); }); diff --git a/packages/api/src/app/metrics.spec.ts b/packages/api/src/app/metrics.spec.ts index 6bb66ed278..80984dcb8b 100644 --- a/packages/api/src/app/metrics.spec.ts +++ b/packages/api/src/app/metrics.spec.ts @@ -1,5 +1,19 @@ /// -import { normalizePath } from './metrics'; +import { EventEmitter } from 'events'; +import express from 'express'; +import type { Request, Response } from 'express'; +import { + createMetrics, + instrumentMongooseQueryMetrics, + normalizePath, + recordGenerationJob, + recordGenerationStreamResumePendingEvents, + recordGenerationStreamSubscription, + recordOpenIDUserLookup, + setGenerationJobsInFlight, +} from './metrics'; + +const request = require('supertest') as (app: express.Express) => any; describe('normalizePath', () => { it.each([ @@ -9,8 +23,17 @@ describe('normalizePath', () => { ['/api/messages/artifact/507f1f77bcf86cd799439012', '/api/messages/artifact/#id'], ['/api/convos/507f1f77bcf86cd799439011', '/api/convos/#id'], ['/api/files/507f1f77bcf86cd799439011', '/api/files/#id'], + ['/api/files/507f1f77bcf86cd799439011/preview', '/api/files/#id/preview'], + ['/api/files/download/user-123/file-456', '/api/files/download/#id/#id'], + ['/api/files/download-url/user-123/file-456', '/api/files/download-url/#id/#id'], + ['/api/files/code/download/session-123/file-456', '/api/files/code/download/#id/#id'], ['/api/agents/507f1f77bcf86cd799439011', '/api/agents/#id'], + ['/api/agents/chat/stream/stream-123', '/api/agents/chat/stream/#id'], + ['/api/agents/chat/status/convo-123', '/api/agents/chat/status/#id'], + ['/api/agents/v1/chat/completions', '/api/agents/v1/chat/completions'], + ['/api/agents/v1/responses', '/api/agents/v1/responses'], ['/api/assistants/507f1f77bcf86cd799439011', '/api/assistants/#id'], + ['/api/skills/507f1f77bcf86cd799439011/files/reference.md', '/api/skills/#id/files'], ['/api/share/some-token-value', '/api/share/#token'], ['/share/shareId-with_nanoidChars', '/share/#id'], ['/share/shareId-with_nanoidChars/edit', '/share/#id/edit'], @@ -45,3 +68,316 @@ describe('normalizePath', () => { expect(normalizePath(input)).toBe(normalized); }); }); + +describe('createMetrics', () => { + afterEach(() => { + delete process.env.METRICS_SECRET; + }); + + it('uses a no-op middleware and unauthorized router when metrics are not configured', async () => { + const app = express(); + const { metricsMiddleware, metricsRouter } = createMetrics(); + const res = new EventEmitter(); + const next = jest.fn(); + + metricsMiddleware( + { headers: {}, method: 'GET', path: '/api/slow-response' } as Request, + res as unknown as Response, + next, + ); + app.use('/metrics', metricsRouter); + + await request(app).get('/metrics').expect(401); + expect(next).toHaveBeenCalledTimes(1); + expect(res.listenerCount('finish')).toBe(0); + expect(res.listenerCount('close')).toBe(0); + }); + + it('tracks request counts, in-flight gauges, and request body sizes', async () => { + const app = express(); + process.env.METRICS_SECRET = 'test-secret'; + const { metricsMiddleware, metricsRouter } = createMetrics(); + app.use(metricsMiddleware); + app.use(express.text({ type: '*/*' })); + app.post('/api/files/:id', (_req, res) => { + res.status(201).send('ok'); + }); + app.use('/metrics', metricsRouter); + + await request(app) + .post('/api/files/507f1f77bcf86cd799439011') + .set('Content-Type', 'text/plain') + .send('x'.repeat(42)) + .expect(201); + + const response = await request(app) + .get('/metrics') + .set('Authorization', 'Bearer test-secret') + .expect(200); + + expect(response.text).toMatch( + /http_requests_total\{method="POST",path="\/api\/files\/#id",status="201"\} 1/, + ); + expect(response.text).toMatch( + /http_requests_in_flight\{method="POST",path="\/api\/files\/#id"\} 0/, + ); + expect(response.text).toMatch( + /http_request_body_bytes_count\{method="POST",path="\/api\/files\/#id"\} 1/, + ); + expect(response.text).toMatch( + /http_request_body_bytes_sum\{method="POST",path="\/api\/files\/#id"\} 42/, + ); + }); + + it('tracks SSE stream counts, active gauges, and stream duration', async () => { + const app = express(); + process.env.METRICS_SECRET = 'test-secret'; + const { metricsMiddleware, metricsRouter } = createMetrics(); + app.use(metricsMiddleware); + app.get('/api/agents/chat/stream/:streamId', (_req, res) => { + res.setHeader('Content-Type', 'text/event-stream'); + res.write('event: message\ndata: {}\n\n'); + res.end(); + }); + app.use('/metrics', metricsRouter); + + await request(app).get('/api/agents/chat/stream/stream-123').expect(200); + + const response = await request(app) + .get('/metrics') + .set('Authorization', 'Bearer test-secret') + .expect(200); + + expect(response.text).toMatch( + /sse_streams_total\{method="GET",path="\/api\/agents\/chat\/stream\/#id",status="200"\} 1/, + ); + expect(response.text).toMatch( + /sse_streams_in_flight\{method="GET",path="\/api\/agents\/chat\/stream\/#id"\} 0/, + ); + expect(response.text).toMatch( + /sse_stream_duration_seconds_count\{method="GET",path="\/api\/agents\/chat\/stream\/#id",status="200"\} 1/, + ); + }); + + it('tracks upload counts, active gauges, duration, and upload bytes', async () => { + const app = express(); + process.env.METRICS_SECRET = 'test-secret'; + const { metricsMiddleware, metricsRouter } = createMetrics(); + app.use(metricsMiddleware); + app.post('/api/files', (_req, res) => { + res.status(201).send('ok'); + }); + app.use('/metrics', metricsRouter); + + await request(app) + .post('/api/files') + .attach('file', Buffer.from('uploaded body'), 'test.txt') + .expect(201); + + const response = await request(app) + .get('/metrics') + .set('Authorization', 'Bearer test-secret') + .expect(200); + + expect(response.text).toMatch( + /upload_requests_total\{method="POST",path="\/api\/files",status="201"\} 1/, + ); + expect(response.text).toMatch( + /upload_requests_in_flight\{method="POST",path="\/api\/files"\} 0/, + ); + expect(response.text).toMatch( + /upload_request_duration_seconds_count\{method="POST",path="\/api\/files",status="201"\} 1/, + ); + expect(response.text).toMatch(/upload_bytes_total\{method="POST",path="\/api\/files"\} \d+/); + }); + + it('does not track non-upload methods on upload paths as uploads', async () => { + const app = express(); + process.env.METRICS_SECRET = 'test-secret'; + const { metricsMiddleware, metricsRouter } = createMetrics(); + app.use(metricsMiddleware); + app.delete('/api/files', (_req, res) => { + res.status(204).end(); + }); + app.use('/metrics', metricsRouter); + + await request(app).delete('/api/files').expect(204); + + const response = await request(app) + .get('/metrics') + .set('Authorization', 'Bearer test-secret') + .expect(200); + + expect(response.text).not.toMatch(/upload_requests_total\{method="DELETE"/); + expect(response.text).not.toMatch(/upload_bytes_total\{method="DELETE"/); + }); + + it('labels requests closed before finish as client-aborted', async () => { + const app = express(); + process.env.METRICS_SECRET = 'test-secret'; + const { metricsMiddleware, metricsRouter } = createMetrics(); + const headers = new Map(); + const res = Object.assign(new EventEmitter(), { + destroyed: false, + getHeader: (name: string) => headers.get(name.toLowerCase()), + setHeader(name: string, value: unknown) { + headers.set(name.toLowerCase(), value); + return this; + }, + statusCode: 200, + writableEnded: false, + writeHead() { + return this; + }, + }); + const next = jest.fn(); + + metricsMiddleware( + { headers: {}, method: 'GET', path: '/api/slow-response' } as Request, + res as unknown as Response, + next, + ); + res.emit('close'); + app.use('/metrics', metricsRouter); + + const response = await request(app) + .get('/metrics') + .set('Authorization', 'Bearer test-secret') + .expect(200); + + expect(next).toHaveBeenCalledTimes(1); + expect(response.text).toMatch( + /http_requests_total\{method="GET",path="\/api\/#path",status="499"\} 1/, + ); + expect(response.text).not.toMatch( + /http_requests_total\{method="GET",path="\/api\/#path",status="200"\}/, + ); + }); + + it('tracks OpenID user lookup outcomes and latency', async () => { + const app = express(); + process.env.METRICS_SECRET = 'test-secret'; + const { metricsRouter } = createMetrics(); + app.use('/metrics', metricsRouter); + + recordOpenIDUserLookup('found', 0.2); + + const response = await request(app) + .get('/metrics') + .set('Authorization', 'Bearer test-secret') + .expect(200); + + expect(response.text).toMatch(/openid_user_lookup_total\{result="found"\} 1/); + expect(response.text).toMatch(/openid_user_lookup_duration_seconds_count\{result="found"\} 1/); + expect(response.text).toMatch(/openid_user_lookup_duration_seconds_sum\{result="found"\} 0.2/); + }); + + it('tracks mongoose query counts and latency by model and operation', async () => { + class FakeQuery { + model = { modelName: 'User' }; + op = 'findOne'; + + exec() { + return Promise.resolve({ _id: 'user-1' }); + } + } + + const app = express(); + process.env.METRICS_SECRET = 'test-secret'; + const { metricsRouter } = createMetrics(); + app.use('/metrics', metricsRouter); + + instrumentMongooseQueryMetrics({ Query: FakeQuery } as never); + await new FakeQuery().exec(); + + const response = await request(app) + .get('/metrics') + .set('Authorization', 'Bearer test-secret') + .expect(200); + + expect(response.text).toMatch( + /mongoose_queries_total\{model="User",operation="findOne",status="success"\} 1/, + ); + expect(response.text).toMatch( + /mongoose_query_duration_seconds_count\{model="User",operation="findOne",status="success"\} 1/, + ); + }); + + it('does not instrument mongoose queries when metrics are not configured', async () => { + class FakeQuery { + model = { modelName: 'User' }; + op = 'findOne'; + + exec() { + return Promise.resolve({ _id: 'user-1' }); + } + } + + const originalExec = FakeQuery.prototype.exec; + + instrumentMongooseQueryMetrics({ Query: FakeQuery } as never); + + expect(FakeQuery.prototype.exec).toBe(originalExec); + }); + + it('tracks mongoose query errors thrown before a promise is returned', async () => { + class ThrowingQuery { + model = { modelName: 'User' }; + op = 'findOne'; + + exec() { + throw new Error('sync database error'); + } + } + + const app = express(); + process.env.METRICS_SECRET = 'test-secret'; + const { metricsRouter } = createMetrics(); + app.use('/metrics', metricsRouter); + + instrumentMongooseQueryMetrics({ Query: ThrowingQuery } as never); + expect(() => new ThrowingQuery().exec()).toThrow('sync database error'); + + const response = await request(app) + .get('/metrics') + .set('Authorization', 'Bearer test-secret') + .expect(200); + + expect(response.text).toMatch( + /mongoose_queries_total\{model="User",operation="findOne",status="error"\} 1/, + ); + expect(response.text).toMatch( + /mongoose_query_duration_seconds_count\{model="User",operation="findOne",status="error"\} 1/, + ); + }); + + it('tracks generation job and stream resume metrics', async () => { + const app = express(); + process.env.METRICS_SECRET = 'test-secret'; + const { metricsRouter } = createMetrics(); + app.use('/metrics', metricsRouter); + + recordGenerationJob('memory', 'created'); + setGenerationJobsInFlight('memory', 2); + recordGenerationStreamSubscription('redis', 'resume', 'not_found'); + recordGenerationStreamSubscription('redis', 'resume_state', 'missing'); + recordGenerationStreamResumePendingEvents('memory', 3); + + const response = await request(app) + .get('/metrics') + .set('Authorization', 'Bearer test-secret') + .expect(200); + + expect(response.text).toMatch(/generation_jobs_total\{store="memory",result="created"\} 1/); + expect(response.text).toMatch(/generation_jobs_in_flight\{store="memory"\} 2/); + expect(response.text).toMatch( + /generation_stream_subscriptions_total\{store="redis",type="resume",result="not_found"\} 1/, + ); + expect(response.text).toMatch( + /generation_stream_subscriptions_total\{store="redis",type="resume_state",result="missing"\} 1/, + ); + expect(response.text).toMatch( + /generation_stream_resume_pending_events_total\{store="memory"\} 3/, + ); + }); +}); diff --git a/packages/api/src/app/metrics.ts b/packages/api/src/app/metrics.ts index 8089a9c3a1..4dd9fddac2 100644 --- a/packages/api/src/app/metrics.ts +++ b/packages/api/src/app/metrics.ts @@ -1,10 +1,18 @@ import { timingSafeEqual } from 'crypto'; import { Router } from 'express'; -import { Registry, collectDefaultMetrics, Counter, Histogram } from 'prom-client'; +import { Registry, collectDefaultMetrics, Counter, Gauge, Histogram } from 'prom-client'; import { logger } from '@librechat/data-schemas'; import type { Request, Response, NextFunction, RequestHandler } from 'express'; +import type { Mongoose } from 'mongoose'; const PATH_NORMALIZATIONS: [RegExp, string][] = [ + [/^\/api\/agents\/chat\/stream\/[^/]+(?=\/|$)/, '/api/agents/chat/stream/#id'], + [/^\/api\/agents\/chat\/status\/[^/]+(?=\/|$)/, '/api/agents/chat/status/#id'], + [/^\/api\/files\/code\/download\/[^/]+\/[^/]+(?=\/|$)/, '/api/files/code/download/#id/#id'], + [/^\/api\/files\/download-url\/[^/]+\/[^/]+(?=\/|$)/, '/api/files/download-url/#id/#id'], + [/^\/api\/files\/download\/[^/]+\/[^/]+(?=\/|$)/, '/api/files/download/#id/#id'], + [/^\/api\/files\/[^/]+\/preview(?=\/|$)/, '/api/files/#id/preview'], + [/^\/api\/skills\/[^/]+\/files(?:\/.*)?(?=\/|$)/, '/api/skills/#id/files'], [/^\/api\/messages\/artifact\/[^/]+(?=\/|$)/, '/api/messages/artifact/#id'], [/^\/api\/messages\/[^/]+\/[^/]+(?=\/|$)/, '/api/messages/#id/#id'], [/^\/api\/convos\/[^/]+\/messages\/[^/]+(?=\/|$)/, '/api/convos/#id/messages/#id'], @@ -22,9 +30,39 @@ const PATH_NORMALIZATIONS: [RegExp, string][] = [ ], ]; -const STATIC_PATHS = new Set(['/', '/health', '/metrics', '/api/auth/login', '/api/config']); +const STATIC_PATHS = new Set([ + '/', + '/health', + '/metrics', + '/api/auth/login', + '/api/config', + '/api/agents/chat/abort', + '/api/agents/chat/active', + '/api/agents/v1/chat/completions', + '/api/agents/v1/responses', + '/api/files', + '/api/files/config', + '/api/files/images', + '/api/files/images/avatar', + '/api/files/speech/stt', +]); + +const UPLOAD_PATHS = new Set([ + '/api/files', + '/api/files/images', + '/api/files/images/avatar', + '/api/files/speech/stt', + '/api/skills/#id/files', +]); + +const UPLOAD_METHODS = new Set(['POST', 'PUT', 'PATCH']); const LOW_CARDINALITY_PATHS: RegExp[] = [ + /^\/api\/agents\/chat\/stream\/#id$/, + /^\/api\/agents\/chat\/status\/#id$/, + /^\/api\/files\/#id\/preview$/, + /^\/api\/files\/(code\/download|download-url|download)\/#id\/#id$/, + /^\/api\/skills\/#id\/files$/, /^\/api\/messages\/#id$/, /^\/api\/messages\/#id\/#id$/, /^\/api\/messages\/artifact\/#id$/, @@ -89,7 +127,215 @@ export interface PrometheusMetrics { metricsRouter: Router; } +export type OpenIDUserLookupResult = 'found' | 'not_found' | 'migration' | 'auth_failed' | 'error'; +export type GenerationJobStore = 'memory' | 'redis'; +export type GenerationJobResult = 'created' | 'completed' | 'error' | 'aborted' | 'abort_failed'; +export type GenerationStreamSubscriptionType = 'initial' | 'resume' | 'resume_state'; +export type GenerationStreamSubscriptionResult = + | 'success' + | 'not_found' + | 'error' + | 'found' + | 'missing'; + +type OpenIDUserLookupMetrics = { + recordLookup: (result: OpenIDUserLookupResult, durationSeconds: number) => void; +}; + +let openIDUserLookupMetrics: OpenIDUserLookupMetrics = { + recordLookup: () => undefined, +}; + +export function recordOpenIDUserLookup( + result: OpenIDUserLookupResult, + durationSeconds: number, +): void { + openIDUserLookupMetrics.recordLookup(result, durationSeconds); +} + +type MongooseQueryMetrics = { + recordQuery: (model: string, operation: string, status: string, durationSeconds: number) => void; +}; + +let mongooseQueryMetrics: MongooseQueryMetrics = { + recordQuery: () => undefined, +}; + +type GenerationJobMetrics = { + recordJob: (store: GenerationJobStore, result: GenerationJobResult) => void; + setJobsInFlight: (store: GenerationJobStore, count: number) => void; + recordSubscription: ( + store: GenerationJobStore, + type: GenerationStreamSubscriptionType, + result: GenerationStreamSubscriptionResult, + ) => void; + recordResumePendingEvents: (store: GenerationJobStore, count: number) => void; +}; + +let generationJobMetrics: GenerationJobMetrics = { + recordJob: () => undefined, + setJobsInFlight: () => undefined, + recordSubscription: () => undefined, + recordResumePendingEvents: () => undefined, +}; + +const resetMetricRecorders = (): void => { + openIDUserLookupMetrics = { + recordLookup: () => undefined, + }; + mongooseQueryMetrics = { + recordQuery: () => undefined, + }; + generationJobMetrics = { + recordJob: () => undefined, + setJobsInFlight: () => undefined, + recordSubscription: () => undefined, + recordResumePendingEvents: () => undefined, + }; +}; + +export function recordGenerationJob(store: GenerationJobStore, result: GenerationJobResult): void { + generationJobMetrics.recordJob(store, result); +} + +export function setGenerationJobsInFlight(store: GenerationJobStore, count: number): void { + generationJobMetrics.setJobsInFlight(store, count); +} + +export function recordGenerationStreamSubscription( + store: GenerationJobStore, + type: GenerationStreamSubscriptionType, + result: GenerationStreamSubscriptionResult, +): void { + generationJobMetrics.recordSubscription(store, type, result); +} + +export function recordGenerationStreamResumePendingEvents( + store: GenerationJobStore, + count: number, +): void { + generationJobMetrics.recordResumePendingEvents(store, count); +} + +const getElapsedSeconds = (startedAt: bigint): number => + Number(process.hrtime.bigint() - startedAt) / 1_000_000_000; + +export const isMetricsConfigured = (): boolean => Boolean(process.env.METRICS_SECRET); + +const createUnauthorizedMetricsRouter = (): Router => { + const metricsRouter = Router(); + metricsRouter.get('/', (_req, res) => { + res.status(401).end(); + }); + return metricsRouter; +}; + +const normalizeMongooseLabel = (value: unknown): string => { + if (typeof value !== 'string' || !value) return 'unknown'; + return value.replace(/[^a-zA-Z0-9_:-]/g, '_').slice(0, 64) || 'unknown'; +}; + +const getHeader = (headers: Record, name: string): unknown => { + const lowerName = name.toLowerCase(); + for (const [key, value] of Object.entries(headers)) { + if (key.toLowerCase() === lowerName) return value; + } + return undefined; +}; + +const headerIncludes = (value: unknown, expected: string): boolean => { + if (Array.isArray(value)) { + return value.some((entry) => headerIncludes(entry, expected)); + } + return typeof value === 'string' && value.toLowerCase().includes(expected); +}; + +const isEventStreamContentType = (value: unknown): boolean => + headerIncludes(value, 'text/event-stream'); + +const isMultipartContentType = (value: unknown): boolean => + headerIncludes(value, 'multipart/form-data'); + +const getRequestContentLength = (req: Request): number | null => { + const contentLength = req.headers['content-length']; + const rawContentLength = Array.isArray(contentLength) ? contentLength[0] : contentLength; + const bodyBytes = rawContentLength == null ? NaN : Number(rawContentLength); + return Number.isFinite(bodyBytes) && bodyBytes >= 0 ? bodyBytes : null; +}; + +const isUploadRequest = (req: Request, normalizedPath: string): boolean => { + if (!UPLOAD_METHODS.has(req.method)) return false; + if (isMultipartContentType(req.headers['content-type'])) return true; + return UPLOAD_PATHS.has(normalizedPath); +}; + +export function recordMongooseQuery( + model: string, + operation: string, + status: string, + durationSeconds: number, +): void { + mongooseQueryMetrics.recordQuery( + normalizeMongooseLabel(model), + normalizeMongooseLabel(operation), + normalizeMongooseLabel(status), + durationSeconds, + ); +} + +export function instrumentMongooseQueryMetrics(mongoose: Mongoose): void { + if (!isMetricsConfigured()) return; + + const instrumented = Symbol.for('librechat.mongooseQueryMetrics.instrumented'); + const queryPrototype = mongoose.Query?.prototype as + | (typeof mongoose.Query.prototype & { [instrumented]?: boolean }) + | undefined; + + if (!queryPrototype || queryPrototype[instrumented]) return; + + const originalExec = queryPrototype.exec; + queryPrototype.exec = function instrumentedExec( + this: typeof queryPrototype & { + model?: { modelName?: string }; + op?: string; + }, + ...args: Parameters + ) { + const startedAt = process.hrtime.bigint(); + const model = normalizeMongooseLabel(this.model?.modelName); + const operation = normalizeMongooseLabel(this.op); + + let result: ReturnType; + try { + result = originalExec.apply(this, args); + } catch (error) { + recordMongooseQuery(model, operation, 'error', getElapsedSeconds(startedAt)); + throw error; + } + + return Promise.resolve(result).then( + (result) => { + recordMongooseQuery(model, operation, 'success', getElapsedSeconds(startedAt)); + return result; + }, + (error) => { + recordMongooseQuery(model, operation, 'error', getElapsedSeconds(startedAt)); + throw error; + }, + ); + } as typeof originalExec; + queryPrototype[instrumented] = true; +} + export function createMetrics(): PrometheusMetrics { + if (!isMetricsConfigured()) { + resetMetricRecorders(); + return { + metricsMiddleware: (_req: Request, _res: Response, next: NextFunction) => next(), + metricsRouter: createUnauthorizedMetricsRouter(), + }; + } + const registry = new Registry(); collectDefaultMetrics({ register: registry }); @@ -108,13 +354,238 @@ export function createMetrics(): PrometheusMetrics { registers: [registry], }); + const httpRequestsInFlight = new Gauge({ + name: 'http_requests_in_flight', + help: 'HTTP requests currently being handled', + labelNames: ['method', 'path'] as const, + registers: [registry], + }); + + const httpRequestBodyBytes = new Histogram({ + name: 'http_request_body_bytes', + help: 'HTTP request body size in bytes from the Content-Length header', + labelNames: ['method', 'path'] as const, + buckets: [1_000, 10_000, 100_000, 1_000_000, 5_000_000, 10_000_000, 25_000_000, 50_000_000], + registers: [registry], + }); + + const sseStreams = new Counter({ + name: 'sse_streams_total', + help: 'Total SSE streams opened', + labelNames: ['method', 'path', 'status'] as const, + registers: [registry], + }); + + const sseStreamsInFlight = new Gauge({ + name: 'sse_streams_in_flight', + help: 'SSE streams currently open', + labelNames: ['method', 'path'] as const, + registers: [registry], + }); + + const sseStreamDuration = new Histogram({ + name: 'sse_stream_duration_seconds', + help: 'SSE stream open duration in seconds', + labelNames: ['method', 'path', 'status'] as const, + buckets: [1, 5, 10, 30, 60, 120, 300, 600, 1_200, 1_800], + registers: [registry], + }); + + const uploadRequests = new Counter({ + name: 'upload_requests_total', + help: 'Total upload requests', + labelNames: ['method', 'path', 'status'] as const, + registers: [registry], + }); + + const uploadRequestsInFlight = new Gauge({ + name: 'upload_requests_in_flight', + help: 'Upload requests currently being handled', + labelNames: ['method', 'path'] as const, + registers: [registry], + }); + + const uploadRequestDuration = new Histogram({ + name: 'upload_request_duration_seconds', + help: 'Upload request duration in seconds', + labelNames: ['method', 'path', 'status'] as const, + buckets: [0.1, 0.3, 0.5, 1, 2, 5, 10, 30, 60, 120, 300], + registers: [registry], + }); + + const uploadBytes = new Counter({ + name: 'upload_bytes_total', + help: 'Upload request bytes from the Content-Length header', + labelNames: ['method', 'path'] as const, + registers: [registry], + }); + + const openIDUserLookupTotal = new Counter({ + name: 'openid_user_lookup_total', + help: 'OpenID user lookup attempts', + labelNames: ['result'] as const, + registers: [registry], + }); + + const openIDUserLookupDuration = new Histogram({ + name: 'openid_user_lookup_duration_seconds', + help: 'OpenID user lookup latency in seconds', + labelNames: ['result'] as const, + buckets: [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60], + registers: [registry], + }); + + openIDUserLookupMetrics = { + recordLookup: (result, durationSeconds) => { + openIDUserLookupTotal.inc({ result }); + openIDUserLookupDuration.observe({ result }, durationSeconds); + }, + }; + + const mongooseQueries = new Counter({ + name: 'mongoose_queries_total', + help: 'Mongoose queries by model, operation, and status', + labelNames: ['model', 'operation', 'status'] as const, + registers: [registry], + }); + + const mongooseQueryDuration = new Histogram({ + name: 'mongoose_query_duration_seconds', + help: 'Mongoose query duration in seconds by model, operation, and status', + labelNames: ['model', 'operation', 'status'] as const, + buckets: [0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60], + registers: [registry], + }); + + mongooseQueryMetrics = { + recordQuery: (model, operation, status, durationSeconds) => { + const labels = { model, operation, status }; + mongooseQueries.inc(labels); + mongooseQueryDuration.observe(labels, durationSeconds); + }, + }; + + const generationJobs = new Counter({ + name: 'generation_jobs_total', + help: 'Generation jobs by backing store and result', + labelNames: ['store', 'result'] as const, + registers: [registry], + }); + + const generationJobsInFlight = new Gauge({ + name: 'generation_jobs_in_flight', + help: 'Generation jobs currently running in this process', + labelNames: ['store'] as const, + registers: [registry], + }); + + const generationStreamSubscriptions = new Counter({ + name: 'generation_stream_subscriptions_total', + help: 'Generation stream subscription attempts by backing store, type, and result', + labelNames: ['store', 'type', 'result'] as const, + registers: [registry], + }); + + const generationStreamResumePendingEvents = new Counter({ + name: 'generation_stream_resume_pending_events_total', + help: 'Pending events delivered while resuming generation streams', + labelNames: ['store'] as const, + registers: [registry], + }); + + generationJobMetrics = { + recordJob: (store, result) => generationJobs.inc({ store, result }), + setJobsInFlight: (store, count) => generationJobsInFlight.set({ store }, count), + recordSubscription: (store, type, result) => + generationStreamSubscriptions.inc({ store, type, result }), + recordResumePendingEvents: (store, count) => + generationStreamResumePendingEvents.inc({ store }, count), + }; + const metricsMiddleware = (req: Request, res: Response, next: NextFunction): void => { const end = httpDuration.startTimer(); - res.on('finish', () => { - const labels = { method: req.method, path: normalizePath(req.path), status: res.statusCode }; - httpRequests.inc(labels); - end(labels); - }); + const labels = { method: req.method, path: normalizePath(req.path) }; + const uploadTracked = isUploadRequest(req, labels.path); + const uploadStartedAt = uploadTracked ? process.hrtime.bigint() : null; + let sseTracked = false; + let sseStartedAt: bigint | null = null; + let completed = false; + + const markSSEStream = () => { + if (completed || res.writableEnded || res.destroyed) return; + if (sseTracked) return; + sseTracked = true; + sseStartedAt = process.hrtime.bigint(); + sseStreamsInFlight.inc(labels); + }; + + const originalSetHeader = res.setHeader; + res.setHeader = function setHeader(this: Response, ...args: Parameters) { + const [name, value] = args; + const result = originalSetHeader.apply(this, args); + if (String(name).toLowerCase() === 'content-type' && isEventStreamContentType(value)) { + markSSEStream(); + } + return result; + } as Response['setHeader']; + + const originalWriteHead = res.writeHead; + res.writeHead = function writeHead(this: Response, ...args: [number, unknown?, unknown?]) { + const [, reasonPhrase, headers] = args; + const responseHeaders = + typeof reasonPhrase === 'object' && reasonPhrase != null ? reasonPhrase : headers; + if ( + isEventStreamContentType( + getHeader((responseHeaders ?? {}) as Record, 'content-type'), + ) || + isEventStreamContentType(res.getHeader('content-type')) + ) { + markSSEStream(); + } + return originalWriteHead.apply(this, args as Parameters); + } as Response['writeHead']; + + httpRequestsInFlight.inc(labels); + if (uploadTracked) { + uploadRequestsInFlight.inc(labels); + } + + const complete = (completedBy: 'finish' | 'close') => { + if (completed) return; + completed = true; + + const requestLabels = { ...labels, status: completedBy === 'close' ? 499 : res.statusCode }; + httpRequests.inc(requestLabels); + end(requestLabels); + httpRequestsInFlight.dec(labels); + + const bodyBytes = getRequestContentLength(req); + if (bodyBytes != null) { + httpRequestBodyBytes.observe(labels, bodyBytes); + } + + if (sseTracked) { + sseStreams.inc(requestLabels); + sseStreamsInFlight.dec(labels); + if (sseStartedAt) { + sseStreamDuration.observe(requestLabels, getElapsedSeconds(sseStartedAt)); + } + } + + if (uploadTracked) { + uploadRequests.inc(requestLabels); + uploadRequestsInFlight.dec(labels); + if (uploadStartedAt) { + uploadRequestDuration.observe(requestLabels, getElapsedSeconds(uploadStartedAt)); + } + if (bodyBytes != null) { + uploadBytes.inc(labels, bodyBytes); + } + } + }; + + res.once('finish', () => complete('finish')); + res.once('close', () => complete('close')); next(); }; diff --git a/packages/api/src/auth/openid.spec.ts b/packages/api/src/auth/openid.spec.ts index 975e736b72..543fa94fe3 100644 --- a/packages/api/src/auth/openid.spec.ts +++ b/packages/api/src/auth/openid.spec.ts @@ -5,6 +5,7 @@ import { ErrorTypes } from 'librechat-data-provider'; import type { IUser, UserMethods } from '@librechat/data-schemas'; import type { CommandStartedEvent } from 'mongodb'; import type { FilterQuery } from 'mongoose'; +import { recordOpenIDUserLookup } from '~/app/metrics'; import { findOpenIDUser, getOpenIdEmail, getOpenIdIssuer, normalizeOpenIdIssuer } from './openid'; function newId() { @@ -19,6 +20,11 @@ jest.mock('@librechat/data-schemas', () => ({ }, })); +jest.mock('~/app/metrics', () => ({ + isMetricsConfigured: jest.fn(() => true), + recordOpenIDUserLookup: jest.fn(), +})); + describe('normalizeOpenIdIssuer', () => { it('normalizes blank, trailing-slash, and discovery-document issuers', () => { expect(normalizeOpenIdIssuer('')).toBeUndefined(); @@ -111,6 +117,7 @@ describe('findOpenIDUser', () => { error: null, migration: false, }); + expect(recordOpenIDUserLookup).toHaveBeenCalledWith('found', expect.any(Number)); }); it('should find user by idOnTheSource', async () => { @@ -734,6 +741,7 @@ describe('findOpenIDUser', () => { findUser: mockFindUser, }), ).rejects.toThrow('Database error'); + expect(recordOpenIDUserLookup).toHaveBeenCalledWith('error', expect.any(Number)); }); it('should reject email fallback when openidId is empty and user has a stored openidId', async () => { diff --git a/packages/api/src/auth/openid.ts b/packages/api/src/auth/openid.ts index 8ee5cbd517..4d346a7591 100644 --- a/packages/api/src/auth/openid.ts +++ b/packages/api/src/auth/openid.ts @@ -2,6 +2,8 @@ import { logger } from '@librechat/data-schemas'; import { ErrorTypes } from 'librechat-data-provider'; import type { IUser, UserMethods } from '@librechat/data-schemas'; import type { FilterQuery } from 'mongoose'; +import { isMetricsConfigured, recordOpenIDUserLookup } from '~/app/metrics'; +import type { OpenIDUserLookupResult } from '~/app/metrics'; export type OpenIdEmailClaims = { email?: unknown; @@ -65,6 +67,17 @@ function hasOpenIdLookupValue(value: string | undefined): value is string { return typeof value === 'string' && value.length > 0; } +function getElapsedSeconds(startedAt: bigint): number { + return Number(process.hrtime.bigint() - startedAt) / 1_000_000_000; +} + +function getOpenIDUserLookupResult(resolution: OpenIdUserResolution): OpenIDUserLookupResult { + if (resolution.error) return 'auth_failed'; + if (resolution.migration) return 'migration'; + if (resolution.user) return 'found'; + return 'not_found'; +} + function getIssuerExactCondition( field: OpenIdLookupField, value: string | undefined, @@ -208,61 +221,79 @@ export async function findOpenIDUser({ idOnTheSource?: string; strategyName?: string; }): Promise { - const normalizedIssuer = normalizeOpenIdIssuer(openidIssuer); - const primaryConditions = getPrimaryLookupConditions(openidId, idOnTheSource, normalizedIssuer); - - let user: IUser | null = null; - if (primaryConditions.length > 0) { - user = await findFirstOpenIdUser(findUser, primaryConditions); - } - - const primaryIssuerResolution = resolveIssuerBoundUser( - user, - normalizedIssuer, - strategyName, - 'OpenID lookup', - ); - if (primaryIssuerResolution) return primaryIssuerResolution; - - if (!user && email) { - user = await findUser({ email }); - logger.warn( - `[${strategyName}] user ${user ? 'found' : 'not found'} with email: ${email} for openidId: ${openidId}`, - ); - - // If user found by email, check if they're allowed to use OpenID provider - if (user && user.provider && user.provider !== 'openid') { - logger.warn( - `[${strategyName}] Attempted OpenID login by user ${user.email}, was registered with "${user.provider}" provider`, + const lookupStartedAt = isMetricsConfigured() ? process.hrtime.bigint() : null; + const finish = (resolution: OpenIdUserResolution): OpenIdUserResolution => { + if (lookupStartedAt != null) { + recordOpenIDUserLookup( + getOpenIDUserLookupResult(resolution), + getElapsedSeconds(lookupStartedAt), ); - return { user: null, error: ErrorTypes.AUTH_FAILED, migration: false }; + } + return resolution; + }; + + try { + const normalizedIssuer = normalizeOpenIdIssuer(openidIssuer); + const primaryConditions = getPrimaryLookupConditions(openidId, idOnTheSource, normalizedIssuer); + + let user: IUser | null = null; + if (primaryConditions.length > 0) { + user = await findFirstOpenIdUser(findUser, primaryConditions); } - if (user?.openidId && user.openidId !== openidId) { - logger.warn( - `[${strategyName}] Rejected email fallback for ${user.email}: stored openidId does not match token sub`, - ); - return { user: null, error: ErrorTypes.AUTH_FAILED, migration: false }; - } - - const emailIssuerResolution = resolveIssuerBoundUser( + const primaryIssuerResolution = resolveIssuerBoundUser( user, normalizedIssuer, strategyName, - 'email fallback', + 'OpenID lookup', ); - if (emailIssuerResolution) return emailIssuerResolution; + if (primaryIssuerResolution) return finish(primaryIssuerResolution); - if (user && !user.openidId) { - logger.info( - `[${strategyName}] Preparing user ${user.email} for migration to OpenID with sub: ${openidId}`, + if (!user && email) { + user = await findUser({ email }); + logger.warn( + `[${strategyName}] user ${user ? 'found' : 'not found'} with email: ${email} for openidId: ${openidId}`, ); - user.provider = 'openid'; - user.openidId = openidId; - if (normalizedIssuer) user.openidIssuer = normalizedIssuer; - return { user, error: null, migration: true }; - } - } - return { user, error: null, migration: false }; + // If user found by email, check if they're allowed to use OpenID provider + if (user && user.provider && user.provider !== 'openid') { + logger.warn( + `[${strategyName}] Attempted OpenID login by user ${user.email}, was registered with "${user.provider}" provider`, + ); + return finish({ user: null, error: ErrorTypes.AUTH_FAILED, migration: false }); + } + + if (user?.openidId && user.openidId !== openidId) { + logger.warn( + `[${strategyName}] Rejected email fallback for ${user.email}: stored openidId does not match token sub`, + ); + return finish({ user: null, error: ErrorTypes.AUTH_FAILED, migration: false }); + } + + const emailIssuerResolution = resolveIssuerBoundUser( + user, + normalizedIssuer, + strategyName, + 'email fallback', + ); + if (emailIssuerResolution) return finish(emailIssuerResolution); + + if (user && !user.openidId) { + logger.info( + `[${strategyName}] Preparing user ${user.email} for migration to OpenID with sub: ${openidId}`, + ); + user.provider = 'openid'; + user.openidId = openidId; + if (normalizedIssuer) user.openidIssuer = normalizedIssuer; + return finish({ user, error: null, migration: true }); + } + } + + return finish({ user, error: null, migration: false }); + } catch (error) { + if (lookupStartedAt != null) { + recordOpenIDUserLookup('error', getElapsedSeconds(lookupStartedAt)); + } + throw error; + } } diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index de6232e85b..366445bb26 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/packages/api/src/stream/GenerationJobManager.ts @@ -10,6 +10,13 @@ import type { IJobStore, } from './interfaces/IJobStore'; import type * as t from '~/types'; +import { + recordGenerationJob, + recordGenerationStreamResumePendingEvents, + recordGenerationStreamSubscription, + setGenerationJobsInFlight, +} from '~/app/metrics'; +import type { GenerationJobStore } from '~/app/metrics'; import { InMemoryEventTransport } from './implementations/InMemoryEventTransport'; import { InMemoryJobStore } from './implementations/InMemoryJobStore'; @@ -87,6 +94,9 @@ class GenerationJobManagerClass { /** Runtime state - always in-memory, not serializable */ private runtimeState = new Map(); + /** Jobs actively generating in this process. */ + private runningJobs = new Set(); + private cleanupInterval: NodeJS.Timeout | null = null; /** Whether we're using Redis stores */ @@ -144,6 +154,7 @@ class GenerationJobManagerClass { isRedis?: boolean; cleanupOnComplete?: boolean; }): void { + const previousStore = this.storeLabel; if (this.cleanupInterval) { logger.warn( '[GenerationJobManager] Reconfiguring after initialization - destroying existing services', @@ -151,10 +162,14 @@ class GenerationJobManagerClass { this.destroy(); } + this.runningJobs.clear(); + setGenerationJobsInFlight(previousStore, 0); + this.jobStore = services.jobStore; this.eventTransport = services.eventTransport; this._isRedis = services.isRedis ?? false; this._cleanupOnComplete = services.cleanupOnComplete ?? true; + this.syncRunningJobMetrics(); logger.info( `[GenerationJobManager] Configured with ${this._isRedis ? 'Redis' : 'in-memory'} stores`, @@ -168,6 +183,14 @@ class GenerationJobManagerClass { return this._isRedis; } + private get storeLabel(): GenerationJobStore { + return this._isRedis ? 'redis' : 'memory'; + } + + private syncRunningJobMetrics(store: GenerationJobStore = this.storeLabel): void { + setGenerationJobsInFlight(store, this.runningJobs.size); + } + /** * Get the job store instance (for advanced use cases). */ @@ -226,6 +249,9 @@ class GenerationJobManagerClass { hasSubscriber: false, }; this.runtimeState.set(streamId, runtime); + this.runningJobs.add(streamId); + this.syncRunningJobMetrics(); + recordGenerationJob(this.storeLabel, 'created'); // Resolve immediately - early event buffer handles late subscribers resolveReady!(); @@ -554,6 +580,9 @@ class GenerationJobManagerClass { completedAt: Date.now(), error, }); + this.runningJobs.delete(streamId); + this.syncRunningJobMetrics(); + recordGenerationJob(this.storeLabel, 'error'); // Keep runtime state so subscribe() can access errorEvent logger.debug( `[GenerationJobManager] Job completed with error (keeping for late subscribers): ${streamId}`, @@ -575,6 +604,9 @@ class GenerationJobManagerClass { }); } + this.runningJobs.delete(streamId); + this.syncRunningJobMetrics(); + recordGenerationJob(this.storeLabel, 'completed'); logger.debug(`[GenerationJobManager] Job completed: ${streamId}`); } @@ -592,6 +624,7 @@ class GenerationJobManagerClass { if (!jobData) { logger.warn(`[GenerationJobManager] Cannot abort - job not found: ${streamId}`); + recordGenerationJob(this.storeLabel, 'abort_failed'); return { text: '', content: [], @@ -682,6 +715,9 @@ class GenerationJobManagerClass { }); } + this.runningJobs.delete(streamId); + this.syncRunningJobMetrics(); + recordGenerationJob(this.storeLabel, 'aborted'); logger.debug(`[GenerationJobManager] Job aborted: ${streamId}`); return { @@ -723,9 +759,11 @@ class GenerationJobManagerClass { onError?: t.ErrorHandler, options?: t.SubscribeOptions, ): Promise<{ unsubscribe: t.UnsubscribeFn } | null> { + const subscriptionType = options?.skipBufferReplay ? 'resume' : 'initial'; // Use lazy initialization to support cross-replica subscriptions const runtime = await this.getOrCreateRuntimeState(streamId); if (!runtime) { + recordGenerationStreamSubscription(this.storeLabel, subscriptionType, 'not_found'); return null; } @@ -761,8 +799,14 @@ class GenerationJobManagerClass { onError, }); - if (subscription.ready) { - await subscription.ready; + try { + if (subscription.ready) { + await subscription.ready; + } + recordGenerationStreamSubscription(this.storeLabel, subscriptionType, 'success'); + } catch (err) { + recordGenerationStreamSubscription(this.storeLabel, subscriptionType, 'error'); + throw err; } const isFirst = this.eventTransport.isFirstSubscriber(streamId); @@ -864,6 +908,11 @@ class GenerationJobManagerClass { : 0; const resumeState = await this.getResumeState(streamId); + recordGenerationStreamSubscription( + this.storeLabel, + 'resume_state', + resumeState ? 'found' : 'missing', + ); let pendingEvents: t.ServerSentEvent[] = []; if (!this._isRedis) { @@ -872,6 +921,7 @@ class GenerationJobManagerClass { pendingEvents = runtime.earlyEventBuffer.slice(bufferLengthAtSnapshot); runtime.earlyEventBuffer = []; if (pendingEvents.length > 0) { + recordGenerationStreamResumePendingEvents(this.storeLabel, pendingEvents.length); logger.debug( `[GenerationJobManager] Captured ${pendingEvents.length} gap events for ${streamId}`, ); @@ -1179,11 +1229,13 @@ class GenerationJobManagerClass { */ private async cleanup(): Promise { const count = await this.jobStore.cleanup(); + let runningJobsChanged = false; // Cleanup runtime state for deleted jobs for (const streamId of this.runtimeState.keys()) { if (!(await this.jobStore.hasJob(streamId))) { this.runtimeState.delete(streamId); + runningJobsChanged = this.runningJobs.delete(streamId) || runningJobsChanged; this.runStepBuffers?.delete(streamId); this.jobStore.clearContentState(streamId); this.eventTransport.cleanup(streamId); @@ -1207,6 +1259,10 @@ class GenerationJobManagerClass { } } + if (runningJobsChanged) { + this.syncRunningJobMetrics(); + } + if (count > 0) { logger.debug(`[GenerationJobManager] Cleaned up ${count} expired jobs`); } @@ -1295,6 +1351,8 @@ class GenerationJobManagerClass { await this.jobStore.destroy(); this.eventTransport.destroy(); this.runtimeState.clear(); + this.runningJobs.clear(); + this.syncRunningJobMetrics(); this.runStepBuffers?.clear(); logger.debug('[GenerationJobManager] Destroyed');