Skip to content

Commit d35d8de

Browse files
authored
feat: add node filter to memory client (#536)
When calling `ListNodes`, the filter can now be applied to the nodes. This supports all the filter options that are supported by the `ListNodes` API.
1 parent 9fcf5d6 commit d35d8de

File tree

5 files changed

+1023
-17
lines changed

5 files changed

+1023
-17
lines changed

.changeset/curly-llamas-glow.md

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
"chainlink-deployments-framework": minor
3+
---
4+
5+
JD Memory Client now supports filtering in `ListNodes`

offchain/jd/memory/memory_client.go

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -349,14 +349,22 @@ func (m *MemoryJobDistributor) GetNode(ctx context.Context, in *nodev1.GetNodeRe
349349
// ListNodes returns all nodes stored in memory.
350350
func (m *MemoryJobDistributor) ListNodes(ctx context.Context, in *nodev1.ListNodesRequest, opts ...grpc.CallOption) (*nodev1.ListNodesResponse, error) {
351351
m.mu.RLock()
352-
nodes := make([]*nodev1.Node, 0, len(m.nodes))
352+
allNodes := make([]*nodev1.Node, 0, len(m.nodes))
353353
for _, node := range m.nodes {
354-
nodes = append(nodes, node)
354+
allNodes = append(allNodes, node)
355355
}
356356
m.mu.RUnlock()
357357

358+
// Apply filtering if filter is provided
359+
var filteredNodes []*nodev1.Node
360+
if in.Filter != nil {
361+
filteredNodes = applyNodeFilter(allNodes, in.Filter)
362+
} else {
363+
filteredNodes = allNodes
364+
}
365+
358366
return &nodev1.ListNodesResponse{
359-
Nodes: nodes,
367+
Nodes: filteredNodes,
360368
}, nil
361369
}
362370

offchain/jd/memory/memory_client_test.go

Lines changed: 40 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,8 @@ import (
1010
jobv1 "github.com/smartcontractkit/chainlink-protos/job-distributor/v1/job"
1111
nodev1 "github.com/smartcontractkit/chainlink-protos/job-distributor/v1/node"
1212
"github.com/smartcontractkit/chainlink-protos/job-distributor/v1/shared/ptypes"
13+
14+
"github.com/smartcontractkit/chainlink-deployments-framework/internal/pointer"
1315
)
1416

1517
func TestMemoryJobDistributor_ProposeJob(t *testing.T) {
@@ -466,22 +468,26 @@ func TestMemoryJobDistributor_GetNode(t *testing.T) {
466468
func TestMemoryJobDistributor_ListNodes(t *testing.T) {
467469
t.Parallel()
468470

471+
client := NewMemoryJobDistributor()
472+
ctx := t.Context()
473+
// Register multiple nodes
474+
_, err := client.RegisterNode(ctx, &nodev1.RegisterNodeRequest{
475+
Name: "Node 1",
476+
PublicKey: "key-1",
477+
Labels: []*ptypes.Label{
478+
{Key: "environment", Value: pointer.To("prod")},
479+
},
480+
})
481+
require.NoError(t, err)
482+
483+
_, err = client.RegisterNode(ctx, &nodev1.RegisterNodeRequest{
484+
Name: "Node 2",
485+
PublicKey: "key-2",
486+
})
487+
require.NoError(t, err)
488+
469489
t.Run("list nodes returns all nodes", func(t *testing.T) {
470490
t.Parallel()
471-
client := NewMemoryJobDistributor()
472-
ctx := t.Context()
473-
// Register multiple nodes
474-
_, err := client.RegisterNode(ctx, &nodev1.RegisterNodeRequest{
475-
Name: "Node 1",
476-
PublicKey: "key-1",
477-
})
478-
require.NoError(t, err)
479-
480-
_, err = client.RegisterNode(ctx, &nodev1.RegisterNodeRequest{
481-
Name: "Node 2",
482-
PublicKey: "key-2",
483-
})
484-
require.NoError(t, err)
485491

486492
// List all nodes
487493
listResp, err := client.ListNodes(ctx, &nodev1.ListNodesRequest{})
@@ -490,6 +496,26 @@ func TestMemoryJobDistributor_ListNodes(t *testing.T) {
490496

491497
assert.Len(t, listResp.Nodes, 2)
492498
})
499+
500+
t.Run("filter nodes by label", func(t *testing.T) {
501+
t.Parallel()
502+
503+
listResp, err := client.ListNodes(ctx, &nodev1.ListNodesRequest{
504+
Filter: &nodev1.ListNodesRequest_Filter{
505+
Selectors: []*ptypes.Selector{
506+
{
507+
Key: "environment",
508+
Op: ptypes.SelectorOp_EQ,
509+
Value: pointer.To("prod"),
510+
},
511+
},
512+
},
513+
})
514+
require.NoError(t, err)
515+
require.NotNil(t, listResp)
516+
assert.Len(t, listResp.Nodes, 1)
517+
assert.Equal(t, "Node 1", listResp.Nodes[0].Name)
518+
})
493519
}
494520

495521
func TestMemoryJobDistributor_UpdateNode(t *testing.T) {

offchain/jd/memory/node_filter.go

Lines changed: 156 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,156 @@
1+
package memory
2+
3+
import (
4+
"slices"
5+
"strings"
6+
7+
nodev1 "github.com/smartcontractkit/chainlink-protos/job-distributor/v1/node"
8+
"github.com/smartcontractkit/chainlink-protos/job-distributor/v1/shared/ptypes"
9+
)
10+
11+
// applyNodeFilter applies the filter to the list of nodes and returns the filtered results.
12+
func applyNodeFilter(
13+
nodes []*nodev1.Node, filter *nodev1.ListNodesRequest_Filter,
14+
) []*nodev1.Node {
15+
var filtered []*nodev1.Node
16+
17+
for _, node := range nodes {
18+
if nodeMatchesFilter(node, filter) {
19+
filtered = append(filtered, node)
20+
}
21+
}
22+
23+
return filtered
24+
}
25+
26+
// nodeMatchesFilter checks if a node matches the given filter criteria.
27+
func nodeMatchesFilter(node *nodev1.Node, filter *nodev1.ListNodesRequest_Filter) bool {
28+
// Check ids
29+
if len(filter.Ids) > 0 {
30+
if !nodeMatchesIds(node, filter.Ids) {
31+
return false
32+
}
33+
}
34+
35+
// Check enabled state
36+
if !nodeMatchesEnabledState(node, filter.Enabled) {
37+
return false
38+
}
39+
40+
// Check selectors
41+
if len(filter.Selectors) > 0 {
42+
for _, selector := range filter.Selectors {
43+
if !nodeMatchesSelector(node, selector) {
44+
return false
45+
}
46+
}
47+
}
48+
49+
return true
50+
}
51+
52+
// nodeMatchesIds checks if a node's ID is in the provided list of IDs.
53+
func nodeMatchesIds(node *nodev1.Node, ids []string) bool {
54+
return slices.Contains(ids, node.Id)
55+
}
56+
57+
// nodeMatchesEnabledState checks if a node matches the enabled state filter.
58+
// ENABLE_STATE_ENABLED: filter for enabled nodes only
59+
// ENABLE_STATE_DISABLED: filter for disabled nodes only
60+
// ENABLE_STATE_UNSPECIFIED: no filtering (default behavior when not set)
61+
func nodeMatchesEnabledState(node *nodev1.Node, enabled nodev1.EnableState) bool {
62+
switch enabled {
63+
case nodev1.EnableState_ENABLE_STATE_ENABLED:
64+
return node.IsEnabled
65+
case nodev1.EnableState_ENABLE_STATE_DISABLED:
66+
return !node.IsEnabled
67+
case nodev1.EnableState_ENABLE_STATE_UNSPECIFIED:
68+
// No filtering by enabled status
69+
return true
70+
default:
71+
// Unknown state, default to true (no filtering)
72+
return true
73+
}
74+
}
75+
76+
// nodeMatchesSelector checks if a node matches a specific selector.
77+
func nodeMatchesSelector(node *nodev1.Node, selector *ptypes.Selector) bool {
78+
// Get the node's labels as a map for easier lookup
79+
nodeLabels := make(map[string]string)
80+
for _, label := range node.Labels {
81+
if label.Value != nil {
82+
nodeLabels[label.Key] = *label.Value
83+
}
84+
}
85+
86+
// Check if the selector key exists in the node's labels
87+
nodeValue, hasKey := nodeLabels[selector.Key]
88+
89+
switch selector.Op {
90+
case ptypes.SelectorOp_EQ:
91+
// Equality check
92+
if selector.Value == nil {
93+
return false
94+
}
95+
96+
return hasKey && nodeValue == *selector.Value
97+
98+
case ptypes.SelectorOp_NOT_EQ:
99+
// Not equal check
100+
if selector.Value == nil {
101+
return false
102+
}
103+
104+
return hasKey && nodeValue != *selector.Value
105+
106+
case ptypes.SelectorOp_IN:
107+
// IN operation - check if node value is in the selector values
108+
if selector.Value == nil {
109+
return false
110+
}
111+
if !hasKey {
112+
return false
113+
}
114+
115+
// Parse comma-separated values
116+
values := strings.Split(*selector.Value, ",")
117+
for _, value := range values {
118+
if strings.TrimSpace(value) == nodeValue {
119+
return true
120+
}
121+
}
122+
123+
return false
124+
125+
case ptypes.SelectorOp_NOT_IN:
126+
// NOT IN operation - check if node value is not in the selector values
127+
if selector.Value == nil {
128+
return false
129+
}
130+
if !hasKey {
131+
return true // Key doesn't exist, so it's not in the list
132+
}
133+
134+
// Parse comma-separated values
135+
values := strings.Split(*selector.Value, ",")
136+
for _, value := range values {
137+
if strings.TrimSpace(value) == nodeValue {
138+
return false // Found in the list, so NOT_IN is false
139+
}
140+
}
141+
142+
return true // Not found in the list, so NOT_IN is true
143+
144+
case ptypes.SelectorOp_EXIST:
145+
// Check if the key exists (regardless of value)
146+
return hasKey
147+
148+
case ptypes.SelectorOp_NOT_EXIST:
149+
// Check if the key does not exist
150+
return !hasKey
151+
152+
default:
153+
// Unknown operation, default to false
154+
return false
155+
}
156+
}

0 commit comments

Comments
 (0)