metropolis/node/core/cluster/manager.go

// Copyright 2020 The Monogon Project Authors.
//
// SPDX-License-Identifier: Apache-2.0
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package cluster

import (
	"context"
	"crypto/ed25519"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"io/ioutil"
	"sync"

	"google.golang.org/protobuf/proto"

	"source.monogon.dev/metropolis/node/core/consensus"
	"source.monogon.dev/metropolis/node/core/localstorage"
	"source.monogon.dev/metropolis/node/core/network"
	"source.monogon.dev/metropolis/pkg/pki"
	"source.monogon.dev/metropolis/pkg/supervisor"
	apb "source.monogon.dev/metropolis/proto/api"
	ppb "source.monogon.dev/metropolis/proto/private"
)

type managerResult struct {
	node *Node
	err  error
}

type state struct {
	mu sync.RWMutex

	oneway       bool
	stateCluster ClusterState
	stateNode    ppb.Node_FSMState

	configuration *ppb.SealedConfiguration

	result  *managerResult
	waiters []chan *managerResult
}

func (s *state) setResult(node *Node, err error) {
	s.result = &managerResult{
		node: node,
		err:  err,
	}
	for _, w := range s.waiters {
		go func(c chan *managerResult) {
			c <- s.result
		}(w)
	}
	s.waiters = nil
}

type Manager struct {
	storageRoot    *localstorage.Root
	networkService *network.Service

	state

	// consensus is the spawned etcd/consensus service, if the Manager brought
	// up a Node that should run one.
	consensus *consensus.Service
}

// NewManager creates a new cluster Manager. The given localstorage Root must
// be places, but not yet started (and will be started as the Manager makes
// progress). The given network Service must already be running.
func NewManager(storageRoot *localstorage.Root, networkService *network.Service) *Manager {
	return &Manager{
		storageRoot:    storageRoot,
		networkService: networkService,

		state: state{
			stateCluster: ClusterUnknown,
			stateNode:    ppb.Node_FSM_STATE_INVALID,
		},
	}
}

func (m *Manager) lock() (*state, func()) {
	m.mu.Lock()
	return &m.state, m.mu.Unlock
}

func (m *Manager) rlock() (*state, func()) {
	m.mu.RLock()
	return &m.state, m.mu.RUnlock
}

func (m *Manager) Wait() (*Node, error) {
	state, unlock := m.lock()

	if state.result != nil {
		unlock()
		return state.result.node, state.result.err
	}

	c := make(chan *managerResult)
	state.waiters = append(state.waiters, c)
	unlock()
	res := <-c
	return res.node, res.err
}

// Run is the runnable of the Manager, to be started using the Supervisor. It
// is one-shot, and should not be restarted.
func (m *Manager) Run(ctx context.Context) error {
	state, unlock := m.lock()
	if state.oneway {
		unlock()
		// TODO(q3k): restart the entire system if this happens
		return fmt.Errorf("cannot restart cluster manager")
	}
	state.oneway = true
	unlock()

	configuration, err := m.storageRoot.ESP.SealedConfiguration.Unseal()
	if err == nil {
		supervisor.Logger(ctx).Info("Sealed configuration present. attempting to join cluster")
		return m.join(ctx, configuration)
	}

	if !errors.Is(err, localstorage.ErrNoSealed) {
		return fmt.Errorf("unexpected sealed config error: %w", err)
	}

	supervisor.Logger(ctx).Info("No sealed configuration, looking for node parameters")

	params, err := m.nodeParams(ctx)
	if err != nil {
		return fmt.Errorf("no parameters available: %w", err)
	}

	switch inner := params.Cluster.(type) {
	case *apb.NodeParameters_ClusterBootstrap_:
		return m.bootstrap(ctx, inner.ClusterBootstrap)
	case *apb.NodeParameters_ClusterRegister_:
		return m.register(ctx, inner.ClusterRegister)
	default:
		return fmt.Errorf("node parameters misconfigured: neither cluster_bootstrap nor cluster_register set")
	}
}

func (m *Manager) bootstrap(ctx context.Context, bootstrap *apb.NodeParameters_ClusterBootstrap) error {
	supervisor.Logger(ctx).Infof("Bootstrapping new cluster, owner public key: %s", hex.EncodeToString(bootstrap.OwnerPublicKey))
	state, unlock := m.lock()
	defer unlock()

	state.configuration = &ppb.SealedConfiguration{}

	// Mount new storage with generated CUK, and save LUK into sealed config proto.
	supervisor.Logger(ctx).Infof("Bootstrapping: 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("Bootstrapping: node public key: %s", hex.EncodeToString([]byte(pub)))

	node := Node{
		clusterUnlockKey: cuk,
		pubkey:           pub,
		state:            ppb.Node_FSM_STATE_UP,
		// TODO(q3k): make this configurable.
		consensusMember:  &NodeRoleConsensusMember{},
		kubernetesWorker: &NodeRoleKubernetesWorker{},
	}

	// Run worker to keep updating /ephemeral/hosts (and thus, /etc/hosts) with
	// our own IP address. This ensures that the node's ID always resolves to
	// its current external IP address.
	supervisor.Run(ctx, "hostsfile", func(ctx context.Context) error {
		supervisor.Signal(ctx, supervisor.SignalHealthy)
		watcher := m.networkService.Watch()
		for {
			status, err := watcher.Get(ctx)
			if err != nil {
				return err
			}
			err = node.ConfigureLocalHostname(ctx, &m.storageRoot.Ephemeral, status.ExternalAddress)
			if err != nil {
				return fmt.Errorf("could not configure hostname: %w", err)
			}
		}
	})

	// Bring up consensus with this node as the only member.
	m.consensus = consensus.New(consensus.Config{
		Data:       &m.storageRoot.Data.Etcd,
		Ephemeral:  &m.storageRoot.Ephemeral.Consensus,
		NewCluster: true,
		Name:       node.ID(),
	})

	supervisor.Logger(ctx).Infof("Bootstrapping: starting consensus...")
	if err := supervisor.Run(ctx, "consensus", m.consensus.Run); err != nil {
		return fmt.Errorf("when starting consensus: %w", err)
	}

	supervisor.Logger(ctx).Info("Bootstrapping: waiting for consensus...")
	if err := m.consensus.WaitReady(ctx); err != nil {
		return fmt.Errorf("consensus service failed to become ready: %w", err)
	}
	supervisor.Logger(ctx).Info("Bootstrapping: consensus ready.")

	kv := m.consensus.KVRoot()
	node.KV = kv

	// Create Metropolis CA and this node's certificate.
	caCertBytes, _, err := PKICA.Ensure(ctx, kv)
	if err != nil {
		return fmt.Errorf("failed to create cluster CA: %w", err)
	}
	nodeCert := PKINamespace.New(PKICA, "", pki.Server([]string{node.ID()}, nil))
	nodeCert.UseExistingKey(priv)
	nodeCertBytes, _, err := nodeCert.Ensure(ctx, kv)
	if err != nil {
		return fmt.Errorf("failed to create node certificate: %w", err)
	}

	if err := m.storageRoot.Data.Node.Credentials.CACertificate.Write(caCertBytes, 0400); err != nil {
		return fmt.Errorf("failed to write CA certificate: %w", err)
	}
	if err := m.storageRoot.Data.Node.Credentials.Certificate.Write(nodeCertBytes, 0400); err != nil {
		return fmt.Errorf("failed to write node certificate: %w", err)
	}
	if err := m.storageRoot.Data.Node.Credentials.Key.Write(priv, 0400); err != nil {
		return fmt.Errorf("failed to write node private key: %w", err)
	}

	// Update our Node obejct in etcd.
	if err := node.Store(ctx, kv); err != nil {
		return fmt.Errorf("failed to store new node in etcd: %w", err)
	}

	state.setResult(&node, nil)

	supervisor.Signal(ctx, supervisor.SignalHealthy)
	supervisor.Signal(ctx, supervisor.SignalDone)
	return nil
}

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 := ioutil.ReadFile("/sys/firmware/qemu_fw_cfg/by_name/dev.monogon.metropolis/parameters.pb/raw")
	if err != nil {
		return nil, fmt.Errorf("could not read firmware enrolment file: %w", err)
	}

	config := apb.NodeParameters{}
	err = proto.Unmarshal(bytes, &config)
	if err != nil {
		return nil, fmt.Errorf("could not unmarshal: %v", err)
	}

	return &config, nil
}

func (m *Manager) nodeParams(ctx context.Context) (*apb.NodeParameters, error) {
	// Retrieve node parameters from qemu's fwcfg interface or ESP.
	// TODO(q3k): probably abstract this away and implement per platform/build/...
	paramsFWCFG, err := m.nodeParamsFWCFG(ctx)
	if err != nil {
		supervisor.Logger(ctx).Warningf("Could not retrieve node parameters from qemu fwcfg: %v", err)
		paramsFWCFG = nil
	} else {
		supervisor.Logger(ctx).Infof("Retrieved node parameters from qemu fwcfg")
	}
	paramsESP, err := m.storageRoot.ESP.NodeParameters.Unmarshal()
	if err != nil {
		supervisor.Logger(ctx).Warningf("Could not retrieve node parameters from ESP: %v", err)
		paramsESP = nil
	} else {
		supervisor.Logger(ctx).Infof("Retrieved node parameters from ESP")
	}
	if paramsFWCFG == nil && paramsESP == nil {
		return nil, fmt.Errorf("could not find node parameters in ESP or qemu fwcfg")
	}
	if paramsFWCFG != nil && paramsESP != nil {
		supervisor.Logger(ctx).Warningf("Node parameters found both inboth ESP and qemu fwcfg, using the latter")
		return paramsFWCFG, nil
	} else if paramsFWCFG != nil {
		return paramsFWCFG, nil
	} else {
		return paramsESP, nil
	}
}

func (m *Manager) join(ctx context.Context, cfg *ppb.SealedConfiguration) error {
	return fmt.Errorf("unimplemented")
}

// Node returns the Node that the Manager brought into a cluster, or nil if the
// Manager is not Running.  This is safe to call from any goroutine.
func (m *Manager) Node() *Node {
	return nil
}