From 847de6df79d4f4c9f80c228aac34eeda8397dbd0 Mon Sep 17 00:00:00 2001 From: Cedric Cordenier Date: Mon, 11 Dec 2023 14:16:49 +0000 Subject: [PATCH] Move medianpoc to feeds --- cmd/chainlink-feeds/main.go | 10 +++ go.mod | 7 ++- go.sum | 10 +-- median/data_source.go | 97 ++++++++++++++++++++++++++++++ median/data_source_test.go | 117 ++++++++++++++++++++++++++++++++++++ median/plugin.go | 92 +++++++++++++++++++++++++++- median/plugin_test.go | 108 +++++++++++++++++++++++++++++++++ 7 files changed, 431 insertions(+), 10 deletions(-) create mode 100644 median/data_source.go create mode 100644 median/data_source_test.go create mode 100644 median/plugin_test.go diff --git a/cmd/chainlink-feeds/main.go b/cmd/chainlink-feeds/main.go index 4232cd7..7d3fb6b 100644 --- a/cmd/chainlink-feeds/main.go +++ b/cmd/chainlink-feeds/main.go @@ -4,6 +4,8 @@ import ( "github.com/hashicorp/go-plugin" "github.com/smartcontractkit/chainlink-common/pkg/loop" + "github.com/smartcontractkit/chainlink-common/pkg/loop/reportingplugins" + "github.com/smartcontractkit/chainlink-common/pkg/types" "github.com/smartcontractkit/chainlink-feeds/median" ) @@ -35,6 +37,14 @@ func main() { GRPCOpts: s.GRPCOpts, }, }, + reportingplugins.PluginServiceName: &reportingplugins.GRPCService[types.MedianProvider]{ + PluginServer: p, + BrokerConfig: loop.BrokerConfig{ + Logger: s.Logger, + StopCh: stop, + GRPCOpts: s.GRPCOpts, + }, + }, }, GRPCServer: s.GRPCOpts.NewServer, }) diff --git a/go.mod b/go.mod index ac9c2e4..43a6dbe 100644 --- a/go.mod +++ b/go.mod @@ -4,8 +4,10 @@ go 1.21.3 require ( github.com/hashicorp/go-plugin v1.5.2 + github.com/shopspring/decimal v1.3.1 github.com/smartcontractkit/chainlink-common v0.1.7-0.20231206181640-faad3f11cfad - github.com/smartcontractkit/libocr v0.0.0-20231020123319-d255366a6545 + github.com/smartcontractkit/libocr v0.0.0-20231130143053-c5102a9c0fb7 + github.com/stretchr/testify v1.8.4 ) require ( @@ -40,7 +42,6 @@ require ( github.com/prometheus/client_model v0.4.1-0.20230718164431-9a2bf3000d16 // indirect github.com/prometheus/common v0.44.0 // indirect github.com/prometheus/procfs v0.11.1 // indirect - github.com/stretchr/testify v1.8.4 // indirect github.com/x448/float16 v0.8.4 // indirect go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.45.0 // indirect go.opentelemetry.io/otel v1.19.0 // indirect @@ -53,7 +54,7 @@ require ( go.uber.org/multierr v1.11.0 // indirect go.uber.org/zap v1.26.0 // indirect golang.org/x/crypto v0.14.0 // indirect - golang.org/x/exp v0.0.0-20230425010034-47ecfdc1ba53 // indirect + golang.org/x/exp v0.0.0-20230810033253-352e893a4cad // indirect golang.org/x/net v0.17.0 // indirect golang.org/x/sys v0.13.0 // indirect golang.org/x/text v0.13.0 // indirect diff --git a/go.sum b/go.sum index 5047c72..d42d541 100644 --- a/go.sum +++ b/go.sum @@ -200,14 +200,16 @@ github.com/prometheus/procfs v0.11.1/go.mod h1:eesXgaPo1q7lBpVMoMy0ZOFTth9hBn4W/ github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4= github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ= github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog= +github.com/shopspring/decimal v1.3.1 h1:2Usl1nmF/WZucqkFZhnfFYxxxu8LG21F6nPQBE5gKV8= +github.com/shopspring/decimal v1.3.1/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o= github.com/smartcontractkit/chainlink-common v0.1.7-0.20231206181640-faad3f11cfad h1:ysPjfbCPJuVxxFZa1Ifv8OPE20pzvnEHjJrPDUo4gT0= github.com/smartcontractkit/chainlink-common v0.1.7-0.20231206181640-faad3f11cfad/go.mod h1:IdlfCN9rUs8Q/hrOYe8McNBIwEOHEsi0jilb3Cw77xs= github.com/smartcontractkit/go-plugin v0.0.0-20231003134350-e49dad63b306 h1:ko88+ZznniNJZbZPWAvHQU8SwKAdHngdDZ+pvVgB5ss= github.com/smartcontractkit/go-plugin v0.0.0-20231003134350-e49dad63b306/go.mod h1:w1sAEES3g3PuV/RzUrgow20W2uErMly84hhD3um1WL4= github.com/smartcontractkit/grpc-proxy v0.0.0-20230731113816-f1be6620749f h1:hgJif132UCdjo8u43i7iPN1/MFnu49hv7lFGFftCHKU= github.com/smartcontractkit/grpc-proxy v0.0.0-20230731113816-f1be6620749f/go.mod h1:MvMXoufZAtqExNexqi4cjrNYE9MefKddKylxjS+//n0= -github.com/smartcontractkit/libocr v0.0.0-20231020123319-d255366a6545 h1:qOsw2ETQD/Sb/W2xuYn2KPWjvvsWA0C+l19rWFq8iNg= -github.com/smartcontractkit/libocr v0.0.0-20231020123319-d255366a6545/go.mod h1:2lyRkw/qLQgUWlrWWmq5nj0y90rWeO6Y+v+fCakRgb0= +github.com/smartcontractkit/libocr v0.0.0-20231130143053-c5102a9c0fb7 h1:AA7vf29c6lFsZm+MtIEtXtg6VUOQV6waJo5MUuHfRjQ= +github.com/smartcontractkit/libocr v0.0.0-20231130143053-c5102a9c0fb7/go.mod h1:WcuWFMskcGK0MYZuH5hEhGJOzdJRUFeNEM4PAKlejI4= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA= @@ -265,8 +267,8 @@ golang.org/x/exp v0.0.0-20191227195350-da58074b4299/go.mod h1:2RIsYlXP63K8oxa1u0 golang.org/x/exp v0.0.0-20200119233911-0405dc783f0a/go.mod h1:2RIsYlXP63K8oxa1u096TMicItID8zy7Y6sNkU49FU4= golang.org/x/exp v0.0.0-20200207192155-f17229e696bd/go.mod h1:J/WKrq2StrnmMY6+EHIKF9dgMWnmCNThgcyBT1FY9mM= golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6/go.mod h1:3jZMyOhIsHpP37uCMkUooju7aAi5cS1Q23tOzKc+0MU= -golang.org/x/exp v0.0.0-20230425010034-47ecfdc1ba53 h1:5llv2sWeaMSnA3w2kS57ouQQ4pudlXrR0dCgw51QK9o= -golang.org/x/exp v0.0.0-20230425010034-47ecfdc1ba53/go.mod h1:V1LtkGg67GoY2N1AnLN78QLrzxkLyJw7RJb1gzOOz9w= +golang.org/x/exp v0.0.0-20230810033253-352e893a4cad h1:g0bG7Z4uG+OgH2QDODnjp6ggkk1bJDsINcuWmJN1iJU= +golang.org/x/exp v0.0.0-20230810033253-352e893a4cad/go.mod h1:FXUEEKJgO7OQYeo8N01OfiKP8RXMtf6e8aTskBGqWdc= golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js= golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0= golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE= diff --git a/median/data_source.go b/median/data_source.go new file mode 100644 index 0000000..f427dc1 --- /dev/null +++ b/median/data_source.go @@ -0,0 +1,97 @@ +package median + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "math/big" + "sync" + "time" + + "github.com/shopspring/decimal" + ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/types" +) + +type BridgeMetaData struct { + LatestAnswer *big.Int `json:"latestAnswer"` + UpdatedAt *big.Int `json:"updatedAt"` // A unix timestamp +} + +func MarshalBridgeMetaData(latestAnswer *big.Int, updatedAt *big.Int) (map[string]interface{}, error) { + b, err := json.Marshal(&BridgeMetaData{LatestAnswer: latestAnswer, UpdatedAt: updatedAt}) + if err != nil { + return nil, err + } + var mp map[string]interface{} + err = json.Unmarshal(b, &mp) + if err != nil { + return nil, err + } + return mp, nil +} + +type DataSource struct { + pipelineRunner types.PipelineRunnerService + spec string + lggr logger.Logger + + current BridgeMetaData + mu sync.RWMutex +} + +func (d *DataSource) Observe(ctx context.Context, reportTimestamp ocrtypes.ReportTimestamp) (*big.Int, error) { + md, err := MarshalBridgeMetaData(d.currentAnswer()) + if err != nil { + d.lggr.Warnw("unable to attach metadata for run", "err", err) + } + + // NOTE: job metadata is automatically attached by the pipeline runner service + vars := types.Vars{ + Vars: map[string]interface{}{ + "jobRun": md, + }, + } + + results, err := d.pipelineRunner.ExecuteRun(ctx, d.spec, vars, types.Options{}) + if err != nil { + return nil, err + } + + finalResults := results.FinalResults() + if len(finalResults) == 0 { + return nil, errors.New("pipeline execution failed: not enough results") + } + + finalResult := finalResults[0] + if finalResult.Error != nil { + return nil, fmt.Errorf("pipeline execution failed: %w", finalResult.Error) + } + + asDecimal, ok := (finalResult.Value).(decimal.Decimal) + if !ok { + return nil, errors.New("cannot convert observation to decimal") + } + + resultAsBigInt := asDecimal.BigInt() + d.updateAnswer(resultAsBigInt) + return resultAsBigInt, nil +} + +func (d *DataSource) currentAnswer() (*big.Int, *big.Int) { + d.mu.RLock() + defer d.mu.RUnlock() + return d.current.LatestAnswer, d.current.UpdatedAt +} + +func (d *DataSource) updateAnswer(latestAnswer *big.Int) { + d.mu.Lock() + defer d.mu.Unlock() + d.current = BridgeMetaData{ + LatestAnswer: latestAnswer, + UpdatedAt: big.NewInt(time.Now().Unix()), + } +} diff --git a/median/data_source_test.go b/median/data_source_test.go new file mode 100644 index 0000000..f8f0ca1 --- /dev/null +++ b/median/data_source_test.go @@ -0,0 +1,117 @@ +package median + +import ( + "context" + "errors" + "math/big" + "testing" + + "github.com/shopspring/decimal" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" + + "github.com/smartcontractkit/chainlink-common/pkg/types" +) + +type mockPipelineRunner struct { + results types.TaskResults + err error + spec string + vars types.Vars + options types.Options +} + +func (m *mockPipelineRunner) ExecuteRun(ctx context.Context, spec string, vars types.Vars, options types.Options) (types.TaskResults, error) { + m.spec = spec + m.vars = vars + m.options = options + return m.results, m.err +} + +func TestDataSource(t *testing.T) { + lggr := logger.Test(t) + expect := int64(3) + pr := &mockPipelineRunner{ + results: types.TaskResults{ + { + TaskValue: types.TaskValue{ + Value: decimal.NewFromInt(expect), + Error: nil, + IsTerminal: true, + }, + Index: 2, + }, + { + TaskValue: types.TaskValue{ + Value: decimal.NewFromInt(6), + Error: nil, + IsTerminal: false, + }, + Index: 1, + }, + }, + } + spec := "SPEC" + ds := &DataSource{ + pipelineRunner: pr, + spec: spec, + lggr: lggr, + } + res, err := ds.Observe(context.Background(), ocrtypes.ReportTimestamp{}) + require.NoError(t, err) + assert.Equal(t, big.NewInt(expect), res) + assert.Equal(t, spec, pr.spec) + assert.Equal(t, big.NewInt(expect), ds.current.LatestAnswer) +} + +func TestDataSource_ResultErrors(t *testing.T) { + lggr := logger.Test(t) + pr := &mockPipelineRunner{ + results: types.TaskResults{ + { + TaskValue: types.TaskValue{ + Error: errors.New("something went wrong"), + IsTerminal: true, + }, + Index: 0, + }, + }, + } + spec := "SPEC" + ds := &DataSource{ + pipelineRunner: pr, + spec: spec, + lggr: lggr, + } + _, err := ds.Observe(context.Background(), ocrtypes.ReportTimestamp{}) + assert.ErrorContains(t, err, "something went wrong") +} + +func TestDataSource_ResultNotAnInt(t *testing.T) { + lggr := logger.Test(t) + + expect := "string-result" + pr := &mockPipelineRunner{ + results: types.TaskResults{ + { + TaskValue: types.TaskValue{ + Value: expect, + IsTerminal: true, + }, + Index: 0, + }, + }, + } + spec := "SPEC" + ds := &DataSource{ + pipelineRunner: pr, + spec: spec, + lggr: lggr, + } + _, err := ds.Observe(context.Background(), ocrtypes.ReportTimestamp{}) + assert.ErrorContains(t, err, "cannot convert observation to decimal") +} diff --git a/median/plugin.go b/median/plugin.go index 54c729c..2746b81 100644 --- a/median/plugin.go +++ b/median/plugin.go @@ -2,6 +2,8 @@ package median import ( "context" + "encoding/json" + "fmt" "math/big" "time" @@ -10,6 +12,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/logger" "github.com/smartcontractkit/chainlink-common/pkg/loop" + "github.com/smartcontractkit/chainlink-common/pkg/loop/reportingplugins" "github.com/smartcontractkit/chainlink-common/pkg/services" "github.com/smartcontractkit/chainlink-common/pkg/types" ) @@ -17,16 +20,34 @@ import ( type Plugin struct { loop.Plugin stop services.StopChan + reportingplugins.MedianProviderServer } func NewPlugin(lggr logger.Logger) *Plugin { - return &Plugin{Plugin: loop.Plugin{Logger: lggr}, stop: make(services.StopChan)} + return &Plugin{ + Plugin: loop.Plugin{Logger: lggr}, + MedianProviderServer: reportingplugins.MedianProviderServer{}, + stop: make(services.StopChan), + } } func (p *Plugin) NewMedianFactory(ctx context.Context, provider types.MedianProvider, dataSource, juelsPerFeeCoin median.DataSource, errorLog loop.ErrorLog) (loop.ReportingPluginFactory, error) { var ctxVals loop.ContextValues ctxVals.SetValues(ctx) lggr := logger.With(p.Logger, ctxVals.Args()...) + factory, err := p.newMedianFactory(lggr, provider, dataSource, juelsPerFeeCoin, errorLog) + if err != nil { + return nil, err + } + + s := &reportingPluginFactoryService{lggr: logger.Named(lggr, "ReportingPluginFactory"), ReportingPluginFactory: factory} + + p.SubService(s) + + return s, nil +} + +func (p *Plugin) newMedianFactory(lggr logger.Logger, provider types.MedianProvider, dataSource, juelsPerFeeCoin median.DataSource, errorLog loop.ErrorLog) (median.NumericalMedianFactory, error) { factory := median.NumericalMedianFactory{ DataSource: dataSource, JuelsPerFeeCoinDataSource: juelsPerFeeCoin, @@ -45,13 +66,78 @@ func (p *Plugin) NewMedianFactory(ctx context.Context, provider types.MedianProv } else { factory.ContractTransmitter = provider.MedianContract() } - s := &reportingPluginFactoryService{lggr: logger.Named(lggr, "ReportingPluginFactory"), ReportingPluginFactory: factory} + return factory, nil +} - p.SubService(s) +type pipelineSpec struct { + Name string `json:"name"` + Spec string `json:"spec"` +} + +type jsonConfig struct { + Pipelines []pipelineSpec `json:"pipelines"` +} + +func (j jsonConfig) defaultPipeline() (string, error) { + return j.getPipeline("__DEFAULT_PIPELINE__") +} +func (j jsonConfig) getPipeline(key string) (string, error) { + for _, v := range j.Pipelines { + if v.Name == key { + return v.Spec, nil + } + } + return "", fmt.Errorf("no pipeline found for %s", key) +} + +func (p *Plugin) NewReportingPluginFactory( + ctx context.Context, + config types.ReportingPluginServiceConfig, + provider types.MedianProvider, + pipelineRunner types.PipelineRunnerService, + telemetry types.TelemetryClient, + errorLog types.ErrorLog, +) (types.ReportingPluginFactory, error) { + f, err := p.newFactory(ctx, config, provider, pipelineRunner, telemetry, errorLog) + if err != nil { + return nil, err + } + s := &reportingPluginFactoryService{lggr: p.Logger, ReportingPluginFactory: f} + p.SubService(s) return s, nil } +func (p *Plugin) newFactory(ctx context.Context, config types.ReportingPluginServiceConfig, provider types.MedianProvider, pipelineRunner types.PipelineRunnerService, telemetry types.TelemetryClient, errorLog types.ErrorLog) (median.NumericalMedianFactory, error) { + jc := &jsonConfig{} + err := json.Unmarshal([]byte(config.PluginConfig), jc) + if err != nil { + return median.NumericalMedianFactory{}, err + } + + dp, err := jc.defaultPipeline() + if err != nil { + return median.NumericalMedianFactory{}, err + } + ds := &DataSource{ + pipelineRunner: pipelineRunner, + spec: dp, + lggr: p.Logger, + } + + jfp, err := jc.getPipeline("juelsPerFeeCoinPipeline") + if err != nil { + return median.NumericalMedianFactory{}, err + } + jds := &DataSource{ + pipelineRunner: pipelineRunner, + spec: jfp, + lggr: p.Logger, + } + + return p.newMedianFactory(p.Logger, provider, ds, jds, errorLog) +} + type reportingPluginFactoryService struct { services.StateMachine lggr logger.Logger diff --git a/median/plugin_test.go b/median/plugin_test.go new file mode 100644 index 0000000..fd078be --- /dev/null +++ b/median/plugin_test.go @@ -0,0 +1,108 @@ +package median + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/smartcontractkit/libocr/offchainreporting2/reportingplugin/median" + ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/types" +) + +type mockErrorLog struct { + types.ErrorLog +} + +type mockOffchainConfigDigester struct { + ocrtypes.OffchainConfigDigester +} + +type mockContractTransmitter struct { + ocrtypes.ContractTransmitter +} + +type mockContractConfigTracker struct { + ocrtypes.ContractConfigTracker +} + +type mockReportCodec struct { + median.ReportCodec +} + +type mockMedianContract struct { + median.MedianContract +} + +type mockOnchainConfigCodec struct { + median.OnchainConfigCodec +} + +type provider struct { + types.Service +} + +func (p provider) OffchainConfigDigester() ocrtypes.OffchainConfigDigester { + return mockOffchainConfigDigester{} +} + +func (p provider) ContractTransmitter() ocrtypes.ContractTransmitter { + return mockContractTransmitter{} +} + +func (p provider) ContractConfigTracker() ocrtypes.ContractConfigTracker { + return mockContractConfigTracker{} +} + +func (p provider) ReportCodec() median.ReportCodec { + return mockReportCodec{} +} + +func (p provider) MedianContract() median.MedianContract { + return mockMedianContract{} +} + +func (p provider) OnchainConfigCodec() median.OnchainConfigCodec { + return mockOnchainConfigCodec{} +} + +func (p provider) ChainReader() types.ChainReader { + return nil +} + +func TestNewPlugin(t *testing.T) { + lggr := logger.Test(t) + p := NewPlugin(lggr) + + defaultSpec := "default-spec" + juelsPerFeeCoinSpec := "jpfc-spec" + config := types.ReportingPluginServiceConfig{ + PluginConfig: fmt.Sprintf( + `{"pipelines": [{"name": "__DEFAULT_PIPELINE__", "spec": "%s"},{"name": "juelsPerFeeCoinPipeline", "spec": "%s"}]}`, + defaultSpec, + juelsPerFeeCoinSpec, + ), + } + pr := &mockPipelineRunner{} + prov := provider{} + + f, err := p.newFactory( + context.Background(), + config, + prov, + pr, + nil, + mockErrorLog{}, + ) + require.NoError(t, err) + + ds := f.DataSource.(*DataSource) + assert.Equal(t, defaultSpec, ds.spec) + jpfcDs := f.JuelsPerFeeCoinDataSource.(*DataSource) + assert.Equal(t, juelsPerFeeCoinSpec, jpfcDs.spec) +}