feat: add metrics to payload disperser (#1441)

Signed-off-by: Cody Littley <[email protected]>

Author: cody-littley

api/clients/v2/disperser_client.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/Layr-Labs/eigenda/api"
 	disperser_rpc "github.com/Layr-Labs/eigenda/api/grpc/disperser/v2"
+	"github.com/Layr-Labs/eigenda/common"
 	"github.com/Layr-Labs/eigenda/core"
 	corev2 "github.com/Layr-Labs/eigenda/core/v2"
 	dispv2 "github.com/Layr-Labs/eigenda/disperser/common/v2"
@@ -23,13 +24,29 @@ type DisperserClientConfig struct {
 	UseSecureGrpcFlag bool
 }
 
+// DisperserClient manages communication with the disperser server.
 type DisperserClient interface {
+	// Close closes the grpc connection to the disperser server.
 	Close() error
-	DisperseBlob(ctx context.Context, data []byte, blobVersion corev2.BlobVersion, quorums []core.QuorumID) (*dispv2.BlobStatus, corev2.BlobKey, error)
+	// DisperseBlob disperses a blob with the given data, blob version, and quorums.
+	DisperseBlob(
+		ctx context.Context,
+		data []byte,
+		blobVersion corev2.BlobVersion,
+		quorums []core.QuorumID) (*dispv2.BlobStatus, corev2.BlobKey, error)
+	// DisperseBlobWithProbe is similar to DisperseBlob, but also takes a SequenceProbe to capture metrics.
+	// If the probe is nil, no metrics are captured.
+	DisperseBlobWithProbe(
+		ctx context.Context,
+		data []byte,
+		blobVersion corev2.BlobVersion,
+		quorums []core.QuorumID,
+		probe *common.SequenceProbe) (*dispv2.BlobStatus, corev2.BlobKey, error)
+	// GetBlobStatus returns the status of a blob with the given blob key.
 	GetBlobStatus(ctx context.Context, blobKey corev2.BlobKey) (*disperser_rpc.BlobStatusReply, error)
+	// GetBlobCommitment returns the blob commitment for a given blob payload.
 	GetBlobCommitment(ctx context.Context, data []byte) (*disperser_rpc.BlobCommitmentReply, error)
 }
-
 type disperserClient struct {
 	config             *DisperserClientConfig
 	signer             corev2.BlobRequestSigner
@@ -39,7 +56,7 @@ type disperserClient struct {
 	client             disperser_rpc.DisperserClient
 	prover             encoding.Prover
 	accountant         *Accountant
-	requestMutex       sync.Mutex // Mutex to ensure only one dispersal request is sent at a time
+	accountantLock     sync.Mutex
 }
 
 var _ DisperserClient = &disperserClient{}
@@ -128,42 +145,68 @@ func (c *disperserClient) DisperseBlob(
 	blobVersion corev2.BlobVersion,
 	quorums []core.QuorumID,
 ) (*dispv2.BlobStatus, corev2.BlobKey, error) {
-	c.requestMutex.Lock()
-	defer c.requestMutex.Unlock()
+	return c.DisperseBlobWithProbe(ctx, data, blobVersion, quorums, nil)
+}
+
+// DisperseBlobWithProbe disperses a blob with the given data, blob version, and quorums. If sequenceProbe is not nil,
+// the probe is used to capture metrics during the dispersal process.
+func (c *disperserClient) DisperseBlobWithProbe(
+	ctx context.Context,
+	data []byte,
+	blobVersion corev2.BlobVersion,
+	quorums []core.QuorumID,
+	probe *common.SequenceProbe,
+) (*dispv2.BlobStatus, corev2.BlobKey, error) {
+
+	if len(quorums) == 0 {
+		return nil, [32]byte{}, api.NewErrorInvalidArg("quorum numbers must be provided")
+	}
+	if c.signer == nil {
+		return nil, [32]byte{}, api.NewErrorInternal("uninitialized signer for authenticated dispersal")
+	}
+	for _, q := range quorums {
+		if q > corev2.MaxQuorumID {
+			return nil, [32]byte{}, api.NewErrorInvalidArg("quorum number must be less than 256")
+		}
+	}
 
 	err := c.initOnceGrpcConnection()
 	if err != nil {
 		return nil, [32]byte{}, api.NewErrorFailover(err)
 	}
+
+	probe.SetStage("acquire_accountant_lock")
+	c.accountantLock.Lock()
+
+	probe.SetStage("prepare_for_dispersal")
+
 	err = c.initOncePopulateAccountant(ctx)
 	if err != nil {
 		return nil, [32]byte{}, api.NewErrorFailover(err)
 	}
 
-	if c.signer == nil {
-		return nil, [32]byte{}, api.NewErrorInternal("uninitialized signer for authenticated dispersal")
-	}
-
 	symbolLength := encoding.GetBlobLengthPowerOf2(uint(len(data)))
 	payment, err := c.accountant.AccountBlob(ctx, time.Now().UnixNano(), uint64(symbolLength), quorums)
 	if err != nil {
+		c.accountantLock.Unlock()
 		return nil, [32]byte{}, fmt.Errorf("error accounting blob: %w", err)
 	}
 
-	if len(quorums) == 0 {
-		return nil, [32]byte{}, api.NewErrorInvalidArg("quorum numbers must be provided")
-	}
-
-	for _, q := range quorums {
-		if q > corev2.MaxQuorumID {
-			return nil, [32]byte{}, api.NewErrorInvalidArg("quorum number must be less than 256")
-		}
+	if payment.CumulativePayment == nil || payment.CumulativePayment.Sign() == 0 {
+		// This request is using reserved bandwidth, no need to prevent parallel dispersal.
+		c.accountantLock.Unlock()
+	} else {
+		// This request is using on-demand bandwidth, current implementation requires sequential dispersal.
+		defer c.accountantLock.Unlock()
 	}
 
 	// check every 32 bytes of data are within the valid range for a bn254 field element
 	_, err = rs.ToFrArray(data)
 	if err != nil {
-		return nil, [32]byte{}, fmt.Errorf("encountered an error to convert a 32-bytes into a valid field element, please use the correct format where every 32bytes(big-endian) is less than 21888242871839275222246405745257275088548364400416034343698204186575808495617 %w", err)
+		return nil, [32]byte{}, fmt.Errorf(
+			"encountered an error to convert a 32-bytes into a valid field element, "+
+				"please use the correct format where every 32bytes(big-endian) is less than "+
+				"21888242871839275222246405745257275088548364400416034343698204186575808495617 %w", err)
 	}
 
 	var blobCommitments encoding.BlobCommitments
@@ -220,6 +263,8 @@ func (c *disperserClient) DisperseBlob(
 		BlobHeader: blobHeaderProto,
 	}
 
+	probe.SetStage("send_to_disperser")
+
 	reply, err := c.client.DisperseBlob(ctx, request)
 	if err != nil {
 		return nil, [32]byte{}, fmt.Errorf("error while calling DisperseBlob: %w", err)
@@ -230,6 +275,8 @@ func (c *disperserClient) DisperseBlob(
 		return nil, [32]byte{}, err
 	}
 
+	probe.SetStage("verify_blob_key")
+
 	if verifyReceivedBlobKey(blobHeader, reply) != nil {
 		return nil, [32]byte{}, fmt.Errorf("verify received blob key: %w", err)
 	}
@@ -245,9 +292,9 @@ func (c *disperserClient) DisperseBlob(
 //
 // This function returns nil if the verification succeeds, and otherwise returns an error describing the failure
 func verifyReceivedBlobKey(
-	// the blob header which was constructed locally and sent to the disperser
+// the blob header which was constructed locally and sent to the disperser
 	blobHeader *corev2.BlobHeader,
-	// the reply received back from the disperser
+// the reply received back from the disperser
 	disperserReply *disperser_rpc.DisperseBlobReply,
 ) error {
 

api/clients/v2/payloaddispersal/payload_disperser.go

@@ -8,8 +8,10 @@ import (
 	"github.com/Layr-Labs/eigenda/api/clients/v2"
 	"github.com/Layr-Labs/eigenda/api/clients/v2/coretypes"
 	dispgrpc "github.com/Layr-Labs/eigenda/api/grpc/disperser/v2"
+	"github.com/Layr-Labs/eigenda/common"
 	core "github.com/Layr-Labs/eigenda/core/v2"
 	"github.com/Layr-Labs/eigensdk-go/logging"
+	"github.com/prometheus/client_golang/prometheus"
 )
 
 // PayloadDisperser provides the ability to disperse payloads to EigenDA via a Disperser grpc service.
@@ -20,9 +22,11 @@ type PayloadDisperser struct {
 	config          PayloadDisperserConfig
 	disperserClient clients.DisperserClient
 	certVerifier    clients.ICertVerifier
+	stageTimer      *common.StageTimer
 }
 
 // NewPayloadDisperser creates a PayloadDisperser from subcomponents that have already been constructed and initialized.
+// If the registry is nil then no metrics will be collected.
 func NewPayloadDisperser(
 	logger logging.Logger,
 	payloadDisperserConfig PayloadDisperserConfig,
@@ -35,6 +39,8 @@ func NewPayloadDisperser(
 	//  disperser, but the disperser's commitments will be verifiable without needing a full-fledged prover
 	disperserClient clients.DisperserClient,
 	certVerifier clients.ICertVerifier,
+	// if nil, then no metrics will be collected
+	registry *prometheus.Registry,
 ) (*PayloadDisperser, error) {
 
 	err := payloadDisperserConfig.checkAndSetDefaults()
@@ -47,6 +53,7 @@ func NewPayloadDisperser(
 		config:          payloadDisperserConfig,
 		disperserClient: disperserClient,
 		certVerifier:    certVerifier,
+		stageTimer:      common.NewStageTimer(registry, "PayloadDisperser", "SendPayload"),
 	}, nil
 }
 
@@ -63,11 +70,17 @@ func (pd *PayloadDisperser) SendPayload(
 	// payload is the raw data to be stored on eigenDA
 	payload *coretypes.Payload,
 ) (*coretypes.EigenDACert, error) {
+
+	probe := pd.stageTimer.NewSequence("convert_to_blob")
+	defer probe.End()
+
 	blob, err := payload.ToBlob(pd.config.PayloadPolynomialForm)
 	if err != nil {
 		return nil, fmt.Errorf("convert payload to blob: %w", err)
 	}
 
+	probe.SetStage("get_quorums")
+
 	timeoutCtx, cancel := context.WithTimeout(ctx, pd.config.ContractCallTimeout)
 	defer cancel()
 	requiredQuorums, err := pd.certVerifier.GetQuorumNumbersRequired(timeoutCtx)
@@ -82,31 +95,37 @@ func (pd *PayloadDisperser) SendPayload(
 	//  serialized bytes. The operations taking place in DisperseBlob require the bytes to be converted into field
 	//  elements anyway, so serializing the blob here is unnecessary work. This will be a larger change that affects
 	//  many areas of code, though.
-	blobStatus, blobKey, err := pd.disperserClient.DisperseBlob(
+	blobStatus, blobKey, err := pd.disperserClient.DisperseBlobWithProbe(
 		timeoutCtx,
 		blob.Serialize(),
 		pd.config.BlobVersion,
 		requiredQuorums,
-	)
+		probe)
 	if err != nil {
 		return nil, fmt.Errorf("disperse blob: %w", err)
 	}
 	pd.logger.Debug("Successful DisperseBlob", "blobStatus", blobStatus.String(), "blobKey", blobKey.Hex())
 
+	probe.SetStage("QUEUED")
+
 	timeoutCtx, cancel = context.WithTimeout(ctx, pd.config.BlobCertifiedTimeout)
 	defer cancel()
-	blobStatusReply, err := pd.pollBlobStatusUntilCertified(timeoutCtx, blobKey, blobStatus.ToProfobuf())
+	blobStatusReply, err := pd.pollBlobStatusUntilCertified(timeoutCtx, blobKey, blobStatus.ToProfobuf(), probe)
 	if err != nil {
 		return nil, fmt.Errorf("poll blob status until certified: %w", err)
 	}
 	pd.logger.Debug("Blob status CERTIFIED", "blobKey", blobKey.Hex())
 
+	probe.SetStage("build_cert")
+
 	eigenDACert, err := pd.buildEigenDACert(ctx, blobKey, blobStatusReply)
 	if err != nil {
 		// error returned from method is sufficiently descriptive
 		return nil, err
 	}
 
+	probe.SetStage("verify_cert")
+
 	timeoutCtx, cancel = context.WithTimeout(ctx, pd.config.ContractCallTimeout)
 	defer cancel()
 	err = pd.certVerifier.VerifyCertV2(timeoutCtx, eigenDACert)
@@ -141,6 +160,7 @@ func (pd *PayloadDisperser) pollBlobStatusUntilCertified(
 	ctx context.Context,
 	blobKey core.BlobKey,
 	initialStatus dispgrpc.BlobStatus,
+	probe *common.SequenceProbe,
 ) (*dispgrpc.BlobStatusReply, error) {
 
 	previousStatus := initialStatus
@@ -182,6 +202,9 @@ func (pd *PayloadDisperser) pollBlobStatusUntilCertified(
 			case dispgrpc.BlobStatus_QUEUED, dispgrpc.BlobStatus_ENCODED, dispgrpc.BlobStatus_GATHERING_SIGNATURES:
 				// TODO (litt): check signing percentage when we are gathering signatures, potentially break
 				//  out of this loop early if we have enough signatures
+
+				// Report all non-terminal statuses to the probe. Repeat reports are no-ops.
+				probe.SetStage(newStatus.String())
 				continue
 			default:
 				return nil, fmt.Errorf(

common/stage_timer.go

@@ -0,0 +1,101 @@
+package common
+
+import (
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+// StageTimer encapsulates metrics to help track the time spent in each stage of the payload dispersal process.
+type StageTimer struct {
+	stageCount   *prometheus.GaugeVec
+	stageLatency *prometheus.SummaryVec
+}
+
+// SequenceProbe can be used to track the amount of time that a particular operation spends doing particular
+// sub-operations (i.e. stages). Multiple instances of a particular operation can be tracked concurrently by the same
+// StageTimer. For each operation, the StageTimer builds a SequenceProbe. Each SequenceProbe is responsible for
+// tracking the lifecycle of a single  iteration of an operation.
+type SequenceProbe struct {
+	stageTimer        *StageTimer
+	currentStage      string
+	currentStageStart time.Time
+}
+
+// NewStageTimer creates a new stageTimer with the given prefix and name.
+func NewStageTimer(registry *prometheus.Registry, prefix, name string) *StageTimer {
+	if registry == nil {
+		return nil
+	}
+
+	stageLatency := promauto.With(registry).NewSummaryVec(
+		prometheus.SummaryOpts{
+			Namespace:  prefix,
+			Name:       name + "_stage_latency_ms",
+			Help:       "the latency of each type of operation",
+			Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+		},
+		[]string{"stage"},
+	)
+
+	stageCount := promauto.With(registry).NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: prefix,
+			Name:      name + "_stage_count",
+			Help:      "the number of operations with a specific stage",
+		},
+		[]string{"stage"},
+	)
+
+	return &StageTimer{
+		stageLatency: stageLatency,
+		stageCount:   stageCount,
+	}
+}
+
+// NewSequence creates a new sequenceProbe with the given initial stage.
+func (s *StageTimer) NewSequence(initialStage string) *SequenceProbe {
+	if s == nil {
+		return nil
+	}
+
+	s.stageCount.WithLabelValues(initialStage).Inc()
+	return &SequenceProbe{
+		stageTimer:        s,
+		currentStage:      initialStage,
+		currentStageStart: time.Now(),
+	}
+}
+
+// SetStage updates the stage of the current sequence.
+func (p *SequenceProbe) SetStage(stage string) {
+	if p == nil {
+		return
+	}
+	if p.currentStage == stage {
+		return
+	}
+
+	now := time.Now()
+	elapsed := now.Sub(p.currentStageStart)
+	p.stageTimer.stageLatency.WithLabelValues(p.currentStage).Observe(ToMilliseconds(elapsed))
+	p.currentStageStart = now
+
+	p.stageTimer.stageCount.WithLabelValues(p.currentStage).Dec()
+	p.stageTimer.stageCount.WithLabelValues(stage).Inc()
+	p.currentStage = stage
+}
+
+// End completes the current sequence. It is important to call this before discarding the sequenceProbe.
+func (p *SequenceProbe) End() {
+	if p == nil {
+		return
+	}
+
+	now := time.Now()
+	elapsed := now.Sub(p.currentStageStart)
+	p.stageTimer.stageLatency.WithLabelValues(p.currentStage).Observe(ToMilliseconds(elapsed))
+
+	p.stageTimer.stageCount.WithLabelValues(p.currentStage).Dec()
+}

disperser/common/v2/blobstore/dynamo_metadata_store.go

@@ -1705,7 +1705,7 @@ func UnmarshalOperatorID(item commondynamodb.Item) (*core.OperatorID, error) {
 	operatorIDStr := obj.OperatorID
 	if strings.HasPrefix(operatorIDStr, dispersalRequestSKPrefix) {
 		operatorIDStr = strings.TrimPrefix(operatorIDStr, dispersalRequestSKPrefix)
-	} else if strings.HasPrefix(operatorIDStr, dispersalResponseSKPrefix) {
+	} else {
 		operatorIDStr = strings.TrimPrefix(operatorIDStr, dispersalResponseSKPrefix)
 	}