mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-10-21 17:44:11 +00:00
🪵 feat: Add OpenTelemetry Logs Pipeline via Winston Bridge (#15565)
Exports application logs as OTel log records when OTEL_LOGS_ENABLED=true, correlated with the active trace, and gates every signal explicitly so a disabled signal never falls back to the SDK's default OTLP exporter. Verified end to end over OTLP/gRPC against an in-process collector.
This commit is contained in:
parent
aec2121923
commit
a6cf61abcf
16 changed files with 533 additions and 100 deletions
15
.env.example
15
.env.example
|
|
@ -326,19 +326,28 @@ NODE_MAX_OLD_SPACE_SIZE=6144
|
|||
# LANGFUSE_FANOUT_BATCH_SEND_SIZE=128
|
||||
# LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT=1000
|
||||
|
||||
#=======================#
|
||||
# OpenTelemetry Tracing #
|
||||
#=======================#
|
||||
#===============#
|
||||
# OpenTelemetry #
|
||||
#===============#
|
||||
|
||||
# Enables backend OpenTelemetry tracing. General backend visibility only;
|
||||
# use Langfuse for GenAI-specific prompt/model observability.
|
||||
# OTEL_TRACING_ENABLED=false
|
||||
# Exports application logs as OpenTelemetry log records, correlated with the
|
||||
# active trace. Logs below OTEL_LOGS_LEVEL are not exported.
|
||||
# OTEL_LOGS_ENABLED=false
|
||||
# OTEL_LOGS_LEVEL=info
|
||||
# OTEL_SERVICE_NAME=librechat
|
||||
# OTEL_SERVICE_VERSION=
|
||||
# Exporter protocol for every signal: http/protobuf (default, port 4318) or grpc (port 4317).
|
||||
# Per-signal overrides: OTEL_EXPORTER_OTLP_TRACES_PROTOCOL, OTEL_EXPORTER_OTLP_LOGS_PROTOCOL.
|
||||
# OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf
|
||||
# OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318
|
||||
# OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=
|
||||
# OTEL_EXPORTER_OTLP_LOGS_ENDPOINT=
|
||||
# OTEL_EXPORTER_OTLP_HEADERS=
|
||||
# OTEL_TRACES_EXPORTER=otlp
|
||||
# OTEL_LOGS_EXPORTER=otlp
|
||||
# OTEL_TRACES_SAMPLER=parentbased_always_on
|
||||
# OTEL_LOG_LEVEL=INFO
|
||||
# OTEL_SDK_DISABLED=false
|
||||
|
|
|
|||
|
|
@ -62,6 +62,7 @@
|
|||
"@opentelemetry/resources": "^2.6.1",
|
||||
"@opentelemetry/sdk-node": "^0.221.0",
|
||||
"@opentelemetry/semantic-conventions": "^1.39.0",
|
||||
"@opentelemetry/winston-transport": "^0.28.0",
|
||||
"@redis/client": "5.10.0",
|
||||
"@smithy/node-http-handler": "^4.4.5",
|
||||
"ai-tokenizer": "^1.0.6",
|
||||
|
|
|
|||
|
|
@ -5,7 +5,9 @@ function isTruthy(value) {
|
|||
}
|
||||
|
||||
function isTelemetryEnabled() {
|
||||
return isTruthy(process.env.OTEL_TRACING_ENABLED) && !isTruthy(process.env.OTEL_SDK_DISABLED);
|
||||
const signalEnabled =
|
||||
isTruthy(process.env.OTEL_TRACING_ENABLED) || isTruthy(process.env.OTEL_LOGS_ENABLED);
|
||||
return signalEnabled && !isTruthy(process.env.OTEL_SDK_DISABLED);
|
||||
}
|
||||
|
||||
if (isTelemetryEnabled()) {
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ describe('telemetry bootstrap', () => {
|
|||
beforeEach(() => {
|
||||
jest.resetModules();
|
||||
process.env = { ...originalEnv };
|
||||
delete process.env.OTEL_LOGS_ENABLED;
|
||||
delete process.env.OTEL_SDK_DISABLED;
|
||||
delete process.env.OTEL_TRACING_ENABLED;
|
||||
jest.doMock('dotenv', () => ({
|
||||
|
|
@ -34,6 +35,7 @@ describe('telemetry bootstrap', () => {
|
|||
});
|
||||
|
||||
it('does not load OpenTelemetry packages when the SDK is disabled', () => {
|
||||
process.env.OTEL_LOGS_ENABLED = 'true';
|
||||
process.env.OTEL_SDK_DISABLED = 'true';
|
||||
process.env.OTEL_TRACING_ENABLED = 'true';
|
||||
jest.doMock(
|
||||
|
|
@ -88,4 +90,28 @@ describe('telemetry bootstrap', () => {
|
|||
expect(telemetry.enabled).toBe(false);
|
||||
expect(telemetry.status).toBe('failed');
|
||||
});
|
||||
|
||||
it('initializes telemetry when only logs are enabled', () => {
|
||||
process.env.OTEL_LOGS_ENABLED = 'true';
|
||||
const initializeTelemetry = jest.fn(() => ({
|
||||
enabled: true,
|
||||
status: 'started',
|
||||
shutdown: jest.fn(),
|
||||
}));
|
||||
jest.doMock(
|
||||
'@librechat/api/telemetry',
|
||||
() => ({
|
||||
initializeTelemetry,
|
||||
telemetryMiddleware: jest.fn(),
|
||||
telemetryErrorMiddleware: jest.fn(),
|
||||
}),
|
||||
{ virtual: true },
|
||||
);
|
||||
|
||||
const telemetry = require('./telemetry');
|
||||
|
||||
expect(initializeTelemetry).toHaveBeenCalledTimes(1);
|
||||
expect(telemetry.enabled).toBe(true);
|
||||
expect(telemetry.status).toBe('started');
|
||||
});
|
||||
});
|
||||
|
|
|
|||
15
package-lock.json
generated
15
package-lock.json
generated
|
|
@ -79,6 +79,7 @@
|
|||
"@opentelemetry/resources": "^2.6.1",
|
||||
"@opentelemetry/sdk-node": "^0.221.0",
|
||||
"@opentelemetry/semantic-conventions": "^1.39.0",
|
||||
"@opentelemetry/winston-transport": "^0.28.0",
|
||||
"@redis/client": "5.10.0",
|
||||
"@smithy/node-http-handler": "^4.4.5",
|
||||
"ai-tokenizer": "^1.0.6",
|
||||
|
|
@ -12854,6 +12855,19 @@
|
|||
"node": ">=14"
|
||||
}
|
||||
},
|
||||
"node_modules/@opentelemetry/winston-transport": {
|
||||
"version": "0.28.0",
|
||||
"resolved": "https://registry.npmjs.org/@opentelemetry/winston-transport/-/winston-transport-0.28.0.tgz",
|
||||
"integrity": "sha512-MHfFCGXB8UlIvIcqjmHQ8DwZa8ie7e9lKoddhfMvUOdT+sX04rsS2KG8zi8stdV7M9T+G4bbEZlrvZblpgqppA==",
|
||||
"license": "Apache-2.0",
|
||||
"dependencies": {
|
||||
"@opentelemetry/api-logs": "^0.218.0",
|
||||
"winston-transport": "4.*"
|
||||
},
|
||||
"engines": {
|
||||
"node": "^18.19.0 || >=20.6.0"
|
||||
}
|
||||
},
|
||||
"node_modules/@oxc-project/types": {
|
||||
"version": "0.147.0",
|
||||
"resolved": "https://registry.npmjs.org/@oxc-project/types/-/types-0.147.0.tgz",
|
||||
|
|
@ -42652,6 +42666,7 @@
|
|||
"@opentelemetry/resources": "^2.6.1",
|
||||
"@opentelemetry/sdk-node": "^0.218.0",
|
||||
"@opentelemetry/semantic-conventions": "^1.39.0",
|
||||
"@opentelemetry/winston-transport": "^0.28.0",
|
||||
"@redis/client": "5.10.0",
|
||||
"@smithy/node-http-handler": "^4.4.5",
|
||||
"ai-tokenizer": "^1.0.6",
|
||||
|
|
|
|||
|
|
@ -126,6 +126,7 @@
|
|||
"@opentelemetry/resources": "^2.6.1",
|
||||
"@opentelemetry/sdk-node": "^0.218.0",
|
||||
"@opentelemetry/semantic-conventions": "^1.39.0",
|
||||
"@opentelemetry/winston-transport": "^0.28.0",
|
||||
"@redis/client": "5.10.0",
|
||||
"@smithy/node-http-handler": "^4.4.5",
|
||||
"ai-tokenizer": "^1.0.6",
|
||||
|
|
|
|||
|
|
@ -5,6 +5,9 @@ describe('getTelemetryConfig', () => {
|
|||
const config = getTelemetryConfig({});
|
||||
|
||||
expect(config.enabled).toBe(false);
|
||||
expect(config.tracingEnabled).toBe(false);
|
||||
expect(config.logsEnabled).toBe(false);
|
||||
expect(config.logsLevel).toBe('info');
|
||||
expect(config.sdkDisabled).toBe(false);
|
||||
expect(config.serviceName).toBe('librechat');
|
||||
expect(config.healthPath).toBe('/health');
|
||||
|
|
@ -17,13 +20,30 @@ describe('getTelemetryConfig', () => {
|
|||
expect(getTelemetryConfig({ OTEL_TRACING_ENABLED: 'false' }).enabled).toBe(false);
|
||||
});
|
||||
|
||||
it('lets OTEL_SDK_DISABLED override tracing enablement', () => {
|
||||
it('enables logs only when OTEL_LOGS_ENABLED is true', () => {
|
||||
const config = getTelemetryConfig({ OTEL_LOGS_ENABLED: 'true' });
|
||||
|
||||
expect(config.enabled).toBe(true);
|
||||
expect(config.logsEnabled).toBe(true);
|
||||
expect(config.tracingEnabled).toBe(false);
|
||||
expect(getTelemetryConfig({ OTEL_LOGS_ENABLED: 'false' }).logsEnabled).toBe(false);
|
||||
});
|
||||
|
||||
it('normalizes OTEL_LOGS_LEVEL', () => {
|
||||
expect(getTelemetryConfig({ OTEL_LOGS_LEVEL: ' Warn ' }).logsLevel).toBe('warn');
|
||||
expect(getTelemetryConfig({ OTEL_LOGS_LEVEL: ' ' }).logsLevel).toBe('info');
|
||||
});
|
||||
|
||||
it('lets OTEL_SDK_DISABLED override tracing and logs enablement', () => {
|
||||
const config = getTelemetryConfig({
|
||||
OTEL_LOGS_ENABLED: 'true',
|
||||
OTEL_SDK_DISABLED: 'true',
|
||||
OTEL_TRACING_ENABLED: 'true',
|
||||
});
|
||||
|
||||
expect(config.enabled).toBe(false);
|
||||
expect(config.tracingEnabled).toBe(false);
|
||||
expect(config.logsEnabled).toBe(false);
|
||||
expect(config.sdkDisabled).toBe(true);
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
const DEFAULT_SERVICE_NAME = 'librechat';
|
||||
export const DEFAULT_HEALTH_PATH = '/health';
|
||||
export const DEFAULT_LOGS_LEVEL = 'info';
|
||||
|
||||
export type TelemetryStatus = 'disabled' | 'failed' | 'started' | 'starting' | 'stopped';
|
||||
|
||||
|
|
@ -7,9 +8,12 @@ export interface TelemetryConfig {
|
|||
enabled: boolean;
|
||||
healthPath: string;
|
||||
ioredisTracingEnabled: boolean;
|
||||
logsEnabled: boolean;
|
||||
logsLevel: string;
|
||||
sdkDisabled: boolean;
|
||||
serviceName: string;
|
||||
serviceVersion?: string;
|
||||
tracingEnabled: boolean;
|
||||
}
|
||||
|
||||
function isTruthy(value?: string | boolean | null): boolean {
|
||||
|
|
@ -29,18 +33,23 @@ function normalizeEnvValue(value?: string): string | undefined {
|
|||
|
||||
export function getTelemetryConfig(env: NodeJS.ProcessEnv = process.env): TelemetryConfig {
|
||||
const sdkDisabled = isTruthy(env.OTEL_SDK_DISABLED);
|
||||
const enabled = isTruthy(env.OTEL_TRACING_ENABLED) && !sdkDisabled;
|
||||
const tracingEnabled = isTruthy(env.OTEL_TRACING_ENABLED) && !sdkDisabled;
|
||||
const logsEnabled = isTruthy(env.OTEL_LOGS_ENABLED) && !sdkDisabled;
|
||||
const serviceName = normalizeEnvValue(env.OTEL_SERVICE_NAME) ?? DEFAULT_SERVICE_NAME;
|
||||
const serviceVersion =
|
||||
normalizeEnvValue(env.OTEL_SERVICE_VERSION) ?? normalizeEnvValue(env.npm_package_version);
|
||||
const ioredisTracingEnabled = isTruthy(env.OTEL_IOREDIS_TRACING_ENABLED);
|
||||
const logsLevel = normalizeEnvValue(env.OTEL_LOGS_LEVEL)?.toLowerCase() ?? DEFAULT_LOGS_LEVEL;
|
||||
|
||||
return {
|
||||
enabled,
|
||||
enabled: tracingEnabled || logsEnabled,
|
||||
serviceName,
|
||||
ioredisTracingEnabled,
|
||||
logsEnabled,
|
||||
logsLevel,
|
||||
sdkDisabled,
|
||||
serviceVersion,
|
||||
tracingEnabled,
|
||||
healthPath: DEFAULT_HEALTH_PATH,
|
||||
};
|
||||
}
|
||||
|
|
|
|||
144
packages/api/src/telemetry/grpc.spec.ts
Normal file
144
packages/api/src/telemetry/grpc.spec.ts
Normal file
|
|
@ -0,0 +1,144 @@
|
|||
import { trace } from '@opentelemetry/api';
|
||||
import { logger } from '@librechat/data-schemas';
|
||||
import { Server, ServerCredentials } from '@grpc/grpc-js';
|
||||
import type { sendUnaryData, ServerUnaryCall, ServiceDefinition } from '@grpc/grpc-js';
|
||||
import { initializeTelemetry, resetTelemetryForTests, shutdownTelemetry } from './sdk';
|
||||
|
||||
const LOGS_EXPORT_PATH = '/opentelemetry.proto.collector.logs.v1.LogsService/Export';
|
||||
const TRACES_EXPORT_PATH = '/opentelemetry.proto.collector.trace.v1.TraceService/Export';
|
||||
const SERVICE_NAME = 'librechat-grpc-spec';
|
||||
const SPAN_NAME = 'grpc-spec-span';
|
||||
const INFO_MESSAGE = 'grpc-spec-info-message';
|
||||
const DEBUG_MESSAGE = 'grpc-spec-debug-message';
|
||||
const SECRET_SUFFIX = 'GRPCSPECSECRET1234567890';
|
||||
const EXPORTER_ENV_KEYS = [
|
||||
'OTEL_EXPORTER_OTLP_ENDPOINT',
|
||||
'OTEL_EXPORTER_OTLP_PROTOCOL',
|
||||
'OTEL_METRICS_EXPORTER',
|
||||
'OTEL_NODE_RESOURCE_DETECTORS',
|
||||
] as const;
|
||||
|
||||
type ExportHandler = (
|
||||
call: ServerUnaryCall<Buffer, Buffer>,
|
||||
callback: sendUnaryData<Buffer>,
|
||||
) => void;
|
||||
|
||||
const identity = (value: Buffer): Buffer => value;
|
||||
|
||||
function createExportService(path: string): ServiceDefinition {
|
||||
return {
|
||||
Export: {
|
||||
path,
|
||||
requestStream: false,
|
||||
responseStream: false,
|
||||
requestSerialize: identity,
|
||||
requestDeserialize: identity,
|
||||
responseSerialize: identity,
|
||||
responseDeserialize: identity,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function createExportHandler(received: Buffer[]): ExportHandler {
|
||||
return (call, callback) => {
|
||||
received.push(call.request);
|
||||
callback(null, Buffer.alloc(0));
|
||||
};
|
||||
}
|
||||
|
||||
function bindServer(server: Server): Promise<number> {
|
||||
return new Promise((resolve, reject) => {
|
||||
server.bindAsync('127.0.0.1:0', ServerCredentials.createInsecure(), (error, port) => {
|
||||
if (error) {
|
||||
reject(error);
|
||||
return;
|
||||
}
|
||||
resolve(port);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Exercises the real Node SDK, OTLP/gRPC exporters and winston bridge against an
|
||||
* in-process gRPC collector, so the environment-driven protocol selection is
|
||||
* verified end to end rather than assumed.
|
||||
*/
|
||||
describe('OTLP/gRPC export', () => {
|
||||
const server = new Server();
|
||||
const receivedLogs: Buffer[] = [];
|
||||
const receivedTraces: Buffer[] = [];
|
||||
const previousEnv: Partial<Record<(typeof EXPORTER_ENV_KEYS)[number], string | undefined>> = {};
|
||||
let startedStatus = '';
|
||||
let traceId = '';
|
||||
|
||||
beforeAll(async () => {
|
||||
server.addService(createExportService(LOGS_EXPORT_PATH), {
|
||||
Export: createExportHandler(receivedLogs),
|
||||
});
|
||||
server.addService(createExportService(TRACES_EXPORT_PATH), {
|
||||
Export: createExportHandler(receivedTraces),
|
||||
});
|
||||
const port = await bindServer(server);
|
||||
server.start();
|
||||
|
||||
for (const key of EXPORTER_ENV_KEYS) {
|
||||
previousEnv[key] = process.env[key];
|
||||
}
|
||||
process.env.OTEL_EXPORTER_OTLP_ENDPOINT = `http://127.0.0.1:${port}`;
|
||||
process.env.OTEL_EXPORTER_OTLP_PROTOCOL = 'grpc';
|
||||
process.env.OTEL_METRICS_EXPORTER = 'none';
|
||||
process.env.OTEL_NODE_RESOURCE_DETECTORS = 'none';
|
||||
|
||||
const controller = initializeTelemetry({
|
||||
OTEL_LOGS_ENABLED: 'true',
|
||||
OTEL_SERVICE_NAME: SERVICE_NAME,
|
||||
OTEL_TRACING_ENABLED: 'true',
|
||||
});
|
||||
startedStatus = controller.status;
|
||||
|
||||
trace.getTracer('librechat.grpc.spec').startActiveSpan(SPAN_NAME, (span) => {
|
||||
traceId = span.spanContext().traceId;
|
||||
logger.info(`${INFO_MESSAGE} sk-${SECRET_SUFFIX}`);
|
||||
logger.debug(DEBUG_MESSAGE);
|
||||
span.end();
|
||||
});
|
||||
|
||||
await shutdownTelemetry();
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await resetTelemetryForTests();
|
||||
for (const key of EXPORTER_ENV_KEYS) {
|
||||
const value = previousEnv[key];
|
||||
if (value === undefined) {
|
||||
delete process.env[key];
|
||||
} else {
|
||||
process.env[key] = value;
|
||||
}
|
||||
}
|
||||
server.forceShutdown();
|
||||
});
|
||||
|
||||
it('starts the Node SDK with gRPC exporters', () => {
|
||||
expect(startedStatus).toBe('started');
|
||||
});
|
||||
|
||||
it('exports winston records over gRPC, redacted and correlated with the active span', () => {
|
||||
const payload = Buffer.concat(receivedLogs);
|
||||
|
||||
expect(receivedLogs.length).toBeGreaterThan(0);
|
||||
expect(payload.includes(INFO_MESSAGE)).toBe(true);
|
||||
expect(payload.includes(SERVICE_NAME)).toBe(true);
|
||||
expect(payload.includes(Buffer.from(traceId, 'hex'))).toBe(true);
|
||||
expect(payload.includes(SECRET_SUFFIX)).toBe(false);
|
||||
expect(payload.includes(DEBUG_MESSAGE)).toBe(false);
|
||||
});
|
||||
|
||||
it('exports spans over gRPC', () => {
|
||||
const payload = Buffer.concat(receivedTraces);
|
||||
|
||||
expect(receivedTraces.length).toBeGreaterThan(0);
|
||||
expect(payload.includes(SPAN_NAME)).toBe(true);
|
||||
expect(payload.includes(Buffer.from(traceId, 'hex'))).toBe(true);
|
||||
});
|
||||
});
|
||||
56
packages/api/src/telemetry/logs.spec.ts
Normal file
56
packages/api/src/telemetry/logs.spec.ts
Normal file
|
|
@ -0,0 +1,56 @@
|
|||
import { logger } from '@librechat/data-schemas';
|
||||
import { OpenTelemetryTransportV3 } from '@opentelemetry/winston-transport';
|
||||
import { attachLogsTransport, detachLogsTransport, getLogsTransport } from './logs';
|
||||
import { getTelemetryConfig } from './config';
|
||||
|
||||
function countOtelTransports(): number {
|
||||
return logger.transports.filter((transport) => transport instanceof OpenTelemetryTransportV3)
|
||||
.length;
|
||||
}
|
||||
|
||||
describe('logs transport lifecycle', () => {
|
||||
let emitWarningSpy: jest.SpyInstance;
|
||||
|
||||
beforeEach(() => {
|
||||
emitWarningSpy = jest.spyOn(process, 'emitWarning').mockImplementation(() => true);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
detachLogsTransport();
|
||||
emitWarningSpy.mockRestore();
|
||||
});
|
||||
|
||||
it('adds one OpenTelemetry transport to the shared logger at the configured level', () => {
|
||||
const config = getTelemetryConfig({ OTEL_LOGS_ENABLED: 'true', OTEL_LOGS_LEVEL: 'warn' });
|
||||
|
||||
attachLogsTransport(config);
|
||||
attachLogsTransport(config);
|
||||
|
||||
expect(countOtelTransports()).toBe(1);
|
||||
expect(getLogsTransport()?.level).toBe('warn');
|
||||
expect(emitWarningSpy).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('falls back to info and warns when OTEL_LOGS_LEVEL is unknown', () => {
|
||||
attachLogsTransport(getTelemetryConfig({ OTEL_LOGS_ENABLED: 'true', OTEL_LOGS_LEVEL: 'loud' }));
|
||||
|
||||
expect(getLogsTransport()?.level).toBe('info');
|
||||
expect(emitWarningSpy).toHaveBeenCalledWith(
|
||||
expect.stringContaining('Ignoring unknown OTEL_LOGS_LEVEL "loud"; using "info"'),
|
||||
{ code: 'LIBRECHAT_OTEL' },
|
||||
);
|
||||
});
|
||||
|
||||
it('removes the transport from the shared logger on detach', () => {
|
||||
attachLogsTransport(getTelemetryConfig({ OTEL_LOGS_ENABLED: 'true' }));
|
||||
const transport = getLogsTransport();
|
||||
|
||||
detachLogsTransport();
|
||||
detachLogsTransport();
|
||||
|
||||
expect(transport).toBeDefined();
|
||||
expect(getLogsTransport()).toBeUndefined();
|
||||
expect(countOtelTransports()).toBe(0);
|
||||
expect(logger.transports).not.toContain(transport);
|
||||
});
|
||||
});
|
||||
57
packages/api/src/telemetry/logs.ts
Normal file
57
packages/api/src/telemetry/logs.ts
Normal file
|
|
@ -0,0 +1,57 @@
|
|||
import winston from 'winston';
|
||||
import { OpenTelemetryTransportV3 } from '@opentelemetry/winston-transport';
|
||||
import { logger, baseLogFormat, jsonTruncateFormat } from '@librechat/data-schemas';
|
||||
import type { TelemetryConfig } from './config';
|
||||
import { emitTelemetryWarning } from './warnings';
|
||||
import { DEFAULT_LOGS_LEVEL } from './config';
|
||||
|
||||
let activeTransport: OpenTelemetryTransportV3 | undefined;
|
||||
|
||||
/**
|
||||
* An unknown level name would resolve to `undefined` inside winston and silently
|
||||
* drop every record, so fall back and say so instead.
|
||||
*/
|
||||
function resolveLogsLevel(level: string): string {
|
||||
if (Object.prototype.hasOwnProperty.call(logger.levels, level)) {
|
||||
return level;
|
||||
}
|
||||
|
||||
emitTelemetryWarning(
|
||||
`Ignoring unknown OTEL_LOGS_LEVEL "${level}"; using "${DEFAULT_LOGS_LEVEL}". ` +
|
||||
`Expected one of: ${Object.keys(logger.levels).join(', ')}.`,
|
||||
);
|
||||
return DEFAULT_LOGS_LEVEL;
|
||||
}
|
||||
|
||||
/**
|
||||
* Bridges the shared winston logger into the OpenTelemetry logs pipeline. The
|
||||
* transport emits through the global LoggerProvider registered by the Node SDK,
|
||||
* and records created while a span is active inherit its trace context.
|
||||
* Records pass through the same redaction as file and JSON console output
|
||||
* before they leave the process.
|
||||
*/
|
||||
export function attachLogsTransport(config: TelemetryConfig): void {
|
||||
if (activeTransport) {
|
||||
return;
|
||||
}
|
||||
|
||||
const transport = new OpenTelemetryTransportV3({
|
||||
level: resolveLogsLevel(config.logsLevel),
|
||||
format: winston.format.combine(baseLogFormat, jsonTruncateFormat()),
|
||||
});
|
||||
logger.add(transport);
|
||||
activeTransport = transport;
|
||||
}
|
||||
|
||||
export function detachLogsTransport(): void {
|
||||
if (!activeTransport) {
|
||||
return;
|
||||
}
|
||||
|
||||
logger.remove(activeTransport);
|
||||
activeTransport = undefined;
|
||||
}
|
||||
|
||||
export function getLogsTransport(): OpenTelemetryTransportV3 | undefined {
|
||||
return activeTransport;
|
||||
}
|
||||
|
|
@ -1,8 +1,10 @@
|
|||
import { Socket } from 'node:net';
|
||||
import { IncomingMessage } from 'node:http';
|
||||
import { Agent as HttpsAgent } from 'node:https';
|
||||
import type { NodeSDKConfiguration } from '@opentelemetry/sdk-node';
|
||||
import type { RequestOptions } from 'node:http';
|
||||
import type { Span } from '@opentelemetry/api';
|
||||
import { attachLogsTransport, detachLogsTransport } from './logs';
|
||||
|
||||
interface HttpInstrumentationOptions {
|
||||
requestHook?: (span: Span, request: object) => void;
|
||||
|
|
@ -16,7 +18,7 @@ interface UndiciInstrumentationOptions {
|
|||
|
||||
const mockStart = jest.fn();
|
||||
const mockShutdown = jest.fn();
|
||||
const mockNodeSDK = jest.fn(() => ({
|
||||
const mockNodeSDK = jest.fn((_configuration?: Partial<NodeSDKConfiguration>) => ({
|
||||
start: mockStart,
|
||||
shutdown: mockShutdown,
|
||||
}));
|
||||
|
|
@ -34,83 +36,67 @@ const mockUndiciInstrumentation = jest.fn((options?: UndiciInstrumentationOption
|
|||
}));
|
||||
const mockResourceFromAttributes = jest.fn((attributes: object) => ({ attributes }));
|
||||
|
||||
jest.mock(
|
||||
'@opentelemetry/sdk-node',
|
||||
() => ({
|
||||
NodeSDK: mockNodeSDK,
|
||||
}),
|
||||
{ virtual: true },
|
||||
);
|
||||
jest.mock('@opentelemetry/sdk-node', () => ({
|
||||
NodeSDK: mockNodeSDK,
|
||||
}));
|
||||
|
||||
jest.mock(
|
||||
'@opentelemetry/instrumentation-express',
|
||||
() => ({
|
||||
ExpressInstrumentation: mockExpressInstrumentation,
|
||||
}),
|
||||
{ virtual: true },
|
||||
);
|
||||
jest.mock('@opentelemetry/instrumentation-express', () => ({
|
||||
ExpressInstrumentation: mockExpressInstrumentation,
|
||||
}));
|
||||
|
||||
jest.mock(
|
||||
'@opentelemetry/instrumentation-http',
|
||||
() => ({
|
||||
HttpInstrumentation: mockHttpInstrumentation,
|
||||
}),
|
||||
{ virtual: true },
|
||||
);
|
||||
jest.mock('@opentelemetry/instrumentation-http', () => ({
|
||||
HttpInstrumentation: mockHttpInstrumentation,
|
||||
}));
|
||||
|
||||
jest.mock(
|
||||
'@opentelemetry/instrumentation-ioredis',
|
||||
() => ({
|
||||
IORedisInstrumentation: mockIORedisInstrumentation,
|
||||
}),
|
||||
{ virtual: true },
|
||||
);
|
||||
jest.mock('@opentelemetry/instrumentation-ioredis', () => ({
|
||||
IORedisInstrumentation: mockIORedisInstrumentation,
|
||||
}));
|
||||
|
||||
jest.mock(
|
||||
'@opentelemetry/instrumentation-mongodb',
|
||||
() => ({
|
||||
MongoDBInstrumentation: mockMongoDBInstrumentation,
|
||||
}),
|
||||
{ virtual: true },
|
||||
);
|
||||
jest.mock('@opentelemetry/instrumentation-mongodb', () => ({
|
||||
MongoDBInstrumentation: mockMongoDBInstrumentation,
|
||||
}));
|
||||
|
||||
jest.mock(
|
||||
'@opentelemetry/instrumentation-mongoose',
|
||||
() => ({
|
||||
MongooseInstrumentation: mockMongooseInstrumentation,
|
||||
}),
|
||||
{ virtual: true },
|
||||
);
|
||||
jest.mock('@opentelemetry/instrumentation-mongoose', () => ({
|
||||
MongooseInstrumentation: mockMongooseInstrumentation,
|
||||
}));
|
||||
|
||||
jest.mock(
|
||||
'@opentelemetry/instrumentation-undici',
|
||||
() => ({
|
||||
UndiciInstrumentation: mockUndiciInstrumentation,
|
||||
}),
|
||||
{ virtual: true },
|
||||
);
|
||||
jest.mock('@opentelemetry/instrumentation-undici', () => ({
|
||||
UndiciInstrumentation: mockUndiciInstrumentation,
|
||||
}));
|
||||
|
||||
jest.mock(
|
||||
'@opentelemetry/resources',
|
||||
() => ({
|
||||
resourceFromAttributes: mockResourceFromAttributes,
|
||||
}),
|
||||
{ virtual: true },
|
||||
);
|
||||
jest.mock('@opentelemetry/resources', () => ({
|
||||
resourceFromAttributes: mockResourceFromAttributes,
|
||||
}));
|
||||
|
||||
jest.mock(
|
||||
'@opentelemetry/semantic-conventions',
|
||||
() => ({
|
||||
ATTR_SERVICE_NAME: 'service.name',
|
||||
ATTR_SERVICE_VERSION: 'service.version',
|
||||
}),
|
||||
{ virtual: true },
|
||||
);
|
||||
jest.mock('@opentelemetry/semantic-conventions', () => ({
|
||||
ATTR_SERVICE_NAME: 'service.name',
|
||||
ATTR_SERVICE_VERSION: 'service.version',
|
||||
}));
|
||||
|
||||
jest.mock('../app/shutdown', () => ({
|
||||
registerShutdownTask: jest.fn(),
|
||||
}));
|
||||
|
||||
jest.mock('./logs', () => ({
|
||||
attachLogsTransport: jest.fn(),
|
||||
detachLogsTransport: jest.fn(),
|
||||
}));
|
||||
|
||||
const mockAttachLogsTransport = attachLogsTransport as jest.MockedFunction<
|
||||
typeof attachLogsTransport
|
||||
>;
|
||||
const mockDetachLogsTransport = detachLogsTransport as jest.MockedFunction<
|
||||
typeof detachLogsTransport
|
||||
>;
|
||||
|
||||
function getSdkConfiguration(): Partial<NodeSDKConfiguration> {
|
||||
const configuration = mockNodeSDK.mock.calls[0]?.[0];
|
||||
if (!configuration) {
|
||||
throw new Error('NodeSDK was not constructed');
|
||||
}
|
||||
return configuration;
|
||||
}
|
||||
|
||||
describe('telemetry SDK lifecycle', () => {
|
||||
let emitWarningSpy: jest.SpyInstance;
|
||||
let getTelemetryRequestSpan: (typeof import('./sdk'))['getTelemetryRequestSpan'];
|
||||
|
|
@ -193,6 +179,84 @@ describe('telemetry SDK lifecycle', () => {
|
|||
expect(mockMongooseInstrumentation).toHaveBeenCalledTimes(1);
|
||||
expect(mockIORedisInstrumentation).not.toHaveBeenCalled();
|
||||
expect(mockUndiciInstrumentation).toHaveBeenCalledTimes(1);
|
||||
|
||||
const configuration = getSdkConfiguration();
|
||||
expect(configuration.instrumentations).toHaveLength(5);
|
||||
expect(configuration.logRecordProcessors).toEqual([]);
|
||||
expect(configuration).not.toHaveProperty('spanProcessors');
|
||||
expect(configuration).not.toHaveProperty('metricReaders');
|
||||
expect(mockAttachLogsTransport).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('starts logs without instrumentations or trace export when only logs are enabled', () => {
|
||||
const controller = initializeTelemetry({
|
||||
OTEL_LOGS_ENABLED: 'true',
|
||||
OTEL_LOGS_LEVEL: 'warn',
|
||||
});
|
||||
|
||||
expect(controller.enabled).toBe(true);
|
||||
expect(controller.status).toBe('started');
|
||||
expect(mockHttpInstrumentation).not.toHaveBeenCalled();
|
||||
expect(mockExpressInstrumentation).not.toHaveBeenCalled();
|
||||
expect(mockMongoDBInstrumentation).not.toHaveBeenCalled();
|
||||
expect(mockMongooseInstrumentation).not.toHaveBeenCalled();
|
||||
expect(mockUndiciInstrumentation).not.toHaveBeenCalled();
|
||||
|
||||
const configuration = getSdkConfiguration();
|
||||
expect(configuration.instrumentations).toEqual([]);
|
||||
expect(configuration.spanProcessors).toEqual([]);
|
||||
expect(configuration.metricReaders).toEqual([]);
|
||||
expect(configuration).not.toHaveProperty('logRecordProcessors');
|
||||
expect(mockAttachLogsTransport).toHaveBeenCalledTimes(1);
|
||||
expect(mockAttachLogsTransport).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ logsEnabled: true, logsLevel: 'warn', tracingEnabled: false }),
|
||||
);
|
||||
});
|
||||
|
||||
it('keeps instrumentations and leaves log export to the SDK when both signals are enabled', () => {
|
||||
initializeTelemetry({
|
||||
OTEL_LOGS_ENABLED: 'true',
|
||||
OTEL_TRACING_ENABLED: 'true',
|
||||
});
|
||||
|
||||
const configuration = getSdkConfiguration();
|
||||
expect(configuration.instrumentations).toHaveLength(5);
|
||||
expect(configuration).not.toHaveProperty('spanProcessors');
|
||||
expect(configuration).not.toHaveProperty('metricReaders');
|
||||
expect(configuration).not.toHaveProperty('logRecordProcessors');
|
||||
expect(mockAttachLogsTransport).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('attaches the logs transport only after an async SDK start succeeds', async () => {
|
||||
let resolveStart: () => void = () => undefined;
|
||||
mockStart.mockReturnValueOnce(
|
||||
new Promise<void>((resolve) => {
|
||||
resolveStart = resolve;
|
||||
}),
|
||||
);
|
||||
|
||||
const controller = initializeTelemetry({ OTEL_LOGS_ENABLED: 'true' });
|
||||
|
||||
expect(controller.status).toBe('starting');
|
||||
expect(mockAttachLogsTransport).not.toHaveBeenCalled();
|
||||
|
||||
resolveStart();
|
||||
await controller.shutdown();
|
||||
|
||||
expect(mockAttachLogsTransport).toHaveBeenCalledTimes(1);
|
||||
expect(mockDetachLogsTransport).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('detaches the logs transport only once the SDK shuts down', async () => {
|
||||
mockShutdown.mockRejectedValueOnce(new Error('shutdown failed'));
|
||||
initializeTelemetry({ OTEL_LOGS_ENABLED: 'true' });
|
||||
mockDetachLogsTransport.mockClear();
|
||||
|
||||
await expect(shutdownTelemetry()).rejects.toThrow('shutdown failed');
|
||||
expect(mockDetachLogsTransport).not.toHaveBeenCalled();
|
||||
|
||||
await shutdownTelemetry();
|
||||
expect(mockDetachLogsTransport).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('enables ioredis instrumentation when explicitly configured', () => {
|
||||
|
|
|
|||
|
|
@ -12,6 +12,8 @@ import type { NodeSDKConfiguration } from '@opentelemetry/sdk-node';
|
|||
import type { Span, Attributes } from '@opentelemetry/api';
|
||||
import type { RequestOptions } from 'node:http';
|
||||
import type { TelemetryConfig, TelemetryStatus } from './config';
|
||||
import { emitTelemetryWarning, getErrorMessage } from './warnings';
|
||||
import { attachLogsTransport, detachLogsTransport } from './logs';
|
||||
import { registerShutdownTask } from '../app/shutdown';
|
||||
import { getTelemetryConfig } from './config';
|
||||
|
||||
|
|
@ -21,7 +23,6 @@ export interface TelemetryController {
|
|||
shutdown: () => Promise<void>;
|
||||
}
|
||||
|
||||
const WARNING_CODE = 'LIBRECHAT_OTEL';
|
||||
const REDACTED_QUERY_VALUE = '[REDACTED]';
|
||||
const SIGNAL_SHUTDOWN_TIMEOUT_MS = 5_000;
|
||||
|
||||
|
|
@ -296,7 +297,7 @@ function getResourceAttributes(config: TelemetryConfig): Attributes {
|
|||
return attributes;
|
||||
}
|
||||
|
||||
function createSdk(config: TelemetryConfig): NodeSDK {
|
||||
function createInstrumentations(config: TelemetryConfig): NodeSDKConfiguration['instrumentations'] {
|
||||
const instrumentations: NodeSDKConfiguration['instrumentations'] = [
|
||||
new HttpInstrumentation({
|
||||
headersToSpanAttributes: {
|
||||
|
|
@ -327,11 +328,31 @@ function createSdk(config: TelemetryConfig): NodeSDK {
|
|||
instrumentations.push(new IORedisInstrumentation());
|
||||
}
|
||||
|
||||
return instrumentations;
|
||||
}
|
||||
|
||||
/**
|
||||
* Each signal is gated explicitly: an empty processor or reader list keeps the
|
||||
* Node SDK from falling back to its environment defaults, which would otherwise
|
||||
* stand up an OTLP exporter for a signal that was never enabled. Enabled signals
|
||||
* keep the standard `OTEL_*_EXPORTER` / `OTEL_EXPORTER_OTLP_*` resolution, so
|
||||
* `OTEL_EXPORTER_OTLP_PROTOCOL=grpc` selects the gRPC exporter for every signal.
|
||||
*/
|
||||
function createSdk(config: TelemetryConfig): NodeSDK {
|
||||
const sdkConfig: Partial<NodeSDKConfiguration> = {
|
||||
resource: resourceFromAttributes(getResourceAttributes(config)),
|
||||
instrumentations,
|
||||
instrumentations: config.tracingEnabled ? createInstrumentations(config) : [],
|
||||
};
|
||||
|
||||
if (!config.tracingEnabled) {
|
||||
sdkConfig.spanProcessors = [];
|
||||
sdkConfig.metricReaders = [];
|
||||
}
|
||||
|
||||
if (!config.logsEnabled) {
|
||||
sdkConfig.logRecordProcessors = [];
|
||||
}
|
||||
|
||||
return new NodeSDK(sdkConfig);
|
||||
}
|
||||
|
||||
|
|
@ -347,14 +368,6 @@ function startSdk(sdk: NodeSDK): void | Promise<void> {
|
|||
return (sdk as NodeSDK & { start: () => void | Promise<void> }).start();
|
||||
}
|
||||
|
||||
function emitWarning(message: string): void {
|
||||
process.emitWarning(message, { code: WARNING_CODE });
|
||||
}
|
||||
|
||||
function getErrorMessage(error: unknown): string {
|
||||
return error instanceof Error ? error.message : String(error);
|
||||
}
|
||||
|
||||
function isControllerEnabled(): boolean {
|
||||
return status === 'starting' || status === 'started';
|
||||
}
|
||||
|
|
@ -386,7 +399,7 @@ function ensureShutdownTaskRegistered(): void {
|
|||
'telemetry',
|
||||
() =>
|
||||
withTimeout(shutdownTelemetry(), SIGNAL_SHUTDOWN_TIMEOUT_MS).catch((error) => {
|
||||
emitWarning(`OpenTelemetry shutdown failed: ${getErrorMessage(error)}`);
|
||||
emitTelemetryWarning(`OpenTelemetry shutdown failed: ${getErrorMessage(error)}`);
|
||||
}),
|
||||
// Exporters close after instrumented modules have emitted their final shutdown spans.
|
||||
{ priority: -100 },
|
||||
|
|
@ -409,6 +422,15 @@ function withTimeout<T>(promise: Promise<T>, timeoutMs: number): Promise<T> {
|
|||
});
|
||||
}
|
||||
|
||||
function markStarted(sdk: NodeSDK, config: TelemetryConfig): void {
|
||||
activeSdk = sdk;
|
||||
status = 'started';
|
||||
ensureShutdownTaskRegistered();
|
||||
if (config.logsEnabled) {
|
||||
attachLogsTransport(config);
|
||||
}
|
||||
}
|
||||
|
||||
export function initializeTelemetry(env: NodeJS.ProcessEnv = process.env): TelemetryController {
|
||||
if (activeSdk || pendingSdk) {
|
||||
return makeController();
|
||||
|
|
@ -430,16 +452,14 @@ export function initializeTelemetry(env: NodeJS.ProcessEnv = process.env): Telem
|
|||
.then(() => {
|
||||
if (pendingSdk === sdk) {
|
||||
pendingSdk = undefined;
|
||||
activeSdk = sdk;
|
||||
status = 'started';
|
||||
ensureShutdownTaskRegistered();
|
||||
markStarted(sdk, config);
|
||||
}
|
||||
})
|
||||
.catch((error) => {
|
||||
if (pendingSdk === sdk) {
|
||||
pendingSdk = undefined;
|
||||
status = 'failed';
|
||||
emitWarning(`OpenTelemetry initialization failed: ${getErrorMessage(error)}`);
|
||||
emitTelemetryWarning(`OpenTelemetry initialization failed: ${getErrorMessage(error)}`);
|
||||
}
|
||||
});
|
||||
startPromise = pendingStart;
|
||||
|
|
@ -451,13 +471,11 @@ export function initializeTelemetry(env: NodeJS.ProcessEnv = process.env): Telem
|
|||
return makeController();
|
||||
}
|
||||
|
||||
activeSdk = sdk;
|
||||
status = 'started';
|
||||
ensureShutdownTaskRegistered();
|
||||
markStarted(sdk, config);
|
||||
return makeController();
|
||||
} catch (error) {
|
||||
status = 'failed';
|
||||
emitWarning(`OpenTelemetry initialization failed: ${getErrorMessage(error)}`);
|
||||
emitTelemetryWarning(`OpenTelemetry initialization failed: ${getErrorMessage(error)}`);
|
||||
return makeController();
|
||||
}
|
||||
}
|
||||
|
|
@ -475,6 +493,7 @@ async function performShutdownTelemetry(): Promise<void> {
|
|||
const sdk = activeSdk;
|
||||
try {
|
||||
await sdk.shutdown();
|
||||
detachLogsTransport();
|
||||
activeSdk = undefined;
|
||||
status = 'stopped';
|
||||
} catch (error) {
|
||||
|
|
@ -505,6 +524,7 @@ export async function resetTelemetryForTests(): Promise<void> {
|
|||
await Promise.resolve(activeSdk.shutdown()).catch(() => undefined);
|
||||
}
|
||||
} finally {
|
||||
detachLogsTransport();
|
||||
activeSdk = undefined;
|
||||
pendingSdk = undefined;
|
||||
startPromise = undefined;
|
||||
|
|
|
|||
9
packages/api/src/telemetry/warnings.ts
Normal file
9
packages/api/src/telemetry/warnings.ts
Normal file
|
|
@ -0,0 +1,9 @@
|
|||
const WARNING_CODE = 'LIBRECHAT_OTEL';
|
||||
|
||||
export function emitTelemetryWarning(message: string): void {
|
||||
process.emitWarning(message, { code: WARNING_CODE });
|
||||
}
|
||||
|
||||
export function getErrorMessage(error: unknown): string {
|
||||
return error instanceof Error ? error.message : String(error);
|
||||
}
|
||||
|
|
@ -34,7 +34,7 @@ const level = (): string => {
|
|||
return env === 'development' ? 'debug' : 'warn';
|
||||
};
|
||||
|
||||
const fileFormat = winston.format.combine(
|
||||
export const baseLogFormat: winston.Logform.Format = winston.format.combine(
|
||||
redactFormat(),
|
||||
winston.format.timestamp({ format: () => new Date().toISOString() }),
|
||||
winston.format.errors({ stack: true }),
|
||||
|
|
@ -56,7 +56,7 @@ if (useFileLogging) {
|
|||
zippedArchive: true,
|
||||
maxSize: '20m',
|
||||
maxFiles: '14d',
|
||||
format: winston.format.combine(fileFormat, winston.format.json()),
|
||||
format: winston.format.combine(baseLogFormat, winston.format.json()),
|
||||
}),
|
||||
);
|
||||
|
||||
|
|
@ -69,7 +69,7 @@ if (useFileLogging) {
|
|||
zippedArchive: true,
|
||||
maxSize: '20m',
|
||||
maxFiles: '14d',
|
||||
format: winston.format.combine(fileFormat, debugTraverse),
|
||||
format: winston.format.combine(baseLogFormat, debugTraverse),
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
|
@ -100,8 +100,8 @@ if (useDebugConsole) {
|
|||
level: consoleLogLevel,
|
||||
silent: consoleSilent,
|
||||
format: useConsoleJson
|
||||
? winston.format.combine(fileFormat, jsonTruncateFormat(), winston.format.json())
|
||||
: winston.format.combine(fileFormat, debugTraverse),
|
||||
? winston.format.combine(baseLogFormat, jsonTruncateFormat(), winston.format.json())
|
||||
: winston.format.combine(baseLogFormat, debugTraverse),
|
||||
}),
|
||||
);
|
||||
} else if (useConsoleJson) {
|
||||
|
|
@ -109,7 +109,7 @@ if (useDebugConsole) {
|
|||
new winston.transports.Console({
|
||||
level: consoleLogLevel,
|
||||
silent: consoleSilent,
|
||||
format: winston.format.combine(fileFormat, jsonTruncateFormat(), winston.format.json()),
|
||||
format: winston.format.combine(baseLogFormat, jsonTruncateFormat(), winston.format.json()),
|
||||
}),
|
||||
);
|
||||
} else {
|
||||
|
|
|
|||
|
|
@ -80,9 +80,9 @@ export {
|
|||
AUDIT_ACTION_CATEGORY,
|
||||
} from './types/admin';
|
||||
export { GENESIS_HASH, PLATFORM_CHAIN_KEY } from './schema/auditLog';
|
||||
export { default as logger } from './config/winston';
|
||||
export { default as logger, baseLogFormat } from './config/winston';
|
||||
export { default as meiliLogger } from './config/meiliLogger';
|
||||
export { redactMessage } from './config/parsers';
|
||||
export { jsonTruncateFormat, redactMessage } from './config/parsers';
|
||||
export {
|
||||
tenantStorage,
|
||||
getTenantId,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue