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, // metrics not available in client context
 	)
 	if err != nil {
 		return err
@@ -108,7 +109,7 @@ func (c *Client) initTxClient(
 	c.State = core
 
 	// setup blob submission service using core
-	blobSvc := blob.NewService(core, nil, nil, nil)
+	blobSvc := blob.NewService(core, nil, nil, nil, nil) // metrics not available in client context
 	err = blobSvc.Start(ctx)
 	if err != nil {
 		return err

blob/metrics.go

@@ -0,0 +1,196 @@
+package blob
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"time"
+
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+	"go.uber.org/fx"
+)
+
+var meter = otel.Meter("blob")
+
+// Metrics tracks blob-related metrics
+type Metrics struct {
+	// Retrieval metrics
+	retrievalCounter  metric.Int64Counter
+	retrievalDuration metric.Float64Histogram
+	retrievalErrors   metric.Int64Counter
+	retrievalNotFound metric.Int64Counter
+
+	// Proof metrics
+	proofCounter  metric.Int64Counter
+	proofDuration metric.Float64Histogram
+	proofErrors   metric.Int64Counter
+
+	// Internal counters (thread-safe)
+	totalRetrievals      atomic.Int64
+	totalRetrievalErrors atomic.Int64
+	totalProofs          atomic.Int64
+	totalProofErrors     atomic.Int64
+}
+
+// WithMetrics registers blob metrics
+func WithMetrics(lc fx.Lifecycle) (*Metrics, error) {
+	// Retrieval metrics
+	retrievalCounter, err := meter.Int64Counter(
+		"blob_retrieval_total",
+		metric.WithDescription("Total number of blob retrieval operations"),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	retrievalDuration, err := meter.Float64Histogram(
+		"blob_retrieval_duration_seconds",
+		metric.WithDescription("Duration of blob retrieval operations"),
+		metric.WithUnit("s"),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	retrievalErrors, err := meter.Int64Counter(
+		"blob_retrieval_errors_total",
+		metric.WithDescription("Total number of blob retrieval errors"),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	retrievalNotFound, err := meter.Int64Counter(
+		"blob_retrieval_not_found_total",
+		metric.WithDescription("Total number of blob not found errors"),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	// Proof metrics
+	proofCounter, err := meter.Int64Counter(
+		"blob_proof_total",
+		metric.WithDescription("Total number of blob proof operations"),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	proofDuration, err := meter.Float64Histogram(
+		"blob_proof_duration_seconds",
+		metric.WithDescription("Duration of blob proof operations"),
+		metric.WithUnit("s"),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	proofErrors, err := meter.Int64Counter(
+		"blob_proof_errors_total",
+		metric.WithDescription("Total number of blob proof errors"),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	metrics := &Metrics{
+		retrievalCounter:  retrievalCounter,
+		retrievalDuration: retrievalDuration,
+		retrievalErrors:   retrievalErrors,
+		retrievalNotFound: retrievalNotFound,
+		proofCounter:      proofCounter,
+		proofDuration:     proofDuration,
+		proofErrors:       proofErrors,
+	}
+
+	// Register observable metrics
+	retrievalTotal, err := meter.Int64ObservableCounter(
+		"blob_retrieval_total_observable",
+		metric.WithDescription("Observable total number of blob retrievals"),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	proofTotal, err := meter.Int64ObservableCounter(
+		"blob_proof_total_observable",
+		metric.WithDescription("Observable total number of blob proofs"),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	callback := func(_ context.Context, observer metric.Observer) error {
+		observer.ObserveInt64(retrievalTotal, metrics.totalRetrievals.Load())
+		observer.ObserveInt64(proofTotal, metrics.totalProofs.Load())
+		return nil
+	}
+
+	clientReg, err := meter.RegisterCallback(callback, retrievalTotal, proofTotal)
+	if err != nil {
+		return nil, err
+	}
+
+	lc.Append(fx.Hook{
+		OnStop: func(context.Context) error {
+			return clientReg.Unregister()
+		},
+	})
+
+	return metrics, nil
+}
+
+// ObserveRetrieval records blob retrieval metrics
+func (m *Metrics) ObserveRetrieval(ctx context.Context, duration time.Duration, err error) {
+	if m == nil {
+		return
+	}
+
+	// Update counters
+	m.totalRetrievals.Add(1)
+	if err != nil {
+		m.totalRetrievalErrors.Add(1)
+	}
+
+	// Record metrics
+	attrs := []attribute.KeyValue{}
+	if err != nil {
+		attrs = append(attrs, attribute.String("error", err.Error()))
+		if errors.Is(err, ErrBlobNotFound) {
+			m.retrievalNotFound.Add(ctx, 1, metric.WithAttributes(attrs...))
+		} else {
+			m.retrievalErrors.Add(ctx, 1, metric.WithAttributes(attrs...))
+		}
+	} else {
+		m.retrievalCounter.Add(ctx, 1, metric.WithAttributes(attrs...))
+	}
+
+	m.retrievalDuration.Record(ctx, duration.Seconds(), metric.WithAttributes(attrs...))
+}
+
+// ObserveProof records blob proof metrics
+func (m *Metrics) ObserveProof(ctx context.Context, duration time.Duration, err error) {
+	if m == nil {
+		return
+	}
+
+	// Update counters
+	m.totalProofs.Add(1)
+	if err != nil {
+		m.totalProofErrors.Add(1)
+	}
+
+	// Record metrics
+	attrs := []attribute.KeyValue{}
+	if err != nil {
+		attrs = append(attrs, attribute.String("error", err.Error()))
+		m.proofErrors.Add(ctx, 1, metric.WithAttributes(attrs...))
+	} else {
+		m.proofCounter.Add(ctx, 1, metric.WithAttributes(attrs...))
+	}
+
+	m.proofDuration.Record(ctx, duration.Seconds(), 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,19 +62,23 @@ 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(
 	submitter Submitter,
 	getter shwap.Getter,
 	headerGetter func(context.Context, uint64) (*header.ExtendedHeader, error),
 	headerSub func(ctx context.Context) (<-chan *header.ExtendedHeader, error),
+	metrics *Metrics,
 ) *Service {
 	return &Service{
 		blobSubmitter: submitter,
 		shareGetter:   getter,
 		headerGetter:  headerGetter,
 		headerSub:     headerSub,
+		metrics:       metrics,
 	}
 }
 
@@ -196,9 +201,12 @@ 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)
+		// Record metrics
+		s.metrics.ObserveRetrieval(ctx, time.Since(start), err)
 	}()
 	span.SetAttributes(
 		attribute.Int64("height", int64(height)),
@@ -210,7 +218,7 @@ func (s *Service) Get(
 	}}
 
 	blob, _, err = s.retrieve(ctx, height, namespace, sharesParser)
-	return
+	return blob, err
 }
 
 // GetProof returns an NMT inclusion proof for a specified namespace to the respective row roots
@@ -222,9 +230,12 @@ 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)
+		// Record metrics
+		s.metrics.ObserveProof(ctx, time.Since(start), err)
 	}()
 	span.SetAttributes(
 		attribute.Int64("height", int64(height)),

blob/service_test.go

@@ -901,7 +901,7 @@ func createServiceWithSub(ctx context.Context, t testing.TB, blobs []*Blob) *Ser
 			nd, err := eds.NamespaceData(ctx, accessor, ns)
 			return nd, err
 		})
-	return NewService(nil, shareGetter, fn, fn2)
+	return NewService(nil, shareGetter, fn, fn2, nil)
 }
 
 func createService(ctx context.Context, t testing.TB, shares []libshare.Share) *Service {
@@ -945,7 +945,7 @@ func createService(ctx context.Context, t testing.TB, shares []libshare.Share) *
 	fn2 := func(ctx context.Context) (<-chan *header.ExtendedHeader, error) {
 		return nil, fmt.Errorf("not implemented")
 	}
-	return NewService(nil, shareGetter, fn, fn2)
+	return NewService(nil, shareGetter, fn, fn2, nil)
 }
 
 // TestProveCommitmentAllCombinations tests proving all the commitments in a block.

nodebuilder/blob/module.go

@@ -31,7 +31,7 @@ func ConstructModule() fx.Option {
 				getByHeightFn func(context.Context, uint64) (*header.ExtendedHeader, error),
 				subscribeFn func(context.Context) (<-chan *header.ExtendedHeader, error),
 			) *blob.Service {
-				return blob.NewService(state, sGetter, getByHeightFn, subscribeFn)
+				return blob.NewService(state, sGetter, getByHeightFn, subscribeFn, nil)
 			},
 			fx.OnStart(func(ctx context.Context, serv *blob.Service) error {
 				return serv.Start(ctx)

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"
@@ -90,8 +91,14 @@ func WithMetrics(metricOpts []otlpmetrichttp.Option, nodeType node.Type) fx.Opti
 			if ca == nil {
 				return
 			}
-			state.WithMetrics(lc, ca)
+			metrics, err := state.WithMetrics(lc, ca)
+			if err != nil {
+				log.Warnf("failed to initialize state metrics: %v", err)
+			} else {
+				ca.SetMetrics(metrics)
+			}
 		}),
+		fx.Provide(blob.WithMetrics),
 		fx.Invoke(fraud.WithMetrics[*header.ExtendedHeader]),
 		fx.Invoke(node.WithMetrics),
 		fx.Invoke(share.WithDiscoveryMetrics),

nodebuilder/state/core.go

@@ -26,6 +26,7 @@ func coreAccessor(
 	network p2p.Network,
 	client *grpc.ClientConn,
 	additionalConns core.AdditionalCoreConns,
+	metrics *state.Metrics,
 ) (
 	*state.CoreAccessor,
 	Module,
@@ -44,7 +45,10 @@ func coreAccessor(
 		}
 	}
 
-	ca, err := state.NewCoreAccessor(keyring, string(keyname), sync, client, network.String(), opts...)
+	ca, err := state.NewCoreAccessor(keyring, string(keyname), sync, client, network.String(), metrics, opts...)
+	if err != nil {
+		return nil, nil, nil, err
+	}
 
 	sBreaker := &modfraud.ServiceBreaker[*state.CoreAccessor, *header.ExtendedHeader]{
 		Service:   ca,

nodebuilder/state/module.go

@@ -29,6 +29,11 @@ func ConstructModule(tp node.Type, cfg *Config, coreCfg *core.Config) fx.Option
 		fx.Provide(func(ks keystore.Keystore) (keyring.Keyring, AccountName, error) {
 			return Keyring(*cfg, ks)
 		}),
+		// Provide nil metrics as fallback when WithMetrics() is not called
+		fx.Provide(func() *state.Metrics {
+			return nil
+		}),
+		// Metrics will be provided by settings.go when enabled
 		fxutil.ProvideIf(coreCfg.IsEndpointConfigured(), fx.Annotate(
 			coreAccessor,
 			fx.OnStart(func(ctx context.Context,

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))