From baf40da22aeedb3f6712b54acc3c60bf492709f6 Mon Sep 17 00:00:00 2001 From: Ravi Kumar L Date: Sun, 21 Jun 2026 19:54:53 +0200 Subject: [PATCH] fix: gate tenant Langfuse fanout --- .env.example | 2 + deploy-compose.langfuse-fanout.yml | 1 + docker-compose.langfuse-fanout.yml | 1 + helm/librechat/readme.md | 4 ++ .../templates/langfuse-fanout-configmap.yaml | 18 +++++++-- otel/langfuse-fanout/README.md | 8 ++++ otel/langfuse-fanout/otelcol.yaml | 18 +++++++-- .../__tests__/run-summarization.test.ts | 36 ++++++++++++++++- packages/api/src/langfuse/config.ts | 39 ++++++++++++------- packages/api/src/langfuse/destinations.ts | 9 ++--- packages/api/src/langfuse/feedback.spec.ts | 27 +++++++++++++ packages/api/src/langfuse/utils.ts | 4 ++ 12 files changed, 140 insertions(+), 27 deletions(-) diff --git a/.env.example b/.env.example index 28bcda90dd..f57ecd39ae 100644 --- a/.env.example +++ b/.env.example @@ -131,6 +131,8 @@ NODE_MAX_OLD_SPACE_SIZE=6144 # See otel/langfuse-fanout/README.md. # LANGFUSE_FANOUT_ENABLED=false # LANGFUSE_FANOUT_COLLECTOR_URL=http://langfuse-fanout-collector:4318 +# Emergency switch: set false to keep central collector export but skip tenant trace/score export. +# LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED=true # Langfuse Cloud base URL options: https://cloud.langfuse.com (EU), # https://us.cloud.langfuse.com (US), https://jp.cloud.langfuse.com (JP), # https://hipaa.cloud.langfuse.com (HIPAA). diff --git a/deploy-compose.langfuse-fanout.yml b/deploy-compose.langfuse-fanout.yml index c83cfbb7a5..f315ab5fc8 100644 --- a/deploy-compose.langfuse-fanout.yml +++ b/deploy-compose.langfuse-fanout.yml @@ -5,6 +5,7 @@ services: environment: - LANGFUSE_FANOUT_ENABLED=true - LANGFUSE_FANOUT_COLLECTOR_URL=http://langfuse-fanout-collector:4318 + - LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED=${LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED:-true} networks: - default - langfuse-fanout diff --git a/docker-compose.langfuse-fanout.yml b/docker-compose.langfuse-fanout.yml index c83cfbb7a5..f315ab5fc8 100644 --- a/docker-compose.langfuse-fanout.yml +++ b/docker-compose.langfuse-fanout.yml @@ -5,6 +5,7 @@ services: environment: - LANGFUSE_FANOUT_ENABLED=true - LANGFUSE_FANOUT_COLLECTOR_URL=http://langfuse-fanout-collector:4318 + - LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED=${LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED:-true} networks: - default - langfuse-fanout diff --git a/helm/librechat/readme.md b/helm/librechat/readme.md index 2bbcb5491f..6632aa9d1d 100755 --- a/helm/librechat/readme.md +++ b/helm/librechat/readme.md @@ -64,6 +64,10 @@ When enabled, the chart also sets `LANGFUSE_FANOUT_ENABLED` and `LANGFUSE_FANOUT_COLLECTOR_URL` for the LibreChat app unless those values are already provided in `librechat.configEnv`. +Set `librechat.configEnv.LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED=false` to keep +central trace export flowing through the collector while disabling tenant trace +and score export. + Langfuse base URLs are startup configuration for the LibreChat server and collector. Tenant API keys can still be added through tenant app configuration at runtime without restarting either component. diff --git a/helm/librechat/templates/langfuse-fanout-configmap.yaml b/helm/librechat/templates/langfuse-fanout-configmap.yaml index d2c3a2ee8d..6112cf1ca6 100644 --- a/helm/librechat/templates/langfuse-fanout-configmap.yaml +++ b/helm/librechat/templates/langfuse-fanout-configmap.yaml @@ -27,6 +27,14 @@ data: check_interval: 1s limit_mib: ${env:LANGFUSE_FANOUT_MEMORY_LIMIT_MIB} spike_limit_mib: ${env:LANGFUSE_FANOUT_MEMORY_SPIKE_LIMIT_MIB} + filter/tenant_export: + error_mode: ignore + traces: + span: + - attributes["langfuse.trace.metadata.librechat.langfuse.tenant_export.enabled"] != "true" + batch/central: + timeout: 1s + send_batch_size: 128 batch/by_auth: timeout: 1s send_batch_size: 128 @@ -53,8 +61,12 @@ data: service: extensions: [headers_setter/tenant_passthrough] pipelines: - traces: + traces/central: receivers: [otlp] - processors: [memory_limiter, batch/by_auth] - exporters: [otlphttp/central, otlphttp/tenant] + processors: [memory_limiter, batch/central] + exporters: [otlphttp/central] + traces/tenant: + receivers: [otlp] + processors: [memory_limiter, filter/tenant_export, batch/by_auth] + exporters: [otlphttp/tenant] {{- end }} diff --git a/otel/langfuse-fanout/README.md b/otel/langfuse-fanout/README.md index f2e9475347..0a048ab357 100644 --- a/otel/langfuse-fanout/README.md +++ b/otel/langfuse-fanout/README.md @@ -15,6 +15,10 @@ disabled unless you explicitly deploy the fanout collector. - The collector also exports the same trace to the tenant Langfuse project by forwarding the tenant `Authorization` header that LibreChat attaches to the OTLP request. +- Tenant export is conditional. LibreChat marks traces as tenant-exportable only + when tenant keys are configured and `LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED` is + not false; unmarked traces are still exported to central but are dropped by + the tenant pipeline. - User feedback scores use Langfuse's direct REST API. Scores are sent to the central project and, when tenant Langfuse keys are configured, the tenant project. @@ -31,6 +35,9 @@ defined in this collector config. starts. - Tenant Langfuse API keys can be added, changed, or disabled in tenant app configuration at runtime without restarting LibreChat or the collector. +- `LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED=false` can be set on LibreChat as an + emergency switch to stop tenant trace and score export while keeping central + collector export active. - This supports Langfuse Cloud and self-hosted Langfuse as long as the relevant central and tenant base URLs are known at collector/server startup. - All tenant traces that use deployment-level fanout are sent to the same @@ -46,6 +53,7 @@ Set the central Langfuse destination in `.env`: LANGFUSE_FANOUT_CENTRAL_BASE_URL=https://cloud.langfuse.com LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=Basic LANGFUSE_FANOUT_TENANT_BASE_URL=https://cloud.langfuse.com +LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED=true ``` Langfuse Cloud base URL options: diff --git a/otel/langfuse-fanout/otelcol.yaml b/otel/langfuse-fanout/otelcol.yaml index 8b8a2b7c39..2625f2a49d 100644 --- a/otel/langfuse-fanout/otelcol.yaml +++ b/otel/langfuse-fanout/otelcol.yaml @@ -18,6 +18,14 @@ processors: check_interval: 1s limit_mib: ${env:LANGFUSE_FANOUT_MEMORY_LIMIT_MIB} spike_limit_mib: ${env:LANGFUSE_FANOUT_MEMORY_SPIKE_LIMIT_MIB} + filter/tenant_export: + error_mode: ignore + traces: + span: + - attributes["langfuse.trace.metadata.librechat.langfuse.tenant_export.enabled"] != "true" + batch/central: + timeout: 1s + send_batch_size: 128 batch/by_auth: timeout: 1s send_batch_size: 128 @@ -44,7 +52,11 @@ exporters: service: extensions: [headers_setter/tenant_passthrough] pipelines: - traces: + traces/central: receivers: [otlp] - processors: [memory_limiter, batch/by_auth] - exporters: [otlphttp/central, otlphttp/tenant] + processors: [memory_limiter, batch/central] + exporters: [otlphttp/central] + traces/tenant: + receivers: [otlp] + processors: [memory_limiter, filter/tenant_export, batch/by_auth] + exporters: [otlphttp/tenant] diff --git a/packages/api/src/agents/__tests__/run-summarization.test.ts b/packages/api/src/agents/__tests__/run-summarization.test.ts index 9d7a573471..b810f25992 100644 --- a/packages/api/src/agents/__tests__/run-summarization.test.ts +++ b/packages/api/src/agents/__tests__/run-summarization.test.ts @@ -205,6 +205,7 @@ beforeEach(() => { jest.clearAllMocks(); delete process.env.LANGFUSE_FANOUT_ENABLED; delete process.env.LANGFUSE_FANOUT_COLLECTOR_URL; + delete process.env.LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED; }); // --------------------------------------------------------------------------- @@ -1194,7 +1195,10 @@ describe('Langfuse run config', () => { publicKey: 'pk-tenant-1', secretKey: 'sk-tenant-1', baseUrl: 'http://langfuse-fanout-collector:4318', - metadata: { 'librechat.tenant.id': 'tenant-1' }, + metadata: { + 'librechat.tenant.id': 'tenant-1', + 'librechat.langfuse.tenant_export.enabled': 'true', + }, tags: ['tenant:tenant-1'], }); }); @@ -1217,10 +1221,14 @@ describe('Langfuse run config', () => { publicKey: 'pk-tenant-1', secretKey: 'sk-tenant-1', baseUrl: 'http://collector-from-env:4318', + metadata: { + 'librechat.tenant.id': 'tenant-1', + 'librechat.langfuse.tenant_export.enabled': 'true', + }, }); }); - it('does not route to fanout when deployment fanout is enabled without tenant keys', async () => { + it('routes central-only traces to the collector when deployment fanout is enabled without tenant keys', async () => { process.env.LANGFUSE_FANOUT_ENABLED = 'true'; process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://collector-from-env:4318'; @@ -1233,6 +1241,30 @@ describe('Langfuse run config', () => { expect(callArgs.langfuse).toEqual({ deterministicTraceId: true, + baseUrl: 'http://collector-from-env:4318', + metadata: { 'librechat.tenant.id': 'tenant-1' }, + tags: ['tenant:tenant-1'], + }); + }); + + it('routes central-only traces to the collector when tenant fanout export is disabled', async () => { + process.env.LANGFUSE_FANOUT_ENABLED = 'true'; + process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://collector-from-env:4318'; + process.env.LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED = 'false'; + + const callArgs = await callAndCaptureRunConfig({ + tenantId: 'tenant-1', + appConfig: { + langfuse: { + publicKey: 'pk-tenant-1', + secretKey: 'sk-tenant-1', + }, + } as AppConfig, + }); + + expect(callArgs.langfuse).toEqual({ + deterministicTraceId: true, + baseUrl: 'http://collector-from-env:4318', metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); diff --git a/packages/api/src/langfuse/config.ts b/packages/api/src/langfuse/config.ts index 642771a1c7..a6c976617b 100644 --- a/packages/api/src/langfuse/config.ts +++ b/packages/api/src/langfuse/config.ts @@ -1,8 +1,13 @@ import type { AppConfig } from '@librechat/data-schemas'; import type { RunConfig } from '@librechat/agents'; -import { normalizeString } from './utils'; +import { isFalseEnv, normalizeString } from './utils'; type LangfuseRunConfig = NonNullable; +const TENANT_EXPORT_METADATA_KEY = 'librechat.langfuse.tenant_export.enabled'; + +function isTenantExportEnabled(): boolean { + return !isFalseEnv(process.env.LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED); +} function mergeTraceMetadata( base: LangfuseRunConfig['metadata'], @@ -56,23 +61,29 @@ export function buildLangfuseConfig({ const publicKey = normalizeString(config?.publicKey); const secretKey = normalizeString(config?.secretKey); const hasTenantCredentials = Boolean(publicKey && secretKey); - if (hasTenantCredentials) { + const fanout = config?.fanout; + const fanoutEnabled = + fanout?.enabled !== false && + (fanout?.enabled === true || process.env.LANGFUSE_FANOUT_ENABLED === 'true'); + const fanoutCollectorUrl = + normalizeString(fanout?.collectorUrl) ?? normalizeString(process.env.LANGFUSE_FANOUT_COLLECTOR_URL); + const tenantExportEnabled = hasTenantCredentials && fanoutEnabled && isTenantExportEnabled(); + + if (hasTenantCredentials && (!fanoutEnabled || tenantExportEnabled)) { langfuse.publicKey = publicKey; langfuse.secretKey = secretKey; + } - const fanout = config?.fanout; - const fanoutEnabled = - fanout?.enabled !== false && - (fanout?.enabled === true || process.env.LANGFUSE_FANOUT_ENABLED === 'true'); - const fanoutCollectorUrl = - normalizeString(fanout?.collectorUrl) ?? - normalizeString(process.env.LANGFUSE_FANOUT_COLLECTOR_URL); - const baseUrl = - fanoutEnabled && fanoutCollectorUrl ? fanoutCollectorUrl : normalizeString(config?.baseUrl); + const baseUrl = fanoutEnabled && fanoutCollectorUrl ? fanoutCollectorUrl : normalizeString(config?.baseUrl); + if (baseUrl) { + langfuse.baseUrl = baseUrl; + } - if (baseUrl) { - langfuse.baseUrl = baseUrl; - } + if (tenantExportEnabled) { + langfuse.metadata = { + ...(langfuse.metadata ?? {}), + [TENANT_EXPORT_METADATA_KEY]: 'true', + }; } return langfuse; diff --git a/packages/api/src/langfuse/destinations.ts b/packages/api/src/langfuse/destinations.ts index 10d79ea4c3..189669da36 100644 --- a/packages/api/src/langfuse/destinations.ts +++ b/packages/api/src/langfuse/destinations.ts @@ -1,5 +1,5 @@ import type { AppConfig } from '@librechat/data-schemas'; -import { normalizeString, toBasicAuthorization } from './utils'; +import { isFalseEnv, normalizeString, toBasicAuthorization } from './utils'; const DEFAULT_BASE_URL = 'https://cloud.langfuse.com'; @@ -9,10 +9,6 @@ export type LangfuseScoreDestination = { authorization: string; }; -function isFalseEnv(value?: string): boolean { - return value != null && ['0', 'false', 'no', 'off'].includes(value.trim().toLowerCase()); -} - function isSampleRateEnabled(value?: string): boolean { if (value == null || value.trim() === '') { return true; @@ -68,6 +64,9 @@ function getTenantScoreDestination(appConfig?: AppConfig): LangfuseScoreDestinat if (!isTracingEnabled()) { return undefined; } + if (isFalseEnv(process.env.LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED)) { + return undefined; + } const config = appConfig?.langfuse; if (config?.enabled === false) { diff --git a/packages/api/src/langfuse/feedback.spec.ts b/packages/api/src/langfuse/feedback.spec.ts index 573e172976..0d46498a4b 100644 --- a/packages/api/src/langfuse/feedback.spec.ts +++ b/packages/api/src/langfuse/feedback.spec.ts @@ -23,6 +23,7 @@ const langfuseEnvKeys = [ 'LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER', 'LANGFUSE_FANOUT_CENTRAL_BASE_URL', 'LANGFUSE_FANOUT_TENANT_BASE_URL', + 'LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED', ]; let fetchMock: jest.SpiedFunction; @@ -278,6 +279,32 @@ describe('Langfuse feedback scores', () => { ); }); + it('skips tenant scores when tenant fanout export is disabled but keeps central scores', async () => { + process.env.LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER = 'Basic central-auth'; + process.env.LANGFUSE_FANOUT_CENTRAL_BASE_URL = 'http://central-langfuse:3000'; + process.env.LANGFUSE_FANOUT_TENANT_BASE_URL = 'http://tenant-langfuse:3000'; + process.env.LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED = 'false'; + const { sendFeedbackScore } = await loadFeedback(); + + await sendFeedbackScore({ + traceId: 'trace-id', + feedback: { rating: 'thumbsUp' }, + appConfig: appConfigWithLangfuse({ + publicKey: 'tenant-public-key', + secretKey: 'tenant-secret-key', + }), + }); + + expect(getFetchMock()).toHaveBeenCalledTimes(1); + expect(getFetchMock()).toHaveBeenCalledWith( + 'http://central-langfuse:3000/api/public/scores', + expect.objectContaining({ + method: 'POST', + headers: expect.objectContaining({ Authorization: 'Basic central-auth' }), + }), + ); + }); + it('deduplicates matching central and tenant score destinations', async () => { process.env.LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER = getTenantAuthorization(); process.env.LANGFUSE_FANOUT_CENTRAL_BASE_URL = 'http://shared-langfuse:3000'; diff --git a/packages/api/src/langfuse/utils.ts b/packages/api/src/langfuse/utils.ts index 14ac26aad0..85d242e0d2 100644 --- a/packages/api/src/langfuse/utils.ts +++ b/packages/api/src/langfuse/utils.ts @@ -2,6 +2,10 @@ export function normalizeString(value: unknown): string | undefined { return typeof value === 'string' && value.trim() !== '' ? value.trim() : undefined; } +export function isFalseEnv(value?: string): boolean { + return value != null && ['0', 'false', 'no', 'off'].includes(value.trim().toLowerCase()); +} + export function toBasicAuthorization(publicKey: string, secretKey: string): string { return `Basic ${Buffer.from(`${publicKey}:${secretKey}`).toString('base64')}`; }