feat(blob|state): Blob submission metrics

api/client/client.go

@@ -97,6 +97,7 @@ func (c *Client) initTxClient(
 		trustedHeadGetter{remote: c.Header},
 		conn,
 		submitCfg.Network.String(),
+		nil,
 	)
 	if err != nil {
 		return err

blob/metrics.go

@@ -0,0 +1,128 @@
+package blob
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+)
+
+var meter = otel.Meter("blob")
+
+// Error type constants for metrics labels
+const (
+	errorTypeNotFound = "not_found"
+	errorTypeTimeout  = "timeout"
+	errorTypeCanceled = "canceled"
+	errorTypeUnknown  = "unknown"
+)
+
+// metrics tracks blob-related metrics
+type metrics struct {
+	// Retrieval metrics
+	retrievalDuration metric.Float64Histogram
+	retrievalTotal    metric.Int64Counter
+
+	// Proof metrics
+	proofDuration metric.Float64Histogram
+	proofTotal    metric.Int64Counter
+}
+
+// WithMetrics initializes metrics for the Service
+func (s *Service) WithMetrics() error {
+	// Retrieval metrics
+	retrievalDuration, err := meter.Float64Histogram(
+		"blob_retrieval_duration_seconds",
+		metric.WithDescription("Duration of blob retrieval operations"),
+		metric.WithUnit("s"),
+	)
+	if err != nil {
+		return err
+	}
+
+	// Proof metrics
+	proofDuration, err := meter.Float64Histogram(
+		"blob_proof_duration_seconds",
+		metric.WithDescription("Duration of blob proof operations"),
+		metric.WithUnit("s"),
+	)
+	if err != nil {
+		return err
+	}
+
+	retrievalTotal, err := meter.Int64Counter(
+		"blob_retrieval_total",
+		metric.WithDescription("Total number of blob retrievals"),
+	)
+	if err != nil {
+		return err
+	}
+
+	proofTotal, err := meter.Int64Counter(
+		"blob_proof_total",
+		metric.WithDescription("Total number of blob proofs"),
+	)
+	if err != nil {
+		return err
+	}
+
+	m := &metrics{
+		retrievalDuration: retrievalDuration,
+		retrievalTotal:    retrievalTotal,
+		proofDuration:     proofDuration,
+		proofTotal:        proofTotal,
+	}
+
+	s.metrics = m
+	return nil
+}
+
+// observeRetrieval records blob retrieval metrics
+func (m *metrics) observeRetrieval(ctx context.Context, duration time.Duration, err error) {
+	if m == nil {
+		return
+	}
+
+	// Record metrics with error type enum to avoid cardinality explosion
+	attrs := []attribute.KeyValue{}
+	if err != nil {
+		errorType := errorTypeUnknown
+		switch {
+		case errors.Is(err, ErrBlobNotFound):
+			errorType = errorTypeNotFound
+		case errors.Is(err, context.DeadlineExceeded):
+			errorType = errorTypeTimeout
+		case errors.Is(err, context.Canceled):
+			errorType = errorTypeCanceled
+		}
+		attrs = append(attrs, attribute.String("error_type", errorType))
+	}
+
+	m.retrievalDuration.Record(ctx, duration.Seconds(), metric.WithAttributes(attrs...))
+	m.retrievalTotal.Add(ctx, 1, metric.WithAttributes(attrs...))
+}
+
+// observeProof records blob proof metrics
+func (m *metrics) observeProof(ctx context.Context, duration time.Duration, err error) {
+	if m == nil {
+		return
+	}
+
+	attrs := []attribute.KeyValue{}
+	if err != nil {
+		errorType := errorTypeUnknown
+		switch {
+		case errors.Is(err, context.DeadlineExceeded):
+			errorType = errorTypeTimeout
+		case errors.Is(err, context.Canceled):
+			errorType = errorTypeCanceled
+		}
+		attrs = append(attrs, attribute.String("error_type", errorType))
+	}
+
+	m.proofDuration.Record(ctx, duration.Seconds(), metric.WithAttributes(attrs...))
+	m.proofTotal.Add(ctx, 1, metric.WithAttributes(attrs...))
+}

blob/service.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"slices"
 	"sync"
+	"time"
 
 	"github.com/cosmos/cosmos-sdk/types"
 	logging "github.com/ipfs/go-log/v2"
@@ -61,6 +62,8 @@ type Service struct {
 	headerGetter func(context.Context, uint64) (*header.ExtendedHeader, error)
 	// headerSub subscribes to new headers to supply to blob subscriptions.
 	headerSub func(ctx context.Context) (<-chan *header.ExtendedHeader, error)
+	// metrics tracks blob-related metrics
+	metrics *metrics
 }
 
 func NewService(
@@ -74,6 +77,7 @@ func NewService(
 		shareGetter:   getter,
 		headerGetter:  headerGetter,
 		headerSub:     headerSub,
+		metrics:       nil, // Will be initialized via WithMetrics() if needed
 	}
 }
 
@@ -84,6 +88,7 @@ func (s *Service) Start(context.Context) error {
 
 func (s *Service) Stop(context.Context) error {
 	s.cancel()
+	// No cleanup needed for metrics
 	return nil
 }
 
@@ -196,9 +201,11 @@ func (s *Service) Get(
 	namespace libshare.Namespace,
 	commitment Commitment,
 ) (blob *Blob, err error) {
+	start := time.Now()
 	ctx, span := tracer.Start(ctx, "get")
 	defer func() {
 		utils.SetStatusAndEnd(span, err)
+		s.metrics.observeRetrieval(ctx, time.Since(start), err)
 	}()
 	span.SetAttributes(
 		attribute.Int64("height", int64(height)),
@@ -222,9 +229,11 @@ func (s *Service) GetProof(
 	namespace libshare.Namespace,
 	commitment Commitment,
 ) (proof *Proof, err error) {
+	start := time.Now()
 	ctx, span := tracer.Start(ctx, "get-proof")
 	defer func() {
 		utils.SetStatusAndEnd(span, err)
+		s.metrics.observeProof(ctx, time.Since(start), err)
 	}()
 	span.SetAttributes(
 		attribute.Int64("height", int64(height)),

core/listener_test.go

@@ -81,7 +81,10 @@ func TestListenerWithWrongChainRPC(t *testing.T) {
 	// create one block to store as Head in local store and then unsubscribe from block events
 	cfg := DefaultTestConfig()
 	cfg.Genesis.ChainID = testChainID
-	fetcher, _ := createCoreFetcher(t, cfg)
+	fetcher, network := createCoreFetcher(t, cfg)
+	t.Cleanup(func() {
+		require.NoError(t, network.Stop())
+	})
 	eds := createEdsPubSub(ctx, t)
 
 	store, err := store.NewStore(store.DefaultParameters(), t.TempDir())

nodebuilder/da/da.go

@@ -59,6 +59,7 @@ type Module interface {
 
 	// Validate validates Commitments against the corresponding Proofs. This should be possible without retrieving the
 	// Blobs.
+	//
 	// Deprecated: This method is deprecated and will be removed in the future.
 	Validate(ctx context.Context, ids []da.ID, proofs []da.Proof, namespace da.Namespace) ([]bool, error)
 }

nodebuilder/header/config.go

@@ -79,7 +79,7 @@ func (cfg *Config) trustedPeers(bpeers p2p.Bootstrappers) (infos []peer.AddrInfo
 		}
 		infos[i] = *p
 	}
-	return infos, err
+	return infos, nil
 }
 
 // Validate performs basic validation of the config.

nodebuilder/settings.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/celestiaorg/go-fraud"
 
+	"github.com/celestiaorg/celestia-node/blob"
 	"github.com/celestiaorg/celestia-node/header"
 	modcore "github.com/celestiaorg/celestia-node/nodebuilder/core"
 	"github.com/celestiaorg/celestia-node/nodebuilder/das"
@@ -86,11 +87,22 @@ func WithMetrics(metricOpts []otlpmetrichttp.Option, nodeType node.Type) fx.Opti
 	baseComponents := fx.Options(
 		fx.Supply(metricOpts),
 		fx.Invoke(initializeMetrics),
-		fx.Invoke(func(lc fx.Lifecycle, ca *state.CoreAccessor) {
+		fx.Invoke(func(ca *state.CoreAccessor) error {
 			if ca == nil {
-				return
+				return nil
 			}
-			state.WithMetrics(lc, ca)
+			err := ca.WithMetrics()
+			if err != nil {
+				return fmt.Errorf("failed to initialize state metrics: %w", err)
+			}
+			return nil
+		}),
+		fx.Invoke(func(serv *blob.Service) error {
+			err := serv.WithMetrics()
+			if err != nil {
+				return fmt.Errorf("failed to initialize blob metrics: %w", err)
+			}
+			return nil
 		}),
 		fx.Invoke(fraud.WithMetrics[*header.ExtendedHeader]),
 		fx.Invoke(node.WithMetrics),

nodebuilder/state/core.go

@@ -49,7 +49,7 @@ func coreAccessor(
 		opts = append(opts, state.WithTxWorkerAccounts(cfg.TxWorkerAccounts))
 	}
 
-	ca, err := state.NewCoreAccessor(keyring, string(keyname), sync, client, network.String(), opts...)
+	ca, err := state.NewCoreAccessor(keyring, string(keyname), sync, client, network.String(), nil, opts...)
 
 	sBreaker := &modfraud.ServiceBreaker[*state.CoreAccessor, *header.ExtendedHeader]{
 		Service:   ca,

nodebuilder/testing.go

@@ -19,7 +19,7 @@ import (
 	"github.com/celestiaorg/celestia-node/libs/fxutil"
 	"github.com/celestiaorg/celestia-node/nodebuilder/node"
 	"github.com/celestiaorg/celestia-node/nodebuilder/p2p"
-	"github.com/celestiaorg/celestia-node/nodebuilder/state"
+	stateModule "github.com/celestiaorg/celestia-node/nodebuilder/state"
 )
 
 const (
@@ -61,7 +61,7 @@ func TestNodeWithConfig(t *testing.T, tp node.Type, cfg *Config, opts ...fx.Opti
 	_, _, err = kr.NewMnemonic(TestKeyringName, keyring.English, "", "", hd.Secp256k1)
 	require.NoError(t, err)
 	cfg.State.DefaultKeyName = TestKeyringName
-	_, accName, err := state.Keyring(cfg.State, ks)
+	_, accName, err := stateModule.Keyring(cfg.State, ks)
 	require.NoError(t, err)
 	require.Equal(t, TestKeyringName, string(accName))
 

pruner/metrics.go

@@ -39,8 +39,12 @@ func (s *Service) WithMetrics() error {
 	}
 
 	callback := func(_ context.Context, observer metric.Observer) error {
-		observer.ObserveInt64(lastPruned, int64(s.checkpoint.LastPrunedHeight))
-		observer.ObserveInt64(failedPrunes, int64(len(s.checkpoint.FailedHeaders)))
+		s.checkpointMu.Lock()
+		if s.checkpoint != nil {
+			observer.ObserveInt64(lastPruned, int64(s.checkpoint.LastPrunedHeight))
+			observer.ObserveInt64(failedPrunes, int64(len(s.checkpoint.FailedHeaders)))
+		}
+		s.checkpointMu.Unlock()
 		return nil
 	}
 

share/eds/retriever_no_race_test.go

@@ -22,7 +22,7 @@ import (
 
 func TestRetriever_ByzantineError(t *testing.T) {
 	const width = 8
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
 	defer cancel()
 
 	bserv := ipld.NewMemBlockservice()

share/root.go

@@ -68,7 +68,7 @@ func RowsWithNamespace(root *AxisRoots, namespace libshare.Namespace) (idxs []in
 			idxs = append(idxs, i)
 		}
 	}
-	return idxs, err
+	return idxs, nil
 }
 
 // RootHashForCoordinates returns the root hash for the given coordinates.