chore: adding da retreiver syncing

Author: chatton

block/internal/syncing/da_retriever_tracing.go

@@ -0,0 +1,57 @@
+package syncing
+
+import (
+	"context"
+
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
+
+	"github.com/evstack/ev-node/block/internal/common"
+)
+
+var _ DARetriever = (*tracedDARetriever)(nil)
+
+// tracedDARetriever wraps a DARetriever with OpenTelemetry tracing.
+type tracedDARetriever struct {
+	inner  DARetriever
+	tracer trace.Tracer
+}
+
+// WithTracingDARetriever wraps a DARetriever with OpenTelemetry tracing.
+func WithTracingDARetriever(inner DARetriever) DARetriever {
+	return &tracedDARetriever{
+		inner:  inner,
+		tracer: otel.Tracer("ev-node/da-retriever"),
+	}
+}
+
+func (t *tracedDARetriever) RetrieveFromDA(ctx context.Context, daHeight uint64) ([]common.DAHeightEvent, error) {
+	ctx, span := t.tracer.Start(ctx, "DARetriever.RetrieveFromDA",
+		trace.WithAttributes(
+			attribute.Int64("da.height", int64(daHeight)),
+		),
+	)
+	defer span.End()
+
+	events, err := t.inner.RetrieveFromDA(ctx, daHeight)
+	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
+		return events, err
+	}
+
+	span.SetAttributes(attribute.Int("event.count", len(events)))
+
+	// add block heights from events
+	if len(events) > 0 {
+		heights := make([]int64, len(events))
+		for i, event := range events {
+			heights[i] = int64(event.Header.Height())
+		}
+		span.SetAttributes(attribute.Int64Slice("block.heights", heights))
+	}
+
+	return events, nil
+}

block/internal/syncing/da_retriever_tracing_test.go

@@ -0,0 +1,123 @@
+package syncing
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/codes"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	"go.opentelemetry.io/otel/sdk/trace/tracetest"
+
+	"github.com/evstack/ev-node/block/internal/common"
+	"github.com/evstack/ev-node/pkg/telemetry/testutil"
+	"github.com/evstack/ev-node/types"
+)
+
+type mockDARetriever struct {
+	retrieveFromDAFn func(ctx context.Context, daHeight uint64) ([]common.DAHeightEvent, error)
+}
+
+func (m *mockDARetriever) RetrieveFromDA(ctx context.Context, daHeight uint64) ([]common.DAHeightEvent, error) {
+	if m.retrieveFromDAFn != nil {
+		return m.retrieveFromDAFn(ctx, daHeight)
+	}
+	return nil, nil
+}
+
+func setupDARetrieverTrace(t *testing.T, inner DARetriever) (DARetriever, *tracetest.SpanRecorder) {
+	t.Helper()
+	sr := tracetest.NewSpanRecorder()
+	tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(sr))
+	t.Cleanup(func() { _ = tp.Shutdown(context.Background()) })
+	otel.SetTracerProvider(tp)
+	return WithTracingDARetriever(inner), sr
+}
+
+func TestTracedDARetriever_RetrieveFromDA_Success(t *testing.T) {
+	mock := &mockDARetriever{
+		retrieveFromDAFn: func(ctx context.Context, daHeight uint64) ([]common.DAHeightEvent, error) {
+			return []common.DAHeightEvent{
+				{
+					Header: &types.SignedHeader{
+						Header: types.Header{
+							BaseHeader: types.BaseHeader{Height: 100},
+						},
+					},
+					DaHeight: daHeight,
+					Source:   common.SourceDA,
+				},
+				{
+					Header: &types.SignedHeader{
+						Header: types.Header{
+							BaseHeader: types.BaseHeader{Height: 101},
+						},
+					},
+					DaHeight: daHeight,
+					Source:   common.SourceDA,
+				},
+			}, nil
+		},
+	}
+	retriever, sr := setupDARetrieverTrace(t, mock)
+	ctx := context.Background()
+
+	events, err := retriever.RetrieveFromDA(ctx, 50)
+	require.NoError(t, err)
+	require.Len(t, events, 2)
+
+	spans := sr.Ended()
+	require.Len(t, spans, 1)
+	span := spans[0]
+	require.Equal(t, "DARetriever.RetrieveFromDA", span.Name())
+	require.Equal(t, codes.Unset, span.Status().Code)
+
+	attrs := span.Attributes()
+	testutil.RequireAttribute(t, attrs, "da.height", int64(50))
+	testutil.RequireAttribute(t, attrs, "event.count", 2)
+}
+
+func TestTracedDARetriever_RetrieveFromDA_NoEvents(t *testing.T) {
+	mock := &mockDARetriever{
+		retrieveFromDAFn: func(ctx context.Context, daHeight uint64) ([]common.DAHeightEvent, error) {
+			return []common.DAHeightEvent{}, nil
+		},
+	}
+	retriever, sr := setupDARetrieverTrace(t, mock)
+	ctx := context.Background()
+
+	events, err := retriever.RetrieveFromDA(ctx, 50)
+	require.NoError(t, err)
+	require.Empty(t, events)
+
+	spans := sr.Ended()
+	require.Len(t, spans, 1)
+	span := spans[0]
+	require.Equal(t, codes.Unset, span.Status().Code)
+
+	attrs := span.Attributes()
+	testutil.RequireAttribute(t, attrs, "event.count", 0)
+}
+
+func TestTracedDARetriever_RetrieveFromDA_Error(t *testing.T) {
+	expectedErr := errors.New("DA retrieval failed")
+	mock := &mockDARetriever{
+		retrieveFromDAFn: func(ctx context.Context, daHeight uint64) ([]common.DAHeightEvent, error) {
+			return nil, expectedErr
+		},
+	}
+	retriever, sr := setupDARetrieverTrace(t, mock)
+	ctx := context.Background()
+
+	_, err := retriever.RetrieveFromDA(ctx, 50)
+	require.Error(t, err)
+	require.Equal(t, expectedErr, err)
+
+	spans := sr.Ended()
+	require.Len(t, spans, 1)
+	span := spans[0]
+	require.Equal(t, codes.Error, span.Status().Code)
+	require.Equal(t, expectedErr.Error(), span.Status().Description)
+}

block/internal/syncing/syncer.go

@@ -201,6 +201,9 @@ func (s *Syncer) Start(ctx context.Context) error {
 
 	// Initialize handlers
 	s.daRetriever = NewDARetriever(s.daClient, s.cache, s.genesis, s.logger)
+	if s.config.Instrumentation.IsTracingEnabled() {
+		s.daRetriever = WithTracingDARetriever(s.daRetriever)
+	}
 	s.fiRetriever = da.NewForcedInclusionRetriever(s.daClient, s.logger, s.config, s.genesis.DAStartHeight, s.genesis.DAEpochForcedInclusion)
 	s.p2pHandler = NewP2PHandler(s.headerStore.Store(), s.dataStore.Store(), s.cache, s.genesis, s.logger)
 	if currentHeight, err := s.store.Height(s.ctx); err != nil {