[hw-7] add metrics, tracing

loms/internal/infra/db/postgres/postgres.go

@@ -15,6 +15,8 @@ func NewPool(ctx context.Context, dsn string) (*pgxpool.Pool, error) {
 		return nil, errors.Wrap(err, "pgxpool.ParseConfig")
 	}
 
+	config.ConnConfig.Tracer = promTracer{}
+
 	pool, err := pgxpool.NewWithConfig(ctx, config)
 	if err != nil {
 		return nil, errors.Wrap(err, "pgxpool.NewWithConfig")
@@ -34,6 +36,8 @@ func NewPools(ctx context.Context, DSNs ...string) ([]*pgxpool.Pool, error) {
 			return nil, errors.Wrap(err, "pgxpool.ParseConfig")
 		}
 
+		cfg.ConnConfig.Tracer = promTracer{}
+
 		pool, err := pgxpool.NewWithConfig(ctx, cfg)
 		if err != nil {
 			closeOpened(pools[:i])

loms/internal/infra/db/postgres/tracer.go

@@ -0,0 +1,38 @@
+package postgres
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"route256/loms/internal/infra/grpc/metrics"
+
+	"github.com/jackc/pgx/v5"
+)
+
+type startKey struct{}
+
+type promTracer struct{}
+
+func (promTracer) TraceQueryStart(ctx context.Context, _ *pgx.Conn, _ pgx.TraceQueryStartData) context.Context {
+	return context.WithValue(ctx, startKey{}, time.Now())
+}
+
+func (promTracer) TraceQueryEnd(ctx context.Context, _ *pgx.Conn, data pgx.TraceQueryEndData) {
+	start, _ := ctx.Value(startKey{}).(time.Time)
+
+	sql := strings.TrimSpace(strings.ToLower(data.CommandTag.String()))
+	parts := strings.SplitN(sql, " ", 2)
+	category := "unknown"
+	if len(parts) > 0 {
+		category = parts[0]
+	}
+
+	errLabel := "none"
+	if data.Err != nil {
+		errLabel = "error"
+	}
+
+	metrics.IncDBQueryCount(category, errLabel)
+	metrics.StoreDBQueryDuration(category, errLabel, time.Since(start))
+}

loms/internal/infra/grpc/metrics/db.go

@@ -0,0 +1,31 @@
+package metrics
+
+import (
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var (
+	dbQueryCounter = promauto.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "app",
+		Name:      "db_queries_total",
+		Help:      "Total DB queries",
+	}, []string{"category", "error"})
+
+	dbQueryDurationHistogram = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "app",
+		Name:      "db_query_duration_seconds",
+		Help:      "Latency of DB queries, seconds",
+		Buckets:   prometheus.DefBuckets,
+	}, []string{"category", "error"})
+)
+
+func IncDBQueryCount(cat, errLabel string) {
+	dbQueryCounter.WithLabelValues(cat, errLabel).Inc()
+}
+
+func StoreDBQueryDuration(cat, errLabel string, d time.Duration) {
+	dbQueryDurationHistogram.WithLabelValues(cat, errLabel).Observe(d.Seconds())
+}

loms/internal/infra/grpc/metrics/metrics.go

@@ -0,0 +1,31 @@
+package metrics
+
+import (
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var (
+	requestCounter = promauto.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "app",
+		Name:      "requests_total",
+		Help:      "Total amount of request by handler",
+	}, []string{"method", "path", "status_code"})
+
+	requestDurationHistogram = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "app",
+		Name:      "request_duration_seconds",
+		Help:      "Latency of handler processing, seconds",
+		Buckets:   prometheus.DefBuckets,
+	}, []string{"method", "path", "status_code"})
+)
+
+func IncRequestHandlerCount(method, path, statusCode string) {
+	requestCounter.WithLabelValues(method, path, statusCode).Inc()
+}
+
+func StoreHandlerRequestDuration(method, path, statusCode string, d time.Duration) {
+	requestDurationHistogram.WithLabelValues(method, path, statusCode).Observe(d.Seconds())
+}

loms/internal/infra/grpc/middleware/metrics.go

@@ -0,0 +1,43 @@
+package mw
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"route256/loms/internal/infra/grpc/metrics"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/status"
+)
+
+const UNKNOWN = "unknown"
+
+func WithMetrics(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp any, err error) {
+	start := time.Now()
+
+	resp, err = handler(ctx, req)
+
+	// "/pkg.Service/Method"
+	service, method := parseFullMethod(info.FullMethod)
+
+	code := status.Code(err).String()
+
+	metrics.IncRequestHandlerCount(service, method, code)
+	metrics.StoreHandlerRequestDuration(service, method, code, time.Since(start))
+
+	return resp, err
+}
+
+func parseFullMethod(full string) (service, method string) {
+	if full == "" {
+		return UNKNOWN, UNKNOWN
+	}
+	full = strings.TrimPrefix(full, "/")
+
+	slash := strings.Index(full, "/")
+	if slash < 0 {
+		return full, UNKNOWN
+	}
+	return full[:slash], full[slash+1:]
+}

loms/internal/infra/grpc/middleware/tracing.go

@@ -0,0 +1,38 @@
+package mw
+
+import (
+	"context"
+	"strings"
+
+	"go.opentelemetry.io/otel/codes"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/status"
+
+	"route256/loms/internal/infra/tracing"
+)
+
+func WithTracing(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
+	service, method := splitFullMethod(info.FullMethod)
+	ctx, span := tracing.Tracer().Start(ctx, service+"/"+method)
+	defer span.End()
+
+	resp, err := handler(ctx, req)
+
+	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, status.Convert(err).Message())
+	} else {
+		span.SetStatus(codes.Ok, "OK")
+	}
+
+	return resp, err
+}
+
+func splitFullMethod(full string) (service, method string) {
+	full = strings.TrimPrefix(full, "/")
+	if i := strings.Index(full, "/"); i >= 0 {
+		return full[:i], full[i+1:]
+	}
+
+	return "unknown", full
+}

loms/internal/infra/messaging/kafka/producer.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"route256/loms/internal/domain/entity"
+	"route256/loms/internal/infra/tracing"
 
 	"github.com/IBM/sarama"
 	"github.com/rs/zerolog/log"
@@ -64,6 +65,9 @@ func (p *statusProducer) runCallbacks() {
 }
 
 func (p *statusProducer) Send(ctx context.Context, id entity.ID, status string) error {
+	ctx, span := tracing.Tracer().Start(ctx, "Producer.Send")
+	defer span.End()
+
 	log.Debug().Msgf("sending event for id: %d; status: %s", id, status)
 
 	newStatus, err := mapOrderStatus(status)
@@ -86,6 +90,9 @@ func (p *statusProducer) Send(ctx context.Context, id entity.ID, status string)
 }
 
 func (p *statusProducer) SendRaw(ctx context.Context, id entity.ID, value []byte) error {
+	ctx, span := tracing.Tracer().Start(ctx, "Producer.SendRaw")
+	defer span.End()
+
 	if len(value) == 0 {
 		return fmt.Errorf("empty message value")
 	}

loms/internal/infra/tracing/init.go

@@ -0,0 +1,48 @@
+package tracing
+
+import (
+	"context"
+
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/sdk/resource"
+	"go.opentelemetry.io/otel/sdk/trace"
+	semconv "go.opentelemetry.io/otel/semconv/v1.17.0"
+	oteltrace "go.opentelemetry.io/otel/trace"
+)
+
+var (
+	globalTracer oteltrace.Tracer
+	provider     *trace.TracerProvider
+)
+
+func NewTracer(serviceName string, option ...trace.TracerProviderOption) oteltrace.Tracer {
+	option = append([]trace.TracerProviderOption{
+		trace.WithResource(resource.NewWithAttributes(
+			semconv.SchemaURL,
+			semconv.ServiceName(serviceName),
+		)),
+	}, option...)
+
+	provider = trace.NewTracerProvider(option...)
+
+	otel.SetTracerProvider(provider)
+
+	globalTracer = otel.GetTracerProvider().Tracer("loms")
+
+	return globalTracer
+}
+
+func Shutdown(ctx context.Context) error {
+	if provider != nil {
+		return provider.Shutdown(ctx)
+	}
+	return nil
+}
+
+func Tracer() oteltrace.Tracer {
+	if globalTracer == nil {
+		return otel.Tracer("loms")
+	}
+
+	return globalTracer
+}

loms/internal/infra/tracing/otel.go

@@ -0,0 +1,22 @@
+package tracing
+
+import (
+	"context"
+
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+)
+
+func InitOTLP(ctx context.Context, serviceName, endpoint string) (shutdown func(context.Context) error, err error) {
+	exp, err := otlptracehttp.New(ctx,
+		otlptracehttp.WithEndpoint(endpoint),
+		otlptracehttp.WithInsecure(),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	NewTracer(serviceName, sdktrace.WithBatcher(exp))
+
+	return Shutdown, nil
+}