m/n/c/cluster: implement register flow
Change-Id: I197cbfa96d34c9912c7fc19710db25276e7440fc
Reviewed-on: https://review.monogon.dev/c/monogon/+/454
Reviewed-by: Lorenz Brun <lorenz@monogon.tech>
diff --git a/metropolis/node/core/cluster/BUILD.bazel b/metropolis/node/core/cluster/BUILD.bazel
index 1d55e8d..45c1164 100644
--- a/metropolis/node/core/cluster/BUILD.bazel
+++ b/metropolis/node/core/cluster/BUILD.bazel
@@ -5,6 +5,7 @@
srcs = [
"cluster.go",
"cluster_bootstrap.go",
+ "cluster_register.go",
"platform.go",
"status.go",
"watcher.go",
@@ -12,13 +13,16 @@
importpath = "source.monogon.dev/metropolis/node/core/cluster",
visibility = ["//metropolis/node/core:__subpackages__"],
deps = [
+ "//metropolis/node:go_default_library",
"//metropolis/node/core/consensus:go_default_library",
"//metropolis/node/core/consensus/client:go_default_library",
"//metropolis/node/core/curator:go_default_library",
+ "//metropolis/node/core/curator/proto/api:go_default_library",
"//metropolis/node/core/identity:go_default_library",
"//metropolis/node/core/localstorage:go_default_library",
"//metropolis/node/core/network:go_default_library",
"//metropolis/node/core/network/hostsfile:go_default_library",
+ "//metropolis/node/core/rpc:go_default_library",
"//metropolis/pkg/event:go_default_library",
"//metropolis/pkg/event/memory:go_default_library",
"//metropolis/pkg/supervisor:go_default_library",
diff --git a/metropolis/node/core/cluster/cluster.go b/metropolis/node/core/cluster/cluster.go
index 3ff1ad4..144a47b 100644
--- a/metropolis/node/core/cluster/cluster.go
+++ b/metropolis/node/core/cluster/cluster.go
@@ -128,10 +128,6 @@
}
}
-func (m *Manager) register(ctx context.Context, bootstrap *apb.NodeParameters_ClusterRegister) error {
- return fmt.Errorf("unimplemented")
-}
-
func (m *Manager) nodeParamsFWCFG(ctx context.Context) (*apb.NodeParameters, error) {
bytes, err := os.ReadFile("/sys/firmware/qemu_fw_cfg/by_name/dev.monogon.metropolis/parameters.pb/raw")
if err != nil {
diff --git a/metropolis/node/core/cluster/cluster_bootstrap.go b/metropolis/node/core/cluster/cluster_bootstrap.go
index 4e1c05b..b9b7ef2 100644
--- a/metropolis/node/core/cluster/cluster_bootstrap.go
+++ b/metropolis/node/core/cluster/cluster_bootstrap.go
@@ -99,7 +99,7 @@
status := Status{
State: cpb.ClusterState_CLUSTER_STATE_HOME,
- hasLocalConsensus: true,
+ HasLocalConsensus: true,
consensusClient: metropolisKV,
// Credentials are set further down once created through a curator
// short-circuit bootstrap function.
diff --git a/metropolis/node/core/cluster/cluster_register.go b/metropolis/node/core/cluster/cluster_register.go
new file mode 100644
index 0000000..81db8c9
--- /dev/null
+++ b/metropolis/node/core/cluster/cluster_register.go
@@ -0,0 +1,144 @@
+package cluster
+
+import (
+ "context"
+ "crypto/ed25519"
+ "crypto/rand"
+ "crypto/x509"
+ "encoding/hex"
+ "fmt"
+ "net"
+ "strconv"
+ "strings"
+ "time"
+
+ "source.monogon.dev/metropolis/node"
+ ipb "source.monogon.dev/metropolis/node/core/curator/proto/api"
+ "source.monogon.dev/metropolis/node/core/identity"
+ "source.monogon.dev/metropolis/node/core/rpc"
+ "source.monogon.dev/metropolis/pkg/supervisor"
+ apb "source.monogon.dev/metropolis/proto/api"
+ cpb "source.monogon.dev/metropolis/proto/common"
+ ppb "source.monogon.dev/metropolis/proto/private"
+)
+
+// register performs the registration flow of the node in the cluster, ie. makes
+// this node part of an existing cluster.
+//
+// This is a temporary implementation that's not well hardened against
+// transitive failures, but is good enough to get us off the ground and able to
+// test multiple nodes in a cluster. It does notably not run either
+// etcd/consensus or the curator, which also prevents Kubernetes from running.
+func (m *Manager) register(ctx context.Context, register *apb.NodeParameters_ClusterRegister) error {
+ // Do a validation pass on the provided NodeParameters.Register data, fail early
+ // if it looks invalid.
+ ca, err := x509.ParseCertificate(register.CaCertificate)
+ if err != nil {
+ return fmt.Errorf("NodeParameters.Register invalid: CaCertificate could not parsed: %w", err)
+ }
+ if err := identity.VerifyCAInsecure(ca); err != nil {
+ return fmt.Errorf("NodeParameters.Register invalid: CaCertificate invalid: %w", err)
+ }
+ if len(register.RegisterTicket) == 0 {
+ return fmt.Errorf("NodeParameters.Register invalid: RegisterTicket not set")
+ }
+ if register.ClusterDirectory == nil || len(register.ClusterDirectory.Nodes) == 0 {
+ return fmt.Errorf("NodeParameters.ClusterDirectory invalid: must contain at least one node")
+ }
+ for i, node := range register.ClusterDirectory.Nodes {
+ if len(node.Addresses) == 0 {
+ return fmt.Errorf("NodeParameters.ClusterDirectory.Nodes[%d] invalid: must have at least one address", i)
+ }
+ for j, add := range node.Addresses {
+ if add.Host == "" || net.ParseIP(add.Host) == nil {
+ return fmt.Errorf("NodeParameters.ClusterDirectory.Nodes[%d].Addresses[%d] invalid: not a parseable address", i, j)
+ }
+ }
+ if len(node.PublicKey) != ed25519.PublicKeySize {
+ return fmt.Errorf("NodeParameters.ClusterDirectory.Nodes[%d] invalid: PublicKey invalid length or not set", i)
+ }
+ }
+
+ // Validation passed, let's take the state lock and start working on registering
+ // us into the cluster.
+
+ state, unlock := m.lock()
+ defer unlock()
+
+ // Tell the user what we're doing.
+ supervisor.Logger(ctx).Infof("Registering into existing cluster.")
+ supervisor.Logger(ctx).Infof(" Cluster CA public key: %s", hex.EncodeToString(ca.PublicKey.(ed25519.PublicKey)))
+ supervisor.Logger(ctx).Infof(" Register Ticket: %s", hex.EncodeToString(register.RegisterTicket))
+ supervisor.Logger(ctx).Infof(" Directory:")
+ for _, node := range register.ClusterDirectory.Nodes {
+ id := identity.NodeID(node.PublicKey)
+ var addresses []string
+ for _, add := range node.Addresses {
+ addresses = append(addresses, add.Host)
+ }
+ supervisor.Logger(ctx).Infof(" Node ID: %s, Addresses: %s", id, strings.Join(addresses, ","))
+ }
+
+ // Mount new storage with generated CUK, and save LUK into sealed config proto.
+ state.configuration = &ppb.SealedConfiguration{}
+ supervisor.Logger(ctx).Infof("Registering: mounting new storage...")
+ cuk, err := m.storageRoot.Data.MountNew(state.configuration)
+ if err != nil {
+ return fmt.Errorf("could not make and mount data partition: %w", err)
+ }
+
+ pub, priv, err := ed25519.GenerateKey(rand.Reader)
+ if err != nil {
+ return fmt.Errorf("could not generate node keypair: %w", err)
+ }
+ supervisor.Logger(ctx).Infof("Registering: node public key: %s", hex.EncodeToString([]byte(pub)))
+
+ // Attempt to connect to first node in cluster directory and to call Register.
+ //
+ // MVP: this should be properly client-side loadbalanced.
+ remote := register.ClusterDirectory.Nodes[0].Addresses[0].Host
+ remote = net.JoinHostPort(remote, strconv.Itoa(int(node.CuratorServicePort)))
+ eph, err := rpc.NewEphemeralClient(remote, priv, ca)
+ if err != nil {
+ return fmt.Errorf("could not create ephemeral client to %q: %w", remote, err)
+ }
+ cur := ipb.NewCuratorClient(eph)
+ _, err = cur.RegisterNode(ctx, &ipb.RegisterNodeRequest{
+ RegisterTicket: register.RegisterTicket,
+ })
+ if err != nil {
+ return fmt.Errorf("register call failed: %w", err)
+ }
+
+ // Attempt to commit in a loop. This will succeed once the node is approved.
+ supervisor.Logger(ctx).Infof("Registering: success, attempting to commit...")
+ var certBytes, caCertBytes []byte
+ for {
+ resC, err := cur.CommitNode(ctx, &ipb.CommitNodeRequest{
+ ClusterUnlockKey: cuk,
+ })
+ if err == nil {
+ supervisor.Logger(ctx).Infof("Registering: Commit succesfull, received certificate")
+ certBytes = resC.NodeCertificate
+ caCertBytes = resC.CaCertificate
+ break
+ }
+ supervisor.Logger(ctx).Infof("Registering: Commit failed, retrying: %v", err)
+ time.Sleep(time.Second)
+ }
+
+ // Node is now UP, build client and report it to downstream code.
+ creds, err := identity.NewNodeCredentials(priv, certBytes, caCertBytes)
+ if err != nil {
+ return fmt.Errorf("NewNodeCredentials failed after receiving certificate from cluster: %w", err)
+ }
+ status := Status{
+ State: cpb.ClusterState_CLUSTER_STATE_HOME,
+ HasLocalConsensus: false,
+ Credentials: creds,
+ }
+ m.status.Set(status)
+ supervisor.Signal(ctx, supervisor.SignalHealthy)
+ supervisor.Signal(ctx, supervisor.SignalDone)
+ return nil
+}
diff --git a/metropolis/node/core/cluster/status.go b/metropolis/node/core/cluster/status.go
index 3dbfb56..481c818 100644
--- a/metropolis/node/core/cluster/status.go
+++ b/metropolis/node/core/cluster/status.go
@@ -20,7 +20,7 @@
// hasLocalConsensus is true if the local node is running a local consensus
// (etcd) server.
- hasLocalConsensus bool
+ HasLocalConsensus bool
// consensusClient is an etcd client to the local consensus server if the node
// has such a server and the cluster state is HOME or SPLIT.
consensusClient client.Namespaced
@@ -43,7 +43,7 @@
// ConsensusClient returns an etcd/consensus client for a given ConsensusUser.
// The node must be running a local consensus/etcd server.
func (s *Status) ConsensusClient(user ConsensusUser) (client.Namespaced, error) {
- if !s.hasLocalConsensus {
+ if !s.HasLocalConsensus {
return nil, ErrNoLocalConsensus
}
diff --git a/metropolis/node/core/debug_service.go b/metropolis/node/core/debug_service.go
index 48a4c9b..f253d0e 100644
--- a/metropolis/node/core/debug_service.go
+++ b/metropolis/node/core/debug_service.go
@@ -53,6 +53,9 @@
}
func (s *debugService) GetDebugKubeconfig(ctx context.Context, req *apb.GetDebugKubeconfigRequest) (*apb.GetDebugKubeconfigResponse, error) {
+ if s.roleserve == nil {
+ return nil, status.Errorf(codes.Unavailable, "node does not run roleserver/kubernetes")
+ }
w := s.roleserve.Watch()
defer w.Close()
for {
diff --git a/metropolis/node/core/identity/certificates.go b/metropolis/node/core/identity/certificates.go
index 95b7e0d..c15f913 100644
--- a/metropolis/node/core/identity/certificates.go
+++ b/metropolis/node/core/identity/certificates.go
@@ -61,15 +61,39 @@
}
}
+// VerifyCAInsecure ensures that the given certificate is a valid certificate
+// that is allowed to act as a CA and which is emitted for an Ed25519 keypair.
+//
+// It does _not_ ensure that the certificate is the local node's CA, and should
+// not be used for security checks, just for data validation checks.
+func VerifyCAInsecure(ca *x509.Certificate) error {
+ // Ensure ca certificate uses ED25519 keypair.
+ if _, ok := ca.PublicKey.(ed25519.PublicKey); !ok {
+ return fmt.Errorf("not issued for ed25519 keypair")
+ }
+ // Ensure CA certificate has the X.509 basic constraints extension. Everything
+ // else is legacy, we might as well weed that out early.
+ if !ca.BasicConstraintsValid {
+ return fmt.Errorf("does not have basic constraints")
+ }
+ // Ensure CA certificate can act as CA per BasicConstraints.
+ if !ca.IsCA {
+ return fmt.Errorf("not permitted to act as CA")
+ }
+ if ca.KeyUsage != 0 && ca.KeyUsage&x509.KeyUsageCertSign == 0 {
+ return fmt.Errorf("not permitted to sign certificates")
+ }
+ return nil
+}
+
// VerifyInCluster ensures that the given certificate has been signed by a CA
// certificate and are both certificates emitted for ed25519 keypairs.
//
// The subject certificate's public key is returned if verification is
// successful, and error is returned otherwise.
func VerifyInCluster(cert, ca *x509.Certificate) (ed25519.PublicKey, error) {
- // Ensure ca certificate uses ED25519 keypair.
- if _, ok := ca.PublicKey.(ed25519.PublicKey); !ok {
- return nil, fmt.Errorf("ca certificate not issued for ed25519 keypair")
+ if err := VerifyCAInsecure(ca); err != nil {
+ return nil, fmt.Errorf("ca certificate invalid: %w", err)
}
// Ensure subject cert is signed by ca.
diff --git a/metropolis/node/core/main.go b/metropolis/node/core/main.go
index 5d495fa..0a63d48 100644
--- a/metropolis/node/core/main.go
+++ b/metropolis/node/core/main.go
@@ -144,71 +144,73 @@
return fmt.Errorf("new couldn't find home in new cluster, aborting: %w", err)
}
- // Here starts some hairy stopgap code. In the future, not all nodes will have
- // direct access to etcd (ie. the ability to retrieve an etcd client via
- // status.ConsensusClient).
- // However, we are not ready to implement this yet, as that would require
- // moving more logic into the curator (eg. some of the Kubernetes PKI logic).
+ // Currently, only the first node of the cluster runs etcd. That means only that
+ // node can run the curator, kubernetes and roleserver.
//
- // For now, we keep Kubernetes PKI initialization logic here, and just assume
- // that every node will have direct access to etcd.
+ // This is a temporary stopgap until we land a roleserver rewrite which fixes
+ // this.
- // Retrieve namespaced etcd KV clients for the two main direct etcd users:
- // - Curator
- // - Kubernetes PKI
- ckv, err := status.ConsensusClient(cluster.ConsensusUserCurator)
- if err != nil {
- close(trapdoor)
- return fmt.Errorf("failed to retrieve consensus curator client: %w", err)
- }
- kkv, err := status.ConsensusClient(cluster.ConsensusUserKubernetesPKI)
- if err != nil {
- close(trapdoor)
- return fmt.Errorf("failed to retrieve consensus kubernetes PKI client: %w", err)
- }
+ var rs *roleserve.Service
+ if status.HasLocalConsensus {
+ // Retrieve namespaced etcd KV clients for the two main direct etcd users:
+ // - Curator
+ // - Kubernetes PKI
+ ckv, err := status.ConsensusClient(cluster.ConsensusUserCurator)
+ if err != nil {
+ close(trapdoor)
+ return fmt.Errorf("failed to retrieve consensus curator client: %w", err)
+ }
+ kkv, err := status.ConsensusClient(cluster.ConsensusUserKubernetesPKI)
+ if err != nil {
+ close(trapdoor)
+ return fmt.Errorf("failed to retrieve consensus kubernetes PKI client: %w", err)
+ }
- // TODO(q3k): restart curator on credentials change?
+ // TODO(q3k): restart curator on credentials change?
- // Start cluster curator. The cluster curator is responsible for lifecycle
- // management of the cluster.
- // In the future, this will only be started on nodes that run etcd.
- c := curator.New(curator.Config{
- Etcd: ckv,
- NodeCredentials: status.Credentials,
- // TODO(q3k): make this configurable?
- LeaderTTL: time.Second * 5,
- Directory: &root.Ephemeral.Curator,
- })
- if err := supervisor.Run(ctx, "curator", c.Run); err != nil {
- close(trapdoor)
- return fmt.Errorf("when starting curator: %w", err)
- }
+ // Start cluster curator. The cluster curator is responsible for lifecycle
+ // management of the cluster.
+ // In the future, this will only be started on nodes that run etcd.
+ c := curator.New(curator.Config{
+ Etcd: ckv,
+ NodeCredentials: status.Credentials,
+ // TODO(q3k): make this configurable?
+ LeaderTTL: time.Second * 5,
+ Directory: &root.Ephemeral.Curator,
+ })
+ if err := supervisor.Run(ctx, "curator", c.Run); err != nil {
+ close(trapdoor)
+ return fmt.Errorf("when starting curator: %w", err)
+ }
- // We are now in a cluster. We can thus access our 'node' object and
- // start all services that we should be running.
- logger.Info("Enrolment success, continuing startup.")
+ // We are now in a cluster. We can thus access our 'node' object and
+ // start all services that we should be running.
+ logger.Info("Enrolment success, continuing startup.")
- // Ensure Kubernetes PKI objects exist in etcd. In the future, this logic will
- // be implemented in the curator.
- kpki := pki.New(lt.MustLeveledFor("pki.kubernetes"), kkv)
- if err := kpki.EnsureAll(ctx); err != nil {
- close(trapdoor)
- return fmt.Errorf("failed to ensure kubernetes PKI present: %w", err)
- }
+ // Ensure Kubernetes PKI objects exist in etcd. In the future, this logic will
+ // be implemented in the curator.
+ kpki := pki.New(lt.MustLeveledFor("pki.kubernetes"), kkv)
+ if err := kpki.EnsureAll(ctx); err != nil {
+ close(trapdoor)
+ return fmt.Errorf("failed to ensure kubernetes PKI present: %w", err)
+ }
- // Start the role service. The role service connects to the curator and runs
- // all node-specific role code (eg. Kubernetes services).
- // supervisor.Logger(ctx).Infof("Starting role service...")
- rs := roleserve.New(roleserve.Config{
- CuratorDial: c.DialCluster,
- StorageRoot: root,
- Network: networkSvc,
- KPKI: kpki,
- NodeID: status.Credentials.ID(),
- })
- if err := supervisor.Run(ctx, "role", rs.Run); err != nil {
- close(trapdoor)
- return fmt.Errorf("failed to start role service: %w", err)
+ // Start the role service. The role service connects to the curator and runs
+ // all node-specific role code (eg. Kubernetes services).
+ // supervisor.Logger(ctx).Infof("Starting role service...")
+ rs = roleserve.New(roleserve.Config{
+ CuratorDial: c.DialCluster,
+ StorageRoot: root,
+ Network: networkSvc,
+ KPKI: kpki,
+ NodeID: status.Credentials.ID(),
+ })
+ if err := supervisor.Run(ctx, "role", rs.Run); err != nil {
+ close(trapdoor)
+ return fmt.Errorf("failed to start role service: %w", err)
+ }
+ } else {
+ logger.Warningf("TODO(q3k): Dummy node - will not run Kubernetes/Curator/Roleserver...")
}
// Start the node debug service.
diff --git a/metropolis/test/e2e/main_test.go b/metropolis/test/e2e/main_test.go
index 4e1c2c5..72fa619 100644
--- a/metropolis/test/e2e/main_test.go
+++ b/metropolis/test/e2e/main_test.go
@@ -18,7 +18,6 @@
import (
"context"
- "crypto/ed25519"
"errors"
"fmt"
"log"
@@ -80,9 +79,10 @@
defer cancel()
// Launch cluster.
- cluster, err := cluster.LaunchCluster(ctx, cluster.ClusterOptions{
- NumNodes: 1,
- })
+ clusterOptions := cluster.ClusterOptions{
+ NumNodes: 2,
+ }
+ cluster, err := cluster.LaunchCluster(ctx, clusterOptions)
if err != nil {
t.Fatalf("LaunchCluster failed: %v", err)
}
@@ -106,22 +106,11 @@
return fmt.Errorf("GetClusterInfo: %w", err)
}
- // Ensure the node is there with its address.
+ // Ensure that the expected node count is present.
nodes := res.ClusterDirectory.Nodes
- if want, got := 1, len(nodes); want != got {
+ if want, got := clusterOptions.NumNodes, len(nodes); want != got {
return fmt.Errorf("wanted %d nodes in cluster directory, got %d", want, got)
}
- node := nodes[0]
- if want, got := ed25519.PublicKeySize, len(node.PublicKey); want != got {
- return fmt.Errorf("wanted %d bytes long public key, got %d", want, got)
- }
- if want, got := 1, len(node.Addresses); want != got {
- return fmt.Errorf("wanted %d node address, got %d", want, got)
- }
- if want, got := "10.1.0.2", node.Addresses[0].Host; want != got {
- return fmt.Errorf("wanted status address %q, got %q", want, got)
- }
-
return nil
})
})
diff --git a/metropolis/test/launch/cluster/BUILD.bazel b/metropolis/test/launch/cluster/BUILD.bazel
index 6c8d794..bf010d2 100644
--- a/metropolis/test/launch/cluster/BUILD.bazel
+++ b/metropolis/test/launch/cluster/BUILD.bazel
@@ -18,8 +18,10 @@
visibility = ["//visibility:public"],
deps = [
"//metropolis/node:go_default_library",
+ "//metropolis/node/core/identity:go_default_library",
"//metropolis/node/core/rpc:go_default_library",
"//metropolis/proto/api:go_default_library",
+ "//metropolis/proto/common:go_default_library",
"//metropolis/test/launch:go_default_library",
"@com_github_cenkalti_backoff_v4//:go_default_library",
"@com_github_grpc_ecosystem_go_grpc_middleware//retry:go_default_library",
diff --git a/metropolis/test/launch/cluster/cluster.go b/metropolis/test/launch/cluster/cluster.go
index 7af1f55..4d5a772 100644
--- a/metropolis/test/launch/cluster/cluster.go
+++ b/metropolis/test/launch/cluster/cluster.go
@@ -27,8 +27,10 @@
"google.golang.org/protobuf/proto"
"source.monogon.dev/metropolis/node"
+ "source.monogon.dev/metropolis/node/core/identity"
"source.monogon.dev/metropolis/node/core/rpc"
apb "source.monogon.dev/metropolis/proto/api"
+ cpb "source.monogon.dev/metropolis/proto/common"
"source.monogon.dev/metropolis/test/launch"
)
@@ -233,6 +235,27 @@
return out.Close()
}
+// getNodes wraps around Management.GetNodes to return a list of nodes in a
+// cluster.
+func getNodes(ctx context.Context, mgmt apb.ManagementClient) ([]*apb.Node, error) {
+ srvN, err := mgmt.GetNodes(ctx, &apb.GetNodesRequest{})
+ if err != nil {
+ return nil, fmt.Errorf("GetNodes: %w", err)
+ }
+ var res []*apb.Node
+ for {
+ node, err := srvN.Recv()
+ if err == io.EOF {
+ break
+ }
+ if err != nil {
+ return nil, fmt.Errorf("GetNodes.Recv: %w", err)
+ }
+ res = append(res, node)
+ }
+ return res, nil
+}
+
// Gets a random EUI-48 Ethernet MAC address
func generateRandomEthernetMAC() (*net.HardwareAddr, error) {
macBuf := make([]byte, 6)
@@ -294,12 +317,9 @@
// The given context will be used to run all qemu instances in the cluster, and
// canceling the context or calling Close() will terminate them.
func LaunchCluster(ctx context.Context, opts ClusterOptions) (*Cluster, error) {
- if opts.NumNodes == 0 {
+ if opts.NumNodes <= 0 {
return nil, errors.New("refusing to start cluster with zero nodes")
}
- if opts.NumNodes > 1 {
- return nil, errors.New("unimplemented")
- }
ctxT, ctxC := context.WithCancel(ctx)
@@ -316,14 +336,15 @@
vmPorts = append(vmPorts, vmPort)
}
- // Make a list of channels that will be populated and closed by all running node
- // qemu processes.
+ // Make a list of channels that will be populated by all running node qemu
+ // processes.
done := make([]chan error, opts.NumNodes)
for i, _ := range done {
done[i] = make(chan error, 1)
}
// Start first node.
+ log.Printf("Cluster: Starting node %d...", 1)
go func() {
err := LaunchNode(ctxT, NodeOptions{
ConnectToSocket: vmPorts[0],
@@ -338,6 +359,7 @@
done[0] <- err
}()
+ // Launch nanoswitch.
portMap, err := launch.ConflictFreePortMap(ClusterPorts)
if err != nil {
ctxC()
@@ -378,7 +400,7 @@
// Retrieve owner certificate - this can take a while because the node is still
// coming up, so do it in a backoff loop.
- log.Printf("Cluster: retrieving owner certificate...")
+ log.Printf("Cluster: retrieving owner certificate (this can take a few seconds while the first node boots)...")
aaa := apb.NewAAAClient(initClient)
var cert *tls.Certificate
err = backoff.Retry(func() error {
@@ -391,15 +413,112 @@
}
log.Printf("Cluster: retrieved owner certificate.")
+ // Build authenticated owner client to new node.
authClient, err := rpc.NewAuthenticatedClient(remote, *cert, nil)
if err != nil {
ctxC()
return nil, fmt.Errorf("NewAuthenticatedClient: %w", err)
}
+ mgmt := apb.NewManagementClient(authClient)
+
+ // Retrieve register ticket to register further nodes.
+ log.Printf("Cluster: retrieving register ticket...")
+ resT, err := mgmt.GetRegisterTicket(ctx, &apb.GetRegisterTicketRequest{})
+ if err != nil {
+ ctxC()
+ return nil, fmt.Errorf("GetRegisterTicket: %w", err)
+ }
+ ticket := resT.Ticket
+ log.Printf("Cluster: retrieved register ticket (%d bytes).", len(ticket))
+
+ // Retrieve cluster info (for directory and ca public key) to register further
+ // nodes.
+ resI, err := mgmt.GetClusterInfo(ctx, &apb.GetClusterInfoRequest{})
+ if err != nil {
+ ctxC()
+ return nil, fmt.Errorf("GetClusterInfo: %w", err)
+ }
+
+ // Now run the rest of the nodes.
+ //
+ // TODO(q3k): parallelize this
+ for i := 1; i < opts.NumNodes; i++ {
+ log.Printf("Cluster: Starting node %d...", i+1)
+ go func(i int) {
+ err := LaunchNode(ctxT, NodeOptions{
+ ConnectToSocket: vmPorts[i],
+ NodeParameters: &apb.NodeParameters{
+ Cluster: &apb.NodeParameters_ClusterRegister_{
+ ClusterRegister: &apb.NodeParameters_ClusterRegister{
+ RegisterTicket: ticket,
+ ClusterDirectory: resI.ClusterDirectory,
+ CaCertificate: resI.CaCertificate,
+ },
+ },
+ },
+ SerialPort: os.Stdout,
+ })
+ done[i] <- err
+ }(i)
+ var newNode *apb.Node
+
+ log.Printf("Cluster: waiting for node %d to appear as NEW...", i)
+ for {
+ nodes, err := getNodes(ctx, mgmt)
+ if err != nil {
+ ctxC()
+ return nil, fmt.Errorf("could not get nodes: %w", err)
+ }
+ for _, n := range nodes {
+ if n.State == cpb.NodeState_NODE_STATE_NEW {
+ newNode = n
+ break
+ }
+ }
+ if newNode != nil {
+ break
+ }
+ time.Sleep(1 * time.Second)
+ }
+ id := identity.NodeID(newNode.Pubkey)
+ log.Printf("Cluster: node %d is %s", i, id)
+
+ log.Printf("Cluster: approving node %d", i)
+ _, err := mgmt.ApproveNode(ctx, &apb.ApproveNodeRequest{
+ Pubkey: newNode.Pubkey,
+ })
+ if err != nil {
+ ctxC()
+ return nil, fmt.Errorf("ApproveNode(%s): %w", id, err)
+ }
+ log.Printf("Cluster: node %d approved, waiting for it to appear as UP...", i)
+ for {
+ nodes, err := getNodes(ctx, mgmt)
+ if err != nil {
+ ctxC()
+ return nil, fmt.Errorf("could not get nodes: %w", err)
+ }
+ found := false
+ for _, n := range nodes {
+ if !bytes.Equal(n.Pubkey, newNode.Pubkey) {
+ continue
+ }
+ if n.State == cpb.NodeState_NODE_STATE_UP {
+ found = true
+ break
+ }
+ time.Sleep(time.Second)
+ }
+ if found {
+ break
+ }
+ }
+ log.Printf("Cluster: node %d (%s) UP!", i, id)
+ }
return &Cluster{
Debug: apb.NewNodeDebugServiceClient(debugConn),
- Management: apb.NewManagementClient(authClient),
+ Management: mgmt,
Owner: *cert,
Ports: portMap,
nodesDone: done,