Skip to content

Commit

Permalink
Reporting number of results returned from Contract Readers
Browse files Browse the repository at this point in the history
  • Loading branch information
@mateusz-sekara
mateusz-sekara committed Nov 15, 2023
1 parent 9488826 commit 07727f7
Show file tree
Hide file tree
Showing 6 changed files with 97 additions and 81 deletions.
28 changes: 15 additions & 13 deletions core/services/ocr2/plugins/ccip/internal/observability/commit_store.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,69 +16,71 @@ func NewObservedCommitStoreReader(origin ccipdata.CommitStoreReader, chainID int
return &ObservedCommitStoreReader{
CommitStoreReader: origin,
metric: metricDetails{
histogram: commitStoreHistogram,
pluginName: pluginName,
chainId: chainID,
interactionDuration: readerHistogram,
resultSetSize: readerDatasetSize,
pluginName: pluginName,
readerName: "CommitStoreReader",
chainId: chainID,
},
}
}

func (o *ObservedCommitStoreReader) GetExpectedNextSequenceNumber(context context.Context) (uint64, error) {
return withObservedContract(o.metric, "GetExpectedNextSequenceNumber", func() (uint64, error) {
return withObservedInteraction(o.metric, "GetExpectedNextSequenceNumber", func() (uint64, error) {
return o.CommitStoreReader.GetExpectedNextSequenceNumber(context)
})
}

func (o *ObservedCommitStoreReader) GetLatestPriceEpochAndRound(context context.Context) (uint64, error) {
return withObservedContract(o.metric, "GetLatestPriceEpochAndRound", func() (uint64, error) {
return withObservedInteraction(o.metric, "GetLatestPriceEpochAndRound", func() (uint64, error) {
return o.CommitStoreReader.GetLatestPriceEpochAndRound(context)
})
}

func (o *ObservedCommitStoreReader) GetAcceptedCommitReportsGteSeqNum(ctx context.Context, seqNum uint64, confs int) ([]ccipdata.Event[ccipdata.CommitStoreReport], error) {
return withObservedContract(o.metric, "GetAcceptedCommitReportsGteSeqNum", func() ([]ccipdata.Event[ccipdata.CommitStoreReport], error) {
return withObservedInteractionAndResults(o.metric, "GetAcceptedCommitReportsGteSeqNum", func() ([]ccipdata.Event[ccipdata.CommitStoreReport], error) {
return o.CommitStoreReader.GetAcceptedCommitReportsGteSeqNum(ctx, seqNum, confs)
})
}

func (o *ObservedCommitStoreReader) GetAcceptedCommitReportsGteTimestamp(ctx context.Context, ts time.Time, confs int) ([]ccipdata.Event[ccipdata.CommitStoreReport], error) {
return withObservedContract(o.metric, "GetAcceptedCommitReportsGteTimestamp", func() ([]ccipdata.Event[ccipdata.CommitStoreReport], error) {
return withObservedInteractionAndResults(o.metric, "GetAcceptedCommitReportsGteTimestamp", func() ([]ccipdata.Event[ccipdata.CommitStoreReport], error) {
return o.CommitStoreReader.GetAcceptedCommitReportsGteTimestamp(ctx, ts, confs)
})
}

func (o *ObservedCommitStoreReader) IsDown(ctx context.Context) (bool, error) {
return withObservedContract(o.metric, "IsDown", func() (bool, error) {
return withObservedInteraction(o.metric, "IsDown", func() (bool, error) {
return o.CommitStoreReader.IsDown(ctx)
})
}

func (o *ObservedCommitStoreReader) IsBlessed(ctx context.Context, root [32]byte) (bool, error) {
return withObservedContract(o.metric, "IsBlessed", func() (bool, error) {
return withObservedInteraction(o.metric, "IsBlessed", func() (bool, error) {
return o.CommitStoreReader.IsBlessed(ctx, root)
})
}

func (o *ObservedCommitStoreReader) EncodeCommitReport(report ccipdata.CommitStoreReport) ([]byte, error) {
return withObservedContract(o.metric, "EncodeCommitReport", func() ([]byte, error) {
return withObservedInteraction(o.metric, "EncodeCommitReport", func() ([]byte, error) {
return o.CommitStoreReader.EncodeCommitReport(report)
})
}

func (o *ObservedCommitStoreReader) DecodeCommitReport(report []byte) (ccipdata.CommitStoreReport, error) {
return withObservedContract(o.metric, "DecodeCommitReport", func() (ccipdata.CommitStoreReport, error) {
return withObservedInteraction(o.metric, "DecodeCommitReport", func() (ccipdata.CommitStoreReport, error) {
return o.CommitStoreReader.DecodeCommitReport(report)
})
}

func (o *ObservedCommitStoreReader) VerifyExecutionReport(ctx context.Context, report ccipdata.ExecReport) (bool, error) {
return withObservedContract(o.metric, "VerifyExecutionReport", func() (bool, error) {
return withObservedInteraction(o.metric, "VerifyExecutionReport", func() (bool, error) {
return o.CommitStoreReader.VerifyExecutionReport(ctx, report)
})
}

func (o *ObservedCommitStoreReader) GetCommitStoreStaticConfig(ctx context.Context) (ccipdata.CommitStoreStaticConfig, error) {
return withObservedContract(o.metric, "GetCommitStoreStaticConfig", func() (ccipdata.CommitStoreStaticConfig, error) {
return withObservedInteraction(o.metric, "GetCommitStoreStaticConfig", func() (ccipdata.CommitStoreStaticConfig, error) {
return o.CommitStoreReader.GetCommitStoreStaticConfig(ctx)
})
}
55 changes: 31 additions & 24 deletions core/services/ocr2/plugins/ccip/internal/observability/metrics.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,45 +21,52 @@ var (
float64(1 * time.Second),
float64(2 * time.Second),
}
labels = []string{"evmChainID", "plugin", "function", "success"}
priceRegistryHistogram = promauto.NewHistogramVec(prometheus.HistogramOpts{
Name: "ccip_price_registry_contract_duration",
Help: "Duration of calls to the Price Registry reader",
labels = []string{"evmChainID", "plugin", "reader", "function", "success"}
readerHistogram = promauto.NewHistogramVec(prometheus.HistogramOpts{
Name: "ccip_reader_duration",
Help: "Duration of calls to Reader instance",
Buckets: latencyBuckets,
}, labels)
commitStoreHistogram = promauto.NewHistogramVec(prometheus.HistogramOpts{
Name: "ccip_commit_store_contract_duration",
Help: "Duration of calls to the Commit Store reader",
Buckets: latencyBuckets,
}, labels)
onRampHistogram = promauto.NewHistogramVec(prometheus.HistogramOpts{
Name: "ccip_onramp_contract_duration",
Help: "Duration of calls to the OnRamp reader",
Buckets: latencyBuckets,
}, labels)
offRampHistogram = promauto.NewHistogramVec(prometheus.HistogramOpts{
Name: "ccip_offramp_contract_duration",
Help: "Duration of calls to the OffRamp contract",
Buckets: latencyBuckets,
readerDatasetSize = promauto.NewGaugeVec(prometheus.GaugeOpts{
Name: "ccip_reader_dataset_size",
Help: "Size of the dataset returned from the Reader instance",
}, labels)
)

type metricDetails struct {
histogram *prometheus.HistogramVec
pluginName string
chainId int64
interactionDuration *prometheus.HistogramVec
resultSetSize *prometheus.GaugeVec
pluginName string
readerName string
chainId int64
}

func withObservedContract[T any](metric metricDetails, function string, contract func() (T, error)) (T, error) {
func withObservedInteraction[T any](metric metricDetails, function string, f func() (T, error)) (T, error) {
contractExecutionStarted := time.Now()
value, err := contract()
metric.histogram.
value, err := f()
metric.interactionDuration.
WithLabelValues(
strconv.FormatInt(metric.chainId, 10),
metric.pluginName,
metric.readerName,
function,
strconv.FormatBool(err == nil),
).
Observe(float64(time.Since(contractExecutionStarted)))
return value, err
}

func withObservedInteractionAndResults[T any](metric metricDetails, function string, f func() ([]T, error)) ([]T, error) {
results, err := withObservedInteraction(metric, function, f)
if err == nil {
metric.resultSetSize.WithLabelValues(
strconv.FormatInt(metric.chainId, 10),
metric.pluginName,
metric.readerName,
function,
strconv.FormatBool(err == nil),
).Set(float64(len(results)))
}
return results, err

}
29 changes: 15 additions & 14 deletions core/services/ocr2/plugins/ccip/internal/observability/metrics_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,31 +15,32 @@ import (
)

func TestProperLabelsArePassed(t *testing.T) {
histogram := offRampHistogram
histogram := readerHistogram
successCounter := 10
failedCounter := 5

details := metricDetails{
histogram: histogram,
pluginName: "plugin",
chainId: 123,
interactionDuration: histogram,
pluginName: "plugin",
readerName: "reader",
chainId: 123,
}

for i := 0; i < successCounter; i++ {
_, err := withObservedContract[string](details, "successFun", successfulContract)
_, err := withObservedInteraction[string](details, "successFun", successfulContract)
require.NoError(t, err)
}

for i := 0; i < failedCounter; i++ {
_, err := withObservedContract[string](details, "failedFun", failedContract)
_, err := withObservedInteraction[string](details, "failedFun", failedContract)
require.Error(t, err)
}

assert.Equal(t, successCounter, counterFromHistogramByLabels(t, histogram, "123", "plugin", "successFun", "true"))
assert.Equal(t, failedCounter, counterFromHistogramByLabels(t, histogram, "123", "plugin", "failedFun", "false"))
assert.Equal(t, successCounter, counterFromHistogramByLabels(t, histogram, "123", "plugin", "reader", "successFun", "true"))
assert.Equal(t, failedCounter, counterFromHistogramByLabels(t, histogram, "123", "plugin", "reader", "failedFun", "false"))

assert.Equal(t, 0, counterFromHistogramByLabels(t, histogram, "123", "plugin", "failedFun", "true"))
assert.Equal(t, 0, counterFromHistogramByLabels(t, histogram, "123", "plugin", "successFun", "false"))
assert.Equal(t, 0, counterFromHistogramByLabels(t, histogram, "123", "plugin", "reader", "failedFun", "true"))
assert.Equal(t, 0, counterFromHistogramByLabels(t, histogram, "123", "plugin", "reader", "successFun", "false"))
}

func TestMetricsSendFromContractDirectly(t *testing.T) {
Expand All @@ -58,10 +59,10 @@ func TestMetricsSendFromContractDirectly(t *testing.T) {
_, _ = observedOfframp.GetDestinationTokens(ctx)
}

assert.Equal(t, expectedCounter, counterFromHistogramByLabels(t, observedOfframp.metric.histogram, "420", "plugin", "GetSupportedTokens", "true"))
assert.Equal(t, expectedCounter, counterFromHistogramByLabels(t, observedOfframp.metric.histogram, "420", "plugin", "GetDestinationTokens", "false"))
assert.Equal(t, 0, counterFromHistogramByLabels(t, observedOfframp.metric.histogram, "420", "plugin", "GetPoolByDestToken", "false"))
assert.Equal(t, 0, counterFromHistogramByLabels(t, observedOfframp.metric.histogram, "420", "plugin", "GetPoolByDestToken", "true"))
assert.Equal(t, expectedCounter, counterFromHistogramByLabels(t, observedOfframp.metric.interactionDuration, "420", "plugin", "OffRampReader", "GetSupportedTokens", "true"))
assert.Equal(t, expectedCounter, counterFromHistogramByLabels(t, observedOfframp.metric.interactionDuration, "420", "plugin", "OffRampReader", "GetDestinationTokens", "false"))
assert.Equal(t, 0, counterFromHistogramByLabels(t, observedOfframp.metric.interactionDuration, "420", "plugin", "OffRampReader", "GetPoolByDestToken", "false"))
assert.Equal(t, 0, counterFromHistogramByLabels(t, observedOfframp.metric.interactionDuration, "420", "plugin", "OffRampReader", "GetPoolByDestToken", "true"))
}

func counterFromHistogramByLabels(t *testing.T, histogramVec *prometheus.HistogramVec, labels ...string) int {
Expand Down
32 changes: 17 additions & 15 deletions core/services/ocr2/plugins/ccip/internal/observability/offramp.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,75 +18,77 @@ func NewObservedOffRampReader(origin ccipdata.OffRampReader, chainID int64, plug
return &ObservedOffRampReader{
OffRampReader: origin,
metric: metricDetails{
histogram: offRampHistogram,
pluginName: pluginName,
chainId: chainID,
interactionDuration: readerHistogram,
resultSetSize: readerDatasetSize,
pluginName: pluginName,
readerName: "OffRampReader",
chainId: chainID,
},
}
}

func (o *ObservedOffRampReader) EncodeExecutionReport(report ccipdata.ExecReport) ([]byte, error) {
return withObservedContract(o.metric, "EncodeExecutionReport", func() ([]byte, error) {
return withObservedInteraction(o.metric, "EncodeExecutionReport", func() ([]byte, error) {
return o.OffRampReader.EncodeExecutionReport(report)
})
}

func (o *ObservedOffRampReader) DecodeExecutionReport(report []byte) (ccipdata.ExecReport, error) {
return withObservedContract(o.metric, "DecodeExecutionReport", func() (ccipdata.ExecReport, error) {
return withObservedInteraction(o.metric, "DecodeExecutionReport", func() (ccipdata.ExecReport, error) {
return o.OffRampReader.DecodeExecutionReport(report)
})
}

func (o *ObservedOffRampReader) GetExecutionStateChangesBetweenSeqNums(ctx context.Context, seqNumMin, seqNumMax uint64, confs int) ([]ccipdata.Event[ccipdata.ExecutionStateChanged], error) {
return withObservedContract(o.metric, "GetExecutionStateChangesBetweenSeqNums", func() ([]ccipdata.Event[ccipdata.ExecutionStateChanged], error) {
func (o *ObservedOffRampReader) withObservedInteractionAndResults(ctx context.Context, seqNumMin, seqNumMax uint64, confs int) ([]ccipdata.Event[ccipdata.ExecutionStateChanged], error) {
return withObservedInteraction(o.metric, "GetExecutionStateChangesBetweenSeqNums", func() ([]ccipdata.Event[ccipdata.ExecutionStateChanged], error) {
return o.OffRampReader.GetExecutionStateChangesBetweenSeqNums(ctx, seqNumMin, seqNumMax, confs)
})
}

func (o *ObservedOffRampReader) GetDestinationTokens(ctx context.Context) ([]common.Address, error) {
return withObservedContract(o.metric, "GetDestinationTokens", func() ([]common.Address, error) {
return withObservedInteractionAndResults(o.metric, "GetDestinationTokens", func() ([]common.Address, error) {
return o.OffRampReader.GetDestinationTokens(ctx)
})
}

func (o *ObservedOffRampReader) GetPoolByDestToken(ctx context.Context, address common.Address) (common.Address, error) {
return withObservedContract(o.metric, "GetPoolByDestToken", func() (common.Address, error) {
return withObservedInteraction(o.metric, "GetPoolByDestToken", func() (common.Address, error) {
return o.OffRampReader.GetPoolByDestToken(ctx, address)
})
}

func (o *ObservedOffRampReader) GetDestinationTokensFromSourceTokens(ctx context.Context, tokenAddresses []common.Address) ([]common.Address, error) {
return withObservedContract(o.metric, "GetDestinationTokensFromSourceTokens", func() ([]common.Address, error) {
return withObservedInteractionAndResults(o.metric, "GetDestinationTokensFromSourceTokens", func() ([]common.Address, error) {
return o.OffRampReader.GetDestinationTokensFromSourceTokens(ctx, tokenAddresses)
})
}

func (o *ObservedOffRampReader) GetSupportedTokens(ctx context.Context) ([]common.Address, error) {
return withObservedContract(o.metric, "GetSupportedTokens", func() ([]common.Address, error) {
return withObservedInteractionAndResults(o.metric, "GetSupportedTokens", func() ([]common.Address, error) {
return o.OffRampReader.GetSupportedTokens(ctx)
})
}

func (o *ObservedOffRampReader) GetSenderNonce(ctx context.Context, sender common.Address) (uint64, error) {
return withObservedContract(o.metric, "GetSenderNonce", func() (uint64, error) {
return withObservedInteraction(o.metric, "GetSenderNonce", func() (uint64, error) {
return o.OffRampReader.GetSenderNonce(ctx, sender)
})
}

func (o *ObservedOffRampReader) CurrentRateLimiterState(ctx context.Context) (evm_2_evm_offramp.RateLimiterTokenBucket, error) {
return withObservedContract(o.metric, "CurrentRateLimiterState", func() (evm_2_evm_offramp.RateLimiterTokenBucket, error) {
return withObservedInteraction(o.metric, "CurrentRateLimiterState", func() (evm_2_evm_offramp.RateLimiterTokenBucket, error) {
return o.OffRampReader.CurrentRateLimiterState(ctx)
})
}

func (o *ObservedOffRampReader) GetExecutionState(ctx context.Context, sequenceNumber uint64) (uint8, error) {
return withObservedContract(o.metric, "GetExecutionState", func() (uint8, error) {
return withObservedInteraction(o.metric, "GetExecutionState", func() (uint8, error) {
return o.OffRampReader.GetExecutionState(ctx, sequenceNumber)
})
}

func (o *ObservedOffRampReader) GetStaticConfig(ctx context.Context) (ccipdata.OffRampStaticConfig, error) {
return withObservedContract(o.metric, "GetStaticConfig", func() (ccipdata.OffRampStaticConfig, error) {
return withObservedInteraction(o.metric, "GetStaticConfig", func() (ccipdata.OffRampStaticConfig, error) {
return o.OffRampReader.GetStaticConfig(ctx)
})
}
18 changes: 10 additions & 8 deletions core/services/ocr2/plugins/ccip/internal/observability/onramp.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,39 +18,41 @@ func NewObservedOnRampReader(origin ccipdata.OnRampReader, chainID int64, plugin
return &ObservedOnRampReader{
OnRampReader: origin,
metric: metricDetails{
histogram: onRampHistogram,
pluginName: pluginName,
chainId: chainID,
interactionDuration: readerHistogram,
resultSetSize: readerDatasetSize,
pluginName: pluginName,
readerName: "OnRampReader",
chainId: chainID,
},
}
}

func (o ObservedOnRampReader) GetSendRequestsGteSeqNum(ctx context.Context, seqNum uint64, confs int) ([]ccipdata.Event[internal.EVM2EVMMessage], error) {
return withObservedContract(o.metric, "GetSendRequestsGteSeqNum", func() ([]ccipdata.Event[internal.EVM2EVMMessage], error) {
return withObservedInteractionAndResults(o.metric, "GetSendRequestsGteSeqNum", func() ([]ccipdata.Event[internal.EVM2EVMMessage], error) {
return o.OnRampReader.GetSendRequestsGteSeqNum(ctx, seqNum, confs)
})
}

func (o ObservedOnRampReader) GetSendRequestsBetweenSeqNums(ctx context.Context, seqNumMin, seqNumMax uint64, confs int) ([]ccipdata.Event[internal.EVM2EVMMessage], error) {
return withObservedContract(o.metric, "GetSendRequestsBetweenSeqNums", func() ([]ccipdata.Event[internal.EVM2EVMMessage], error) {
return withObservedInteractionAndResults(o.metric, "GetSendRequestsBetweenSeqNums", func() ([]ccipdata.Event[internal.EVM2EVMMessage], error) {
return o.OnRampReader.GetSendRequestsBetweenSeqNums(ctx, seqNumMin, seqNumMax, confs)
})
}

func (o ObservedOnRampReader) RouterAddress() (common.Address, error) {
return withObservedContract(o.metric, "RouterAddress", func() (common.Address, error) {
return withObservedInteraction(o.metric, "RouterAddress", func() (common.Address, error) {
return o.OnRampReader.RouterAddress()
})
}

func (o ObservedOnRampReader) Address() (common.Address, error) {
return withObservedContract(o.metric, "Address", func() (common.Address, error) {
return withObservedInteraction(o.metric, "Address", func() (common.Address, error) {
return o.OnRampReader.Address()
})
}

func (o ObservedOnRampReader) GetDynamicConfig() (ccipdata.OnRampDynamicConfig, error) {
return withObservedContract(o.metric, "GetDynamicConfig", func() (ccipdata.OnRampDynamicConfig, error) {
return withObservedInteraction(o.metric, "GetDynamicConfig", func() (ccipdata.OnRampDynamicConfig, error) {
return o.OnRampReader.GetDynamicConfig()
})
}
Loading

0 comments on commit 07727f7

Please sign in to comment.