Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: add data stream clients #16437

Merged
merged 5 commits into from
Feb 19, 2025
+679 −49
Merged

feat: add data stream clients #16437

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
e0f2ac0
feat: add data stream clients
ChrisAmora Feb 17, 2025
c490d78
fix: lint
ChrisAmora Feb 17, 2025
6868c63
fix: simplify code
ChrisAmora Feb 17, 2025
bf8f4fb
fix: pr suggestions
ChrisAmora Feb 18, 2025
f6577fc
fix: test
ChrisAmora Feb 18, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
106 changes: 71 additions & 35 deletions deployment/data-streams/changeset/jd_register_nodes.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,37 +3,90 @@ package changeset
import (
"errors"
"fmt"
"strconv"

nodev1 "github.com/smartcontractkit/chainlink-protos/job-distributor/v1/node"
"github.com/smartcontractkit/chainlink-protos/job-distributor/v1/shared/ptypes"
"github.com/smartcontractkit/chainlink/deployment"
)

type NodeType int

const (
NodeTypeOracle NodeType = iota
NodeTypeBootstrap
)

func (nt NodeType) String() string {
switch nt {
case NodeTypeOracle:
return "oracle"
case NodeTypeBootstrap:
return "bootstrap"
default:
return "unknown"
}
}

type RegisterNodesInput struct {
EnvLabel string
ProductName string
DONs DONConfigMap
// Will be deleted after migration to DONConfigMap
DONs DONConfigMap `json:"dons,omitempty"`
DONsList []DONConfig `json:"dons_list,omitempty"`
}

type DONConfigMap map[string]DONConfig

type DONConfig struct {
ID int `json:"id"`
ChainSelector string `json:"chainSelector"`
Name string `json:"name"`
ChannelConfigStore string `json:"channelConfigStore"`
Verifier string `json:"verifier"`
Configurator string `json:"configurator"`
Nodes []NodeCfg `json:"nodes"`
BootstrapNodes []NodeCfg `json:"bootstrapNodes"`
}

type NodeCfg struct {
Name string `json:"name"`
CSAKey string `json:"csa_key"`
IsBootstrap bool `json:"isBootstrap"`
IsBootstrap bool `json:"isBootstrap,omitempty"`
}

func validateNodeSlice(nodes []NodeCfg, nodeType string, donIndex int) error {
for _, node := range nodes {
if node.Name == "" {
return fmt.Errorf("DON[%d] has %s node with empty Name", donIndex, nodeType)
}
if node.CSAKey == "" {
return fmt.Errorf("DON[%d] %s node %s has empty CSAKey", donIndex, nodeType, node.Name)
}
}
return nil
}

func registerNodesForDON(e deployment.Environment, nodes []NodeCfg, baseLabels []*ptypes.Label, nodeType NodeType) {
ntStr := nodeType.String()
for _, node := range nodes {
labels := append([]*ptypes.Label(nil), baseLabels...)
labels = append(labels, &ptypes.Label{
Key: "nodeType",
Value: &ntStr,
})
nodeID, err := e.Offchain.RegisterNode(e.GetContext(), &nodev1.RegisterNodeRequest{
Name: node.Name,
PublicKey: node.CSAKey,
Labels: labels,
})
if err != nil {
e.Logger.Errorw("failed to register node", "node", node.Name, "error", err)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we return the error here? or is it intentional to continue if not registered succesfully?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is Intentional, but I left a comment so we can revisit this if necessary in the future.

} else {
e.Logger.Infow("registered node", "name", node.Name, "id", nodeID)
}
}
}

// RegisterNodesWithJD registers each node from the config with the Job Distributor.
// It logs errors but continues to register remaining nodes even if some fail (we may revisit this in the future).
func RegisterNodesWithJD(e deployment.Environment, cfg RegisterNodesInput) (deployment.ChangesetOutput, error) {
baseLabels := []*ptypes.Label{
{
Expand All @@ -46,27 +99,9 @@ func RegisterNodesWithJD(e deployment.Environment, cfg RegisterNodesInput) (depl
},
}

for _, don := range cfg.DONs {
for _, node := range don.Nodes {
labels := append([]*ptypes.Label(nil), baseLabels...)
isBootstrapStr := strconv.FormatBool(node.IsBootstrap)

labels = append(labels, &ptypes.Label{
Key: "isBootstrap",
Value: &isBootstrapStr,
})

nodeID, err := e.Offchain.RegisterNode(e.GetContext(), &nodev1.RegisterNodeRequest{
Name: node.Name,
PublicKey: node.CSAKey,
Labels: labels,
})
if err != nil {
e.Logger.Errorw("failed to register node", "node", node.Name, "error", err)
} else {
e.Logger.Infow("registered node", "name", node.Name, "id", nodeID)
}
}
for _, don := range cfg.DONsList {
registerNodesForDON(e, don.Nodes, baseLabels, NodeTypeOracle)
registerNodesForDON(e, don.BootstrapNodes, baseLabels, NodeTypeBootstrap)
}

return deployment.ChangesetOutput{}, nil
Expand All @@ -80,17 +115,18 @@ func (cfg RegisterNodesInput) Validate() error {
return errors.New("ProductName must not be empty")
}

for donName, don := range cfg.DONs {
for i, don := range cfg.DONsList {
if don.Name == "" {
return fmt.Errorf("DON[%s] has empty Name", donName)
return fmt.Errorf("DON[%d] has empty Name", i)
}
if err := validateNodeSlice(don.Nodes, "node", i); err != nil {
return err
}
if len(don.BootstrapNodes) == 0 {
return fmt.Errorf("DON[%d] has no bootstrap nodes", i)
}
for _, node := range don.Nodes {
if node.Name == "" {
return fmt.Errorf("DON[%s] has node with empty Name", donName)
}
if node.CSAKey == "" {
return fmt.Errorf("DON[%s] node %s has empty CSAKey", donName, node.Name)
}
if err := validateNodeSlice(don.BootstrapNodes, "bootstrap", i); err != nil {
return err
}
}
return nil
Expand Down
53 changes: 39 additions & 14 deletions deployment/data-streams/changeset/jd_register_nodes_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@ import (
"testing"

"github.com/stretchr/testify/require"

"go.uber.org/zap/zapcore"

"github.com/smartcontractkit/chainlink/deployment"
Expand All @@ -21,7 +20,6 @@ func TestRegisterNodesWithJD(t *testing.T) {
nodeP2pKey := e.NodeIDs[0]

jobClient, ok := e.Offchain.(*memory.JobClient)

require.True(t, ok, "expected Offchain to be of type *memory.JobClient")
require.Lenf(t, jobClient.Nodes, 1, "expected exactly 1 node")
require.Emptyf(t, jobClient.RegisteredNodes, "no registered nodes expected")
Expand All @@ -34,10 +32,10 @@ func TestRegisterNodesWithJD(t *testing.T) {
RegisterNodesInput{
EnvLabel: "test-env",
ProductName: "test-product",
DONs: DONConfigMap{
"don1": {
DONsList: []DONConfig{
{
Name: "don1",
Nodes: []NodeCfg{
BootstrapNodes: []NodeCfg{
{Name: "node1", CSAKey: csaKey},
},
},
Expand All @@ -55,11 +53,14 @@ func TestRegisterNodesInput_Validate(t *testing.T) {
cfg := RegisterNodesInput{
EnvLabel: "test-env",
ProductName: "test-product",
DONs: DONConfigMap{
"don1": {
DONsList: []DONConfig{
{
Name: "MyDON",
Nodes: []NodeCfg{
{Name: "node1", CSAKey: "0xabc", IsBootstrap: false},
{Name: "node1", CSAKey: "0xabc"},
},
BootstrapNodes: []NodeCfg{
{Name: "bootstrap1", CSAKey: "0xdef"},
},
},
},
Expand All @@ -70,13 +71,17 @@ func TestRegisterNodesInput_Validate(t *testing.T) {

t.Run("missing product name", func(t *testing.T) {
cfg := RegisterNodesInput{
EnvLabel: "test-env",
DONs: DONConfigMap{
"don2": {
EnvLabel: "test-env",
ProductName: "",
DONsList: []DONConfig{
{
Name: "AnotherDON",
Nodes: []NodeCfg{
{Name: "node1", CSAKey: "0xdef"},
},
BootstrapNodes: []NodeCfg{
{Name: "node2", CSAKey: "0xabc"},
},
},
},
}
Expand All @@ -88,16 +93,36 @@ func TestRegisterNodesInput_Validate(t *testing.T) {
cfg := RegisterNodesInput{
EnvLabel: "test-env",
ProductName: "test-product",
DONs: DONConfigMap{
"don3": {
DONsList: []DONConfig{
{
Name: "EmptyCSA",
Nodes: []NodeCfg{
{Name: "node1", CSAKey: "", IsBootstrap: true},
{Name: "node1", CSAKey: ""},
},
BootstrapNodes: []NodeCfg{
{Name: "bootstrap1", CSAKey: ""},
},
},
},
}
err := cfg.Validate()
require.Error(t, err, "expected an error when CSAKey is empty")
})

t.Run("missing BootstrapNode", func(t *testing.T) {
cfg := RegisterNodesInput{
EnvLabel: "test-env",
ProductName: "test-product",
DONsList: []DONConfig{
{
Name: "EmptyCSA",
Nodes: []NodeCfg{
{Name: "node1", CSAKey: "0xaaa"},
},
},
},
}
err := cfg.Validate()
require.Error(t, err, "expected an error when BooststrapNodes is empty")
})
}
Loading
Loading