Track LMCache hit rate from linked vLLM/SGLang engines (Pro)

apps/api/src/common/interfaces/storage-port.interface.ts

@@ -436,6 +436,12 @@ export interface KvCacheFootprintQuery extends KvCacheRangeQuery {
 export interface KvCacheSampleQuery extends KvCacheRangeQuery {
   engineId?: string;
   modelName?: string;
+  bucketMs?: number;
+}
+
+export interface KvCacheSampleTotals {
+  requestedTokens: number;
+  hitTokens: number;
 }
 
 export interface MemorySnapshotQueryOptions {
@@ -655,7 +661,9 @@ export interface StoragePort {
   pruneOldKvCacheFootprintSnapshots(cutoffTimestamp: number, connectionId?: string): Promise<number>;
   saveKvCacheEngineSamples(samples: KvCacheEngineSample[]): Promise<void>;
   getKvCacheEngineSamples(query: KvCacheSampleQuery): Promise<KvCacheEngineSample[]>;
+  getKvCacheSampleTotals(query: Omit<KvCacheSampleQuery, 'limit' | 'bucketMs'>): Promise<KvCacheSampleTotals>;
   pruneOldKvCacheEngineSamples(cutoffTimestamp: number, connectionId?: string): Promise<number>;
+  deleteKvCacheEngineSamples(engineId: string): Promise<void>;
   saveKvCacheEngine(engine: StoredKvCacheEngine): Promise<StoredKvCacheEngine>;
   getKvCacheEngines(connectionId?: string): Promise<StoredKvCacheEngine[]>;
   getKvCacheEngine(id: string): Promise<StoredKvCacheEngine | null>;

apps/api/src/common/utils/outbound-url-guard.ts

@@ -0,0 +1,158 @@
+import { BadRequestException, Logger } from '@nestjs/common';
+import { promises as dns } from 'dns';
+import { isIP, type LookupFunction } from 'net';
+
+export interface OutboundUrlOptions {
+  label: string;
+  allowPrivateNetworks?: boolean;
+  resolveDns?: 'always' | 'production' | 'never';
+}
+
+export class BlockedAddressError extends Error {}
+
+const logger = new Logger('OutboundUrlGuard');
+
+const BLOCKED_IP_PATTERNS = [
+  /^127\./,
+  /^10\./,
+  /^172\.(1[6-9]|2[0-9]|3[01])\./,
+  /^192\.168\./,
+  /^100\.(6[4-9]|[7-9][0-9]|1[01][0-9]|12[0-7])\./,
+  /^169\.254\./,
+  /^::1?$/,
+  /^0\./,
+  /^fe[89ab][0-9a-f]:/i,
+  /^f[cd][0-9a-f]{2}:/i,
+];
+
+const ALWAYS_BLOCKED = [/^169\.254\./, /^fe[89ab][0-9a-f]:/i, /^fd00:ec2::254$/, /^100\.100\.100\.200$/, /^192\.0\.0\.192$/];
+
+const LOOPBACK = [/^127\./, /^0\./, /^::1?$/];
+
+const MAPPED_DOTTED = /^::ffff:(?:0:)?(\d{1,3}(?:\.\d{1,3}){3})$/;
+const MAPPED_HEX = /^::ffff:(?:0:)?([0-9a-f]{1,4}):([0-9a-f]{1,4})$/;
+
+const bareHost = (hostname: string) => hostname.replace(/^\[|\]$/g, '').toLowerCase();
+
+function canonicalIp(ip: string): string {
+  const address = bareHost(ip);
+  if (isIP(address) !== 6) return address;
+  try {
+    return bareHost(new URL(`http://[${address}]`).hostname);
+  } catch {
+    return address;
+  }
+}
+
+function unmapIpv4(address: string): string {
+  const dotted = MAPPED_DOTTED.exec(address);
+  if (dotted) return dotted[1];
+  const hex = MAPPED_HEX.exec(address);
+  if (!hex) return address;
+  const high = parseInt(hex[1], 16);
+  const low = parseInt(hex[2], 16);
+  return `${high >> 8}.${high & 255}.${low >> 8}.${low & 255}`;
+}
+
+const capitalize = (text: string) => text.charAt(0).toUpperCase() + text.slice(1);
+
+const isLocalhostName = (hostname: string) => /(^|\.)localhost\.?$/.test(hostname);
+
+function blockedAddressMessage(label: string, hostname: string, address: string): string {
+  logger.warn(`${capitalize(label)} ${hostname} resolves to blocked IP address ${address}`);
+  return `${capitalize(label)} resolves to a blocked address`;
+}
+
+export function isBlockedIp(ip: string, allowPrivateNetworks = false): boolean {
+  const address = unmapIpv4(canonicalIp(ip));
+  if (ALWAYS_BLOCKED.some((pattern) => pattern.test(address))) return true;
+  if (allowPrivateNetworks) return false;
+  return BLOCKED_IP_PATTERNS.some((pattern) => pattern.test(address));
+}
+
+function isLoopback(ip: string): boolean {
+  const address = unmapIpv4(canonicalIp(ip));
+  return LOOPBACK.some((pattern) => pattern.test(address));
+}
+
+export async function assertSafeOutboundUrl(rawUrl: string, options: OutboundUrlOptions): Promise<URL> {
+  const { label, allowPrivateNetworks, resolveDns = 'always' } = options;
+  try {
+    const parsed = new URL(rawUrl);
+
+    if (!['http:', 'https:'].includes(parsed.protocol)) {
+      throw new BadRequestException('Only HTTP and HTTPS protocols are allowed');
+    }
+
+    if (parsed.username || parsed.password) {
+      logger.warn(`${capitalize(label)} contains credentials, consider using custom headers instead: ${parsed.hostname}`);
+    }
+
+    const isProduction = process.env.NODE_ENV === 'production';
+    const host = unmapIpv4(bareHost(parsed.hostname));
+    const isLocalhost = host === 'localhost' || LOOPBACK.some((pattern) => pattern.test(host));
+
+    if (isLocalhost && (!isProduction || allowPrivateNetworks)) {
+      logger.debug(`Allowing localhost ${label} in ${process.env.NODE_ENV || 'development'} mode: ${rawUrl}`);
+      return parsed;
+    }
+
+    if (parsed.hostname === 'localhost' || parsed.hostname === '0.0.0.0') {
+      throw new BadRequestException(`Cannot use localhost or 0.0.0.0 as ${label} in production`);
+    }
+
+    if (isIP(host) && isBlockedIp(host, allowPrivateNetworks)) {
+      throw new BadRequestException(`Cannot use private IP addresses as ${label}`);
+    }
+
+    if (!allowPrivateNetworks && (parsed.hostname.startsWith('127.') || isLocalhostName(host))) {
+      throw new BadRequestException('Suspicious hostname detected');
+    }
+
+    if (!isIP(host) && resolveDns !== 'never' && (resolveDns !== 'production' || isProduction)) {
+      try {
+        const answers = await dns.lookup(host, { all: true, verbatim: true });
+        for (const { address: addr } of answers) {
+          if (isBlockedIp(addr, allowPrivateNetworks)) {
+            throw new BadRequestException(blockedAddressMessage(label, host, addr));
+          }
+        }
+      } catch (dnsError: any) {
+        if (dnsError instanceof BadRequestException) {
+          throw dnsError;
+        }
+        logger.warn(`Failed to resolve DNS for ${label}: ${parsed.hostname}`);
+        throw new BadRequestException(`Failed to resolve ${label} hostname`);
+      }
+    }
+
+    return parsed;
+  } catch (error) {
+    if (error instanceof BadRequestException) {
+      throw error;
+    }
+    throw new BadRequestException(`Invalid ${label}`);
+  }
+}
+
+export function guardedLookup(options: Omit<OutboundUrlOptions, 'resolveDns'>): LookupFunction {
+  const { label, allowPrivateNetworks } = options;
+  return (hostname, lookupOptions, callback) => {
+    dns.lookup(hostname, { family: lookupOptions.family, hints: lookupOptions.hints, all: true, verbatim: true }).then(
+      (answers) => {
+        const allowLoopback = allowPrivateNetworks || process.env.NODE_ENV !== 'production';
+        const blocked = answers.find(({ address }) => isBlockedIp(address, allowPrivateNetworks) && !(allowLoopback && isLoopback(address)));
+        if (blocked) {
+          callback(new BlockedAddressError(blockedAddressMessage(label, hostname, blocked.address)), '', 0);
+        } else if (answers.length === 0) {
+          callback(Object.assign(new Error(`getaddrinfo ENOTFOUND ${hostname}`), { code: 'ENOTFOUND' }), '', 0);
+        } else if (lookupOptions.all) {
+          (callback as unknown as (error: null, addresses: typeof answers) => void)(null, answers);
+        } else {
+          callback(null, answers[0].address, answers[0].family);
+        }
+      },
+      (error: NodeJS.ErrnoException) => callback(error, '', 0),
+    );
+  };
+}

apps/api/src/common/utils/pinned-request.ts

@@ -0,0 +1,58 @@
+import { request as httpRequest, type IncomingMessage } from 'http';
+import { request as httpsRequest } from 'https';
+import type { LookupFunction } from 'net';
+import { Readable } from 'stream';
+
+export interface PinnedRequestInit {
+  method?: string;
+  headers: Record<string, string>;
+  body?: string;
+  signal: AbortSignal;
+  lookup?: LookupFunction;
+}
+
+export class UnexpectedStatusError extends Error {}
+
+const NULL_BODY_STATUSES = new Set([204, 205, 304]);
+
+function toResponse(message: IncomingMessage): Response {
+  const headers = new Headers();
+  for (const [name, value] of Object.entries(message.headers)) {
+    if (value === undefined) continue;
+    for (const item of Array.isArray(value) ? value : [value]) headers.append(name, item);
+  }
+  const status = message.statusCode ?? 502;
+  if (status < 200 || status > 599) throw new UnexpectedStatusError(`unexpected HTTP status ${status}`);
+  if (NULL_BODY_STATUSES.has(status)) {
+    message.resume();
+    return new Response(null, { status, headers });
+  }
+  return new Response(Readable.toWeb(message) as ReadableStream<Uint8Array>, { status, headers });
+}
+
+export const pinnedRequest = (url: string, init: PinnedRequestInit): Promise<Response> =>
+  new Promise((resolve, reject) => {
+    const target = new URL(url);
+    const send = target.protocol === 'https:' ? httpsRequest : httpRequest;
+    const headers = init.body === undefined ? init.headers : { 'Content-Length': String(Buffer.byteLength(init.body)), ...init.headers };
+    const req = send(
+      target,
+      {
+        method: init.method ?? 'GET',
+        agent: false,
+        headers,
+        signal: init.signal,
+        ...(init.lookup ? { lookup: init.lookup } : {}),
+      },
+      (message) => {
+        try {
+          resolve(toResponse(message));
+        } catch (error) {
+          message.destroy();
+          reject(error);
+        }
+      },
+    );
+    req.on('error', reject);
+    req.end(init.body);
+  });

apps/api/src/external-metrics/__tests__/otel-metrics-ingest.controller.spec.ts

@@ -19,6 +19,7 @@ const clean: IngestResult = {
     unmapped_metric: 0,
     invalid_value: 0,
     cardinality_limit: 0,
+    unknown_engine: 0,
   },
 };
 

apps/api/src/external-metrics/__tests__/otel-metrics-ingest.service.spec.ts

@@ -120,6 +120,7 @@ describe('OtelMetricsIngestService', () => {
       unmapped_metric: 1,
       invalid_value: 1,
       cardinality_limit: 0,
+      unknown_engine: 0,
     });
     expect(toPartialSuccess(result)).toEqual({
       rejectedDataPoints: 8,
@@ -420,4 +421,78 @@ describe('OtelMetricsIngestService discovery', () => {
     expect(discovered.list(NOW_MS)).toEqual([]);
     expect(result.dropped.unknown_instance).toBe(1);
   });
+
+  describe('KV cache routing', () => {
+    const lmcacheSum = (name: string): OtlpMetric => ({
+      name,
+      sum: { dataPoints: [{ attributes: [], asInt: '5' }, { attributes: [], asInt: '6' }], aggregationTemporality: 2 },
+    });
+
+    function buildWithSink(sinkResult = { accepted: 2, dropped: { unmapped_metric: 3 } as Record<string, number> }) {
+      const registry = { findByHostPort: jest.fn() } as unknown as ConnectionRegistry;
+      const store = new ExternalMetricsStore();
+      const applySpy = jest.spyOn(store, 'apply');
+      const sink = { ingest: jest.fn().mockReturnValue(sinkResult) };
+      const service = new OtelMetricsIngestService(registry, store, undefined, undefined, sink);
+      return { service, registry, applySpy, sink };
+    }
+
+    it('routes a resource with the engine attribute to the sink', () => {
+      const { service, registry, applySpy, sink } = buildWithSink();
+      const metrics = [lmcacheSum('lmcache:num_hit_tokens_total')];
+      const result = service.ingest(resource([str('betterdb.lmcache.engine', 'eng-1')], metrics), NOW_MS);
+      expect(sink.ingest).toHaveBeenCalledWith({ 'betterdb.lmcache.engine': 'eng-1' }, metrics, NOW_MS);
+      expect(registry.findByHostPort).not.toHaveBeenCalled();
+      expect(applySpy).not.toHaveBeenCalled();
+      expect(result.accepted).toBe(2);
+      expect(result.dropped.unmapped_metric).toBe(3);
+    });
+
+    it('routes a resource whose metrics are all lmcache without the attribute', () => {
+      const { service, sink } = buildWithSink();
+      service.ingest(resource([], [lmcacheSum('lmcache:num_hit_tokens_total')]), NOW_MS);
+      expect(sink.ingest).toHaveBeenCalledTimes(1);
+    });
+
+    it('drops an lmcache resource as unknown_engine when no sink is present', () => {
+      const { service } = build();
+      const result = service.ingest(
+        resource([str('betterdb.lmcache.engine', 'eng-1')], [lmcacheSum('lmcache:num_hit_tokens_total')]),
+        NOW_MS,
+      );
+      expect(result.dropped.unknown_engine).toBe(2);
+      expect(result.accepted).toBe(0);
+    });
+
+    it('keeps Valkey metrics of a tagged resource on the Valkey path and routes the rest to the sink', () => {
+      const { service, registry, sink } = buildWithSink({ accepted: 2, dropped: {} });
+      (registry.findByHostPort as jest.Mock).mockReturnValue({ id: 'ext', connectionType: 'external' });
+      const kv = lmcacheSum('lmcache:num_hit_tokens_total');
+      const result = service.ingest(
+        resource([str('betterdb.lmcache.engine', 'eng-1'), ...identity], [gauge('redis.memory.used', 1024), kv]),
+        NOW_MS,
+      );
+      expect(sink.ingest).toHaveBeenCalledWith(expect.objectContaining({ 'betterdb.lmcache.engine': 'eng-1' }), [kv], NOW_MS);
+      expect(registry.findByHostPort).toHaveBeenCalledWith('cache.internal', 6379);
+      expect(result.accepted).toBe(3);
+    });
+
+    it('does not call the sink for a tagged resource that only carries Valkey metrics', () => {
+      const { service, registry, sink } = buildWithSink();
+      (registry.findByHostPort as jest.Mock).mockReturnValue({ id: 'ext', connectionType: 'external' });
+      const result = service.ingest(
+        resource([str('betterdb.lmcache.engine', 'eng-1'), ...identity], [gauge('valkey.memory.used', 1024)]),
+        NOW_MS,
+      );
+      expect(sink.ingest).not.toHaveBeenCalled();
+      expect(result.accepted).toBe(1);
+    });
+
+    it('keeps the Valkey path when a sink is present', () => {
+      const { service, sink } = buildWithSink();
+      const result = service.ingest(resource([], [gauge('redis.memory.used', 1)]), NOW_MS);
+      expect(sink.ingest).not.toHaveBeenCalled();
+      expect(result.dropped.unidentified).toBe(1);
+    });
+  });
 });

apps/api/src/external-metrics/kv-cache-otlp-sink.ts

@@ -0,0 +1,18 @@
+import type { DropReason, OtlpMetric } from './otlp-metrics-types';
+
+export const KV_CACHE_OTLP_SINK = Symbol('KV_CACHE_OTLP_SINK');
+export const LMCACHE_ENGINE_ATTRIBUTE = 'betterdb.lmcache.engine';
+
+export interface KvCacheOtlpSinkResult {
+  accepted: number;
+  dropped: Partial<Record<DropReason, number>>;
+}
+
+export interface KvCacheOtlpSink {
+  ingest(resourceAttributes: Record<string, string>, metrics: OtlpMetric[], nowMs: number): KvCacheOtlpSinkResult;
+}
+
+export function isKvCacheResource(resourceAttributes: Record<string, string>, metrics: OtlpMetric[]): boolean {
+  if (resourceAttributes[LMCACHE_ENGINE_ATTRIBUTE] !== undefined) return true;
+  return metrics.length > 0 && metrics.every((metric) => (metric.name ?? '').startsWith('lmcache:'));
+}