m/n/core: implement node heartbeats
This change introduces cluster member node health monitoring by
implementing a bidirectional RPC stream the nodes will periodically
send their heartbeat updates through. Management.GetNodes call was
modified to include the new node health information.
Relevant data available through the management API is non-persistent,
and stored within current Curator leader's local state. As such, it
will become briefly unavailable in an event of leader re-election. The
information returned, however, is guaranteed to be correct.
Change-Id: I916ac48f496941a7decc09d672ecf72a914b0d88
Reviewed-on: https://review.monogon.dev/c/monogon/+/694
Reviewed-by: Sergiusz Bazanski <serge@monogon.tech>
Tested-by: Jenkins CI
diff --git a/metropolis/test/e2e/main_test.go b/metropolis/test/e2e/main_test.go
index c0af1fd..89df286 100644
--- a/metropolis/test/e2e/main_test.go
+++ b/metropolis/test/e2e/main_test.go
@@ -20,6 +20,7 @@
"context"
"errors"
"fmt"
+ "io"
"log"
"net"
"net/http"
@@ -149,6 +150,40 @@
}
return nil
})
+ testEventual(t, "Heartbeat test successful", ctx, 60*time.Second, func(ctx context.Context) error {
+ // Ensure all cluster nodes are capable of sending heartbeat updates.
+ // This test assumes the expected count of nodes is already present in
+ // the cluster.
+ for {
+ srvN, err := mgmt.GetNodes(ctx, &apb.GetNodesRequest{})
+ if err != nil {
+ return fmt.Errorf("GetNodes: %w", err)
+ }
+
+ // Count the unhealthy nodes.
+ var unhealthy int
+ for {
+ node, err := srvN.Recv()
+ if err == io.EOF {
+ break
+ }
+ if err != nil {
+ return fmt.Errorf("GetNodes.Recv: %w", err)
+ }
+
+ if node.Health != apb.Node_HEALTHY {
+ unhealthy++
+ }
+ }
+
+ // If all nodes tested in this iteration are healthy, the test has
+ // been passed.
+ if unhealthy == 0 {
+ break
+ }
+ }
+ return nil
+ })
})
t.Run("Kubernetes", func(t *testing.T) {
t.Parallel()