Skip to content

赵月辉 / fastgpt-migrated

Go to a project
Commit 463733fc by Ryo Committed by GitHub
Options

refactor: record message into otel body instead of attributes (#6911) 

* fix: reduce otel log body duplication

* refactor: record message into otel body instead of attributes

* fix: trace and metric initialization

* fix: remove redundant traceId and spanId in the body
parent edbf57b2
Showing with 466 additions and 203 deletions
packages/service/common/tracing/client.ts
......@@ -6,7 +6,6 @@ import {
getCurrentSpanContext,
getTracer
} from '@fastgpt-sdk/otel/tracing';
import { withContext } from '../logger';
import { serviceEnv } from '../../env';
type SpanAttributeValue = string | number | boolean;
......@@ -102,24 +101,14 @@ export async function withActiveSpan<T>(
attributes: normalizeAttributes(options.attributes)
},
async (span: SpanLike) => {
const spanContext = span.spanContext();
return withContext(
{
traceId: spanContext.traceId,
spanId: spanContext.spanId
},
async () => {
try {
return await callback(span);
} catch (error) {
setSpanError(span, error);
throw error;
} finally {
span.end();
}
}
);
try {
return await callback(span);
} catch (error) {
setSpanError(span, error);
throw error;
} finally {
span.end();
}
}
);
}
......
pro @ 91c9337c
Subproject commit f1062d7ebf01ddb991f0c2dfead9f6d581eaed65
Subproject commit 91c9337cb64290f7fbaf68939545e432ddacbc7a
projects/app/src/instrumentation-node.ts
......@@ -32,6 +32,8 @@ export async function registerNodeInstrumentation() {
{ instrumentationCheck },
{ getErrText },
{ configureLogger, getLogger, LogCategories },
{ configureMetrics },
{ configureTracing },
{ InitialErrorEnum }
] = await Promise.all([
import('@fastgpt/service/common/mongo/init'),
......@@ -53,13 +55,16 @@ export async function registerNodeInstrumentation() {
import('@/service/common/system/health'),
import('@fastgpt/global/common/error/utils'),
import('@fastgpt/service/common/logger'),
import('@fastgpt/service/common/metrics'),
import('@fastgpt/service/common/tracing'),
import('@fastgpt/service/common/system/constants')
]);
await runInitializationStep({
step: 'configure-logger',
action: () => configureLogger()
});
await Promise.all([
runInitializationStep({ step: 'configure-tracing', action: () => configureTracing() }),
runInitializationStep({ step: 'configure-metrics', action: () => configureMetrics() }),
runInitializationStep({ step: 'configure-logger', action: () => configureLogger() })
]);
const logger = getLogger(LogCategories.SYSTEM);
logger.info('Starting system initialization...');
......
sdk/otel/README.md
......@@ -80,6 +80,14 @@ import { configureTracingFromEnv, getTracer } from '@fastgpt-sdk/otel/tracing';
- `OTEL_TRACES_SAMPLER`
- `OTEL_TRACES_SAMPLER_ARG`
## logger OTel 输出
OTel logger 默认会把日志事件写成结构化 `body`，把业务上下文保留在 body 中，`attributes` 只保留日志元信息。
- `body` 形如 `{ __log_message, ...properties }`：`__log_message` 是人读摘要，properties 是结构化上下文。
- `attributes` 不再复制 properties，只保留 `category` 等日志元信息，避免 body 与 attributes 重复。
- 遇到 Map、Set、Error、BigInt、循环引用、class instance 等 JS 特有值时，会先安全归一成普通对象/数组/标量，避免出现 `[object Object]` 或序列化异常。
## 说明
- 这个包当前是“整理好的统一入口”，不是“已经迁移完成的替换方案”。
......
sdk/otel/package.json
......@@ -30,6 +30,8 @@
"scripts": {
"build": "tsdown",
"dev": "tsdown --watch",
"lint": "eslint .",
"test": "vitest run --config vitest.config.ts",
"prepublishOnly": "pnpm build"
},
"keywords": [
......
sdk/otel/src/logger/index.ts
export { configureLogger, disposeLogger, getLogger } from './client';
export { withContext, withCategoryPrefix } from '@logtape/logtape';
export { getOpenTelemetrySink } from './otel';
export type {
BodyFormatter,
ExceptionAttributeMode,
ObjectRenderer,
OpenTelemetrySink,
OpenTelemetrySinkOptions
} from './otel';
export type { OpenTelemetrySink, OpenTelemetrySinkOptions } from './otel';
export type {
ConsoleLoggerOptions,
LogCategory,
......
sdk/otel/src/logger/otel.ts
import { getLogger, type Logger, type LogRecord, type Sink } from '@logtape/logtape';
import { diag, type DiagLogger, DiagLogLevel } from '@opentelemetry/api';
import { context, diag, type DiagLogger, DiagLogLevel } from '@opentelemetry/api';
import {
type AnyValue,
type Logger as OTLogger,
......@@ -37,18 +37,7 @@ type ILoggerProvider = LoggerProviderBase & {
shutdown?: () => Promise<void>;
};
export type ObjectRenderer = 'json' | 'inspect';
type Message = (string | null | undefined)[];
export type BodyFormatter = (message: Message) => AnyValue;
export type ExceptionAttributeMode = 'semconv' | 'raw' | false;
interface OpenTelemetrySinkOptionsBase {
messageType?: 'string' | 'array' | BodyFormatter;
objectRenderer?: ObjectRenderer;
exceptionAttributes?: ExceptionAttributeMode;
diagnostics?: boolean;
loggerName?: string;
}
......@@ -94,34 +83,18 @@ async function initializeLoggerProvider(
return loggerProvider;
}
function emitLogRecord(
logger: OTLogger,
record: LogRecord,
options: OpenTelemetrySinkOptions
): void {
const objectRenderer = options.objectRenderer ?? 'inspect';
const exceptionMode = options.exceptionAttributes ?? 'semconv';
const { category, level, message, timestamp, properties } = record;
function emitLogRecord(logger: OTLogger, record: LogRecord): void {
const { category, level, timestamp } = record;
const severityNumber = mapLevelToSeverityNumber(level);
const attributes = convertToAttributes(properties ?? {}, objectRenderer, exceptionMode);
attributes['category'] = [...category];
logger.emit({
severityNumber,
severityText: level,
body:
typeof options.messageType === 'function'
? convertMessageToCustomBodyFormat(
message,
objectRenderer,
exceptionMode,
options.messageType
)
: options.messageType === 'array'
? convertMessageToArray(message, objectRenderer, exceptionMode)
: convertMessageToString(message, objectRenderer, exceptionMode),
attributes,
body: convertRecordToStructuredBody(record),
attributes: {
category: [...category]
},
context: context.active(),
timestamp: new Date(timestamp)
} satisfies OTLogRecord);
}
......@@ -151,7 +124,7 @@ export function getOpenTelemetrySink(options: OpenTelemetrySinkOptions = {}): Op
if (category[0] === 'logtape' && category[1] === 'meta' && category[2] === 'otel') {
return;
}
emitLogRecord(logger, record, options);
emitLogRecord(logger, record);
},
{
ready: Promise.resolve(),
......@@ -177,7 +150,7 @@ export function getOpenTelemetrySink(options: OpenTelemetrySinkOptions = {}): Op
}
if (logger != null) {
emitLogRecord(logger, record, options);
emitLogRecord(logger, record);
return;
}
......@@ -193,14 +166,13 @@ export function getOpenTelemetrySink(options: OpenTelemetrySinkOptions = {}): Op
loggerProvider = provider;
logger = provider.getLogger(getOpenTelemetryLoggerName(options));
for (const pendingRecord of pendingRecords) {
emitLogRecord(logger, pendingRecord, options);
emitLogRecord(logger, pendingRecord);
}
pendingRecords = [];
})
.catch((error) => {
initError = error as Error;
pendingRecords = [];
// eslint-disable-next-line no-console
console.error('Failed to initialize OpenTelemetry logger:', error);
});
}
......@@ -227,43 +199,41 @@ export function getOpenTelemetrySink(options: OpenTelemetrySinkOptions = {}): Op
return sink;
}
function convertValueToAnyValue(
value: unknown,
objectRenderer: ObjectRenderer,
exceptionMode: ExceptionAttributeMode
): AnyValue | null {
type SafeNormalizeOptions = {
seen?: WeakSet<object>;
depth?: number;
maxDepth?: number;
maxKeys?: number;
bytesAsSummary?: boolean;
};
const defaultMaxNormalizeDepth = 8;
const defaultMaxObjectKeys = 128;
const reservedStructuredBodyKeys = new Set(['traceId', 'spanId']);
function convertValueToAnyValue(value: unknown): AnyValue | null {
if (value == null) return null;
if (typeof value === 'string' || typeof value === 'number' || typeof value === 'boolean') {
return value;
}
if (Array.isArray(value)) {
let primitiveType: string | null = null;
let isHomogeneous = true;
for (const item of value) {
if (item == null) continue;
const itemType = typeof item;
if (itemType !== 'string' && itemType !== 'number' && itemType !== 'boolean') {
isHomogeneous = false;
break;
}
if (primitiveType === null) {
primitiveType = itemType;
} else if (primitiveType !== itemType) {
isHomogeneous = false;
break;
}
}
const normalized = normalizeLogValue(value, { bytesAsSummary: false });
if (normalized == null) return null;
if (isHomogeneous && primitiveType !== null) {
return value as AnyValue;
}
if (
typeof normalized === 'string' ||
typeof normalized === 'number' ||
typeof normalized === 'boolean'
) {
return normalized;
}
if (normalized instanceof Uint8Array) return normalized;
if (Array.isArray(normalized)) {
const converted: AnyValue[] = [];
for (const item of value) {
const convertedItem = convertValueToAnyValue(item, objectRenderer, exceptionMode);
for (const item of normalized) {
const convertedItem = convertValueToAnyValue(item);
if (convertedItem !== null) {
converted.push(convertedItem);
}
......@@ -271,15 +241,10 @@ function convertValueToAnyValue(
return converted;
}
if (value instanceof Date) {
return value.toISOString();
}
if (value instanceof Error) {
const errorObj = serializeValue(value) as Record<string, unknown>;
if (typeof normalized === 'object') {
const converted: Record<string, AnyValue> = {};
for (const [key, val] of Object.entries(errorObj)) {
const convertedVal = convertValueToAnyValue(val, objectRenderer, exceptionMode);
for (const [key, val] of Object.entries(normalized as Record<string, unknown>)) {
const convertedVal = convertValueToAnyValue(val);
if (convertedVal !== null) {
converted[key] = convertedVal;
}
......@@ -287,58 +252,14 @@ function convertValueToAnyValue(
return converted;
}
if (typeof value === 'object') {
const proto = Object.getPrototypeOf(value);
const isPlainObject = proto === Object.prototype || proto === null;
if (isPlainObject) {
const converted: Record<string, AnyValue> = {};
for (const [key, val] of Object.entries(value as Record<string, unknown>)) {
const convertedVal = convertValueToAnyValue(val, objectRenderer, exceptionMode);
if (convertedVal !== null) {
converted[key] = convertedVal;
}
}
return converted;
}
if (objectRenderer === 'inspect') {
return nodeInspect(value);
}
return JSON.stringify(value);
}
return String(value);
}
function convertToAttributes(
properties: Record<string, unknown>,
objectRenderer: ObjectRenderer,
exceptionMode: ExceptionAttributeMode
): Record<string, AnyValue> {
const attributes: Record<string, AnyValue> = {};
for (const [name, value] of Object.entries(properties)) {
if (value == null) continue;
if (value instanceof Error && exceptionMode === 'semconv') {
attributes['exception.type'] = value.name;
attributes['exception.message'] = value.message;
if (typeof value.stack === 'string') {
attributes['exception.stacktrace'] = value.stack;
}
continue;
}
const convertedValue = convertValueToAnyValue(value, objectRenderer, exceptionMode);
if (convertedValue !== null) {
attributes[name] = convertedValue;
}
}
return attributes;
return String(normalized);
}
function serializeValue(value: unknown): unknown {
function serializeValue(value: unknown, seen = new WeakSet<object>()): unknown {
if (value instanceof Error) {
if (seen.has(value)) return '[Circular]';
seen.add(value);
const serialized: Record<string, unknown> = {
name: value.name,
message: value.message
......@@ -350,16 +271,16 @@ function serializeValue(value: unknown): unknown {
const cause = (value as { cause?: unknown }).cause;
if (cause !== undefined) {
serialized.cause = serializeValue(cause);
serialized.cause = serializeValue(cause, seen);
}
if (typeof AggregateError !== 'undefined' && value instanceof AggregateError) {
serialized.errors = value.errors.map(serializeValue);
serialized.errors = value.errors.map((error) => serializeValue(error, seen));
}
for (const key of Object.keys(value)) {
if (!(key in serialized)) {
serialized[key] = serializeValue((value as unknown as Record<string, unknown>)[key]);
serialized[key] = serializeValue((value as unknown as Record<string, unknown>)[key], seen);
}
}
......@@ -367,13 +288,16 @@ function serializeValue(value: unknown): unknown {
}
if (Array.isArray(value)) {
return value.map(serializeValue);
return value.map((item) => serializeValue(item, seen));
}
if (value !== null && typeof value === 'object') {
if (seen.has(value)) return '[Circular]';
seen.add(value);
const serialized: Record<string, unknown> = {};
for (const [key, val] of Object.entries(value)) {
serialized[key] = serializeValue(val);
serialized[key] = serializeValue(val, seen);
}
return serialized;
}
......@@ -381,66 +305,153 @@ function serializeValue(value: unknown): unknown {
return value;
}
function convertToString(
value: unknown,
objectRenderer: ObjectRenderer,
exceptionMode: ExceptionAttributeMode
): string | null | undefined {
if (value === null || value === undefined || typeof value === 'string') {
function normalizeLogValue(value: unknown, options: SafeNormalizeOptions = {}): unknown {
const {
seen = new WeakSet<object>(),
depth = 0,
maxDepth = defaultMaxNormalizeDepth,
maxKeys = defaultMaxObjectKeys,
bytesAsSummary = false
} = options;
if (value == null) return null;
const valueType = typeof value;
if (valueType === 'string' || valueType === 'number' || valueType === 'boolean') {
return value;
}
if (objectRenderer === 'inspect') return nodeInspect(value);
if (typeof value === 'number' || typeof value === 'boolean') {
return value.toString();
if (valueType === 'bigint') return value.toString();
if (valueType === 'symbol') return value.toString();
if (valueType === 'function') {
return `[Function ${(value as { name?: string }).name || 'anonymous'}]`;
}
if (value instanceof Date) return value.toISOString();
if (value instanceof Error && (exceptionMode === 'raw' || exceptionMode === 'semconv')) {
return JSON.stringify(serializeValue(value));
if (value instanceof Error) return serializeValue(value, seen);
if (value instanceof Uint8Array) {
return bytesAsSummary ? `[${value.constructor.name} length=${value.byteLength}]` : value;
}
return JSON.stringify(value);
}
function convertMessageToArray(
message: readonly unknown[],
objectRenderer: ObjectRenderer,
exceptionMode: ExceptionAttributeMode
): AnyValue {
const body: (string | null | undefined)[] = [];
for (let i = 0; i < message.length; i += 2) {
const msg = message[i] as string;
body.push(msg);
if (message.length <= i + 1) break;
const val = message[i + 1];
body.push(convertToString(val, objectRenderer, exceptionMode));
if (typeof value !== 'object') return String(value);
if (seen.has(value)) return '[Circular]';
if (depth >= maxDepth) return '[MaxDepth]';
seen.add(value);
if (Array.isArray(value)) {
return value.map((item) =>
normalizeLogValue(item, { seen, depth: depth + 1, maxDepth, maxKeys, bytesAsSummary })
);
}
return body;
if (value instanceof Map) {
const normalized: Record<string, unknown> = {};
let count = 0;
for (const [key, val] of value.entries()) {
if (count >= maxKeys) {
normalized.__truncated = true;
break;
}
normalized[String(key)] = normalizeLogValue(val, {
seen,
depth: depth + 1,
maxDepth,
maxKeys,
bytesAsSummary
});
count += 1;
}
return normalized;
}
if (value instanceof Set) {
return Array.from(value.values()).map((item) =>
normalizeLogValue(item, { seen, depth: depth + 1, maxDepth, maxKeys, bytesAsSummary })
);
}
const proto = Object.getPrototypeOf(value);
const isPlainObject = proto === Object.prototype || proto === null;
const normalized: Record<string, unknown> = {};
if (!isPlainObject && value.constructor?.name) {
normalized.__type = value.constructor.name;
}
const entries = Object.entries(value as Record<string, unknown>);
if (entries.length === 0 && !isPlainObject) {
return nodeInspect(value);
}
for (const [index, [key, val]] of entries.entries()) {
if (index >= maxKeys) {
normalized.__truncated = true;
break;
}
normalized[key] = normalizeLogValue(val, {
seen,
depth: depth + 1,
maxDepth,
maxKeys,
bytesAsSummary
});
}
return normalized;
}
function convertMessageToString(
message: readonly unknown[],
objectRenderer: ObjectRenderer,
exceptionMode: ExceptionAttributeMode
): AnyValue {
function convertMessageToText(record: LogRecord): AnyValue {
const { message } = record;
let body = '';
for (let i = 0; i < message.length; i += 2) {
const msg = message[i] as string;
body += msg;
if (message.length <= i + 1) break;
const val = message[i + 1];
const extra = convertToString(val, objectRenderer, exceptionMode);
body += extra ?? JSON.stringify(extra);
const value = message[i + 1];
if (value === null || value === undefined) continue;
if (typeof value === 'string' || typeof value === 'number' || typeof value === 'boolean') {
body += value.toString();
continue;
}
if (value instanceof Date) {
body += value.toISOString();
}
}
return body;
const trimmed = body.trimEnd().replace(/[:：,，]\s*$/, '');
return trimmed || getStructuredLogFallback(record);
}
function getRawMessageText(record: LogRecord): string {
if (typeof record.rawMessage === 'string') {
return record.rawMessage;
}
return Array.from(record.rawMessage).join('');
}
function getStructuredLogFallback(record: LogRecord): string {
const rawMessage = getRawMessageText(record);
return rawMessage === '{*}' ? 'structured log' : rawMessage;
}
function convertMessageToCustomBodyFormat(
message: readonly unknown[],
objectRenderer: ObjectRenderer,
exceptionMode: ExceptionAttributeMode,
bodyFormatter: BodyFormatter
): AnyValue {
const body = message.map((msg) => convertToString(msg, objectRenderer, exceptionMode));
return bodyFormatter(body);
function convertRecordToStructuredBody(record: LogRecord): AnyValue {
const body: Record<string, AnyValue> = {
__log_message: convertMessageToText(record)
};
for (const [key, value] of Object.entries(record.properties ?? {})) {
if (reservedStructuredBodyKeys.has(key)) continue;
const convertedValue = convertValueToAnyValue(value);
if (convertedValue !== null) {
body[key] = convertedValue;
}
}
return body;
}
class DiagLoggerAdaptor implements DiagLogger {
......
sdk/otel/test/logger/otel.test.ts 0 → 100644
import type { LogRecord as OTelLogRecord } from '@opentelemetry/api-logs';
import { describe, expect, it } from 'vitest';
import { createLoggerOptionsFromEnv } from '../../src/logger/env';
import { getOpenTelemetrySink } from '../../src/logger/otel';
function createRecord(overrides: Partial<import('@logtape/logtape').LogRecord> = {}) {
return {
category: ['test'],
level: 'info',
message: ['Message'],
rawMessage: 'Message',
timestamp: 1_700_000_000_000,
properties: {},
...overrides
} satisfies import('@logtape/logtape').LogRecord;
}
function createMemoryLoggerProvider(records: OTelLogRecord[]) {
return {
getLogger: () => ({
emit: (record: OTelLogRecord) => {
records.push(record);
}
})
};
}
describe('getOpenTelemetrySink', () => {
it('places structured properties in the log body without duplicating attributes', () => {
const emitted: OTelLogRecord[] = [];
const sink = getOpenTelemetrySink({
loggerProvider: createMemoryLoggerProvider(emitted)
});
sink(
createRecord({
message: ['Completions body'],
rawMessage: 'Completions body',
properties: {
completionsBody: {
messages: [
{
role: 'user',
content: 'hello'
}
],
model: 'deepseek-v3.2',
stream: true
}
}
})
);
expect(emitted).toHaveLength(1);
expect(emitted[0]?.body).toEqual({
__log_message: 'Completions body',
completionsBody: {
messages: [
{
role: 'user',
content: 'hello'
}
],
model: 'deepseek-v3.2',
stream: true
}
});
expect(emitted[0]?.attributes).toEqual({
category: ['test']
});
});
it('keeps scalar placeholders in the structured body message', () => {
const emitted: OTelLogRecord[] = [];
const sink = getOpenTelemetrySink({
loggerProvider: createMemoryLoggerProvider(emitted)
});
sink(
createRecord({
message: ['User ', 'Ada', ' used ', 3, ' credits'],
rawMessage: 'User {user} used {count} credits',
properties: {
user: 'Ada',
count: 3
}
})
);
expect(emitted[0]?.body).toEqual({
__log_message: 'User Ada used 3 credits',
user: 'Ada',
count: 3
});
});
it('uses a fallback body for purely structured logs', () => {
const emitted: OTelLogRecord[] = [];
const sink = getOpenTelemetrySink({
loggerProvider: createMemoryLoggerProvider(emitted)
});
sink(
createRecord({
message: ['', { a: 1 }, ''],
rawMessage: '{*}',
properties: {
a: 1
}
})
);
expect(emitted[0]?.body).toEqual({
__log_message: 'structured log',
a: 1
});
expect(emitted[0]?.attributes?.a).toBeUndefined();
});
it('normalizes JS-specific values in the structured body', () => {
class CustomValue {
constructor(readonly id: string) {}
}
const emitted: OTelLogRecord[] = [];
const sink = getOpenTelemetrySink({
loggerProvider: createMemoryLoggerProvider(emitted)
});
const circular: Record<string, unknown> = { name: 'root' };
circular.self = circular;
sink(
createRecord({
properties: {
nested: {
map: new Map([['key', { value: 1 }]]),
set: new Set(['a', 'b']),
bigint: 1n,
custom: new CustomValue('custom-1'),
circular
}
}
})
);
expect((emitted[0]?.body as Record<string, unknown>).nested).toEqual({
map: {
key: {
value: 1
}
},
set: ['a', 'b'],
bigint: '1',
custom: {
__type: 'CustomValue',
id: 'custom-1'
},
circular: {
name: 'root',
self: '[Circular]'
}
});
});
it('normalizes top-level errors in the structured body', () => {
const emitted: OTelLogRecord[] = [];
const sink = getOpenTelemetrySink({
loggerProvider: createMemoryLoggerProvider(emitted)
});
const error = new Error('boom');
sink(
createRecord({
properties: {
error
}
})
);
const body = emitted[0]?.body as Record<string, Record<string, unknown>>;
expect(body.error?.name).toBe('Error');
expect(body.error?.message).toBe('boom');
expect(body.error?.stack).toEqual(expect.any(String));
});
it('keeps a property named message without overwriting the log message', () => {
const emitted: OTelLogRecord[] = [];
const sink = getOpenTelemetrySink({
loggerProvider: createMemoryLoggerProvider(emitted)
});
sink(
createRecord({
message: ['Request completed'],
rawMessage: 'Request completed',
properties: {
message: 'business message'
}
})
);
expect(emitted[0]?.body).toEqual({
__log_message: 'Request completed',
message: 'business message'
});
});
it('does not duplicate trace context fields in the structured body', () => {
const emitted: OTelLogRecord[] = [];
const sink = getOpenTelemetrySink({
loggerProvider: createMemoryLoggerProvider(emitted)
});
sink(
createRecord({
message: ['Workflow node run'],
rawMessage: 'Workflow node run',
properties: {
traceId: 'trace-id',
spanId: 'span-id',
requestId: 'request-id'
}
})
);
expect(emitted[0]?.body).toEqual({
__log_message: 'Workflow node run',
requestId: 'request-id'
});
});
});
describe('createLoggerOptionsFromEnv', () => {
it('creates OTel logger options without body or attribute modes', () => {
const options = createLoggerOptionsFromEnv({
env: {
LOG_ENABLE_OTEL: 'true'
},
defaultServiceName: 'test-service'
});
expect(options.otel && 'bodyMode' in options.otel).toBe(false);
expect(options.otel && 'attributeObjectMode' in options.otel).toBe(false);
});
});
sdk/otel/vitest.config.ts 0 → 100644
import { defineConfig } from 'vitest/config';
export default defineConfig({
test: {
include: ['test/**/*.test.ts'],
pool: 'threads',
reporters: ['default']
}
});
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment