mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-10-20 17:13:44 +00:00
⚖️ feat: Add Operational Prometheus Metrics (#13265)
* add operational prometheus metrics * fix metrics review findings * fix metrics edge cases * gate mongoose metrics instrumentation * gate metrics setup when unconfigured
This commit is contained in:
Danny Avila
GitHub
parent
bd64251eb9
commit
2ce3921f78
7 changed files with 967 additions and 59 deletions
|
|
@ -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');
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
});
|
||||
|
|
|
|||
|
|
@ -1,5 +1,19 @@
|
|||
/// <reference types="jest" />
|
||||
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<string, unknown>();
|
||||
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/,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>, 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<typeof originalExec>
|
||||
) {
|
||||
const startedAt = process.hrtime.bigint();
|
||||
const model = normalizeMongooseLabel(this.model?.modelName);
|
||||
const operation = normalizeMongooseLabel(this.op);
|
||||
|
||||
let result: ReturnType<typeof originalExec>;
|
||||
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<Response['setHeader']>) {
|
||||
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<string, unknown>, 'content-type'),
|
||||
) ||
|
||||
isEventStreamContentType(res.getHeader('content-type'))
|
||||
) {
|
||||
markSSEStream();
|
||||
}
|
||||
return originalWriteHead.apply(this, args as Parameters<typeof originalWriteHead>);
|
||||
} 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();
|
||||
};
|
||||
|
||||
|
|
|
|||
|
|
@ -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 () => {
|
||||
|
|
|
|||
|
|
@ -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<OpenIdUserResolution> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<string, RuntimeJobState>();
|
||||
|
||||
/** Jobs actively generating in this process. */
|
||||
private runningJobs = new Set<string>();
|
||||
|
||||
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<void> {
|
||||
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');
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue