gs: Improve gatewayserver io tracing for ttigw & lbs lns

Author: vlasebian

pkg/gatewayserver/io/semtechws/ws.go

@@ -29,6 +29,9 @@ import (
 
 	"github.com/gorilla/mux"
 	"github.com/gorilla/websocket"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
 	"go.thethings.network/lorawan-stack/v3/pkg/auth/rights"
 	"go.thethings.network/lorawan-stack/v3/pkg/errors"
 	"go.thethings.network/lorawan-stack/v3/pkg/frequencyplans"
@@ -37,6 +40,7 @@ import (
 	"go.thethings.network/lorawan-stack/v3/pkg/log"
 	"go.thethings.network/lorawan-stack/v3/pkg/random"
 	"go.thethings.network/lorawan-stack/v3/pkg/ratelimit"
+	"go.thethings.network/lorawan-stack/v3/pkg/telemetry/tracing"
 	"go.thethings.network/lorawan-stack/v3/pkg/ttnpb"
 	"go.thethings.network/lorawan-stack/v3/pkg/types"
 	"go.thethings.network/lorawan-stack/v3/pkg/unique"
@@ -45,7 +49,10 @@ import (
 	"google.golang.org/grpc/metadata"
 )
 
-const pingIntervalJitter = 0.1
+const (
+	pingIntervalJitter = 0.1
+	tracerNamespace    = "go.thethings.network/lorawan-stack/pkg/gatewayserver/io/semtechws"
+)
 
 var (
 	errGatewayID          = errors.DefineInvalidArgument("invalid_gateway_id", "invalid gateway ID `{id}`")
@@ -118,6 +125,15 @@ func (s *srv) handleConnectionInfo(w http.ResponseWriter, r *http.Request) {
 	))
 	logger := log.FromContext(ctx)
 
+	tracer := tracing.FromContext(s.ctx).Tracer(tracerNamespace)
+	ctx, span := tracer.Start(ctx, "HandleConnectionInfo",
+		trace.WithAttributes(
+			attribute.String("protocol", s.Protocol()),
+			attribute.String("remote_addr", r.RemoteAddr),
+		),
+	)
+	defer span.End()
+
 	assertAuth := func(ctx context.Context, ids *ttnpb.GatewayIdentifiers) error {
 		ctx, hasAuth := withForwardedAuth(ctx, ids, r.Header.Get("Authorization"))
 		if !hasAuth {
@@ -188,6 +204,21 @@ func (s *srv) handleTraffic(w http.ResponseWriter, r *http.Request) (err error)
 		downstreamCh = make(chan []byte, 1)
 	)
 
+	tracer := tracing.FromContext(s.ctx).Tracer(tracerNamespace)
+	ctx, span := tracer.Start(ctx, "HandleTraffic",
+		trace.WithAttributes(
+			attribute.String("protocol", s.Protocol()),
+			attribute.String("remote_addr", r.RemoteAddr),
+		),
+	)
+	defer func() {
+		if err != nil {
+			span.RecordError(err)
+			span.SetStatus(codes.Error, err.Error())
+		}
+		span.End()
+	}()
+
 	ctx = log.NewContextWithFields(ctx, log.Fields(
 		"endpoint", eps.Traffic,
 		"remote_addr", r.RemoteAddr,
@@ -352,15 +383,23 @@ func (s *srv) handleTraffic(w http.ResponseWriter, r *http.Request) (err error)
 					return err
 				}
 			case down := <-conn.Down():
-				dnmsg, err := s.formatter.FromDownlink(ctx, down, conn.BandID(), time.Now(), downlinkTokens)
-				if err != nil {
-					logger.WithError(err).Warn("Failed to marshal downlink message")
+				_, dnSpan := tracer.Start(ctx, "HandleDownMessage")
+				dnmsg, dnErr := s.formatter.FromDownlink(ctx, down, conn.BandID(), time.Now(), downlinkTokens)
+				if dnErr != nil {
+					dnSpan.RecordError(dnErr)
+					dnSpan.SetStatus(codes.Error, dnErr.Error())
+					dnSpan.End()
+					logger.WithError(dnErr).Warn("Failed to marshal downlink message")
 					continue
 				}
-				if err = ws.WriteMessage(websocket.TextMessage, dnmsg); err != nil {
-					logger.WithError(err).Warn("Failed to send downlink message")
-					return err
+				if dnErr = ws.WriteMessage(websocket.TextMessage, dnmsg); dnErr != nil {
+					dnSpan.RecordError(dnErr)
+					dnSpan.SetStatus(codes.Error, dnErr.Error())
+					dnSpan.End()
+					logger.WithError(dnErr).Warn("Failed to send downlink message")
+					return dnErr
 				}
+				dnSpan.End()
 			case downstream := <-downstreamCh:
 				if err := ws.WriteMessage(websocket.TextMessage, downstream); err != nil {
 					logger.WithError(err).Warn("Failed to send message downstream")
@@ -381,7 +420,20 @@ func (s *srv) handleTraffic(w http.ResponseWriter, r *http.Request) (err error)
 			logger.WithError(err).Debug("Failed to read message")
 			return err
 		}
-		downstream, err := s.formatter.HandleUp(ctx, data, ids, conn, time.Now(), downlinkTokens)
+		var downstream []byte
+		func() {
+			_, msgSpan := tracer.Start(ctx, "HandleUpMessage",
+				trace.WithAttributes(
+					attribute.Int("message_size", len(data)),
+				),
+			)
+			defer msgSpan.End()
+			downstream, err = s.formatter.HandleUp(ctx, data, ids, conn, time.Now(), downlinkTokens)
+			if err != nil {
+				msgSpan.RecordError(err)
+				msgSpan.SetStatus(codes.Error, err.Error())
+			}
+		}()
 		if err != nil {
 			return err
 		}

pkg/gatewayserver/io/ttigw/ttigw.go

@@ -28,6 +28,9 @@ import (
 	"github.com/coder/websocket"
 	"github.com/gorilla/mux"
 	"go.opentelemetry.io/contrib/instrumentation/github.com/gorilla/mux/otelmux"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
 	lorav1 "go.thethings.industries/pkg/api/gen/tti/gateway/data/lora/v1"
 	ttica "go.thethings.industries/pkg/ca"
 	"go.thethings.network/lorawan-stack/v3/pkg/auth/mtls"
@@ -51,13 +54,15 @@ const (
 	pingIntervalJitter = 0.1
 	pingTimeout        = 30 * time.Second
 	subprotocol        = "v1.lora.data.gateway.thethings.industries"
+	tracerNamespace    = "go.thethings.network/lorawan-stack/pkg/gatewayserver/io/ttigw"
 )
 
 // Frontend implements the The Things Industries V1 gateway frontend.
 type Frontend struct {
 	http.Handler
 	server io.Server
 	cfg    Config
+	tracer trace.Tracer
 }
 
 var _ io.Frontend = (*Frontend)(nil)
@@ -87,6 +92,7 @@ func New(ctx context.Context, server io.Server, cfg Config) (*Frontend, error) {
 		Handler: router,
 		server:  server,
 		cfg:     cfg,
+		tracer:  tracing.FromContext(ctx).Tracer(tracerNamespace),
 	}
 
 	router.HandleFunc("/api/protocols/tti/v1", f.handleGet).Methods(http.MethodGet)
@@ -238,8 +244,23 @@ func (f *Frontend) ping(ctx context.Context, wsConn *websocket.Conn, srvConn *io
 	}
 }
 
-func (f *Frontend) handleConnection(wsConn *websocket.Conn, srvConn *io.Connection) error {
+func (f *Frontend) handleConnection(wsConn *websocket.Conn, srvConn *io.Connection) (err error) {
 	ctx := srvConn.Context()
+
+	ctx, span := f.tracer.Start(ctx, "HandleConnection",
+		trace.WithAttributes(
+			attribute.String("gateway_uid", unique.ID(ctx, srvConn.Gateway().Ids)),
+			attribute.String("protocol", f.Protocol()),
+		),
+	)
+	defer func() {
+		if err != nil {
+			span.RecordError(err)
+			span.SetStatus(codes.Error, err.Error())
+		}
+		span.End()
+	}()
+
 	logger := log.FromContext(ctx)
 
 	gtwConfig, err := buildLoRaGatewayConfig(srvConn.PrimaryFrequencyPlan())
@@ -276,10 +297,10 @@ func (f *Frontend) handleConnection(wsConn *websocket.Conn, srvConn *io.Connecti
 		return sendMessages(ctx, wsConn, msgCh)
 	})
 	wg.Go(func() error {
-		return enqueueNetworkServerMessages(ctx, srvConn, gtwConfig, msgCh, dlTokens)
+		return enqueueNetworkServerMessages(ctx, f.tracer, srvConn, gtwConfig, msgCh, dlTokens)
 	})
 	wg.Go(func() error {
-		return readGatewayMessages(ctx, wsConn, srvConn, gtwConfig, dlTokens)
+		return readGatewayMessages(ctx, f.tracer, wsConn, srvConn, gtwConfig, dlTokens)
 	})
 	return wg.Wait()
 }
@@ -308,6 +329,7 @@ func sendMessages(
 
 func enqueueNetworkServerMessages(
 	ctx context.Context,
+	tracer trace.Tracer,
 	srvConn *io.Connection,
 	gtwConfig *lorav1.GatewayConfig,
 	msgCh chan<- *lorav1.NetworkServerMessage,
@@ -320,12 +342,17 @@ func enqueueNetworkServerMessages(
 			return ctx.Err()
 		case down := <-srvConn.Down():
 			logger.Debug("Send downlink message")
+			_, dnSpan := tracer.Start(ctx, "HandleDownMessage")
 			dlToken := dlTokens.Next(down, time.Now())
 			msg, err := fromDownlinkMessage(gtwConfig, down)
 			if err != nil {
+				dnSpan.RecordError(err)
+				dnSpan.SetStatus(codes.Error, err.Error())
+				dnSpan.End()
 				logger.WithError(err).Warn("Failed to convert downlink message")
 				continue
 			}
+			dnSpan.End()
 			select {
 			case <-ctx.Done():
 				return ctx.Err()
@@ -349,6 +376,7 @@ var (
 
 func readGatewayMessages(
 	ctx context.Context,
+	tracer trace.Tracer,
 	wsConn *websocket.Conn,
 	srvConn *io.Connection,
 	gtwConfig *lorav1.GatewayConfig,
@@ -404,9 +432,17 @@ func readGatewayMessages(
 			continue
 		}
 
+		_, msgSpan := tracer.Start(ctx, "HandleUpMessage",
+			trace.WithAttributes(
+				attribute.Int("message_size", len(buf)),
+			),
+		)
 		if err := processGatewayMessage(ctx, srvConn, gtwConfig, dlTokens, &envelope, receivedAt); err != nil {
+			msgSpan.RecordError(err)
+			msgSpan.SetStatus(codes.Error, err.Error())
 			logger.WithError(err).Warn("Failed to handle message")
 		}
+		msgSpan.End()
 	}
 }
 

pkg/telemetry/tracing/tracing.go

@@ -24,6 +24,7 @@ import (
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 	semconv "go.opentelemetry.io/otel/semconv/v1.12.0"
 	otrace "go.opentelemetry.io/otel/trace"
+	"go.opentelemetry.io/otel/trace/noop"
 	"go.thethings.network/lorawan-stack/v3/pkg/version"
 )
 
@@ -49,7 +50,7 @@ func initResource(ctx context.Context) (*resource.Resource, error) {
 // If tracing is not enabled it returns a noop tracer provider instead.
 func Initialize(ctx context.Context, config *Config) (otrace.TracerProvider, func(context.Context) error, error) {
 	if !config.Enable {
-		return otrace.NewNoopTracerProvider(), func(_ context.Context) error { return nil }, nil
+		return noop.NewTracerProvider(), func(_ context.Context) error { return nil }, nil
 	}
 
 	exp, err := exporterFromConfig(ctx, config)