test otel

cmd/cart/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"log"
@@ -206,9 +207,15 @@ func main() {
 	done := make(chan bool, 1)
 	signal.Notify(sigs, syscall.SIGTERM)
 
+	otelShutdown, err := setupOTelSDK(context.Background())
+	if err != nil {
+		log.Fatalf("Unable to start otel %v", err)
+	}
+
 	go func() {
 		sig := <-sigs
 		fmt.Println("Shutting down due to signal:", sig)
+		otelShutdown(context.Background())
 		diskStorage.Close()
 		pool.Close()
 

cmd/cart/otel.go

@@ -0,0 +1,122 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/exporters/stdout/stdoutlog"
+	"go.opentelemetry.io/otel/exporters/stdout/stdoutmetric"
+	"go.opentelemetry.io/otel/exporters/stdout/stdouttrace"
+	"go.opentelemetry.io/otel/log/global"
+	"go.opentelemetry.io/otel/propagation"
+	"go.opentelemetry.io/otel/sdk/log"
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/trace"
+)
+
+// setupOTelSDK bootstraps the OpenTelemetry pipeline.
+// If it does not return an error, make sure to call shutdown for proper cleanup.
+func setupOTelSDK(ctx context.Context) (func(context.Context) error, error) {
+	var shutdownFuncs []func(context.Context) error
+	var err error
+
+	// shutdown calls cleanup functions registered via shutdownFuncs.
+	// The errors from the calls are joined.
+	// Each registered cleanup will be invoked once.
+	shutdown := func(ctx context.Context) error {
+		var err error
+		for _, fn := range shutdownFuncs {
+			err = errors.Join(err, fn(ctx))
+		}
+		shutdownFuncs = nil
+		return err
+	}
+
+	// handleErr calls shutdown for cleanup and makes sure that all errors are returned.
+	handleErr := func(inErr error) {
+		err = errors.Join(inErr, shutdown(ctx))
+	}
+
+	// Set up propagator.
+	prop := newPropagator()
+	otel.SetTextMapPropagator(prop)
+
+	// Set up trace provider.
+	tracerProvider, err := newTracerProvider()
+	if err != nil {
+		handleErr(err)
+		return shutdown, err
+	}
+	shutdownFuncs = append(shutdownFuncs, tracerProvider.Shutdown)
+	otel.SetTracerProvider(tracerProvider)
+
+	// Set up meter provider.
+	meterProvider, err := newMeterProvider()
+	if err != nil {
+		handleErr(err)
+		return shutdown, err
+	}
+	shutdownFuncs = append(shutdownFuncs, meterProvider.Shutdown)
+	otel.SetMeterProvider(meterProvider)
+
+	// Set up logger provider.
+	loggerProvider, err := newLoggerProvider()
+	if err != nil {
+		handleErr(err)
+		return shutdown, err
+	}
+	shutdownFuncs = append(shutdownFuncs, loggerProvider.Shutdown)
+	global.SetLoggerProvider(loggerProvider)
+
+	return shutdown, err
+}
+
+func newPropagator() propagation.TextMapPropagator {
+	return propagation.NewCompositeTextMapPropagator(
+		propagation.TraceContext{},
+		propagation.Baggage{},
+	)
+}
+
+func newTracerProvider() (*trace.TracerProvider, error) {
+	traceExporter, err := stdouttrace.New(
+		stdouttrace.WithPrettyPrint())
+	if err != nil {
+		return nil, err
+	}
+
+	tracerProvider := trace.NewTracerProvider(
+		trace.WithBatcher(traceExporter,
+			// Default is 5s. Set to 1s for demonstrative purposes.
+			trace.WithBatchTimeout(time.Second)),
+	)
+	return tracerProvider, nil
+}
+
+func newMeterProvider() (*metric.MeterProvider, error) {
+	metricExporter, err := stdoutmetric.New()
+	if err != nil {
+		return nil, err
+	}
+
+	meterProvider := metric.NewMeterProvider(
+		metric.WithReader(metric.NewPeriodicReader(metricExporter,
+			// Default is 1m. Set to 3s for demonstrative purposes.
+			metric.WithInterval(3*time.Second))),
+	)
+	return meterProvider, nil
+}
+
+func newLoggerProvider() (*log.LoggerProvider, error) {
+	logExporter, err := stdoutlog.New()
+	if err != nil {
+		return nil, err
+	}
+
+	loggerProvider := log.NewLoggerProvider(
+		log.WithProcessor(log.NewBatchProcessor(logExporter)),
+	)
+	return loggerProvider, nil
+}

cmd/cart/pool-server.go

@@ -17,6 +17,9 @@ import (
 	"github.com/gogo/protobuf/proto"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
+	"go.opentelemetry.io/contrib/bridges/otelslog"
+	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
+	"go.opentelemetry.io/otel"
 )
 
 var (
@@ -449,7 +452,10 @@ func CartIdHandler(fn func(cartId cart.CartId, w http.ResponseWriter, r *http.Re
 func (s *PoolServer) ProxyHandler(fn func(w http.ResponseWriter, r *http.Request, cartId cart.CartId) error) func(cartId cart.CartId, w http.ResponseWriter, r *http.Request) error {
 	return func(cartId cart.CartId, w http.ResponseWriter, r *http.Request) error {
 		if ownerHost, ok := s.OwnerHost(uint64(cartId)); ok {
+			ctx, span := tracer.Start(r.Context(), "proxy")
+			defer span.End()
 			handled, err := ownerHost.Proxy(uint64(cartId), w, r)
+			logger.InfoContext(ctx, fmt.Sprintf("owner host %s", ownerHost.Name()))
 			grainLookups.Inc()
 			if err == nil && handled {
 				return nil
@@ -461,6 +467,15 @@ func (s *PoolServer) ProxyHandler(fn func(w http.ResponseWriter, r *http.Request
 	}
 }
 
+var (
+	tracer = otel.Tracer(name)
+
+	// meter   = otel.Meter(name)
+	logger = otelslog.NewLogger(name)
+
+// rollCnt metric.Int64Counter
+)
+
 type AddVoucherRequest struct {
 	VoucherCode string `json:"code"`
 }
@@ -555,20 +570,26 @@ func (s *PoolServer) Serve() *http.ServeMux {
 		w.WriteHeader(http.StatusOK)
 	})
 
-	mux.HandleFunc("GET /", CookieCartIdHandler(s.ProxyHandler(s.GetCartHandler)))
-	mux.HandleFunc("GET /add/{sku}", CookieCartIdHandler(s.ProxyHandler(s.AddSkuToCartHandler)))
-	mux.HandleFunc("POST /add", CookieCartIdHandler(s.ProxyHandler(s.AddMultipleItemHandler)))
-	mux.HandleFunc("POST /", CookieCartIdHandler(s.ProxyHandler(s.AddSkuRequestHandler)))
-	mux.HandleFunc("POST /set", CookieCartIdHandler(s.ProxyHandler(s.SetCartItemsHandler)))
-	mux.HandleFunc("DELETE /{itemId}", CookieCartIdHandler(s.ProxyHandler(s.DeleteItemHandler)))
-	mux.HandleFunc("PUT /", CookieCartIdHandler(s.ProxyHandler(s.QuantityChangeHandler)))
-	mux.HandleFunc("DELETE /", CookieCartIdHandler(s.ProxyHandler(s.RemoveCartCookie)))
-	mux.HandleFunc("POST /delivery", CookieCartIdHandler(s.ProxyHandler(s.SetDeliveryHandler)))
-	mux.HandleFunc("DELETE /delivery/{deliveryId}", CookieCartIdHandler(s.ProxyHandler(s.RemoveDeliveryHandler)))
-	mux.HandleFunc("PUT /delivery/{deliveryId}/pickupPoint", CookieCartIdHandler(s.ProxyHandler(s.SetPickupPointHandler)))
-	mux.HandleFunc("PUT /voucher", CookieCartIdHandler(s.ProxyHandler(s.AddVoucherHandler)))
-	mux.HandleFunc("PUT /subscription-details", CookieCartIdHandler(s.ProxyHandler(s.SubscriptionDetailsHandler)))
-	mux.HandleFunc("DELETE /voucher/{voucherId}", CookieCartIdHandler(s.ProxyHandler(s.RemoveVoucherHandler)))
+	handleFunc := func(pattern string, handlerFunc func(http.ResponseWriter, *http.Request)) {
+		// Configure the "http.route" for the HTTP instrumentation.
+		handler := otelhttp.WithRouteTag(pattern, http.HandlerFunc(handlerFunc))
+		mux.Handle(pattern, handler)
+	}
+
+	handleFunc("GET /", CookieCartIdHandler(s.ProxyHandler(s.GetCartHandler)))
+	handleFunc("GET /add/{sku}", CookieCartIdHandler(s.ProxyHandler(s.AddSkuToCartHandler)))
+	handleFunc("POST /add", CookieCartIdHandler(s.ProxyHandler(s.AddMultipleItemHandler)))
+	handleFunc("POST /", CookieCartIdHandler(s.ProxyHandler(s.AddSkuRequestHandler)))
+	handleFunc("POST /set", CookieCartIdHandler(s.ProxyHandler(s.SetCartItemsHandler)))
+	handleFunc("DELETE /{itemId}", CookieCartIdHandler(s.ProxyHandler(s.DeleteItemHandler)))
+	handleFunc("PUT /", CookieCartIdHandler(s.ProxyHandler(s.QuantityChangeHandler)))
+	handleFunc("DELETE /", CookieCartIdHandler(s.ProxyHandler(s.RemoveCartCookie)))
+	handleFunc("POST /delivery", CookieCartIdHandler(s.ProxyHandler(s.SetDeliveryHandler)))
+	handleFunc("DELETE /delivery/{deliveryId}", CookieCartIdHandler(s.ProxyHandler(s.RemoveDeliveryHandler)))
+	handleFunc("PUT /delivery/{deliveryId}/pickupPoint", CookieCartIdHandler(s.ProxyHandler(s.SetPickupPointHandler)))
+	handleFunc("PUT /voucher", CookieCartIdHandler(s.ProxyHandler(s.AddVoucherHandler)))
+	handleFunc("PUT /subscription-details", CookieCartIdHandler(s.ProxyHandler(s.SubscriptionDetailsHandler)))
+	handleFunc("DELETE /voucher/{voucherId}", CookieCartIdHandler(s.ProxyHandler(s.RemoveVoucherHandler)))
 
 	//mux.HandleFunc("GET /checkout", CookieCartIdHandler(s.ProxyHandler(s.HandleCheckout)))
 	//mux.HandleFunc("GET /confirmation/{orderId}", CookieCartIdHandler(s.ProxyHandler(s.HandleConfirmation)))