8. Observability очередей

package tracing

import (
    amqp "github.com/rabbitmq/amqp091-go"
    "go.opentelemetry.io/otel/propagation"
)

// AMQPHeadersCarrier адаптирует amqp.Table к интерфейсу TextMapCarrier
type AMQPHeadersCarrier amqp.Table

// Get извлекает значение строкового заголовка
func (c AMQPHeadersCarrier) Get(key string) string {
    val, ok := c[key]
    if !ok {
        return ""
    }
    strVal, ok := val.(string)
    if !ok {
        return ""
    }
    return strVal
}

// Set сохраняет заголовок в amqp.Table
func (c AMQPHeadersCarrier) Set(key string, value string) {
    c[key] = value
}

// Keys возвращает список всех ключей заголовков
func (c AMQPHeadersCarrier) Keys() []string {
    keys := make([]string, 0, len(c))
    for k := range c {
        keys = append(keys, k)
    }
    return keys
}

package producer

import (
    "context"
    "fmt"

    amqp "github.com/rabbitmq/amqp091-go"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/trace"
)

type EventPublisher struct {
    channel *amqp.Channel
    tracer  trace.Tracer
}

func NewEventPublisher(ch *amqp.Channel, tracer trace.Tracer) *EventPublisher {
    return &EventPublisher{channel: ch, tracer: tracer}
}

func (p *EventPublisher) PublishOrderCreated(ctx context.Context, orderID string, body []byte) error {
    // Создаем дочерний Span операции отправки в брокер
    ctx, span := p.tracer.Start(ctx, "orders.publish",
        trace.WithSpanKind(trace.SpanKindProducer),
    )
    defer span.End()

    headers := make(amqp.Table)

    // Внедряем W3C Trace Context (traceparent) в таблицу заголовков
    propagator := otel.GetTextMapPropagator()
    propagator.Inject(ctx, AMQPHeadersCarrier(headers))

    // Добавляем бизнес-идентификатор для быстрой фильтрации в Wireshark / Management UI
    headers["MessageID"] = fmt.Sprintf("msg-%s", orderID)

    msg := amqp.Publishing{
        DeliveryMode: amqp.Persistent,
        ContentType:  "application/json",
        Headers:      headers,
        Body:         body,
    }

    err := p.channel.PublishWithContext(ctx,
        "orders.exchange", // exchange
        "order.created",   // routing key
        false,             // mandatory
        false,             // immediate
        msg,
    )
    if err != nil {
        span.RecordError(err)
        return fmt.Errorf("failed to publish message: %w", err)
    }

    return nil
}

package consumer

import (
    "context"
    "log/slog"

    amqp "github.com/rabbitmq/amqp091-go"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/trace"
)

type OrderConsumer struct {
    tracer trace.Tracer
    logger *slog.Logger
}

func (c *OrderConsumer) HandleDelivery(d amqp.Delivery) {
    // 1. Извлекаем родительский контекст трассировки из Headers брокера
    propagator := otel.GetTextMapPropagator()
    parentCtx := propagator.Extract(context.Background(), AMQPHeadersCarrier(d.Headers))

    // 2. Стартуем новый локальный Span, который в OTel UI будет дочерним к Producer Span
    ctx, span := c.tracer.Start(parentCtx, "orders.consume",
        trace.WithSpanKind(trace.SpanKindConsumer),
    )
    defer span.End()

    // 3. Выполняем доменную бизнес-логику
    if err := c.processOrder(ctx, d.Body); err != nil {
        span.RecordError(err)
        c.logger.ErrorContext(ctx, "failed to process message",
            slog.String("error", err.Error()),
            slog.Uint64("delivery_tag", d.DeliveryTag),
        )
        _ = d.Nack(false, false) // Отправляем в DLQ без зацикливания
        return
    }

    _ = d.Ack(false)
}

func (c *OrderConsumer) processOrder(ctx context.Context, payload []byte) error {
    // Вся цепочка downstream-вызовов (SQL, Redis, HTTP) наследует ctx с TraceID!
    return nil
}

package metrics

import (
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promauto"
)

var (
    // EventsPublishedTotal считает интенсивность исходящего потока
    EventsPublishedTotal = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "events_published_total",
            Help: "Общее количество опубликованных сообщений в брокер",
        },
        []string{"topic", "status"}, // status: success / error
    )

    // EventsConsumedTotal считает интенсивность вычитки и распределение исходов
    EventsConsumedTotal = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "events_consumed_total",
            Help: "Общее количество вычитанных сообщений из очереди",
        },
        []string{"topic", "consumer_group", "status"}, // status: ack, nack, dlq
    )

    // EventProcessingDuration измеряет задержку выполнения бизнес-логики консьюмера
    EventProcessingDuration = promauto.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "event_processing_duration_seconds",
            Help:    "Время обработки одного сообщения консьюмером в секундах",
            Buckets: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0},
        },
        []string{"topic", "consumer_group"},
    )
)

start := time.Now()
err := handler.Process(ctx, msg)
duration := time.Since(start).Seconds()

EventProcessingDuration.WithLabelValues("orders", "billing-group").Observe(duration)

if err != nil {
    EventsConsumedTotal.WithLabelValues("orders", "billing-group", "nack").Inc()
} else {
    EventsConsumedTotal.WithLabelValues("orders", "billing-group", "ack").Inc()
}

package logging

import (
    "context"
    "log/slog"
    "os"

    "go.opentelemetry.io/otel/trace"
)

// TraceContextHandler автоматически внедряет trace_id и span_id в каждую запись лога
type TraceContextHandler struct {
    slog.Handler
}

func NewTraceJSONHandler() *TraceContextHandler {
    baseHandler := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
        Level: slog.LevelInfo,
    })
    return &TraceContextHandler{Handler: baseHandler}
}

func (h *TraceContextHandler) Handle(ctx context.Context, r slog.Record) error {
    if ctx != nil {
        span := trace.SpanFromContext(ctx)
        spanCtx := span.SpanContext()
        if spanCtx.IsValid() {
            r.AddAttrs(
                slog.String("trace_id", spanCtx.TraceID().String()),
                slog.String("span_id", spanCtx.SpanID().String()),
            )
        }
    }
    return h.Handler.Handle(ctx, r)
}

func init() {
    handler := NewTraceJSONHandler()
    slog.SetDefault(slog.New(handler))
}

func ProcessMessage(ctx context.Context, msg []byte) {
    // В логе автоматически появятся "trace_id" и "span_id" из переданного ctx!
    slog.InfoContext(ctx, "starting message processing",
        slog.Int("payload_bytes", len(msg)),
    )

    if err := doDB(ctx); err != nil {
        slog.ErrorContext(ctx, "database update failed",
            slog.String("error", err.Error()),
        )
    }
}

{
  "time": "2026-09-15T11:45:00.123456789Z",
  "level": "ERROR",
  "msg": "database update failed",
  "error": "connection refused",
  "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
  "span_id": "00f067aa0ba902b7"
}