feat: add otel instrumentation (#51)

Author: w-h-a

cmd/engine.go

@@ -1,6 +1,7 @@
 package cmd
 
 import (
+	"context"
 	"log/slog"
 	"sync"
 	"time"
@@ -19,12 +20,119 @@ import (
 	"github.com/w-h-a/workflow/internal/engine/config"
 	"github.com/w-h-a/workflow/internal/engine/services/coordinator"
 	"github.com/w-h-a/workflow/internal/engine/services/worker"
+	"go.opentelemetry.io/contrib/bridges/otelslog"
+	"go.opentelemetry.io/contrib/instrumentation/host"
+	"go.opentelemetry.io/contrib/instrumentation/runtime"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
+	"go.opentelemetry.io/otel/exporters/stdout/stdoutlog"
+	globallog "go.opentelemetry.io/otel/log/global"
+	"go.opentelemetry.io/otel/propagation"
+	logsdk "go.opentelemetry.io/otel/sdk/log"
+	metricsdk "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/resource"
+	tracesdk "go.opentelemetry.io/otel/sdk/trace"
+	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
 )
 
 func StartEngine(ctx *cli.Context) error {
 	// cfg
 	config.New()
 
+	// resource
+	name := config.Name()
+
+	resource, err := resource.New(
+		context.Background(),
+		resource.WithAttributes(
+			semconv.ServiceName(name),
+		),
+		resource.WithProcess(),
+	)
+	if err != nil {
+		return err
+	}
+
+	// logs
+	logsExporter, err := initLogsExporter()
+	if err != nil {
+		return err
+	}
+
+	lp := logsdk.NewLoggerProvider(
+		logsdk.WithResource(resource),
+		logsdk.WithProcessor(
+			logsdk.NewBatchProcessor(logsExporter),
+		),
+	)
+
+	defer lp.Shutdown(context.Background())
+
+	globallog.SetLoggerProvider(lp)
+
+	logger := otelslog.NewLogger(
+		config.Name(),
+		otelslog.WithLoggerProvider(lp),
+	)
+
+	slog.SetDefault(logger)
+
+	// traces
+	traceExporter, err := initTracesExporter()
+	if err != nil {
+		return err
+	}
+
+	tp := tracesdk.NewTracerProvider(
+		tracesdk.WithResource(resource),
+		tracesdk.WithSampler(tracesdk.AlwaysSample()),
+		tracesdk.WithSpanProcessor(
+			tracesdk.NewBatchSpanProcessor(
+				traceExporter,
+			),
+		),
+	)
+
+	defer tp.Shutdown(context.Background())
+
+	otel.SetTracerProvider(tp)
+	otel.SetTextMapPropagator(
+		propagation.NewCompositeTextMapPropagator(
+			propagation.TraceContext{},
+			propagation.Baggage{},
+		),
+	)
+
+	// metrics
+	metricsExporter, err := initMetricsExporter()
+	if err != nil {
+		return err
+	}
+
+	mp := metricsdk.NewMeterProvider(
+		metricsdk.WithResource(resource),
+		metricsdk.WithReader(
+			metricsdk.NewPeriodicReader(
+				metricsExporter,
+				metricsdk.WithInterval(15*time.Second),
+				metricsdk.WithProducer(runtime.NewProducer()),
+			),
+		),
+	)
+
+	defer mp.Shutdown(context.Background())
+
+	otel.SetMeterProvider(mp)
+
+	if err := host.Start(); err != nil {
+		return err
+	}
+
+	if err := runtime.Start(runtime.WithMinimumReadMemStatsInterval(time.Second)); err != nil {
+		return err
+	}
+
 	// broker client
 	brokerClient := initBroker()
 
@@ -105,7 +213,7 @@ func StartEngine(ctx *cli.Context) error {
 	}
 
 	// block
-	err := <-ch
+	err = <-ch
 	if err != nil {
 		slog.ErrorContext(ctx.Context, "failed to start", "error", err)
 		return err
@@ -138,6 +246,26 @@ func StartEngine(ctx *cli.Context) error {
 	return nil
 }
 
+func initLogsExporter() (logsdk.Exporter, error) {
+	return stdoutlog.New()
+}
+
+func initTracesExporter() (tracesdk.SpanExporter, error) {
+	return otlptracehttp.New(
+		context.Background(),
+		otlptracehttp.WithEndpoint(config.TracesAddress()),
+		otlptracehttp.WithInsecure(),
+	)
+}
+
+func initMetricsExporter() (metricsdk.Exporter, error) {
+	return otlpmetrichttp.New(
+		context.Background(),
+		otlpmetrichttp.WithEndpoint(config.MetricsAddress()),
+		otlpmetrichttp.WithInsecure(),
+	)
+}
+
 func initBroker() broker.Broker {
 	switch config.Broker() {
 	case string(broker.Rabbit):

docker-compose.yaml

@@ -1,4 +1,35 @@
 services:
+  jaeger:
+    container_name: jaeger
+    image: jaegertracing/all-in-one:1.62.0
+    ports:
+      - '16686:16686'
+    healthcheck:
+      test: ["CMD-SHELL", "wget --spider http://localhost:16686/"] # Updated command
+      interval: 10s
+      timeout: 5s
+      retries: 5
+      start_period: 30s
+
+  prometheus:
+    container_name: prometheus
+    image: prom/prometheus:v3.0.0
+    command: 
+      - "--config.file=/etc/prometheus/prometheus.yaml"
+      - "--web.enable-otlp-receiver"
+      - "--web.enable-remote-write-receiver"
+      - "--enable-feature=remote-write-receiver"
+    ports:
+      - '9090:9090'
+    volumes:
+      - ./prometheus.yaml:/etc/prometheus/prometheus.yaml
+    healthcheck:
+      test: ["CMD-SHELL", "wget --spider http://localhost:9090/metrics"] # Updated command
+      interval: 10s
+      timeout: 5s
+      retries: 5
+      start_period: 30s
+
   rabbitmq:
     container_name: rabbitmq
     image: rabbitmq:3.13-management
@@ -39,13 +70,20 @@ services:
       - VERSION=0.1.0-alpha.0
       - MODE=worker
       - HTTP_ADDRESS=:4001
+      - TRACES_ADDRESS=jaeger:4318
+      - METRICS_ADDRESS=prometheus:9090
+      - OTEL_EXPORTER_OTLP_METRICS_ENDPOINT=http://prometheus:9090/api/v1/otlp/v1/metrics
       - BROKER=rabbit
       - BROKER_LOCATION=amqp://guest:guest@rabbitmq:5672
       - RUNNER=docker
       - RUNNER_HOST=unix:///var/run/docker.sock
     volumes:
       - /var/run/docker.sock:/var/run/docker.sock
     depends_on:
+      jaeger:
+        condition: service_healthy
+      prometheus:
+        condition: service_healthy
       rabbitmq:
         condition: service_healthy
 
@@ -61,11 +99,18 @@ services:
       - VERSION=0.1.0-alpha.0
       - MODE=coordinator
       - HTTP_ADDRESS=:4000
+      - TRACES_ADDRESS=jaeger:4318
+      - METRICS_ADDRESS=prometheus:9090
+      - OTEL_EXPORTER_OTLP_METRICS_ENDPOINT=http://prometheus:9090/api/v1/otlp/v1/metrics
       - BROKER=rabbit
       - BROKER_LOCATION=amqp://guest:guest@rabbitmq:5672
       - READ_WRITER=postgres
       - READ_WRITER_LOCATION=postgres://tasks:tasks@postgres:5432/tasks?sslmode=disable
     depends_on:
+      jaeger:
+        condition: service_healthy
+      prometheus:
+        condition: service_healthy
       rabbitmq:
         condition: service_healthy
       postgres:

go.mod

@@ -11,34 +11,61 @@ require (
 	github.com/urfave/cli/v2 v2.27.7
 	github.com/w-h-a/pkg v0.41.0
 	go.nhat.io/otelsql v0.16.0
+	go.opentelemetry.io/contrib/bridges/otelslog v0.13.0
+	go.opentelemetry.io/contrib/instrumentation/host v0.63.0
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.62.0
-	go.opentelemetry.io/otel v1.37.0
+	go.opentelemetry.io/contrib/instrumentation/runtime v0.63.0
+	go.opentelemetry.io/otel v1.38.0
+	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.38.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.37.0
+	go.opentelemetry.io/otel/exporters/stdout/stdoutlog v0.14.0
+	go.opentelemetry.io/otel/log v0.14.0
+	go.opentelemetry.io/otel/sdk v1.38.0
+	go.opentelemetry.io/otel/sdk/log v0.14.0
+	go.opentelemetry.io/otel/sdk/metric v1.38.0
+	go.opentelemetry.io/otel/trace v1.38.0
 )
 
 require (
 	github.com/Microsoft/go-winio v0.4.14 // indirect
+	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
 	github.com/containerd/log v0.1.0 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.7 // indirect
 	github.com/distribution/reference v0.6.0 // indirect
 	github.com/docker/go-connections v0.5.0 // indirect
 	github.com/docker/go-units v0.5.0 // indirect
+	github.com/ebitengine/purego v0.8.4 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/go-logr/logr v1.4.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-ole/go-ole v1.3.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 // indirect
+	github.com/lufia/plan9stats v0.0.0-20250827001030-24949be3fa54 // indirect
 	github.com/moby/docker-image-spec v1.3.1 // indirect
 	github.com/moby/term v0.5.2 // indirect
 	github.com/morikuni/aec v1.0.0 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/opencontainers/image-spec v1.1.1 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
+	github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
+	github.com/shirou/gopsutil/v4 v4.25.7 // indirect
+	github.com/tklauser/go-sysconf v0.3.15 // indirect
+	github.com/tklauser/numcpus v0.10.0 // indirect
 	github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1 // indirect
+	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.37.0 // indirect
-	go.opentelemetry.io/otel/metric v1.37.0 // indirect
-	go.opentelemetry.io/otel/trace v1.37.0 // indirect
-	golang.org/x/sys v0.33.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.37.0 // indirect
+	go.opentelemetry.io/otel/metric v1.38.0 // indirect
+	go.opentelemetry.io/proto/otlp v1.7.1 // indirect
+	golang.org/x/net v0.43.0 // indirect
+	golang.org/x/sys v0.35.0 // indirect
+	golang.org/x/text v0.28.0 // indirect
 	golang.org/x/time v0.12.0 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20250825161204-c5933d9347a5 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250825161204-c5933d9347a5 // indirect
+	google.golang.org/grpc v1.75.0 // indirect
+	google.golang.org/protobuf v1.36.8 // indirect
 	gotest.tools/v3 v3.5.2 // indirect
 )