تحكم في خدماتك مع OTEL و Jaeger و Prometheus

services:

​

  jaeger:

    image: jaegertracing/all-in-one:1.52

    ports:

      - "6831:6831/udp" # UDP port for the Jaeger agent

      - "16686:16686"   # Web UI

      - "14268:14268"   # HTTP port for spans

    networks:

      - internal

​

  prometheus:

    image: prom/prometheus:latest

    volumes:

      - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro

    ports:

      - "9090:9090"

    depends_on:

      - kafka

      - jaeger

      - otel-collector

    command:

      --config.file=/etc/prometheus/prometheus.yml

    networks:

      - internal

​

  otel-collector:

    image: otel/opentelemetry-collector-contrib:0.91.0

    command: ['--config=/etc/otel-collector.yaml']

    ports:

      - "4317:4317" # OTLP gRPC receiver

    volumes:

      - ./otel-collector.yaml:/etc/otel-collector.yaml

    depends_on:

      - jaeger

    networks:

      - internal

​

receivers:

  # Add the OTLP receiver listening on port 4317.

  otlp:

    protocols:

      grpc:

        endpoint: "0.0.0.0:4317"

​

processors:

  batch:

  # https://github.com/open-telemetry/opentelemetry-collector/tree/main/processor/memorylimiterprocessor

  memory_limiter:

    check_interval: 1s

    limit_percentage: 80

    spike_limit_percentage: 15

​

extensions:

  health_check: {}

​

exporters:

  otlp:

    endpoint: "jaeger:4317"

    tls:

      insecure: true

  prometheus:

    endpoint: 0.0.0.0:9090

  debug:

    verbosity: detailed

​

service:

  extensions: [health_check]

  pipelines:

    traces:

      receivers: [otlp]

      processors: [memory_limiter, batch]

      exporters: [otlp]

​

    metrics:

      receivers: [otlp]

      processors: [memory_limiter, batch]

      exporters: [prometheus]

​

"go.opentelemetry.io/otel"

"go.opentelemetry.io/otel/exporters/otlp/otlptrace"

"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"

"go.opentelemetry.io/otel/sdk/resource"

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

semconv "go.opentelemetry.io/otel/semconv/v1.17.0"

​

func InitTracer(ctx context.Context) func() {

    exp, err := otlptrace.New(

        ctx,

        otlptracegrpc.NewClient(

            otlptracegrpc.WithEndpoint(endpoint),

            otlptracegrpc.WithInsecure(),

        ),

    )

    if err != nil {

        log.Fatalf("failed to create OTLP trace exporter: %v", err)

    }

​

    res, err := resource.New(ctx,

        resource.WithAttributes(

            semconv.ServiceNameKey.String("auth-service"),

            semconv.ServiceVersionKey.String("1.0.0"),

            semconv.DeploymentEnvironmentKey.String("stg"),

        ),

    )

    if err != nil {

        log.Fatalf("failed to create resource: %v", err)

    }

​

    tp := trace.NewTracerProvider(

        trace.WithBatcher(exp),

        trace.WithResource(res),

    )

​

    otel.SetTracerProvider(tp)

​

    return func() {

        err := tp.Shutdown(ctx)

        if err != nil {

            log.Printf("error shutting down tracer provider: %v", err)

        }

    }

}

​

tracer := otel.Tracer("auth-service")

ctx, span := tracer.Start(ctx, "GetUserInfo")

defer span.End()

​

tracedLogger := logging.AddTraceContextToLogger(ctx)

​

tracedLogger.Info("find user info",

    zap.String("operation", "find user"),

    zap.String("username", username),

)

​

user, err := s.userRepo.GetUserInfo(ctx, username)

if err != nil {

    s.logger.Error(errNotFound)

    span.RecordError(err)

    span.SetStatus(otelCodes.Error, "Failed to fetch user info")

​

    return nil, status.Errorf(grpcCodes.NotFound, errNotFound, err)

}

​

span.SetStatus(otelCodes.Ok, "User info retrieved successfully")

​

func (r *UserRepository) GetUserInfo(ctx context.Context, username string) (*models.User, error) {

    tracer := otel.Tracer("auth-service")

    ctx, span := tracer.Start(ctx, "UserRepository.GetUserInfo",

        trace.WithAttributes(

            attribute.String("db.statement", query),

            attribute.String("db.user", username),

        ),

    )

    defer span.End()

​

    var user models.User

    // Some code that queries the DB...

    // err := doDatabaseCall()

​

    if err != nil {

        span.RecordError(err)

        span.SetStatus(codes.Error, "Failed to execute query")

        return nil, fmt.Errorf("failed to fetch user info: %w", err)

    }

​

    span.SetStatus(codes.Ok, "Query executed successfully")

    return &user, nil

}

​

server := grpc.NewServer(

    grpc.StatsHandler(otelgrpc.NewServerHandler()),

)

​

pb.RegisterAuthServiceServer(server, authService)

​

shutdown := tracing.InitTracer(ctx)

defer shutdown()

​

conn, err := grpc.Dial(

    "auth-service:50051",

    grpc.WithInsecure(),

    grpc.WithStatsHandler(otelgrpc.NewClientHandler()),

)

if err != nil {

    logger.Fatal("error", zap.Error(err))

}

​

func (r *mongoPurchaseRepo) SavePurchase(ctx context.Context, purchase entity.Purchase) error {

    span := r.handleTracing(ctx, purchase)

    defer span.End()

​

    // Insert the record into MongoDB, including the current span's Trace ID

    _, err := r.collection.InsertOne(ctx, bson.M{

        "_id":             purchase.ID,

        "user_id":         purchase.UserID,

        "username":        purchase.Username,

        "amount":          purchase.Amount,

        "currency":        purchase.Currency,

        "payment_method":  purchase.PaymentMethod,

        // ...

        "trace_id":        span.SpanContext().TraceID().String(),

    })

​

    return err

}

​

// If we find a Trace ID in the payload, attach it to the context

newCtx := ctx

if traceID != "" {

    log.Printf("Found Trace ID: %s", traceID)

    newCtx = context.WithValue(ctx, "trace-id", traceID)

}

​

// Create a new span

tracer := otel.Tracer("purchase-processor")

newCtx, span := tracer.Start(newCtx, "handler.processPayload")

defer span.End()

​

if traceID != "" {

    span.SetAttributes(

        attribute.String("trace.id", traceID),

    )

}

​

// Parse the "after" field into a Purchase struct...

var purchase model.Purchase

if err := mapstructure.Decode(afterDoc, &purchase); err != nil {

    log.Printf("Failed to map 'after' payload to Purchase struct: %v", err)

    return err

}

​

// If we find a Trace ID in the payload, attach it to the context

newCtx := ctx

if traceID != "" {

    log.Printf("Found Trace ID: %s", traceID)

    newCtx = context.WithValue(ctx, "trace-id", traceID)

}

​

// Create a new span

tracer := otel.Tracer("purchase-processor")

newCtx, span := tracer.Start(newCtx, "handler.processPayload")

defer span.End()

​

if traceID != "" {

    span.SetAttributes(

        attribute.String("trace.id", traceID),

    )

}

​

// Parse the "after" field into a Purchase struct...

var purchase model.Purchase

if err := mapstructure.Decode(afterDoc, &purchase); err != nil {

    log.Printf("Failed to map 'after' payload to Purchase struct: %v", err)

    return err

}

​

// saramaHeadersCarrier is a helper to set/get headers in a Sarama message.

type saramaHeadersCarrier *[]sarama.RecordHeader

​

func (c saramaHeadersCarrier) Get(key string) string {

    for _, h := range *c {

        if string(h.Key) == key {

            return string(h.Value)

        }

    }

    return ""

}

​

func (c saramaHeadersCarrier) Set(key string, value string) {

    *c = append(*c, sarama.RecordHeader{

        Key:   []byte(key),

        Value: []byte(value),

    })

}

​

// Before sending a message to Kafka:

func produceMessageWithTraceID(ctx context.Context, producer sarama.SyncProducer, topic string, value []byte) error {

    span := trace.SpanFromContext(ctx)

    traceID := span.SpanContext().TraceID().String()

​

    headers := make([]sarama.RecordHeader, 0)

    carrier := saramaHeadersCarrier(&headers)

    carrier.Set("trace-id", traceID)

​

    msg := &sarama.ProducerMessage{

        Topic:   topic,

        Value:   sarama.ByteEncoder(value),

        Headers: headers,

    }

​

    _, _, err := producer.SendMessage(msg)

    return err

}

​