m/n/core/curator: fix clusternet sync issues
Clusternet sync was broken whenever a node just started a curator
watcher, as the curator's codepath to serve the backlog wasn't copying
over clusternet data.
This shouldn't have happened, especially as we implemented a unified
function to convert node data into node update data, we just forgot to
use it during the initial backlog generation code on the curator.
I've spent some time trying to come up with a testcase that would
automatically catch any further bug of this type, but that's not really
doable without having more formalized type casts between all the
different types a node can be encoded in (curator in-memory, curator
proto state, api node object). But we do still update one of the curator
tests to catch this particular regression.
Change-Id: I203d9a41b735db63d076c7e68a9fc6fe2f795ab4
Reviewed-on: https://review.monogon.dev/c/monogon/+/1912
Reviewed-by: Lorenz Brun <lorenz@monogon.tech>
Tested-by: Jenkins CI
diff --git a/metropolis/node/core/curator/BUILD.bazel b/metropolis/node/core/curator/BUILD.bazel
index 83ffbe6..ff7599f 100644
--- a/metropolis/node/core/curator/BUILD.bazel
+++ b/metropolis/node/core/curator/BUILD.bazel
@@ -77,12 +77,15 @@
"//metropolis/pkg/supervisor",
"//metropolis/proto/api",
"//metropolis/proto/common",
+ "@com_github_google_go_cmp//cmp",
"@io_etcd_go_etcd_client_v3//:client",
"@io_etcd_go_etcd_tests_v3//integration",
"@org_golang_google_grpc//:go_default_library",
"@org_golang_google_grpc//grpclog",
"@org_golang_google_grpc//test/bufconn",
"@org_golang_google_protobuf//proto",
+ "@org_golang_google_protobuf//testing/protocmp",
+ "@org_golang_google_protobuf//types/known/timestamppb",
"@org_uber_go_zap//:zap",
],
)
diff --git a/metropolis/node/core/curator/impl_leader_curator.go b/metropolis/node/core/curator/impl_leader_curator.go
index e3cc2cf..ff08453 100644
--- a/metropolis/node/core/curator/impl_leader_curator.go
+++ b/metropolis/node/core/curator/impl_leader_curator.go
@@ -126,11 +126,7 @@
// TODO(q3k): formalize message limits, set const somewhere.
we := &ipb.WatchEvent{}
for _, n := range nodes {
- we.Nodes = append(we.Nodes, &ipb.Node{
- Id: n.ID(),
- Roles: n.proto().Roles,
- Status: n.status,
- })
+ n.appendToEvent(we)
if proto.Size(we) > (2 << 20) {
if err := srv.Send(we); err != nil {
return err
@@ -193,17 +189,23 @@
return &res, nil
}
+// appendToId records a node state represented by Node into a Curator
+// WatchEvent.
+func (n *Node) appendToEvent(ev *ipb.WatchEvent) {
+ np := n.proto()
+ ev.Nodes = append(ev.Nodes, &ipb.Node{
+ Id: n.ID(),
+ Roles: np.Roles,
+ Status: np.Status,
+ Clusternet: np.Clusternet,
+ })
+}
+
// appendToId records a node update represented by nodeAtID into a Curator
// WatchEvent, either a Node or NodeTombstone.
func (kv nodeAtID) appendToEvent(ev *ipb.WatchEvent) {
if node := kv.value; node != nil {
- np := node.proto()
- ev.Nodes = append(ev.Nodes, &ipb.Node{
- Id: node.ID(),
- Roles: np.Roles,
- Status: np.Status,
- Clusternet: np.Clusternet,
- })
+ node.appendToEvent(ev)
} else {
ev.NodeTombstones = append(ev.NodeTombstones, &ipb.WatchEvent_NodeTombstone{
NodeId: kv.id,
diff --git a/metropolis/node/core/curator/impl_leader_test.go b/metropolis/node/core/curator/impl_leader_test.go
index 5490b5b..f013460 100644
--- a/metropolis/node/core/curator/impl_leader_test.go
+++ b/metropolis/node/core/curator/impl_leader_test.go
@@ -15,11 +15,14 @@
"testing"
"time"
+ "github.com/google/go-cmp/cmp"
"go.etcd.io/etcd/tests/v3/integration"
"go.uber.org/zap"
"google.golang.org/grpc"
"google.golang.org/grpc/test/bufconn"
"google.golang.org/protobuf/proto"
+ "google.golang.org/protobuf/testing/protocmp"
+ "google.golang.org/protobuf/types/known/timestamppb"
common "source.monogon.dev/metropolis/node"
"source.monogon.dev/metropolis/node/core/consensus"
@@ -534,11 +537,52 @@
}
fakeNodeID := identity.NodeID(fakeNodePub)
fakeNodeKey, _ := nodeEtcdPrefix.Key(fakeNodeID)
+
fakeNode := &ppb.Node{
PublicKey: fakeNodePub,
Roles: &cpb.NodeRoles{},
+ Status: &cpb.NodeStatus{
+ ExternalAddress: "1.2.3.4",
+ RunningCurator: &cpb.NodeStatus_RunningCurator{
+ Port: 1337,
+ },
+ Timestamp: timestamppb.New(time.Now()),
+ },
+ Clusternet: &cpb.NodeClusterNetworking{
+ WireguardPubkey: "deadbeef",
+ Prefixes: []*cpb.NodeClusterNetworking_Prefix{
+ {
+ Cidr: "10.0.0.0/8",
+ },
+ },
+ },
}
+
+ // Check a given node against fakeNode.
+ checkFakeNode := func(n *ipb.Node) bool {
+ t.Helper()
+ okay := true
+ if n.Id != fakeNodeID {
+ t.Errorf("Wanted faked node ID %q, got %q", fakeNodeID, n.Id)
+ okay = false
+ }
+ if diff := cmp.Diff(fakeNode.Status, n.Status, protocmp.Transform()); diff != "" {
+ t.Errorf("Status differs: %s", diff)
+ okay = false
+ }
+ if diff := cmp.Diff(fakeNode.Clusternet, n.Clusternet, protocmp.Transform()); diff != "" {
+ t.Errorf("Clusternet differs: %s", diff)
+ okay = false
+ }
+ if diff := cmp.Diff(fakeNode.Roles, n.Roles, protocmp.Transform()); diff != "" {
+ t.Errorf("Roles differs: %s", diff)
+ okay = false
+ }
+ return okay
+ }
+
fakeNodeInit, err := proto.Marshal(fakeNode)
+
if err != nil {
t.Fatalf("Marshal: %v", err)
}
@@ -553,9 +597,7 @@
if n == nil {
continue
}
- if n.Id != fakeNodeID {
- t.Errorf("Wanted faked node ID %q, got %q", fakeNodeID, n.Id)
- }
+ checkFakeNode(n)
break
}
@@ -581,6 +623,7 @@
if n := nodes[fakeNodeID]; n == nil {
t.Errorf("Node %q should exist, got %v", fakeNodeID, n)
}
+ checkFakeNode(nodes[fakeNodeID])
if len(nodes) != 2 {
t.Errorf("Exptected two nodes in map, got %d", len(nodes))
}