Serverless 可观测性#

import { Context } from 'aws-lambda';

interface LogEntry {
  timestamp: string;
  level: 'DEBUG' | 'INFO' | 'WARN' | 'ERROR';
  requestId: string;
  functionName: string;
  message: string;
  [key: string]: any;
}

const createLogger = (context: Context) => {
  const log = (level: LogEntry['level'], message: string, data?: object) => {
    const entry: LogEntry = {
      timestamp: new Date().toISOString(),
      level,
      requestId: context.awsRequestId,
      functionName: context.functionName,
      message,
      ...data,
    };

    console.log(JSON.stringify(entry));
  };

  return {
    debug: (msg: string, data?: object) => log('DEBUG', msg, data),
    info: (msg: string, data?: object) => log('INFO', msg, data),
    warn: (msg: string, data?: object) => log('WARN', msg, data),
    error: (msg: string, data?: object) => log('ERROR', msg, data),
  };
};

export const handler = async (event: any, context: Context) => {
  const logger = createLogger(context);

  logger.info('Processing request', {
    path: event.path,
    method: event.httpMethod,
    bodySize: event.body?.length,
  });

  try {
    const result = await processEvent(event);
    logger.info('Request completed', { resultSize: JSON.stringify(result).length });
    return result;
  } catch (error) {
    logger.error('Request failed', {
      error: error.message,
      stack: error.stack,
    });
    throw error;
  }
};

import { CloudWatchClient, PutMetricDataCommand } from '@aws-sdk/client-cloudwatch';

const cloudwatch = new CloudWatchClient({});

interface MetricData {
  namespace: string;
  name: string;
  value: number;
  unit?: 'Count' | 'Milliseconds' | 'Bytes' | 'Seconds';
  dimensions?: Record<string, string>;
}

export const recordMetric = async (metric: MetricData) => {
  await cloudwatch.send(new PutMetricDataCommand({
    Namespace: metric.namespace,
    MetricData: [{
      MetricName: metric.name,
      Dimensions: Object.entries(metric.dimensions || {}).map(([Name, Value]) => ({
        Name,
        Value,
      })),
      Value: metric.value,
      Unit: metric.unit || 'Count',
      Timestamp: new Date(),
    }],
  }));
};

// 使用示例
export const recordOrderProcessed = async (orderId: string, duration: number) => {
  await recordMetric({
    namespace: 'ECommerce/Orders',
    name: 'OrderProcessed',
    value: 1,
    dimensions: { region: process.env.AWS_REGION! },
  });

  await recordMetric({
    namespace: 'ECommerce/Orders',
    name: 'ProcessingDuration',
    value: duration,
    unit: 'Milliseconds',
    dimensions: { region: process.env.AWS_REGION! },
  });
};

import { register, Counter, Histogram, Gauge } from 'prom-client';

// 请求计数器
const httpRequestsTotal = new Counter({
  name: 'http_requests_total',
  help: 'Total number of HTTP requests',
  labelNames: ['method', 'path', 'status'],
});

// 请求延迟直方图
const httpRequestDuration = new Histogram({
  name: 'http_request_duration_seconds',
  help: 'Duration of HTTP requests in seconds',
  labelNames: ['method', 'path'],
  buckets: [0.01, 0.05, 0.1, 0.5, 1, 5],
});

// 当前处理中的请求数
const httpRequestsInProgress = new Gauge({
  name: 'http_requests_in_progress',
  help: 'Number of HTTP requests currently being processed',
  labelNames: ['method'],
});

export const metricsHandler = async (req: Request): Promise<Response> => {
  // 记录请求开始
  const startTime = Date.now();
  httpRequestsInProgress.inc({ method: req.method });

  try {
    const response = await handleRequest(req);

    // 记录请求完成
    httpRequestsTotal.inc({
      method: req.method,
      path: new URL(req.url).pathname,
      status: response.status.toString(),
    });

    httpRequestDuration.observe(
      { method: req.method, path: new URL(req.url).pathname },
      (Date.now() - startTime) / 1000
    );

    return response;
  } finally {
    httpRequestsInProgress.dec({ method: req.method });
  }
};

// /metrics 端点
export const metricsEndpoint = async (): Promise<Response> => {
  return new Response(await register.metrics(), {
    headers: { 'Content-Type': register.contentType },
  });
};

import { Segment, Subsegment, captureAWSv3Client } from 'aws-xray-sdk';
import { DynamoDBClient } from '@aws-sdk/client-dynamodb';
import { DynamoDBDocumentClient, GetCommand } from '@aws-sdk/lib-dynamodb';

// 为 AWS SDK 启用 X-Ray
const dynamoDB = captureAWSv3Client(new DynamoDBClient({}));

export const handler = async (event: any, context: any) => {
  // X-Ray 自动捕获 Lambda 调用

  const userId = event.pathParameters?.userId;

  // 创建子段标记自定义逻辑
  const subsegment = context.awsXray?.createSubsegment('fetchUserData');

  try {
    const startTime = Date.now();

    // DynamoDB 调用被自动追踪
    const response = await dynamoDB.send(new DynamoDBDocumentClient({
      command: new GetCommand({
        TableName: 'users',
        Key: { id: userId },
      }),
    }));

    const duration = Date.now() - startTime;
    subsegment?.addMetadata('table', 'users');
    subsegment?.addMetadata('duration_ms', duration);

    return {
      statusCode: 200,
      body: JSON.stringify(response.Item),
    };
  } catch (error) {
    subsegment?.addAnnotation('error', true);
    throw error;
  } finally {
    subsegment?.close();
  }
};

import { NodeSDK } from '@opentelemetry/sdk-node';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
import { AwsLambda instrumentation } from '@opentelemetry/instrumentation-aws-lambda';
import { getNodeAutoInstrumentations } from '@opentelemetry/auto-instrumentations-node';

const sdk = new NodeSDK({
  traceExporter: new OTLPTraceExporter({
    url: 'https://otel-collector:4318/v1/traces',
  }),
  instrumentations: [
    new AwsLambdaInstrumentation(),
    getNodeAutoInstrumentations({
      '@opentelemetry/instrumentation-http': { enabled: true },
      '@opentelemetry/instrumentation-express': { enabled: true },
    }),
  ],
});

sdk.start();

// 你的 Lambda 函数
export const handler = async (event: any) => {
  return processEvent(event);
};

import { Context } from 'aws-lambda';
import { propagation, context } from '@opentelemetry/api';

// 提取追踪上下文
export const handler = async (event: any, lambdaContext: Context) => {
  // 从 HTTP 头或 SQS 消息提取上下文
  const extractedContext = propagation.extract(context.active(), event.headers || event.MessageAttributes);

  // 在上下文中执行
  return context.with(extractedContext, async () => {
    const tracer = trace.getTracer('my-service');

    return tracer.startActiveSpan('processOrder', (span) => {
      try {
        span.setAttributes({
          'order.id': event.orderId,
          'user.id': event.userId,
        });

        const result = processOrder(event);

        span.setStatus({ code: SpanStatusCode.OK });
        return result;
      } catch (error) {
        span.recordException(error);
        span.setStatus({ code: SpanStatusCode.ERROR });
        throw error;
      } finally {
        span.end();
      }
    });
  });
};

// 调用其他服务时注入上下文
const callDownstreamService = async (data: any) => {
  const headers: Record<string, string> = {};

  // 将追踪上下文注入到 HTTP 头
  propagation.inject(context.active(), headers);

  return fetch('https://api.example.com', {
    headers: {
      ...headers,
      'Content-Type': 'application/json',
    },
    method: 'POST',
    body: JSON.stringify(data),
  });
};