feat: add low-noise Redis observability (#14309)

* feat(api): add low-noise Redis observability

* fix(api): preserve Redis proxy constructor

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';

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) };

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;

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';

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);

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<Pick<Span, 'setAttributes'>> {
+  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);
+  });
+});

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<object, Map<string, object>>();
+
+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<string>;
+  span: Span;
+  useCases: Map<string, RedisUseCaseSummary>;
+}
+
+const requestTelemetry = new AsyncLocalStorage<RedisRequestTelemetry>();
+const activeRedisObservation = new AsyncLocalStorage<boolean>();
+
+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<string>(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<T>(
+  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<T>(
+  client: RedisClient,
+  namespace: string,
+  operationName: string,
+  operation: () => T | PromiseLike<T>,
+  isErrorResult?: (result: T) => boolean,
+): Promise<T> {
+  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<T extends object>(
+  cache: T,
+  namespace: string,
+  client: RedisClient = 'keyv',
+): T {
+  const instrumented = cache as Record<PropertyKey, unknown>;
+  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<T extends object>(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<T extends object>(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<string, object>();
+  byUseCase.set(useCase, proxy);
+  instrumentedClients.set(client, byUseCase);
+  return proxy;
+}

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<string | null> {
     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<boolean> {
     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');

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<Record<string, ParsedServerConfig>> {
-    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`);

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<void> {
     // 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 {

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) => {

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,

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<string> | 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();

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(