-
Notifications
You must be signed in to change notification settings - Fork 1.7k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Automation Telemetry: Send BlockNumber and Node Version (#9927)
* rebase on toml feature flag * fixes + send node version rebase on feature flag PR * fixes * comments * minor changes * lint * constant added * logger * toggle by feature flag * minor fixes * marshal error exits start() * marshal error fix * add automation custom telem to job * rename toml tag * test * removed print statement * merge * configDigest go routine getter * configDigest var * refactor1 * send NodeVersion msg every new ConfigDigest * lint * refactor * move registry creation outside delegate.go * plugin config bool flag * custom telem for 2.1 * block subscriber + thread controller * ContractConfigTracker * reset forge-std * use toml config flag * set toml flag default to true * plugin config bool flag * make generate * refactor * hourly node version msg * fix tests * lint fix * goimports fixed * goimport fix 2 --------- Co-authored-by: FelixFan1992 <[email protected]>
- Loading branch information
1 parent
5f09e55
commit b2c1a17
Showing
23 changed files
with
431 additions
and
21 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
156 changes: 156 additions & 0 deletions
156
core/services/ocr2/plugins/ocr2keeper/custom_telemetry.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,156 @@ | ||
| package ocr2keeper | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/hex" | ||
| "time" | ||
|
|
||
| "github.com/smartcontractkit/libocr/commontypes" | ||
| "github.com/smartcontractkit/libocr/offchainreporting2plus/types" | ||
| "google.golang.org/protobuf/proto" | ||
|
|
||
| ocr2keepers "github.com/smartcontractkit/ocr2keepers/pkg/v3/types" | ||
|
|
||
| "github.com/smartcontractkit/chainlink/v2/core/logger" | ||
| evm21 "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ocr2keeper/evm21" | ||
| "github.com/smartcontractkit/chainlink/v2/core/services/synchronization/telem" | ||
| "github.com/smartcontractkit/chainlink/v2/core/static" | ||
| "github.com/smartcontractkit/chainlink/v2/core/utils" | ||
| ) | ||
|
|
||
| type AutomationCustomTelemetryService struct { | ||
| utils.StartStopOnce | ||
| monitoringEndpoint commontypes.MonitoringEndpoint | ||
| blockSubscriber *evm21.BlockSubscriber | ||
| blockSubChanID int | ||
| threadCtrl utils.ThreadControl | ||
| lggr logger.Logger | ||
| configDigest [32]byte | ||
| contractConfigTracker types.ContractConfigTracker | ||
| } | ||
|
|
||
| // NewAutomationCustomTelemetryService creates a telemetry service for new blocks and node version | ||
| func NewAutomationCustomTelemetryService(me commontypes.MonitoringEndpoint, | ||
| lggr logger.Logger, blocksub *evm21.BlockSubscriber, configTracker types.ContractConfigTracker) (*AutomationCustomTelemetryService, error) { | ||
| return &AutomationCustomTelemetryService{ | ||
| monitoringEndpoint: me, | ||
| threadCtrl: utils.NewThreadControl(), | ||
| lggr: lggr.Named("AutomationCustomTelem"), | ||
| contractConfigTracker: configTracker, | ||
| blockSubscriber: blocksub, | ||
| }, nil | ||
| } | ||
|
|
||
| // Start starts Custom Telemetry Service, sends 1 NodeVersion message to endpoint at start and sends new BlockNumber messages | ||
| func (e *AutomationCustomTelemetryService) Start(ctx context.Context) error { | ||
| return e.StartOnce("AutomationCustomTelemetryService", func() error { | ||
| e.lggr.Infof("Starting: Custom Telemetry Service") | ||
| _, configDetails, err := e.contractConfigTracker.LatestConfigDetails(ctx) | ||
| if err != nil { | ||
| e.lggr.Errorf("Error occurred while getting newestConfigDetails for initialization %s", err) | ||
| } else { | ||
| e.configDigest = configDetails | ||
| e.sendNodeVersionMsg() | ||
| } | ||
| e.threadCtrl.Go(func(ctx context.Context) { | ||
| ticker := time.NewTicker(1 * time.Minute) | ||
| defer ticker.Stop() | ||
| for { | ||
| select { | ||
| case <-ticker.C: | ||
| _, newConfigDigest, err := e.contractConfigTracker.LatestConfigDetails(ctx) | ||
| if err != nil { | ||
| e.lggr.Errorf("Error occurred while getting newestConfigDetails in configDigest loop %s", err) | ||
| } | ||
| if newConfigDigest != e.configDigest { | ||
| e.configDigest = newConfigDigest | ||
| e.sendNodeVersionMsg() | ||
| } | ||
| case <-ctx.Done(): | ||
| return | ||
| } | ||
| } | ||
| }) | ||
|
|
||
| chanID, blockSubscriberChan, blockSubErr := e.blockSubscriber.Subscribe() | ||
| if blockSubErr != nil { | ||
| e.lggr.Errorf("Block Subscriber Error: Subscribe(): %s", blockSubErr) | ||
|
|
||
| } else { | ||
| e.blockSubChanID = chanID | ||
| e.threadCtrl.Go(func(ctx context.Context) { | ||
| e.lggr.Infof("Started: Sending BlockNumber Messages") | ||
| for { | ||
| select { | ||
| case blockHistory := <-blockSubscriberChan: | ||
| latestBlockKey, err := blockHistory.Latest() | ||
| if err != nil { | ||
| e.lggr.Errorf("BlockSubscriber BlockHistory.Latest() failed: %s", err) | ||
| continue | ||
| } | ||
| e.sendBlockNumberMsg(latestBlockKey) | ||
| case <-ctx.Done(): | ||
| return | ||
| } | ||
| } | ||
| }) | ||
| } | ||
| return nil | ||
| }) | ||
| } | ||
|
|
||
| // Close stops go routines and closes channels | ||
| func (e *AutomationCustomTelemetryService) Close() error { | ||
| // use utils | ||
| return e.StopOnce("AutomationCustomTelemetryService", func() error { | ||
| e.lggr.Infof("Stopping: custom telemetry service") | ||
| e.threadCtrl.Close() | ||
| err := e.blockSubscriber.Unsubscribe(e.blockSubChanID) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| e.lggr.Infof("Stopped: Custom telemetry service") | ||
| return nil | ||
| }) | ||
| } | ||
|
|
||
| func (e *AutomationCustomTelemetryService) sendNodeVersionMsg() { | ||
| vMsg := &telem.NodeVersion{ | ||
| Timestamp: uint64(time.Now().UTC().UnixMilli()), | ||
| NodeVersion: static.Version, | ||
| ConfigDigest: e.configDigest[:], | ||
| } | ||
| wrappedVMsg := &telem.AutomationTelemWrapper{ | ||
| Msg: &telem.AutomationTelemWrapper_NodeVersion{ | ||
| NodeVersion: vMsg, | ||
| }, | ||
| } | ||
| bytes, err := proto.Marshal(wrappedVMsg) | ||
| if err != nil { | ||
| e.lggr.Errorf("Error occurred while marshalling the Node Version Message %s: %v", wrappedVMsg.String(), err) | ||
| } else { | ||
| e.monitoringEndpoint.SendLog(bytes) | ||
| e.lggr.Infof("NodeVersion Message Sent to Endpoint: %d", vMsg.Timestamp) | ||
| } | ||
| } | ||
|
|
||
| func (e *AutomationCustomTelemetryService) sendBlockNumberMsg(blockKey ocr2keepers.BlockKey) { | ||
| blockNumMsg := &telem.BlockNumber{ | ||
| Timestamp: uint64(time.Now().UTC().UnixMilli()), | ||
| BlockNumber: uint64(blockKey.Number), | ||
| BlockHash: hex.EncodeToString(blockKey.Hash[:]), | ||
| ConfigDigest: e.configDigest[:], | ||
| } | ||
| wrappedBlockNumMsg := &telem.AutomationTelemWrapper{ | ||
| Msg: &telem.AutomationTelemWrapper_BlockNumber{ | ||
| BlockNumber: blockNumMsg, | ||
| }, | ||
| } | ||
| b, err := proto.Marshal(wrappedBlockNumMsg) | ||
| if err != nil { | ||
| e.lggr.Errorf("Error occurred while marshalling the Block Num Message %s: %v", wrappedBlockNumMsg.String(), err) | ||
| } else { | ||
| e.monitoringEndpoint.SendLog(b) | ||
| e.lggr.Infof("BlockNumber Message Sent to Endpoint: %d", blockNumMsg.Timestamp) | ||
| } | ||
| } |
17 changes: 17 additions & 0 deletions
17
core/services/ocr2/plugins/ocr2keeper/custom_telemetry_test.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,17 @@ | ||
| package ocr2keeper | ||
|
|
||
| import ( | ||
| "testing" | ||
| ) | ||
|
|
||
| func TestNewAutomationCustomTelemetryService(t *testing.T) { | ||
| // me := &MockMonitoringEndpoint{} | ||
| // lggr := &MockLogger{} | ||
| // blocksub := &MockBlockSubscriber{} | ||
| // configTracker := &MockContractConfigTracker{} | ||
|
|
||
| // service, err := NewAutomationCustomTelemetryService(me, lggr, blocksub, configTracker) | ||
| // if err != nil { | ||
| // t.Errorf("Expected no error, but got: %v", err) | ||
| // } | ||
| } |
Oops, something went wrong.