diff --git a/packages/api/src/app/metrics.spec.ts b/packages/api/src/app/metrics.spec.ts index ebb1267504..7714330f2f 100644 --- a/packages/api/src/app/metrics.spec.ts +++ b/packages/api/src/app/metrics.spec.ts @@ -11,6 +11,7 @@ import { recordGenerationStreamResumePendingEvents, recordGenerationStreamSubscription, recordOpenIDUserLookup, + recordRedisOperation, recordRumProxyRequest, setGenerationJobsInFlight, } from './metrics'; @@ -272,6 +273,31 @@ describe('createMetrics', () => { expect(response.text).toMatch(/openid_user_lookup_duration_seconds_sum\{result="found"\} 0.2/); }); + it('tracks Redis operation outcomes and latency by use case', async () => { + const app = express(); + process.env.METRICS_SECRET = 'test-secret'; + const { metricsRouter } = createMetrics(); + app.use('/metrics', metricsRouter); + + recordRedisOperation('keyv', 'auth_user_doc', 'get', 'success', 0.02); + recordRedisOperation('ioredis', 'rate_limit', 'eval', 'error', 0.05); + + const response = await request(app) + .get('/metrics') + .set('Authorization', 'Bearer test-secret') + .expect(200); + + expect(response.text).toMatch( + /redis_operations_total\{client="keyv",use_case="auth_user_doc",operation="get",status="success"\} 1/, + ); + expect(response.text).toMatch( + /redis_operation_duration_seconds_count\{client="ioredis",use_case="rate_limit",operation="eval",status="error"\} 1/, + ); + expect(response.text).toMatch( + /redis_operation_duration_seconds_sum\{client="ioredis",use_case="rate_limit",operation="eval",status="error"\} 0.05/, + ); + }); + it('tracks RUM proxy request outcomes', async () => { const app = express(); process.env.METRICS_SECRET = 'test-secret'; diff --git a/packages/api/src/app/metrics.ts b/packages/api/src/app/metrics.ts index da2ce635d3..b10556cba8 100644 --- a/packages/api/src/app/metrics.ts +++ b/packages/api/src/app/metrics.ts @@ -148,6 +148,8 @@ export type RumProxyResult = | 'collector_5xx' | 'collector_error' | 'collector_timeout'; +export type RedisClient = 'ioredis' | 'keyv'; +export type RedisOperationStatus = 'success' | 'error'; type OpenIDUserLookupMetrics = { recordLookup: (result: OpenIDUserLookupResult, durationSeconds: number) => void; @@ -198,6 +200,20 @@ let rumProxyMetrics: RumProxyMetrics = { recordRequest: () => undefined, }; +type RedisOperationMetrics = { + recordOperation: ( + client: RedisClient, + useCase: string, + operation: string, + status: RedisOperationStatus, + durationSeconds: number, + ) => void; +}; + +let redisOperationMetrics: RedisOperationMetrics = { + recordOperation: () => undefined, +}; + const resetMetricRecorders = (): void => { openIDUserLookupMetrics = { recordLookup: () => undefined, @@ -214,6 +230,9 @@ const resetMetricRecorders = (): void => { rumProxyMetrics = { recordRequest: () => undefined, }; + redisOperationMetrics = { + recordOperation: () => undefined, + }; }; export function recordGenerationJob(store: GenerationJobStore, result: GenerationJobResult): void { @@ -243,6 +262,16 @@ export function recordRumProxyRequest(endpoint: RumProxyEndpoint, result: RumPro rumProxyMetrics.recordRequest(endpoint, result); } +export function recordRedisOperation( + client: RedisClient, + useCase: string, + operation: string, + status: RedisOperationStatus, + durationSeconds: number, +): void { + redisOperationMetrics.recordOperation(client, useCase, operation, status, durationSeconds); +} + const getElapsedSeconds = (startedAt: bigint): number => Number(process.hrtime.bigint() - startedAt) / 1_000_000_000; @@ -526,6 +555,21 @@ export function createMetrics(): PrometheusMetrics { registers: [registry], }); + const redisOperations = new Counter({ + name: 'redis_operations_total', + help: 'Logical Redis operations by client, use case, operation, and status', + labelNames: ['client', 'use_case', 'operation', 'status'] as const, + registers: [registry], + }); + + const redisOperationDuration = new Histogram({ + name: 'redis_operation_duration_seconds', + help: 'Logical Redis operation latency in seconds', + labelNames: ['client', 'use_case', 'operation', 'status'] as const, + buckets: [0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5], + registers: [registry], + }); + generationJobMetrics = { recordJob: (store, result) => generationJobs.inc({ store, result }), setJobsInFlight: (store, count) => generationJobsInFlight.set({ store }, count), @@ -539,6 +583,14 @@ export function createMetrics(): PrometheusMetrics { recordRequest: (endpoint, result) => rumProxyRequests.inc({ endpoint, result }), }; + redisOperationMetrics = { + recordOperation: (client, useCase, operation, status, durationSeconds) => { + const labels = { client, use_case: useCase, operation, status }; + redisOperations.inc(labels); + redisOperationDuration.observe(labels, durationSeconds); + }, + }; + const metricsMiddleware = (req: Request, res: Response, next: NextFunction): void => { const end = httpDuration.startTimer(); const labels = { method: req.method, path: normalizePath(req.path) }; diff --git a/packages/api/src/cache/cacheFactory.ts b/packages/api/src/cache/cacheFactory.ts index 5c5db865db..68f7215c4d 100644 --- a/packages/api/src/cache/cacheFactory.ts +++ b/packages/api/src/cache/cacheFactory.ts @@ -16,6 +16,12 @@ import { RedisStore as ConnectRedis } from 'connect-redis'; import type { SendCommandFn } from 'rate-limit-redis'; import { keyvRedisClient, ioredisClient } from './redisClients'; import { batchDeleteKeys, scanKeys } from './redisUtils'; +import { + instrumentIORedisClient, + instrumentRedisCache, + observeRedisOperation, + RedisUseCases, +} from './redisTelemetry'; import { cacheConfig } from './cacheConfig'; import { violationFile } from './keyvFiles'; @@ -79,7 +85,7 @@ export const standardCache = (namespace: string, ttl?: number, fallbackStore?: o logger.debug(`Cleared ${keysToDelete.length} keys from namespace ${namespace}`); }; - return cache; + return instrumentRedisCache(cache, namespace); } catch (err) { logger.error(`Failed to create Redis cache for namespace ${namespace}:`, err); throw err; @@ -124,7 +130,10 @@ export const sessionCache = (namespace: string, ttl?: number): MemoryStore | Con const MemoryStore = createMemoryStore(session); return new MemoryStore({ ttl, checkPeriod: Time.ONE_DAY }); } - const store = new ConnectRedis({ client: ioredisClient, ttl, prefix: namespace }); + const redisClient = ioredisClient + ? instrumentIORedisClient(ioredisClient, namespace) + : ioredisClient; + const store = new ConnectRedis({ client: redisClient, ttl, prefix: namespace }); if (ioredisClient) { ioredisClient.on('error', (err) => { logger.error(`Session store Redis error for namespace ${namespace}:`, err); @@ -152,11 +161,14 @@ export const limiterCache = (prefix: string): RedisStore | undefined => { try { const sendCommand: SendCommandFn = (async (...args: string[]) => { - if (ioredisClient == null) { + const redisClient = ioredisClient; + if (redisClient == null) { throw new Error('Redis client not available'); } try { - return await ioredisClient.call(args[0], ...args.slice(1)); + return await observeRedisOperation('ioredis', RedisUseCases.RATE_LIMIT, args[0], () => + redisClient.call(args[0], ...args.slice(1)), + ); } catch (err) { logger.error('Redis command execution failed:', err); throw err; diff --git a/packages/api/src/cache/index.ts b/packages/api/src/cache/index.ts index ff0046b71f..a390f87809 100644 --- a/packages/api/src/cache/index.ts +++ b/packages/api/src/cache/index.ts @@ -5,3 +5,4 @@ export { default as keyvMongo } from './keyvMongo'; export * from './cacheFactory'; export * from './principals'; export * from './redisUtils'; +export * from './redisTelemetry'; diff --git a/packages/api/src/cache/principals.ts b/packages/api/src/cache/principals.ts index 5730a3c7ed..11e9dc9411 100644 --- a/packages/api/src/cache/principals.ts +++ b/packages/api/src/cache/principals.ts @@ -2,6 +2,7 @@ import { randomUUID } from 'crypto'; import { Time, CacheKeys } from 'librechat-data-provider'; import type { Keyv } from 'keyv'; import { keyvRedisClient, ioredisClient } from './redisClients'; +import { instrumentIORedisClient } from './redisTelemetry'; import { standardCache } from './cacheFactory'; import { cacheConfig } from './cacheConfig'; import { math } from '~/utils'; @@ -42,7 +43,9 @@ export function userPrincipalsCache(): UserPrincipalsCache | undefined { } const cache: UserPrincipalsCache = standardCache(CacheKeys.USER_PRINCIPALS, cacheTtl); - const redisClient = ioredisClient; + const redisClient = ioredisClient + ? instrumentIORedisClient(ioredisClient, CacheKeys.USER_PRINCIPALS) + : ioredisClient; const isRedisBacked = keyvRedisClient != null && !cacheConfig.FORCED_IN_MEMORY_CACHE_NAMESPACES?.includes(CacheKeys.USER_PRINCIPALS); diff --git a/packages/api/src/cache/redisTelemetry.spec.ts b/packages/api/src/cache/redisTelemetry.spec.ts new file mode 100644 index 0000000000..32b746d5c0 --- /dev/null +++ b/packages/api/src/cache/redisTelemetry.spec.ts @@ -0,0 +1,196 @@ +import { CacheKeys } from 'librechat-data-provider'; +import type { Span } from '@opentelemetry/api'; +import { + createRedisRequestTelemetry, + finishRedisRequestTelemetry, + instrumentIORedisClient, + instrumentRedisCache, + normalizeRedisUseCase, + observeRedisOperation, + RedisUseCases, + runWithRedisRequestTelemetry, +} from './redisTelemetry'; +import { isMetricsConfigured, recordRedisOperation } from '~/app/metrics'; + +jest.mock('~/app/metrics', () => ({ + isMetricsConfigured: jest.fn(() => false), + recordRedisOperation: jest.fn(), +})); + +const mockIsMetricsConfigured = jest.mocked(isMetricsConfigured); +const mockRecordRedisOperation = jest.mocked(recordRedisOperation); + +function createSpan(): jest.Mocked> { + return { + setAttributes: jest.fn().mockReturnThis(), + }; +} + +describe('redisTelemetry', () => { + beforeEach(() => { + jest.clearAllMocks(); + mockIsMetricsConfigured.mockReturnValue(false); + }); + + it('normalizes only bounded cache and explicit use cases', () => { + expect(normalizeRedisUseCase(CacheKeys.AUTH_USER_DOC)).toBe('auth_user_doc'); + expect(normalizeRedisUseCase('MCP::ServersRegistry::Servers::tenant-123')).toBe( + RedisUseCases.MCP_REGISTRY, + ); + expect(normalizeRedisUseCase('violations:concurrent')).toBe(RedisUseCases.VIOLATIONS); + expect(normalizeRedisUseCase('user-controlled-namespace')).toBe('other'); + }); + + it('aggregates successful and failed operations onto the request span', async () => { + const span = createSpan(); + const telemetry = createRedisRequestTelemetry(span as unknown as Span); + + await runWithRedisRequestTelemetry(telemetry, async () => { + await expect( + observeRedisOperation('keyv', CacheKeys.AUTH_USER_DOC, 'GET', async () => 'cached'), + ).resolves.toBe('cached'); + await expect( + observeRedisOperation('ioredis', RedisUseCases.RATE_LIMIT, 'EVAL', async () => { + throw new Error('redis unavailable'); + }), + ).rejects.toThrow('redis unavailable'); + }); + + finishRedisRequestTelemetry(telemetry); + + expect(mockRecordRedisOperation).toHaveBeenNthCalledWith( + 1, + 'keyv', + 'auth_user_doc', + 'get', + 'success', + expect.any(Number), + ); + expect(mockRecordRedisOperation).toHaveBeenNthCalledWith( + 2, + 'ioredis', + 'rate_limit', + 'eval', + 'error', + expect.any(Number), + ); + + const attributes = Object.assign({}, ...span.setAttributes.mock.calls.map(([value]) => value)); + expect(attributes).toMatchObject({ + 'librechat.redis.calls': 2, + 'librechat.redis.errors': 1, + 'librechat.redis.operations': ['eval', 'get'], + 'librechat.redis.use_cases': ['auth_user_doc', 'rate_limit'], + 'librechat.redis.auth_user_doc.calls': 1, + 'librechat.redis.auth_user_doc.errors': 0, + 'librechat.redis.rate_limit.calls': 1, + 'librechat.redis.rate_limit.errors': 1, + }); + expect(attributes['librechat.redis.duration_ms']).toEqual(expect.any(Number)); + expect(attributes['librechat.redis.max_call_ms']).toEqual(expect.any(Number)); + }); + + it('does not time operations when neither metrics nor request tracing is active', async () => { + await expect( + observeRedisOperation('keyv', CacheKeys.APP_CONFIG, 'get', async () => 'value'), + ).resolves.toBe('value'); + + expect(mockRecordRedisOperation).not.toHaveBeenCalled(); + }); + + it('counts nested Keyv delegation as one logical operation', async () => { + const span = createSpan(); + const telemetry = createRedisRequestTelemetry(span as unknown as Span); + const cache = instrumentRedisCache( + { + getMany: jest.fn(async () => ['cache-value']), + async get() { + return (await this.getMany())[0]; + }, + }, + CacheKeys.TOOL_CACHE, + ); + + await runWithRedisRequestTelemetry(telemetry, async () => { + await expect(cache.get()).resolves.toBe('cache-value'); + }); + + expect(telemetry.calls).toBe(1); + expect(telemetry.operations).toEqual(new Set(['get'])); + expect(mockRecordRedisOperation).toHaveBeenCalledTimes(1); + }); + + it('records resolved ioredis pipeline command errors', async () => { + const span = createSpan(); + const telemetry = createRedisRequestTelemetry(span as unknown as Span); + const commandError = new Error('command failed'); + const pipeline = { + exec: jest.fn(async () => [[commandError, null]]), + }; + const redis = instrumentIORedisClient( + { + pipeline: jest.fn(() => pipeline), + }, + RedisUseCases.GENERATION_STREAM, + ); + + await runWithRedisRequestTelemetry(telemetry, async () => { + await expect(redis.pipeline().exec()).resolves.toEqual([[commandError, null]]); + }); + + expect(telemetry.errors).toBe(1); + expect(mockRecordRedisOperation).toHaveBeenCalledWith( + 'ioredis', + RedisUseCases.GENERATION_STREAM, + 'pipeline', + 'error', + expect.any(Number), + ); + }); + + it('preserves the ioredis client constructor', () => { + class FakeRedisClient {} + + const client = new FakeRedisClient(); + const redis = instrumentIORedisClient(client, RedisUseCases.GENERATION_STREAM); + + expect(redis.constructor).toBe(client.constructor); + expect(redis.constructor.name).toBe('FakeRedisClient'); + expect(redis).toBeInstanceOf(FakeRedisClient); + }); + + it('instruments Keyv methods and ioredis pipelines without changing their results', async () => { + const span = createSpan(); + const telemetry = createRedisRequestTelemetry(span as unknown as Span); + const cache = instrumentRedisCache( + { + get: jest.fn(async () => 'cache-value'), + }, + CacheKeys.TOOL_CACHE, + ); + const pipeline = { + get: jest.fn().mockReturnThis(), + exec: jest.fn(async () => [['ok', 'pipeline-value']]), + }; + const redis = instrumentIORedisClient( + { + get: jest.fn(async () => 'redis-value'), + on: jest.fn().mockReturnThis(), + pipeline: jest.fn(() => pipeline), + }, + RedisUseCases.GENERATION_STREAM, + ); + + await runWithRedisRequestTelemetry(telemetry, async () => { + await expect(cache.get()).resolves.toBe('cache-value'); + await expect(redis.get()).resolves.toBe('redis-value'); + await expect(redis.on().get()).resolves.toBe('redis-value'); + await expect(redis.pipeline().get('key').exec()).resolves.toEqual([['ok', 'pipeline-value']]); + }); + + expect(telemetry.calls).toBe(4); + expect(telemetry.operations).toEqual(new Set(['get', 'pipeline'])); + expect(telemetry.useCases.has('tool_cache')).toBe(true); + expect(telemetry.useCases.has(RedisUseCases.GENERATION_STREAM)).toBe(true); + }); +}); diff --git a/packages/api/src/cache/redisTelemetry.ts b/packages/api/src/cache/redisTelemetry.ts new file mode 100644 index 0000000000..d75f224773 --- /dev/null +++ b/packages/api/src/cache/redisTelemetry.ts @@ -0,0 +1,332 @@ +import { AsyncLocalStorage } from 'async_hooks'; +import { CacheKeys } from 'librechat-data-provider'; +import type { Span } from '@opentelemetry/api'; +import { + isMetricsConfigured, + recordRedisOperation, + type RedisClient, + type RedisOperationStatus, +} from '~/app/metrics'; + +const REDIS_CACHE_METHODS = [ + 'clear', + 'delete', + 'deleteMany', + 'get', + 'getMany', + 'getManyRaw', + 'getRaw', + 'has', + 'hasMany', + 'set', + 'setMany', +] as const; + +/** Keep this list aligned with direct commands used by instrumented ioredis clients. */ +const IOREDIS_COMMANDS = new Set([ + 'call', + 'del', + 'eval', + 'evalsha', + 'exists', + 'expire', + 'get', + 'hgetall', + 'incr', + 'lrange', + 'mget', + 'publish', + 'psubscribe', + 'punsubscribe', + 'sadd', + 'scan', + 'scard', + 'set', + 'smembers', + 'srem', + 'subscribe', + 'unsubscribe', + 'xack', + 'xgroup', + 'xrange', + 'xreadgroup', +]); + +const INSTRUMENTED_CACHE = Symbol('librechat.redisTelemetry.instrumentedCache'); +const MAX_DETAILED_TRACE_USE_CASES = 10; +const instrumentedClients = new WeakMap>(); + +export const RedisUseCases = { + GENERATION_STREAM: 'generation_stream', + LEADER_ELECTION: 'leader_election', + MCP_REGISTRY: 'mcp_registry', + RATE_LIMIT: 'rate_limit', + VIOLATIONS: 'violations', +} as const; + +type RedisUseCaseSummary = { + calls: number; + durationMs: number; + errors: number; + maxCallMs: number; +}; + +export interface RedisRequestTelemetry { + calls: number; + durationMs: number; + ended: boolean; + errors: number; + maxCallMs: number; + operations: Set; + span: Span; + useCases: Map; +} + +const requestTelemetry = new AsyncLocalStorage(); +const activeRedisObservation = new AsyncLocalStorage(); + +const normalizeLabel = (value: string): string => + value + .trim() + .toLowerCase() + .replace(/[^a-z0-9]+/g, '_') + .replace(/^_+|_+$/g, '') + .slice(0, 64) || 'unknown'; + +const cacheUseCases = new Set(Object.values(CacheKeys).map((value) => normalizeLabel(value))); +const explicitUseCases = new Set(Object.values(RedisUseCases)); + +export function normalizeRedisUseCase(namespace: string): string { + const normalized = normalizeLabel(namespace.replace(/:+$/, '')); + if (cacheUseCases.has(normalized) || explicitUseCases.has(normalized)) { + return normalized; + } + if (normalized.startsWith('mcp_serversregistry')) { + return RedisUseCases.MCP_REGISTRY; + } + if (normalized.startsWith('violations_')) { + return RedisUseCases.VIOLATIONS; + } + return 'other'; +} + +export function createRedisRequestTelemetry(span: Span): RedisRequestTelemetry { + return { + calls: 0, + durationMs: 0, + ended: false, + errors: 0, + maxCallMs: 0, + operations: new Set(), + span, + useCases: new Map(), + }; +} + +export function runWithRedisRequestTelemetry( + telemetry: RedisRequestTelemetry, + callback: () => T, +): T { + return requestTelemetry.run(telemetry, callback); +} + +const roundedMilliseconds = (value: number): number => Math.round(value * 1000) / 1000; + +export function finishRedisRequestTelemetry(telemetry: RedisRequestTelemetry): void { + if (telemetry.ended) { + return; + } + telemetry.ended = true; + if (telemetry.calls === 0) { + return; + } + + telemetry.span.setAttributes({ + 'librechat.redis.calls': telemetry.calls, + 'librechat.redis.duration_ms': roundedMilliseconds(telemetry.durationMs), + 'librechat.redis.errors': telemetry.errors, + 'librechat.redis.max_call_ms': roundedMilliseconds(telemetry.maxCallMs), + 'librechat.redis.operations': [...telemetry.operations].sort(), + 'librechat.redis.use_cases': [...telemetry.useCases.keys()].sort(), + }); + + const detailedUseCases = [...telemetry.useCases.entries()] + .sort(([, left], [, right]) => right.durationMs - left.durationMs) + .slice(0, MAX_DETAILED_TRACE_USE_CASES); + + for (const [useCase, summary] of detailedUseCases) { + const prefix = `librechat.redis.${useCase}`; + telemetry.span.setAttributes({ + [`${prefix}.calls`]: summary.calls, + [`${prefix}.duration_ms`]: roundedMilliseconds(summary.durationMs), + [`${prefix}.errors`]: summary.errors, + [`${prefix}.max_call_ms`]: roundedMilliseconds(summary.maxCallMs), + }); + } +} + +function addRequestObservation( + telemetry: RedisRequestTelemetry | undefined, + useCase: string, + operation: string, + status: RedisOperationStatus, + durationMs: number, +): void { + if (!telemetry || telemetry.ended) { + return; + } + + telemetry.calls += 1; + telemetry.durationMs += durationMs; + telemetry.maxCallMs = Math.max(telemetry.maxCallMs, durationMs); + telemetry.operations.add(operation); + if (status === 'error') { + telemetry.errors += 1; + } + + const summary = telemetry.useCases.get(useCase) ?? { + calls: 0, + durationMs: 0, + errors: 0, + maxCallMs: 0, + }; + summary.calls += 1; + summary.durationMs += durationMs; + summary.maxCallMs = Math.max(summary.maxCallMs, durationMs); + if (status === 'error') { + summary.errors += 1; + } + telemetry.useCases.set(useCase, summary); +} + +export async function observeRedisOperation( + client: RedisClient, + namespace: string, + operationName: string, + operation: () => T | PromiseLike, + isErrorResult?: (result: T) => boolean, +): Promise { + if (activeRedisObservation.getStore()) { + return await operation(); + } + + const telemetry = requestTelemetry.getStore(); + if ((!telemetry || telemetry.ended) && !isMetricsConfigured()) { + return await operation(); + } + + const useCase = normalizeRedisUseCase(namespace); + const redisOperation = normalizeLabel(operationName); + const startedAt = process.hrtime.bigint(); + let status: RedisOperationStatus = 'success'; + try { + const result = await activeRedisObservation.run(true, operation); + if (isErrorResult?.(result)) { + status = 'error'; + } + return result; + } catch (error) { + status = 'error'; + throw error; + } finally { + const durationSeconds = Number(process.hrtime.bigint() - startedAt) / 1_000_000_000; + addRequestObservation(telemetry, useCase, redisOperation, status, durationSeconds * 1000); + recordRedisOperation(client, useCase, redisOperation, status, durationSeconds); + } +} + +export function instrumentRedisCache( + cache: T, + namespace: string, + client: RedisClient = 'keyv', +): T { + const instrumented = cache as Record; + if (instrumented[INSTRUMENTED_CACHE]) { + return cache; + } + + for (const method of REDIS_CACHE_METHODS) { + const original = instrumented[method]; + if (typeof original !== 'function') { + continue; + } + instrumented[method] = (...args: unknown[]) => + observeRedisOperation(client, namespace, method, () => Reflect.apply(original, cache, args)); + } + + instrumented[INSTRUMENTED_CACHE] = true; + return cache; +} + +function instrumentPipeline(pipeline: T, namespace: string): T { + return new Proxy(pipeline, { + get(target, property, receiver) { + const value = Reflect.get(target, property, receiver); + if (property !== 'exec' || typeof value !== 'function') { + if (typeof value !== 'function') { + return value; + } + return (...args: unknown[]) => { + const result = Reflect.apply(value, target, args); + return result === target ? receiver : result; + }; + } + return (...args: unknown[]) => + observeRedisOperation( + 'ioredis', + namespace, + 'pipeline', + () => Reflect.apply(value, target, args), + pipelineResultHasErrors, + ); + }, + }); +} + +function pipelineResultHasErrors(result: unknown): boolean { + return ( + Array.isArray(result) && + result.some((entry) => Array.isArray(entry) && entry.length > 0 && entry[0] != null) + ); +} + +export function instrumentIORedisClient(client: T, namespace: string): T { + const useCase = normalizeRedisUseCase(namespace); + const existing = instrumentedClients.get(client)?.get(useCase); + if (existing) { + return existing as T; + } + + const proxy = new Proxy(client, { + get(target, property, receiver) { + const value = Reflect.get(target, property, receiver); + if (property === 'constructor') { + return value; + } + if (typeof property !== 'string' || typeof value !== 'function') { + return value; + } + if (property === 'pipeline' || property === 'multi') { + return (...args: unknown[]) => + instrumentPipeline(Reflect.apply(value, target, args) as object, useCase); + } + if (!IOREDIS_COMMANDS.has(property)) { + return (...args: unknown[]) => { + const result = Reflect.apply(value, target, args); + return result === target ? receiver : result; + }; + } + return (...args: unknown[]) => { + const operation = property === 'call' && typeof args[0] === 'string' ? args[0] : property; + return observeRedisOperation('ioredis', useCase, operation, () => + Reflect.apply(value, target, args), + ); + }; + }, + }); + + const byUseCase = instrumentedClients.get(client) ?? new Map(); + byUseCase.set(useCase, proxy); + instrumentedClients.set(client, byUseCase); + return proxy; +} diff --git a/packages/api/src/cluster/LeaderElection.ts b/packages/api/src/cluster/LeaderElection.ts index e40a95cde8..4c5ceddfe8 100644 --- a/packages/api/src/cluster/LeaderElection.ts +++ b/packages/api/src/cluster/LeaderElection.ts @@ -1,4 +1,5 @@ import { logger } from '@librechat/data-schemas'; +import { observeRedisOperation, RedisUseCases } from '~/cache/redisTelemetry'; import { cacheConfig as cache } from '~/cache/cacheConfig'; import { keyvRedisClient } from '~/cache/redisClients'; import { clusterConfig as cluster } from './config'; @@ -79,10 +80,12 @@ export class LeaderElection { end `; - await keyvRedisClient!.eval(script, { - keys: [LeaderElection.LEADER_KEY], - arguments: [this.UUID], - }); + await observeRedisOperation('keyv', RedisUseCases.LEADER_ELECTION, 'eval', () => + keyvRedisClient!.eval(script, { + keys: [LeaderElection.LEADER_KEY], + arguments: [this.UUID], + }), + ); } catch (error) { logger.error('Failed to release leadership lock:', error); } @@ -95,7 +98,9 @@ export class LeaderElection { */ public static async getLeaderUUID(): Promise { if (!cache.USE_REDIS) return null; - return await keyvRedisClient!.get(LeaderElection.LEADER_KEY); + return await observeRedisOperation('keyv', RedisUseCases.LEADER_ELECTION, 'get', () => + keyvRedisClient!.get(LeaderElection.LEADER_KEY), + ); } /** @@ -115,10 +120,12 @@ export class LeaderElection { */ private async electSelf(): Promise { try { - const result = await keyvRedisClient!.set(LeaderElection.LEADER_KEY, this.UUID, { - NX: true, - EX: cluster.LEADER_LEASE_DURATION, - }); + const result = await observeRedisOperation('keyv', RedisUseCases.LEADER_ELECTION, 'set', () => + keyvRedisClient!.set(LeaderElection.LEADER_KEY, this.UUID, { + NX: true, + EX: cluster.LEADER_LEASE_DURATION, + }), + ); if (result !== 'OK') return false; @@ -152,10 +159,16 @@ export class LeaderElection { end `; - const result = await keyvRedisClient!.eval(script, { - keys: [LeaderElection.LEADER_KEY], - arguments: [this.UUID, cluster.LEADER_LEASE_DURATION.toString()], - }); + const result = await observeRedisOperation( + 'keyv', + RedisUseCases.LEADER_ELECTION, + 'eval', + () => + keyvRedisClient!.eval(script, { + keys: [LeaderElection.LEADER_KEY], + arguments: [this.UUID, cluster.LEADER_LEASE_DURATION.toString()], + }), + ); if (result === 0) { logger.warn('Lost leadership, clearing refresh timer'); diff --git a/packages/api/src/mcp/registry/cache/ServerConfigsCacheRedis.ts b/packages/api/src/mcp/registry/cache/ServerConfigsCacheRedis.ts index af1316056d..dcca34cbef 100644 --- a/packages/api/src/mcp/registry/cache/ServerConfigsCacheRedis.ts +++ b/packages/api/src/mcp/registry/cache/ServerConfigsCacheRedis.ts @@ -1,9 +1,9 @@ -import type Keyv from 'keyv'; import { fromPairs } from 'lodash'; import { logger } from '@librechat/data-schemas'; +import type Keyv from 'keyv'; import type { IServerConfigsRepositoryInterface } from '~/mcp/registry/ServerConfigsRepositoryInterface'; import type { ParsedServerConfig, AddServerResult } from '~/mcp/types'; -import { standardCache, keyvRedisClient } from '~/cache'; +import { keyvRedisClient, observeRedisOperation, RedisUseCases, standardCache } from '~/cache'; import { BaseRegistryCache } from './BaseRegistryCache'; /** @@ -69,17 +69,26 @@ export class ServerConfigsCacheRedis } public async getAll(): Promise> { - if (!keyvRedisClient || !('scanIterator' in keyvRedisClient)) { + const redisClient = keyvRedisClient; + if (!redisClient || !('scanIterator' in redisClient)) { throw new Error('Redis client with scanIterator not available.'); } const startTime = Date.now(); const pattern = `*${this.cache.namespace}:*`; - const keys: string[] = []; - for await (const key of keyvRedisClient.scanIterator({ MATCH: pattern })) { - keys.push(key); - } + const keys = await observeRedisOperation( + 'keyv', + RedisUseCases.MCP_REGISTRY, + 'scan', + async () => { + const scannedKeys: string[] = []; + for await (const key of redisClient.scanIterator({ MATCH: pattern })) { + scannedKeys.push(key); + } + return scannedKeys; + }, + ); if (keys.length === 0) { logger.debug(`[ServerConfigsCacheRedis] getAll(${this.namespace}): no keys found`); diff --git a/packages/api/src/middleware/concurrency.ts b/packages/api/src/middleware/concurrency.ts index 22302e79d0..6183390502 100644 --- a/packages/api/src/middleware/concurrency.ts +++ b/packages/api/src/middleware/concurrency.ts @@ -1,6 +1,6 @@ import { logger } from '@librechat/data-schemas'; import { CacheKeys, Time, ViolationTypes } from 'librechat-data-provider'; -import { standardCache, cacheConfig, ioredisClient } from '~/cache'; +import { standardCache, cacheConfig, instrumentIORedisClient, ioredisClient } from '~/cache'; import { isEnabled, math } from '~/utils'; const { USE_REDIS } = cacheConfig; @@ -118,8 +118,9 @@ export async function checkAndIncrementPendingRequest( // A single EVAL round-trip atomically increments, checks, and decrements if over-limit. if (USE_REDIS && ioredisClient) { const key = buildKey(userId); + const redisClient = instrumentIORedisClient(ioredisClient, CacheKeys.PENDING_REQ); try { - const result = (await ioredisClient.eval( + const result = (await redisClient.eval( CHECK_AND_INCREMENT_SCRIPT, 1, key, @@ -193,8 +194,9 @@ export async function decrementPendingRequest(userId: string): Promise { // Use atomic Lua script to decrement and clean up zero/negative keys in one round-trip if (USE_REDIS && ioredisClient) { const key = buildKey(userId); + const redisClient = instrumentIORedisClient(ioredisClient, CacheKeys.PENDING_REQ); try { - const newCount = (await ioredisClient.eval(DECREMENT_SCRIPT, 1, key)) as number; + const newCount = (await redisClient.eval(DECREMENT_SCRIPT, 1, key)) as number; if (newCount === 0) { logger.debug(`[concurrency] User ${userId} pending requests cleared`); } else { diff --git a/packages/api/src/stream/implementations/RedisEventTransport.ts b/packages/api/src/stream/implementations/RedisEventTransport.ts index 43a573f1e8..d0eaebad47 100644 --- a/packages/api/src/stream/implementations/RedisEventTransport.ts +++ b/packages/api/src/stream/implementations/RedisEventTransport.ts @@ -1,6 +1,7 @@ -import type { Redis, Cluster } from 'ioredis'; import { logger } from '@librechat/data-schemas'; +import type { Redis, Cluster } from 'ioredis'; import type { IEventTransport } from '~/stream/interfaces/IJobStore'; +import { instrumentIORedisClient, RedisUseCases } from '~/cache/redisTelemetry'; /** * Redis key prefixes for pub/sub channels @@ -112,8 +113,8 @@ export class RedisEventTransport implements IEventTransport { * @param subscriber - Redis client for subscribing (must be dedicated) */ constructor(publisher: Redis | Cluster, subscriber: Redis | Cluster) { - this.publisher = publisher; - this.subscriber = subscriber; + this.publisher = instrumentIORedisClient(publisher, RedisUseCases.GENERATION_STREAM); + this.subscriber = instrumentIORedisClient(subscriber, RedisUseCases.GENERATION_STREAM); // Set up message handler for all subscriptions this.subscriber.on('message', (channel: string, message: string) => { diff --git a/packages/api/src/stream/implementations/RedisJobStore.ts b/packages/api/src/stream/implementations/RedisJobStore.ts index 3000ac843c..d8a6acaccc 100644 --- a/packages/api/src/stream/implementations/RedisJobStore.ts +++ b/packages/api/src/stream/implementations/RedisJobStore.ts @@ -16,6 +16,7 @@ import { STEER_QUEUE_MAX_DEPTH, isPendingActionStale, } from '~/stream/interfaces/IJobStore'; +import { instrumentIORedisClient, RedisUseCases } from '~/cache/redisTelemetry'; import { toPendingSteer } from '~/stream/SteeringLifecycle'; /** @@ -372,7 +373,7 @@ export class RedisJobStore implements IJobStore { private cleanupIntervalMs = 60000; constructor(redis: Redis | Cluster, options?: RedisJobStoreOptions) { - this.redis = redis; + this.redis = instrumentIORedisClient(redis, RedisUseCases.GENERATION_STREAM); this.ttl = { completed: options?.completedTtl ?? DEFAULT_TTL.completed, running: options?.runningTtl ?? DEFAULT_TTL.running, diff --git a/packages/api/src/telemetry/middleware.spec.ts b/packages/api/src/telemetry/middleware.spec.ts index 271d8c0a11..d723372beb 100644 --- a/packages/api/src/telemetry/middleware.spec.ts +++ b/packages/api/src/telemetry/middleware.spec.ts @@ -1,10 +1,12 @@ import { EventEmitter } from 'node:events'; +import { CacheKeys } from 'librechat-data-provider'; import { SpanStatusCode, trace } from '@opentelemetry/api'; import type { NextFunction, Response } from 'express'; import type { Span } from '@opentelemetry/api'; import type { ServerRequest } from '~/types'; -import { getTelemetryRequestSpan } from './sdk'; import { telemetryErrorMiddleware, telemetryMiddleware } from './middleware'; +import { observeRedisOperation } from '~/cache/redisTelemetry'; +import { getTelemetryRequestSpan } from './sdk'; jest.mock('./sdk', () => ({ getTelemetryRequestSpan: jest.fn(), @@ -116,6 +118,33 @@ describe('telemetryMiddleware', () => { ); }); + it('adds aggregated Redis activity to the stored request span', async () => { + const requestSpan = createSpan(); + const res = createResponse(200); + let operation: Promise | undefined; + mockGetTelemetryRequestSpan.mockReturnValue(requestSpan); + + telemetryMiddleware(createRequest(), res as Response, () => { + operation = observeRedisOperation('keyv', CacheKeys.AUTH_USER_DOC, 'get', async () => + Promise.resolve('cached-user'), + ); + }); + await operation; + res.emit('finish'); + + const attributes = Object.assign( + {}, + ...requestSpan.setAttributes.mock.calls.map(([value]) => value), + ); + expect(attributes).toMatchObject({ + 'librechat.redis.calls': 1, + 'librechat.redis.errors': 0, + 'librechat.redis.operations': ['get'], + 'librechat.redis.use_cases': ['auth_user_doc'], + 'librechat.redis.auth_user_doc.calls': 1, + }); + }); + it('records safe route and identity attributes without body content', () => { const span = createSpan(); const req = createRequest(); diff --git a/packages/api/src/telemetry/middleware.ts b/packages/api/src/telemetry/middleware.ts index d9c110c1b9..7362f45b65 100644 --- a/packages/api/src/telemetry/middleware.ts +++ b/packages/api/src/telemetry/middleware.ts @@ -2,6 +2,11 @@ import { SpanStatusCode, trace } from '@opentelemetry/api'; import type { Span, Attributes } from '@opentelemetry/api'; import type { NextFunction, Response } from 'express'; import type { ServerRequest } from '~/types'; +import { + createRedisRequestTelemetry, + finishRedisRequestTelemetry, + runWithRedisRequestTelemetry, +} from '~/cache/redisTelemetry'; import { getTelemetryRequestSpan } from './sdk'; import { DEFAULT_HEALTH_PATH } from './config'; @@ -112,6 +117,7 @@ export function telemetryMiddleware(req: ServerRequest, res: Response, next: Nex span.setAttributes({ 'http.request.method': req.method, }); + const redisTelemetry = createRedisRequestTelemetry(span); let completed = false; const complete = () => { @@ -119,6 +125,7 @@ export function telemetryMiddleware(req: ServerRequest, res: Response, next: Nex return; } completed = true; + finishRedisRequestTelemetry(redisTelemetry); setCompletionAttributes(span, req, res); }; @@ -127,12 +134,13 @@ export function telemetryMiddleware(req: ServerRequest, res: Response, next: Nex return; } completed = true; + finishRedisRequestTelemetry(redisTelemetry); setCompletionAttributes(span, req, res, !res.writableEnded); }; res.once('finish', complete); res.once('close', close); - next(); + runWithRedisRequestTelemetry(redisTelemetry, next); } export function telemetryErrorMiddleware(