Add Basic Metering Report to Workflow Engine (smartcontractkit#16398)

* Add Basic Metering Report to Workflow Engine

This commit adds a metering report struct to the workflow engine such that each step can be individually added to the
report. Metering units and values are generally treated as string values, but post values are verified to be valid
numeric.

The metering report can provide a calculated median value for all posted units.

* update mathematical operations on metering spend value

* fix tests and greater than comparison

Author: EasterTheBunny

core/services/workflows/engine.go

@@ -143,6 +143,7 @@ type Engine struct {
 	clock          clockwork.Clock
 	ratelimiter    *ratelimiter.RateLimiter
 	workflowLimits *syncerlimiter.Limits
+	meterReport    *MeteringReport
 }
 
 func (e *Engine) Start(_ context.Context) error {
@@ -564,6 +565,8 @@ func generateExecutionID(workflowID, eventID string) (string, error) {
 
 // startExecution kicks off a new workflow execution when a trigger event is received.
 func (e *Engine) startExecution(ctx context.Context, executionID string, event *values.Map) error {
+	e.meterReport = NewMeteringReport()
+
 	lggr := e.logger.With("event", event, platform.KeyWorkflowExecutionID, executionID)
 	lggr.Debug("executing on a trigger event")
 	ec := &store.WorkflowExecution{

core/services/workflows/metering.go

@@ -0,0 +1,110 @@
+package workflows
+
+import (
+	"sort"
+	"sync"
+
+	"github.com/shopspring/decimal"
+)
+
+type MeteringReportStepRef string
+
+type MeteringSpendUnit string
+
+func (s MeteringSpendUnit) String() string {
+	return string(s)
+}
+
+func (s MeteringSpendUnit) DecimalToSpendValue(value decimal.Decimal) MeteringSpendValue {
+	return MeteringSpendValue{value: value, roundingPlace: 18}
+}
+
+func (s MeteringSpendUnit) IntToSpendValue(value int64) MeteringSpendValue {
+	return MeteringSpendValue{value: decimal.NewFromInt(value), roundingPlace: 18}
+}
+
+type MeteringSpendValue struct {
+	value         decimal.Decimal
+	roundingPlace uint8
+}
+
+func (v MeteringSpendValue) Add(value MeteringSpendValue) MeteringSpendValue {
+	return MeteringSpendValue{
+		value:         v.value.Add(value.value),
+		roundingPlace: v.roundingPlace,
+	}
+}
+
+func (v MeteringSpendValue) Div(value MeteringSpendValue) MeteringSpendValue {
+	return MeteringSpendValue{
+		value:         v.value.Div(value.value),
+		roundingPlace: v.roundingPlace,
+	}
+}
+
+func (v MeteringSpendValue) GreaterThan(value MeteringSpendValue) bool {
+	return v.value.GreaterThan(value.value)
+}
+
+func (v MeteringSpendValue) String() string {
+	return v.value.StringFixedBank(int32(v.roundingPlace))
+}
+
+type MeteringReportStep struct {
+	Peer2PeerID string
+	SpendUnit   MeteringSpendUnit
+	SpendValue  MeteringSpendValue
+}
+
+type MeteringReport struct {
+	mu    sync.RWMutex
+	steps map[MeteringReportStepRef]MeteringReportStep
+}
+
+func NewMeteringReport() *MeteringReport {
+	return &MeteringReport{
+		steps: make(map[MeteringReportStepRef]MeteringReportStep),
+	}
+}
+
+func (r *MeteringReport) MedianSpend() map[MeteringSpendUnit]MeteringSpendValue {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+
+	values := map[MeteringSpendUnit][]MeteringSpendValue{}
+	medians := map[MeteringSpendUnit]MeteringSpendValue{}
+
+	for _, step := range r.steps {
+		vals, ok := values[step.SpendUnit]
+		if !ok {
+			vals = []MeteringSpendValue{}
+		}
+
+		values[step.SpendUnit] = append(vals, step.SpendValue)
+	}
+
+	for unit, set := range values {
+		sort.Slice(set, func(i, j int) bool {
+			return set[j].GreaterThan(set[i])
+		})
+
+		if len(set)%2 > 0 {
+			medians[unit] = set[len(set)/2]
+
+			continue
+		}
+
+		medians[unit] = set[len(set)/2-1].Add(set[len(set)/2]).Div(unit.IntToSpendValue(2))
+	}
+
+	return medians
+}
+
+func (r *MeteringReport) AddStep(ref MeteringReportStepRef, step MeteringReportStep) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	r.steps[ref] = step
+
+	return nil
+}

core/services/workflows/metering_test.go

@@ -0,0 +1,129 @@
+package workflows_test
+
+import (
+	"strconv"
+	"testing"
+
+	"github.com/shopspring/decimal"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/exp/maps"
+
+	"github.com/smartcontractkit/chainlink/v2/core/services/workflows"
+)
+
+func TestMeteringReport(t *testing.T) {
+	t.Parallel()
+
+	testUnitA := workflows.MeteringSpendUnit("a")
+	testUnitB := workflows.MeteringSpendUnit("b")
+
+	t.Run("MedianSpend returns median for multiple spend units", func(t *testing.T) {
+		t.Parallel()
+
+		report := workflows.NewMeteringReport()
+		steps := []workflows.MeteringReportStep{
+			{"abc", testUnitA, testUnitA.IntToSpendValue(1)},
+			{"xyz", testUnitA, testUnitA.IntToSpendValue(2)},
+			{"abc", testUnitA, testUnitA.IntToSpendValue(3)},
+			{"abc", testUnitB, testUnitB.DecimalToSpendValue(decimal.NewFromFloat(0.1))},
+			{"xyz", testUnitB, testUnitB.DecimalToSpendValue(decimal.NewFromFloat(0.2))},
+			{"abc", testUnitB, testUnitB.DecimalToSpendValue(decimal.NewFromFloat(0.3))},
+		}
+
+		for idx, step := range steps {
+			require.NoError(t, report.AddStep(workflows.MeteringReportStepRef(strconv.Itoa(idx)), step))
+		}
+
+		expected := map[workflows.MeteringSpendUnit]workflows.MeteringSpendValue{
+			testUnitA: testUnitB.IntToSpendValue(2),
+			testUnitB: testUnitB.DecimalToSpendValue(decimal.NewFromFloat(0.2)),
+		}
+
+		median := report.MedianSpend()
+
+		require.Len(t, median, 2)
+		require.Contains(t, maps.Keys(median), testUnitA)
+		require.Contains(t, maps.Keys(median), testUnitB)
+
+		assert.Equal(t, expected[testUnitA].String(), median[testUnitA].String())
+		assert.Equal(t, expected[testUnitB].String(), median[testUnitB].String())
+	})
+
+	t.Run("MedianSpend returns median single spend value", func(t *testing.T) {
+		t.Parallel()
+
+		report := workflows.NewMeteringReport()
+		steps := []workflows.MeteringReportStep{
+			{"abc", testUnitA, testUnitA.IntToSpendValue(1)},
+		}
+
+		for idx, step := range steps {
+			require.NoError(t, report.AddStep(workflows.MeteringReportStepRef(strconv.Itoa(idx)), step))
+		}
+
+		expected := map[workflows.MeteringSpendUnit]workflows.MeteringSpendValue{
+			testUnitA: testUnitA.IntToSpendValue(1),
+		}
+
+		median := report.MedianSpend()
+
+		require.Len(t, median, 1)
+		require.Contains(t, maps.Keys(median), testUnitA)
+
+		assert.Equal(t, expected[testUnitA].String(), median[testUnitA].String())
+	})
+
+	t.Run("MedianSpend returns median odd number of spend values", func(t *testing.T) {
+		t.Parallel()
+
+		report := workflows.NewMeteringReport()
+		steps := []workflows.MeteringReportStep{
+			{"abc", testUnitA, testUnitA.IntToSpendValue(1)},
+			{"abc", testUnitA, testUnitA.IntToSpendValue(3)},
+			{"xyz", testUnitA, testUnitA.IntToSpendValue(2)},
+		}
+
+		for idx, step := range steps {
+			require.NoError(t, report.AddStep(workflows.MeteringReportStepRef(strconv.Itoa(idx)), step))
+		}
+
+		expected := map[workflows.MeteringSpendUnit]workflows.MeteringSpendValue{
+			testUnitA: testUnitA.IntToSpendValue(2),
+		}
+
+		median := report.MedianSpend()
+
+		require.Len(t, median, 1)
+		require.Contains(t, maps.Keys(median), testUnitA)
+
+		assert.Equal(t, expected[testUnitA].String(), median[testUnitA].String())
+	})
+
+	t.Run("MedianSpend returns median as average for even number of spend values", func(t *testing.T) {
+		t.Parallel()
+
+		report := workflows.NewMeteringReport()
+		steps := []workflows.MeteringReportStep{
+			{"xyz", testUnitA, testUnitA.IntToSpendValue(42)},
+			{"abc", testUnitA, testUnitA.IntToSpendValue(1)},
+			{"abc", testUnitA, testUnitA.IntToSpendValue(3)},
+			{"xyz", testUnitA, testUnitA.IntToSpendValue(2)},
+		}
+
+		for idx, step := range steps {
+			require.NoError(t, report.AddStep(workflows.MeteringReportStepRef(strconv.Itoa(idx)), step))
+		}
+
+		expected := map[workflows.MeteringSpendUnit]workflows.MeteringSpendValue{
+			testUnitA: testUnitA.DecimalToSpendValue(decimal.NewFromFloat(2.5)),
+		}
+
+		median := report.MedianSpend()
+
+		require.Len(t, median, 1)
+		require.Contains(t, maps.Keys(median), testUnitA)
+
+		assert.Equal(t, expected[testUnitA].String(), median[testUnitA].String())
+	})
+}