📖 数据密集型设计

分布式追踪与链路追踪

深入探讨分布式追踪原理、链路追踪实现及分布式系统监控

一、分布式追踪概述

分布式追踪是分布式系统中定位问题和性能优化的关键技术，通过追踪请求在多个服务间的流转，能够了解系统的调用链和性能瓶颈。链路追踪是分布式追踪的核心实现方式。

二、分布式追踪系统对比

2.1 追踪系统对比

特性 Jaeger Zipkin OpenTelemetry
后端存储 Cassandra/ES Cassandra/ES 可配置
采样策略 多种 简单 灵活
可视化 丰富 基本 集成
语言支持 多语言 多语言 标准API
标准兼容 OpenTracing OpenTracing 统一标准

2.2 追踪模型

graph TD A[Trace] --> B[Span 1] A --> C[Span 2] A --> D[Span 3] B --> B1[服务A] B1 --> B2[操作: HTTP请求] B2 --> B3[耗时: 50ms] C --> C1[服务B] C1 --> C2[操作: 数据库查询] C2 --> C3[耗时: 200ms] D --> D1[服务C] D1 --> D2[操作: 缓存查询] D2 --> D3[耗时: 10ms] B --> C C --> D E[上下文传播] --> E1[TraceId] E1 --> E2[SpanId] E2 --> E3[ParentSpanId] E3 --> E4[Baggage]

三、Jaeger链路追踪

3.1 Jaeger架构

graph TD A[客户端] --> B[Jaeger Agent] B --> C[Jaeger Collector] C --> D[存储层] D --> D1[Cassandra] D --> D2[Elasticsearch] D --> D3[Memory] E[Query Service] --> D E --> F[UI] G[采样策略] --> G1[概率采样] G1 --> G2[速率限制采样] G2 --> G3[基于操作采样] H[数据模型] --> H1[Trace] H1 --> H2[Span] H2 --> H3[Tag] H3 --> H4[Log]

3.2 Jaeger客户端实现

public class JaegerTracerFactory
{
    public ITracer CreateTracer(string serviceName)
    {
        var configuration = new Configuration(serviceName)
            .WithSampler(new ConstSampler(true))
            .WithReporter(new RemoteReporter.Builder()
                .WithSender(new UdpSender("localhost", 6831, 0))
                .Build());
        
        return configuration.GetTracer();
    }
    
    public ITracer CreateTracerWithSampler(string serviceName, ISampler sampler)
    {
        var configuration = new Configuration(serviceName)
            .WithSampler(sampler)
            .WithReporter(new RemoteReporter.Builder()
                .WithSender(new UdpSender("localhost", 6831, 0))
                .Build());
        
        return configuration.GetTracer();
    }
    
    public ISampler CreateProbabilisticSampler(double probability)
    {
        return new ProbabilisticSampler(probability);
    }
    
    public ISampler CreateRateLimitingSampler(int maxTracesPerSecond)
    {
        return new RateLimitingSampler(maxTracesPerSecond);
    }
}

public class TracedService
{
    private readonly ITracer _tracer;
    
    public TracedService(ITracer tracer)
    {
        _tracer = tracer;
    }
    
    public async Task ProcessRequestAsync(Request request)
    {
        using var scope = _tracer.BuildSpan("ProcessRequest").StartActive(true);
        
        scope.Span.SetTag("request.id", request.Id);
        scope.Span.SetTag("request.type", request.Type);
        
        try
        {
            var result = await DoWorkAsync(request);
            
            scope.Span.SetTag("result.status", "success");
            
            return result;
        }
        catch (Exception ex)
        {
            scope.Span.SetTag("result.status", "error");
            scope.Span.Log(new Dictionary { { "error", ex.Message } });
            
            throw;
        }
    }
    
    private async Task DoWorkAsync(Request request)
    {
        using var scope = _tracer.BuildSpan("DoWork").AsChildOf(_tracer.ActiveSpan.Context).StartActive(true);
        
        scope.Span.SetTag("work.type", request.WorkType);
        
        await Task.Delay(100);
        
        return new Result { Id = request.Id, Success = true };
    }
}

四、OpenTelemetry

4.1 OpenTelemetry实现

public class OpenTelemetrySetup
{
    public void ConfigureOpenTelemetry(IHostBuilder hostBuilder)
    {
        hostBuilder.ConfigureServices((context, services) =>
        {
            services.AddOpenTelemetryTracing(builder =>
            {
                builder
                    .AddAspNetCoreInstrumentation()
                    .AddHttpClientInstrumentation()
                    .AddSqlClientInstrumentation()
                    .AddSource("MyApplication")
                    .SetResourceBuilder(ResourceBuilder.CreateDefault()
                        .AddService(context.Configuration["ServiceName"]))
                    .AddJaegerExporter(options =>
                    {
                        options.AgentHost = context.Configuration["Jaeger:AgentHost"];
                        options.AgentPort = int.Parse(context.Configuration["Jaeger:AgentPort"]);
                    });
            });
        });
    }
}

public class OpenTelemetryTracedService
{
    private readonly ActivitySource _activitySource;
    
    public OpenTelemetryTracedService(ActivitySource activitySource)
    {
        _activitySource = activitySource;
    }
    
    public async Task ProcessAsync(Request request)
    {
        using var activity = _activitySource.StartActivity("ProcessAsync");
        
        activity?.SetTag("request.id", request.Id);
        activity?.SetTag("request.type", request.Type);
        
        var result = await InternalProcessAsync(request);
        
        activity?.SetStatus(ActivityStatusCode.Ok);
        
        return result;
    }
    
    private async Task InternalProcessAsync(Request request)
    {
        using var activity = _activitySource.StartActivity("InternalProcessAsync");
        
        activity?.SetParentId(Activity.Current?.Id);
        
        await Task.Delay(50);
        
        return new Result { Id = request.Id };
    }
}

五、分布式追踪实现

5.1 上下文传播

public class TraceContextPropagation
{
    public const string TraceIdHeader = "trace-id";
    public const string SpanIdHeader = "span-id";
    public const string ParentSpanIdHeader = "parent-span-id";
    public const string SampledHeader = "sampled";
    
    public TraceContext Extract(HttpRequest request)
    {
        return new TraceContext
        {
            TraceId = request.Headers.TryGetValue(TraceIdHeader, out var traceId) ? traceId : GenerateTraceId(),
            SpanId = request.Headers.TryGetValue(SpanIdHeader, out var spanId) ? spanId : GenerateSpanId(),
            ParentSpanId = request.Headers.TryGetValue(ParentSpanIdHeader, out var parentSpanId) ? parentSpanId : null,
            Sampled = request.Headers.TryGetValue(SampledHeader, out var sampled) && sampled == "1"
        };
    }
    
    public void Inject(TraceContext context, HttpRequestMessage request)
    {
        request.Headers.Add(TraceIdHeader, context.TraceId);
        request.Headers.Add(SpanIdHeader, context.SpanId);
        
        if (!string.IsNullOrEmpty(context.ParentSpanId))
        {
            request.Headers.Add(ParentSpanIdHeader, context.ParentSpanId);
        }
        
        request.Headers.Add(SampledHeader, context.Sampled ? "1" : "0");
    }
    
    private string GenerateTraceId()
    {
        return Guid.NewGuid().ToString("N");
    }
    
    private string GenerateSpanId()
    {
        return Guid.NewGuid().ToString("N").Substring(0, 16);
    }
}

public class TraceContext
{
    public string TraceId { get; set; }
    public string SpanId { get; set; }
    public string ParentSpanId { get; set; }
    public bool Sampled { get; set; }
    public Dictionary Baggage { get; set; } = new();
}

5.2 自定义追踪

public class CustomTracer
{
    private readonly ITraceRepository _traceRepository;
    private readonly TraceContextPropagation _propagation;
    
    public CustomTracer(ITraceRepository traceRepository, TraceContextPropagation propagation)
    {
        _traceRepository = traceRepository;
        _propagation = propagation;
    }
    
    public async Task StartSpanAsync(string operationName, TraceContext context = null)
    {
        var span = new TraceSpan
        {
            TraceId = context?.TraceId ?? _propagation.GenerateTraceId(),
            SpanId = _propagation.GenerateSpanId(),
            ParentSpanId = context?.SpanId,
            OperationName = operationName,
            StartTime = DateTime.UtcNow,
            Tags = new Dictionary(),
            Logs = new List()
        };
        
        await _traceRepository.RecordSpanAsync(span);
        
        return span;
    }
    
    public async Task EndSpanAsync(TraceSpan span)
    {
        span.EndTime = DateTime.UtcNow;
        span.Duration = (span.EndTime - span.StartTime).TotalMilliseconds;
        
        await _traceRepository.UpdateSpanAsync(span);
    }
    
    public async Task TraceAsync(string operationName, Func> action, TraceContext context = null)
    {
        var span = await StartSpanAsync(operationName, context);
        
        try
        {
            var result = await action();
            
            span.Status = SpanStatus.Ok;
            
            return result;
        }
        catch (Exception ex)
        {
            span.Status = SpanStatus.Error;
            span.Logs.Add(new TraceLog { Timestamp = DateTime.UtcNow, Message = ex.Message });
            
            throw;
        }
        finally
        {
            await EndSpanAsync(span);
        }
    }
    
    public void AddTag(TraceSpan span, string key, string value)
    {
        span.Tags[key] = value;
    }
    
    public void AddLog(TraceSpan span, string message)
    {
        span.Logs.Add(new TraceLog { Timestamp = DateTime.UtcNow, Message = message });
    }
}

public class TraceSpan
{
    public string TraceId { get; set; }
    public string SpanId { get; set; }
    public string ParentSpanId { get; set; }
    public string OperationName { get; set; }
    public DateTime StartTime { get; set; }
    public DateTime EndTime { get; set; }
    public double Duration { get; set; }
    public SpanStatus Status { get; set; }
    public Dictionary Tags { get; set; } = new();
    public List Logs { get; set; } = new();
}

public enum SpanStatus { Ok, Error, Unknown }
public class TraceLog { public DateTime Timestamp; public string Message; }

六、采样策略

6.1 采样策略实现

public class SamplingStrategy
{
    private readonly Random _random = new();
    
    public bool ShouldSample(string traceId, SamplingConfig config)
    {
        switch (config.Strategy)
        {
            case SamplingStrategyType.Constant:
                return config.SampleAll;
            
            case SamplingStrategyType.Probabilistic:
                return ShouldSampleProbabilistic(config.Probability);
            
            case SamplingStrategyType.RateLimiting:
                return ShouldSampleRateLimited(config.MaxTracesPerSecond);
            
            case SamplingStrategyType.RuleBased:
                return ShouldSampleByRules(traceId, config.Rules);
            
            default:
                return false;
        }
    }
    
    private bool ShouldSampleProbabilistic(double probability)
    {
        return _random.NextDouble() < probability;
    }
    
    private bool ShouldSampleRateLimited(int maxTracesPerSecond)
    {
        return _rateLimiter.TryAcquire();
    }
    
    private bool ShouldSampleByRules(string traceId, List rules)
    {
        foreach (var rule in rules)
        {
            if (rule.Matches(traceId))
            {
                return rule.Sample;
            }
        }
        
        return false;
    }
}

public class SamplingConfig
{
    public SamplingStrategyType Strategy { get; set; }
    public bool SampleAll { get; set; }
    public double Probability { get; set; }
    public int MaxTracesPerSecond { get; set; }
    public List Rules { get; set; } = new();
}

public enum SamplingStrategyType { Constant, Probabilistic, RateLimiting, RuleBased }

public class SamplingRule
{
    public string OperationName { get; set; }
    public string ServiceName { get; set; }
    public bool Sample { get; set; }
    
    public bool Matches(string traceId)
    {
        return true;
    }
}

七、分布式追踪最佳实践

7.1 追踪实现原则

原则 描述 实现方式
全链路追踪 覆盖所有服务 自动instrumentation
上下文传播 传递trace信息 HTTP headers
合理采样 控制追踪开销 概率/规则采样
丰富标签 添加业务信息 自定义tags
可视化分析 直观展示数据 Jaeger UI

7.2 链路追踪最佳实践

  • 使用标准追踪协议
  • 实现自动追踪
  • 设置合理的采样率
  • 添加关键业务标签
  • 监控追踪系统本身

八、总结

分布式追踪与链路追踪是分布式系统监控的关键技术。通过选择合适的追踪系统（Jaeger、Zipkin、OpenTelemetry），能够实现全链路追踪和性能分析。上下文传播是实现分布式追踪的核心，采样策略能够控制追踪开销。遵循分布式追踪最佳实践，能够快速定位系统问题，优化系统性能。