feat(cannon): Additional tracing (#223)

* feat(cannon): Add network to root tracing span * Add processor tracing
ethpandaops · Oct 3, 2023 · 5b81903 · 5b81903
1 parent 5abfb4f
commit 5b81903
Show file tree

Hide file tree

Showing 12 changed files with 64 additions and 8 deletions.
diff --git a/pkg/cannon/deriver/beacon/eth/v2/attester_slashing.go b/pkg/cannon/deriver/beacon/eth/v2/attester_slashing.go
@@ -96,7 +96,11 @@ func (a *AttesterSlashingDeriver) run(rctx context.Context) {
 			return
 		default:
 			operation := func() error {
-				ctx, span := observability.Tracer().Start(rctx, fmt.Sprintf("Derive %s", a.Name()))
+				ctx, span := observability.Tracer().Start(rctx,
+					fmt.Sprintf("Derive %s", a.Name()),
+					trace.WithAttributes(
+						attribute.String("network", string(a.beacon.Metadata().Network.Name))),
+				)
 				defer span.End()
 
 				time.Sleep(100 * time.Millisecond)

diff --git a/pkg/cannon/deriver/beacon/eth/v2/beacon_block.go b/pkg/cannon/deriver/beacon/eth/v2/beacon_block.go
@@ -103,7 +103,10 @@ func (b *BeaconBlockDeriver) run(rctx context.Context) {
 			return
 		default:
 			operation := func() error {
-				ctx, span := tracer.Start(rctx, fmt.Sprintf("Derive %s", b.Name()))
+				ctx, span := tracer.Start(rctx, fmt.Sprintf("Derive %s", b.Name()),
+					trace.WithAttributes(
+						attribute.String("network", string(b.beacon.Metadata().Network.Name))),
+				)
 				defer span.End()
 
 				time.Sleep(100 * time.Millisecond)

diff --git a/pkg/cannon/deriver/beacon/eth/v2/bls_to_execution_change.go b/pkg/cannon/deriver/beacon/eth/v2/bls_to_execution_change.go
@@ -98,7 +98,10 @@ func (b *BLSToExecutionChangeDeriver) run(rctx context.Context) {
 			return
 		default:
 			operation := func() error {
-				ctx, span := observability.Tracer().Start(rctx, fmt.Sprintf("Derive %s", b.Name()))
+				ctx, span := observability.Tracer().Start(rctx, fmt.Sprintf("Derive %s", b.Name()),
+					trace.WithAttributes(
+						attribute.String("network", string(b.beacon.Metadata().Network.Name))),
+				)
 				defer span.End()
 
 				time.Sleep(100 * time.Millisecond)

diff --git a/pkg/cannon/deriver/beacon/eth/v2/deposit.go b/pkg/cannon/deriver/beacon/eth/v2/deposit.go
@@ -96,7 +96,10 @@ func (b *DepositDeriver) run(rctx context.Context) {
 			return
 		default:
 			operation := func() error {
-				ctx, span := observability.Tracer().Start(rctx, fmt.Sprintf("Derive %s", b.Name()))
+				ctx, span := observability.Tracer().Start(rctx, fmt.Sprintf("Derive %s", b.Name()),
+					trace.WithAttributes(
+						attribute.String("network", string(b.beacon.Metadata().Network.Name))),
+				)
 				defer span.End()
 
 				time.Sleep(100 * time.Millisecond)

diff --git a/pkg/cannon/deriver/beacon/eth/v2/execution_transaction.go b/pkg/cannon/deriver/beacon/eth/v2/execution_transaction.go
@@ -99,7 +99,10 @@ func (b *ExecutionTransactionDeriver) run(rctx context.Context) {
 			return
 		default:
 			operation := func() error {
-				ctx, span := observability.Tracer().Start(rctx, fmt.Sprintf("Derive %s", b.Name()))
+				ctx, span := observability.Tracer().Start(rctx, fmt.Sprintf("Derive %s", b.Name()),
+					trace.WithAttributes(
+						attribute.String("network", string(b.beacon.Metadata().Network.Name))),
+				)
 				defer span.End()
 
 				time.Sleep(100 * time.Millisecond)

diff --git a/pkg/cannon/deriver/beacon/eth/v2/proposer_slashing.go b/pkg/cannon/deriver/beacon/eth/v2/proposer_slashing.go
@@ -96,7 +96,10 @@ func (b *ProposerSlashingDeriver) run(rctx context.Context) {
 			return
 		default:
 			operation := func() error {
-				ctx, span := observability.Tracer().Start(rctx, fmt.Sprintf("Derive %s", b.Name()))
+				ctx, span := observability.Tracer().Start(rctx, fmt.Sprintf("Derive %s", b.Name()),
+					trace.WithAttributes(
+						attribute.String("network", string(b.beacon.Metadata().Network.Name))),
+				)
 				defer span.End()
 
 				time.Sleep(100 * time.Millisecond)

diff --git a/pkg/cannon/deriver/beacon/eth/v2/voluntary_exit.go b/pkg/cannon/deriver/beacon/eth/v2/voluntary_exit.go
@@ -96,7 +96,10 @@ func (b *VoluntaryExitDeriver) run(rctx context.Context) {
 			return
 		default:
 			operation := func() error {
-				ctx, span := observability.Tracer().Start(rctx, fmt.Sprintf("Derive %s", b.Name()))
+				ctx, span := observability.Tracer().Start(rctx, fmt.Sprintf("Derive %s", b.Name()),
+					trace.WithAttributes(
+						attribute.String("network", string(b.beacon.Metadata().Network.Name))),
+				)
 				defer span.End()
 
 				time.Sleep(100 * time.Millisecond)

diff --git a/pkg/cannon/deriver/beacon/eth/v2/withdrawal.go b/pkg/cannon/deriver/beacon/eth/v2/withdrawal.go
@@ -96,7 +96,10 @@ func (b *WithdrawalDeriver) run(rctx context.Context) {
 			return
 		default:
 			operation := func() error {
-				ctx, span := observability.Tracer().Start(rctx, fmt.Sprintf("Derive %s", b.Name()))
+				ctx, span := observability.Tracer().Start(rctx, fmt.Sprintf("Derive %s", b.Name()),
+					trace.WithAttributes(
+						attribute.String("network", string(b.beacon.Metadata().Network.Name))),
+				)
 				defer span.End()
 
 				time.Sleep(100 * time.Millisecond)

diff --git a/pkg/output/http/exporter.go b/pkg/output/http/exporter.go
@@ -8,8 +8,12 @@ import (
 	"io"
 	"net/http"
 
+	"github.com/ethpandaops/xatu/pkg/observability"
 	"github.com/ethpandaops/xatu/pkg/proto/xatu"
 	"github.com/sirupsen/logrus"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/protobuf/encoding/protojson"
 )
 
@@ -38,6 +42,9 @@ func NewItemExporter(name string, config *Config, log logrus.FieldLogger) (ItemE
 }
 
 func (e ItemExporter) ExportItems(ctx context.Context, items []*xatu.DecoratedEvent) error {
+	_, span := observability.Tracer().Start(ctx, "HTTPItemExporter.ExportItems", trace.WithAttributes(attribute.Int64("num_events", int64(len(items)))))
+	defer span.End()
+
 	e.log.WithField("events", len(items)).Debug("Sending batch of events to HTTP sink")
 
 	if err := e.sendUpstream(ctx, items); err != nil {
@@ -46,6 +53,8 @@ func (e ItemExporter) ExportItems(ctx context.Context, items []*xatu.DecoratedEv
 			WithField("num_events", len(items)).
 			Error("Failed to send events upstream")
 
+		span.SetStatus(codes.Error, err.Error())
+
 		return err
 	}
 

diff --git a/pkg/output/stdout/exporter.go b/pkg/output/stdout/exporter.go
@@ -3,8 +3,11 @@ package stdout
 import (
 	"context"
 
+	"github.com/ethpandaops/xatu/pkg/observability"
 	"github.com/ethpandaops/xatu/pkg/proto/xatu"
 	"github.com/sirupsen/logrus"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/protobuf/encoding/protojson"
 )
 
@@ -21,6 +24,9 @@ func NewItemExporter(name string, config *Config, log logrus.FieldLogger) (ItemE
 }
 
 func (e ItemExporter) ExportItems(ctx context.Context, items []*xatu.DecoratedEvent) error {
+	_, span := observability.Tracer().Start(ctx, "StdOutItemExporter.ExportItems", trace.WithAttributes(attribute.Int64("num_events", int64(len(items)))))
+	defer span.End()
+
 	e.log.WithField("events", len(items)).Debug("Sending batch of events to stdout sink")
 
 	if err := e.sendUpstream(ctx, items); err != nil {

diff --git a/pkg/output/xatu/exporter.go b/pkg/output/xatu/exporter.go
@@ -5,8 +5,12 @@ import (
 	"fmt"
 	"net"
 
+	"github.com/ethpandaops/xatu/pkg/observability"
 	pb "github.com/ethpandaops/xatu/pkg/proto/xatu"
 	"github.com/sirupsen/logrus"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
@@ -50,6 +54,9 @@ func NewItemExporter(name string, config *Config, log logrus.FieldLogger) (ItemE
 }
 
 func (e ItemExporter) ExportItems(ctx context.Context, items []*pb.DecoratedEvent) error {
+	_, span := observability.Tracer().Start(ctx, "XatuItemExporter.ExportItems", trace.WithAttributes(attribute.Int64("num_events", int64(len(items)))))
+	defer span.End()
+
 	e.log.WithField("events", len(items)).Debug("Sending batch of events to xatu sink")
 
 	if err := e.sendUpstream(ctx, items); err != nil {
@@ -58,6 +65,8 @@ func (e ItemExporter) ExportItems(ctx context.Context, items []*pb.DecoratedEven
 			WithField("num_events", len(items)).
 			Error("Failed to send events upstream")
 
+		span.SetStatus(codes.Error, err.Error())
+
 		return err
 	}
 

diff --git a/pkg/processor/batch.go b/pkg/processor/batch.go
@@ -14,6 +14,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/ethpandaops/xatu/pkg/observability"
 	"github.com/sirupsen/logrus"
 )
 
@@ -162,6 +163,9 @@ func NewBatchItemProcessor[T any](exporter ItemExporter[T], name string, log log
 
 // OnEnd method enqueues a item for later processing.
 func (bvp *BatchItemProcessor[T]) Write(ctx context.Context, s []*T) error {
+	_, span := observability.Tracer().Start(ctx, "BatchItemProcessor.Write")
+	defer span.End()
+
 	bvp.metrics.SetItemsQueued(bvp.name, float64(len(bvp.queue)))
 
 	if bvp.o.ShippingMethod == ShippingMethodSync {
@@ -185,6 +189,9 @@ func (bvp *BatchItemProcessor[T]) Write(ctx context.Context, s []*T) error {
 // ImmediatelyExportItems immediately exports the items to the exporter.
 // Useful for propogating errors from the exporter.
 func (bvp *BatchItemProcessor[T]) ImmediatelyExportItems(ctx context.Context, items []*T) error {
+	_, span := observability.Tracer().Start(ctx, "BatchItemProcessor.ImmediatelyExportItems")
+	defer span.End()
+
 	bvp.batchMutex.Lock()
 	defer bvp.batchMutex.Unlock()