使用 Langfuse SDK 上报 Trace 数据到 CLS

npm install @langfuse/tracing @langfuse/otel @langfuse/openai \\
  @opentelemetry/sdk-node \\
  @opentelemetry/exporter-trace-otlp-http \\
  dotenv

import "dotenv/config";

import { NodeSDK } from "@opentelemetry/sdk-node";
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
import { LangfuseSpanProcessor } from "@langfuse/otel";

const secretId = process.env.TENCENTCLOUD_SECRET_ID!;
const secretKey = process.env.TENCENTCLOUD_SECRET_KEY!;
const region = process.env.CLS_DEFAULT_REGION!;
const topicId = process.env.CLS_TOPIC_ID!;

const auth = Buffer.from(`${secretId}:${secretKey}`).toString("base64");

export const sdk = new NodeSDK({
  spanProcessors: [
    new LangfuseSpanProcessor({
      exporter: new OTLPTraceExporter({
        url: `https://${region}.cls.tencentcs.com/v1/traces`,
        headers: {
          Authorization: `Basic ${auth}`,
          topic_id: topicId,
        },
      }),
      flushAt: 512,
      flushInterval: 5000,
    }),
  ],
});

sdk.start();

process.on("SIGTERM", async () => {
  await sdk.shutdown();
});

import { sdk } from "./instrumentation";

import OpenAI from "openai";
import { observeOpenAI } from "@langfuse/openai";
import { startActiveObservation, propagateAttributes } from "@langfuse/tracing";

const openai = observeOpenAI(new OpenAI());

async function chat(userMessage: string) {
  return startActiveObservation("chat", async (span) => {
    span.update({ input: [{ role: "user", content: userMessage }] });

    return propagateAttributes(
      { userId: "u-123", sessionId: "s-456", tags: ["model:gpt-4o"] },
      async () => {
        const res = await openai.chat.completions.create({
          model: "gpt-4o",
          messages: [{ role: "user", content: userMessage }],
        });
        span.update({ output: res.choices[0].message });
        return res;
      },
    );
  });
}

await chat("你好");
await sdk.shutdown();

npm install @langfuse/tracing @langfuse/otel @langfuse/openai \\
  @opentelemetry/sdk-node @opentelemetry/sdk-trace-base \\
  @opentelemetry/api @opentelemetry/core \\
  tencentcloud-cls-sdk-nodejs dotenv

import {
  SpanExporter,
  ReadableSpan,
} from "@opentelemetry/sdk-trace-base";
import { ExportResult, ExportResultCode, hrTimeToNanoseconds } from "@opentelemetry/core";
import {
  Producer,
  LogItem,
  Content,
  TencentCloudClsSDKException,
} from "tencentcloud-cls-sdk-nodejs";

export interface ClsExporterOptions {
  endpoint: string;
  topicId: string;
  secretId: string;
  secretKey: string;
  token?: string;
}

export class TencentCLSSpanExporter implements SpanExporter {
  private producer: Producer;

  constructor(opts: ClsExporterOptions) {
    this.producer = new Producer({
      endpoint: opts.endpoint,
      topic_id: opts.topicId,
      credential: {
        secretId: opts.secretId,
        secretKey: opts.secretKey,
        token: opts.token ?? "",
      },
      time: 2,
      count: 1000,
      maxMemLogCount: 10000,
      onSendLogsError: (err) => console.warn("[cls] send error", err),
    });
  }

  export(spans: ReadableSpan[], cb: (r: ExportResult) => void): void {
    Promise.all(spans.map((span) => this.producer.send(this.spanToLogItem(span))))
      .then(() => cb({ code: ExportResultCode.SUCCESS }))
      .catch((err: Error) => {
        if (err instanceof TencentCloudClsSDKException) {
          console.warn("[cls] sdk exception:", err.toString());
        }
        cb({ code: ExportResultCode.FAILED, error: err });
      });
  }

  private spanToLogItem(span: ReadableSpan): LogItem {
    const item = new LogItem();
    const endNs = hrTimeToNanoseconds(span.endTime);
    const startNs = hrTimeToNanoseconds(span.startTime);
    item.setTime(Math.floor(endNs / 1e6));

    const ctx = span.spanContext();
    item.pushBack(new Content("traceID", ctx.traceId));
    item.pushBack(new Content("spanID", ctx.spanId));
    item.pushBack(new Content("parentSpanID", span.parentSpanContext?.spanId ?? ""));
    item.pushBack(new Content("name", span.name));
    item.pushBack(new Content("kind", String(span.kind)));
    item.pushBack(new Content("start", String(startNs)));
    item.pushBack(new Content("end", String(endNs)));
    item.pushBack(new Content("duration", String(endNs - startNs)));
    item.pushBack(new Content("statusCode", String(span.status.code)));
    item.pushBack(new Content("statusMessage", span.status.message ?? ""));
    item.pushBack(new Content("attribute", JSON.stringify(span.attributes)));
    item.pushBack(new Content("resource", JSON.stringify(span.resource.attributes)));
    return item;
  }

  shutdown(): Promise<void> {
    return Promise.resolve();
  }

  forceFlush(): Promise<void> {
    return Promise.resolve();
  }
}

import "dotenv/config";

import { NodeSDK } from "@opentelemetry/sdk-node";
import { LangfuseSpanProcessor } from "@langfuse/otel";
import { TencentCLSSpanExporter } from "./cls-span-exporter";

export const sdk = new NodeSDK({
  spanProcessors: [
    new LangfuseSpanProcessor({
      exporter: new TencentCLSSpanExporter({
        endpoint: process.env.CLS_ENDPOINT!,
        topicId: process.env.CLS_TOPIC_ID!,
        secretId: process.env.TENCENTCLOUD_SECRET_ID!,
        secretKey: process.env.TENCENTCLOUD_SECRET_KEY!,
        token: process.env.TENCENTCLOUD_TOKEN,
      }),
      flushAt: 512,        // 攒批阈值
      flushInterval: 5,    // 秒
    }),
  ],
});

sdk.start();

process.on("SIGTERM", () => sdk.shutdown());