Kiến trúc Phân tán Tracing Go Microservices (2026): OpenTelemetry & Tail-Based Sampling

// SAI LẦM: Chặt đứt cây trace thành span mồ côi!
go func() {
    ctx := context.Background() // Mất toàn bộ TraceID gốc!
    processPayment(ctx, orderID)
}()

// CHUẨN MỰC: Kế thừa Trace Context an toàn
go func(parentCtx context.Context) {
    // Trích xuất SpanContext hiện tại và gắn vào một context mới độc lập với timeout
    spanContext := trace.SpanContextFromContext(parentCtx)
    detachedCtx := trace.ContextWithSpanContext(context.Background(), spanContext)
    
    tr := otel.Tracer("order-worker")
    ctx, span := tr.Start(detachedCtx, "processPaymentAsync")
    defer span.End()

    processPayment(ctx, orderID)
}(ctx)

// pkg/telemetry/grpc_interceptors.go
package telemetry

import (
	"context"

	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/codes"
	"go.opentelemetry.io/otel/propagation"
	semconv "go.opentelemetry.io/otel/semconv/v1.24.0"
	"go.opentelemetry.io/otel/trace"
	"google.golang.org/grpc"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
)

// metadataSupplier bọc metadata.MD của gRPC để triển khai interface propagation.TextMapCarrier
type metadataSupplier struct {
	md *metadata.MD
}

func (s *metadataSupplier) Get(key string) string {
	values := s.md.Get(key)
	if len(values) == 0 {
		return ""
	}
	return values[0]
}

func (s *metadataSupplier) Set(key string, value string) {
	s.md.Set(key, value)
}

func (s *metadataSupplier) Keys() []string {
	keys := make([]string, 0, len(*s.md))
	for k := range *s.md {
		keys = append(keys, k)
	}
	return keys
}

// UnaryClientInterceptor tự động tiêm W3C trace context vào metadata gửi đi
func UnaryClientInterceptor(tracer trace.Tracer) grpc.UnaryClientInterceptor {
	return func(
		ctx context.Context,
		method string,
		req, reply interface{},
		cc *grpc.ClientConn,
		invoker grpc.UnaryInvoker,
		opts ...grpc.CallOption,
	) error {
		ctx, span := tracer.Start(ctx, method,
			trace.WithSpanKind(trace.SpanKindClient),
			trace.WithAttributes(semconv.RPCMethodKey.String(method)),
		)
		defer span.End()

		md, ok := metadata.FromOutgoingContext(ctx)
		if !ok {
			md = metadata.New(nil)
		} else {
			md = md.Copy()
		}

		// Tiêm W3C traceparent vào gRPC Metadata Carrier
		otel.GetTextMapPropagator().Inject(ctx, &metadataSupplier{md: &md})
		ctx = metadata.NewOutgoingContext(ctx, md)

		err := invoker(ctx, method, req, reply, cc, opts...)
		if err != nil {
			s, _ := status.FromError(err)
			span.SetStatus(codes.Error, s.Message())
			span.RecordError(err)
		} else {
			span.SetStatus(codes.Ok, "gRPC Call OK")
		}

		return err
	}
}

// UnaryServerInterceptor trích xuất W3C trace context từ metadata nhận được
func UnaryServerInterceptor(tracer trace.Tracer) grpc.UnaryServerInterceptor {
	return func(
		ctx context.Context,
		req interface{},
		info *grpc.UnaryServerInfo,
		handler grpc.UnaryHandler,
	) (interface{}, error) {
		md, ok := metadata.FromIncomingContext(ctx)
		if !ok {
			md = metadata.New(nil)
		}

		// Trích xuất traceparent từ incoming metadata
		propagator := otel.GetTextMapPropagator()
		extractedCtx := propagator.Extract(ctx, &metadataSupplier{md: &md})

		ctx, span := tracer.Start(extractedCtx, info.FullMethod,
			trace.WithSpanKind(trace.SpanKindServer),
			trace.WithAttributes(semconv.RPCMethodKey.String(info.FullMethod)),
		)
		defer span.End()

		resp, err := handler(ctx, req)
		if err != nil {
			s, _ := status.FromError(err)
			span.SetStatus(codes.Error, s.Message())
			span.RecordError(err)
		} else {
			span.SetStatus(codes.Ok, "Handled Successfully")
		}

		return resp, err
	}
}

// pkg/telemetry/kafka_carrier.go
package telemetry

import (
	"context"

	"github.com/IBM/sarama"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/propagation"
)

type KafkaHeaderCarrier struct {
	headers *[]sarama.RecordHeader
}

func NewKafkaHeaderCarrier(headers *[]sarama.RecordHeader) *KafkaHeaderCarrier {
	return &KafkaHeaderCarrier{headers: headers}
}

func (c *KafkaHeaderCarrier) Get(key string) string {
	for _, h := range *c.headers {
		if string(h.Key) == key {
			return string(h.Value)
		}
	}
	return ""
}

func (c *KafkaHeaderCarrier) Set(key string, value string) {
	// Kiểm tra xem key đã tồn tại chưa để ghi đè
	for i, h := range *c.headers {
		if string(h.Key) == key {
			(*c.headers)[i].Value = []byte(value)
			return
		}
	}
	*c.headers = append(*c.headers, sarama.RecordHeader{
		Key:   []byte(key),
		Value: []byte(value),
	})
}

func (c *KafkaHeaderCarrier) Keys() []string {
	keys := make([]string, 0, len(*c.headers))
	for _, h := range *c.headers {
		keys = append(keys, string(h.Key))
	}
	return keys
}

// InjectTraceContextToKafkaMsg tiêm active trace context vào Sarama ProducerMessage
func InjectTraceContextToKafkaMsg(ctx context.Context, msg *sarama.ProducerMessage) {
	carrier := NewKafkaHeaderCarrier(&msg.Headers)
	otel.GetTextMapPropagator().Inject(ctx, carrier)
}

// ExtractTraceContextFromKafkaMsg trích xuất trace context từ Sarama ConsumerMessage
func ExtractTraceContextFromKafkaMsg(ctx context.Context, msg *sarama.ConsumerMessage) context.Context {
	headers := make([]sarama.RecordHeader, len(msg.Headers))
	for i, h := range msg.Headers {
		headers[i] = *h
	}
	carrier := NewKafkaHeaderCarrier(&headers)
	return otel.GetTextMapPropagator().Extract(ctx, carrier)
}

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  # 1. BẮT BUỘC: Giới hạn bộ nhớ để ngăn chặn crash OOM
  memory_limiter:
    check_interval: 1s
    limit_percentage: 80
    spike_limit_percentage: 20

  # 2. Gom cụm span để tối ưu hóa mạng
  batch:
    send_batch_size: 1024
    timeout: 2s

  # 3. TẨY XÓA DỮ LIỆU NHẠY CẢM (PII Redaction) bằng ngôn ngữ OTTL
  transform:
    error_mode: ignore
    trace_statements:
      - context: span
        statements:
          - replace_pattern(attributes["http.target"], "access_token=[^&]+", "access_token=REDACTED")
          - replace_pattern(attributes["db.statement"], "password\\s*=\\s*'[^']+'", "password='REDACTED'")
          - delete_key(attributes, "user.social_security_number")

  # 4. BỘ LẤY MẪU ĐUÔI (Tail-Based Sampling): Tiết kiệm 80% dung lượng lưu trữ
  tail_sampling:
    decision_wait: 10s
    num_traces: 50000
    expected_new_traces_per_sec: 2000
    policies:
      # Chính sách 1: Lưu giữ 100% các vết trace có mã lỗi HTTP 5xx
      - name: sample_http_errors
        type: numeric_attribute
        numeric_attribute:
          key: http.status_code
          min_value: 500
          max_value: 599

      # Chính sách 2: Lưu giữ 100% các vết trace bị lỗi gRPC
      - name: sample_grpc_errors
        type: status_code
        status_code:
          statuses: [ERROR]

      # Chính sách 3: Lưu giữ 100% các trace bị chậm (Latency > 1.500ms)
      - name: sample_slow_traces
        type: latency
        latency:
          threshold_ms: 1500

      # Chính sách 4: Chỉ lấy mẫu 5% các trace thành công thông thường để làm đường cơ sở (Baseline)
      - name: probabilistic_sample_success
        type: probabilistic
        probabilistic:
          sampling_percentage: 5.0

exporters:
  otlp/tempo:
    endpoint: tempo-distributor.monitoring.svc:4317
    tls:
      insecure: true

  prometheus:
    endpoint: 0.0.0.0:8889
    namespace: otel
    enable_open_metrics: true

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, transform, tail_sampling, batch]
      exporters: [otlp/tempo]
    metrics:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [prometheus]

// pkg/telemetry/metrics.go
package telemetry

import (
	"context"

	"github.com/prometheus/client_golang/prometheus"
	"go.opentelemetry.io/otel/trace"
)

var (
	HttpRequestDuration = prometheus.NewHistogramVec(
		prometheus.HistogramOpts{
			Name:    "http_request_duration_seconds",
			Help:    "Phân phối độ trễ của các yêu cầu HTTP tính bằng giây.",
			Buckets: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0},
		},
		[]string{"method", "route", "status_code"},
	)
)

func init() {
	prometheus.MustRegister(HttpRequestDuration)
}

// ObserveLatencyWithExemplar ghi nhận độ trễ và đính kèm TraceID làm Exemplar
func ObserveLatencyWithExemplar(ctx context.Context, durationSec float64, method, route, statusCode string) {
	span := trace.SpanFromContext(ctx)
	if !span.SpanContext().IsValid() {
		HttpRequestDuration.WithLabelValues(method, route, statusCode).Observe(durationSec)
		return
	}

	traceID := span.SpanContext().TraceID().String()
	
	// Đính kèm Trace ID vào Prometheus Exemplar
	if observer, ok := HttpRequestDuration.WithLabelValues(method, route, statusCode).(prometheus.ExemplarObserver); ok {
		observer.ObserveWithExemplar(durationSec, prometheus.Labels{
			"trace_id": traceID,
		})
	} else {
		HttpRequestDuration.WithLabelValues(method, route, statusCode).Observe(durationSec)
	}
}