telemetry: add metrics to PDP operation failures

Author: frrist

pkg/pdp/service/roots_add.go

@@ -117,6 +117,7 @@ func (p *PDPService) AddRoots(ctx context.Context, id uint64, request []types.Ro
 			log.Errorw("failed to add roots", "id", id, "request", request, "err", retErr)
 			span.RecordError(retErr)
 			span.SetStatus(codes.Error, "failed to add roots")
+			PDPAddPieceFailureCounter.Inc(ctx)
 		} else {
 			span.SetAttributes(attribute.Stringer("tx", res))
 			log.Infow("added roots", "id", id, "request", request, "response", res)

pkg/pdp/service/telemetry.go

@@ -2,8 +2,32 @@ package service
 
 import (
 	"go.opentelemetry.io/otel"
+
+	"github.com/storacha/piri/pkg/telemetry"
 )
 
 var (
 	tracer = otel.Tracer("github.com/storacha/piri/pkg/pdp/service")
 )
+
+var (
+	PDPAddPieceFailureCounter *telemetry.Counter
+)
+
+func init() {
+	tel := telemetry.Global()
+	newCounter := func(name, desc string) *telemetry.Counter {
+		counter, err := tel.NewCounter(telemetry.CounterConfig{
+			Name:        name,
+			Description: desc,
+		})
+		if err != nil {
+			log.Warnw("failed to init telemetry counter", "name", name, "error", err)
+			return nil
+		}
+		return counter
+	}
+
+	PDPAddPieceFailureCounter = newCounter("pdp_add_piece_failure", "records failure to add a pdp piece")
+
+}

pkg/pdp/tasks/next_pdp.go

@@ -146,6 +146,11 @@ func adjustNextProveAt(nextProveAt int64, minRequiredEpoch int64, provingPeriod
 
 func (n *NextProvingPeriodTask) Do(taskID scheduler.TaskID) (done bool, err error) {
 	ctx := context.Background()
+	defer func() {
+		if err != nil {
+			PDPNextFailureCounter.Inc(ctx)
+		}
+	}()
 	// Select the proof set where challenge_request_task_id equals taskID and prove_at_epoch is not NULL
 	var pdp models.PDPProofSet
 	err = n.db.WithContext(ctx).

pkg/pdp/tasks/prove_pdp.go

@@ -162,6 +162,11 @@ func NewProveTask(
 
 func (p *ProveTask) Do(taskID scheduler.TaskID) (done bool, err error) {
 	ctx := context.Background()
+	defer func() {
+		if err != nil {
+			PDPProveFailureCounter.Inc(ctx)
+		}
+	}()
 
 	// Retrieve proof set and challenge epoch for the task
 	var proveTask models.PDPProveTask

pkg/pdp/tasks/sender_eth.go

@@ -12,12 +12,14 @@ import (
 	ethtypes "github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/storacha/filecoin-services/go/evmerrors"
-	"github.com/storacha/piri/pkg/pdp/types"
+	"go.opentelemetry.io/otel/attribute"
 	"go.uber.org/multierr"
 	"golang.org/x/xerrors"
 	"gorm.io/gorm"
 	"gorm.io/gorm/clause"
 
+	"github.com/storacha/piri/pkg/pdp/types"
+
 	"github.com/storacha/piri/pkg/pdp/promise"
 	"github.com/storacha/piri/pkg/pdp/scheduler"
 	"github.com/storacha/piri/pkg/pdp/service/models"
@@ -85,12 +87,20 @@ func (s *SenderETH) Send(ctx context.Context, fromAddress common.Address, tx *et
 				if dataErr.ErrorData() != nil {
 					if parsedErr, failure := evmerrors.ParseRevert(dataErr.ErrorData().(string)); failure == nil {
 						log.Errorw("parsed contract revert during gas estimation", "error", parsedErr)
+						// NB(forrest): ErrorSelector returns the contract error code as hex,
+						// selector values are finite and bounded, so adding it to the counter keeps
+						// cardinality low while giving actionable diagnostics
+						MessageEstimateGasFailureCounter.Inc(ctx, attribute.String("selector",
+							parsedErr.ErrorSelector()), attribute.String("method", reason))
 						return common.Hash{}, types.NewError(types.KindInvalidInput, parsedErr.Error())
 					} else {
 						log.Warnw("failed to parse revert during gas estimation", "parse_error", failure, "original_error", err)
 					}
 				}
 			}
+			// NB(forrest): otherwise we consider the selector unknown
+			MessageEstimateGasFailureCounter.Inc(ctx, attribute.String("selector", "unknown"),
+				attribute.String("method", reason))
 			return common.Hash{}, fmt.Errorf("failed to estimate gas: %w", err)
 		}
 		if gasLimit == 0 {
@@ -360,6 +370,7 @@ func (s *SendTaskETH) Do(taskID scheduler.TaskID) (done bool, err error) {
 	var sendError string
 	if err != nil {
 		sendError = err.Error()
+		MessageSendFailureCounter.Inc(ctx, attribute.String("method", dbTx.SendReason))
 	}
 
 	err = s.db.Model(&models.MessageSendsEth{}).

pkg/pdp/tasks/telemetry.go

@@ -0,0 +1,36 @@
+package tasks
+
+import (
+	"github.com/storacha/piri/pkg/telemetry"
+)
+
+var (
+	MessageEstimateGasFailureCounter *telemetry.Counter
+	MessageSendFailureCounter        *telemetry.Counter
+	PDPProveFailureCounter           *telemetry.Counter
+	PDPNextFailureCounter            *telemetry.Counter
+	PDPAddPieceFailureCounter        *telemetry.Counter
+)
+
+func init() {
+	tel := telemetry.Global()
+	newCounter := func(name, desc string) *telemetry.Counter {
+		counter, err := tel.NewCounter(telemetry.CounterConfig{
+			Name:        name,
+			Description: desc,
+		})
+		if err != nil {
+			log.Warnw("failed to init telemetry counter", "name", name, "error", err)
+			return nil
+		}
+		return counter
+	}
+
+	MessageEstimateGasFailureCounter = newCounter("pdp_message_estimate_gas_failure",
+		"records failure to estimate gas for sending messages; similar to a send failure")
+	MessageSendFailureCounter = newCounter("pdp_message_send_failure", "records failure to send a message")
+	PDPNextFailureCounter = newCounter("pdp_next_failure", "records failure in next pdp task")
+	PDPProveFailureCounter = newCounter("pdp_prove_failure", "records failure to submit a pdp proof")
+	PDPAddPieceFailureCounter = newCounter("pdp_add_piece_failure", "records failure to add a pdp piece")
+
+}