add tracing

lib/gat/handlers/pool/pools/hybrid/pool.go

@@ -3,6 +3,10 @@ package hybrid
 import (
 	"context"
 	"fmt"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
 	"sync"
 	"time"
 
@@ -29,6 +33,8 @@ type Pool struct {
 
 	clients map[fed.BackendKey]*Client
 	mu      sync.RWMutex
+
+	tracer trace.Tracer
 }
 
 func NewPool(ctx context.Context, config Config) *Pool {
@@ -39,6 +45,9 @@ func NewPool(ctx context.Context, config Config) *Pool {
 
 		primary: spool.MakePool(c),
 		replica: spool.MakePool(c),
+		tracer: otel.Tracer("hybrid-pool", trace.WithInstrumentationAttributes(
+			attribute.String("component", "gfx.cafe/gfx/pggat/lib/gat/handlers/pool/pools/hybrid/pool.go"),
+		)),
 	}
 	go p.primary.ScaleLoop(ctx)
 	go p.replica.ScaleLoop(ctx)
@@ -62,6 +71,25 @@ func (T *Pool) RemoveRecipe(ctx context.Context, name string) {
 }
 
 func (T *Pool) Pair(ctx context.Context, client *Client, server *spool.Server) (err, serverErr error) {
+	ctx, span := T.tracer.Start(ctx, "Pair")
+	defer span.End()
+
+	// returning 2 errors is questionable
+	err, serverErr = T.pair(ctx, client, server)
+	if (err != nil) || (serverErr != nil) {
+		if serverErr != nil {
+			span.RecordError(err)
+			span.SetStatus(codes.Error, err.Error())
+		} else {
+			span.RecordError(err)
+			span.SetStatus(codes.Error, err.Error())
+		}
+	}
+
+	return
+}
+
+func (T *Pool) pair(ctx context.Context, client *Client, server *spool.Server) (err, serverErr error) {
 	client.SetState(metrics.ConnStatePairing, server, true)
 	server.SetState(metrics.ConnStatePairing, client.ID)
 
@@ -450,6 +478,20 @@ func (T *Pool) serveOnly(ctx context.Context, l prom.PoolHybridLabels, conn *fed
 }
 
 func (T *Pool) Serve(ctx context.Context, conn *fed.Conn) error {
+	ctx, span := T.tracer.Start(ctx, "Serve")
+	defer span.End()
+
+	// returning 2 errors is questionable
+	err := T.serve(ctx, conn)
+	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
+	}
+
+	return err
+}
+
+func (T *Pool) serve(ctx context.Context, conn *fed.Conn) error {
 	labels := prom.PoolHybridLabels{
 		Database: conn.Database,
 		User:     conn.User,
@@ -478,6 +520,9 @@ func (T *Pool) Serve(ctx context.Context, conn *fed.Conn) error {
 }
 
 func (T *Pool) Cancel(ctx context.Context, key fed.BackendKey) {
+	ctx, span := T.tracer.Start(ctx, "Cancel")
+	defer span.End()
+
 	peer, replica := func() (*spool.Server, bool) {
 		T.mu.RLock()
 		defer T.mu.RUnlock()
@@ -503,6 +548,9 @@ func (T *Pool) Cancel(ctx context.Context, key fed.BackendKey) {
 }
 
 func (T *Pool) ReadMetrics(ctx context.Context, m *metrics.Pool) {
+	ctx, span := T.tracer.Start(ctx, "ReadMetrics")
+	defer span.End()
+
 	T.primary.ReadMetrics(ctx, m)
 	T.replica.ReadMetrics(ctx, m)
 
@@ -520,6 +568,9 @@ func (T *Pool) ReadMetrics(ctx context.Context, m *metrics.Pool) {
 }
 
 func (T *Pool) Close(ctx context.Context) {
+	ctx, span := T.tracer.Start(ctx, "Close")
+	defer span.End()
+
 	T.primary.Close(ctx)
 	T.replica.Close(ctx)
 }