metropolis/node/core/network/hostsfile/hostsfile.go - monogon - Gitiles

 // hostsfile implements a service which owns and writes all node-local
 // files/interfaces used by the system to resolve the local node's name and the
 // names of other nodes in the cluster:
 //
 //  1. All cluster node names are written into /etc/hosts for DNS resolution.
 //  2. The local node's name is written into /etc/machine-id.
 //  3. The local node's name is set as the UNIX hostname of the machine (via the
 //     sethostname call).
 //  4. The local node's ClusterDirectory is updated with the same set of
 //     addresses as the one used in /etc/hosts.
 //
 // The hostsfile Service can start up in two modes: with cluster connectivity
 // and without cluster connectivity. Without cluster connectivity, only
 // information about the current node (as retrieved from the network service)
 // will be used to populate local data. In cluster mode, information about other
 // nodes is also used.
 package hostsfile

 import (
 	"bytes"
 	"context"
 	"fmt"
 	"net"
 	"sort"

 	"golang.org/x/sys/unix"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/proto"

 	ipb "source.monogon.dev/metropolis/node/core/curator/proto/api"
 	"source.monogon.dev/metropolis/node/core/localstorage"
 	"source.monogon.dev/metropolis/node/core/network"
 	"source.monogon.dev/metropolis/node/core/roleserve"
 	"source.monogon.dev/metropolis/pkg/supervisor"
 	cpb "source.monogon.dev/metropolis/proto/common"
 )

 type Config struct {
 	// Network is a handle to the Network service, used to update the hostsfile
 	// service with information about the local node's external IP address.
 	Network *network.Service
 	// Ephemeral is the root of the ephemeral storage of the node, into which the
 	// service will write its managed files.
 	Ephemeral *localstorage.EphemeralDirectory
 	// ESP is the root of the node's EFI System Partition.
 	ESP *localstorage.ESPDirectory

 	// Roleserver is an instance of the roleserver service which will be queried for
 	// ClusterMembership and a Curator client.
 	Roleserver *roleserve.Service
 }

 // Service is the hostsfile service instance. See package-level documentation
 // for more information.
 type Service struct {
 	Config

 	// localC is a channel populated by the local sub-runnable with the newest
 	// available information about the local node's address. It is automatically
 	// created and closed by Run.
 	localC chan string
 	// clusterC is a channel populated by the cluster sub-runnable with the newest
 	// available information about the cluster nodes. It is automatically created and
 	// closed by Run.
 	clusterC chan nodeMap
 }

 type ClusterDialer func(ctx context.Context) (*grpc.ClientConn, error)

 // nodeInfo contains all of a single node's data needed to build its entry in
 // either hostsfile or ClusterDirectory.
 type nodeInfo struct {
 	// address is the node's IP address.
 	address string
 	// local is true if address belongs to the local node.
 	local bool
 }

 // nodeMap is a map from node ID (effectively DNS name) to node IP address.
 type nodeMap map[string]nodeInfo

 // hosts generates a complete /etc/hosts file based on the contents of the
 // nodeMap. Apart from the addresses in the nodeMap, entries for localhost
 // pointing to 127.0.0.1 and ::1 will also be generated.
 func (m nodeMap) hosts(ctx context.Context) []byte {
 	var nodeIdsSorted []string
 	for k, _ := range m {
 		nodeIdsSorted = append(nodeIdsSorted, k)
 	}
 	sort.Slice(nodeIdsSorted, func(i, j int) bool {
 		return nodeIdsSorted[i] < nodeIdsSorted[j]
 	})

 	lines := [][]byte{
 		[]byte("127.0.0.1 localhost"),
 		[]byte("::1 localhost"),
 	}
 	for _, nid := range nodeIdsSorted {
 		addr := m[nid].address
 		line := fmt.Sprintf("%s %s", addr, nid)
 		supervisor.Logger(ctx).Infof("Hosts entry: %s", line)
 		lines = append(lines, []byte(line))
 	}
 	lines = append(lines, []byte(""))

 	return bytes.Join(lines, []byte("\n"))
 }

 // clusterDirectory builds a ClusterDirectory based on nodeMap contents. If m
 // is empty, an empty ClusterDirectory is returned.
 func (m nodeMap) clusterDirectory(ctx context.Context) *cpb.ClusterDirectory {
 	var directory cpb.ClusterDirectory
 	for _, ni := range m {
 		// Skip local addresses.
 		if ni.local {
 			continue
 		}

 		supervisor.Logger(ctx).Infof("ClusterDirectory entry: %s", ni.address)
 		addresses := []*cpb.ClusterDirectory_Node_Address{
 			{Host: ni.address},
 		}
 		node := &cpb.ClusterDirectory_Node{
 			Addresses: addresses,
 		}
 		directory.Nodes = append(directory.Nodes, node)
 	}
 	return &directory
 }

 func (s *Service) Run(ctx context.Context) error {
 	s.localC = make(chan string)
 	defer close(s.localC)
 	s.clusterC = make(chan nodeMap)
 	defer close(s.clusterC)

 	cmw := s.Roleserver.ClusterMembership.Watch()
 	defer cmw.Close()
 	supervisor.Logger(ctx).Infof("Waiting for node ID...")
 	nodeID, err := roleserve.GetNodeID(ctx, cmw)
 	if err != nil {
 		return err
 	}
 	supervisor.Logger(ctx).Infof("Got node ID, starting...")

 	if err := supervisor.Run(ctx, "local", s.runLocal); err != nil {
 		return err
 	}
 	if err := supervisor.Run(ctx, "cluster", s.runCluster); err != nil {
 		return err
 	}

 	// Immediately update machine-id and hostname, we don't need network addresses
 	// for that.
 	if err := s.Ephemeral.MachineID.Write([]byte(nodeID), 0644); err != nil {
 		return fmt.Errorf("failed to write /ephemeral/machine-id: %w", err)
 	}
 	if err := unix.Sethostname([]byte(nodeID)); err != nil {
 		return fmt.Errorf("failed to set runtime hostname: %w", err)
 	}
 	// Immediately write an /etc/hosts just containing localhost, even if we don't
 	// yet have a network address.
 	nodes := make(nodeMap)
 	if err := s.Ephemeral.Hosts.Write(nodes.hosts(ctx), 0644); err != nil {
 		return fmt.Errorf("failed to write %s: %w", s.Ephemeral.Hosts.FullPath(), err)
 	}

 	supervisor.Signal(ctx, supervisor.SignalHealthy)
 	// Update nodeMap in a loop, issuing writes/updates when any change occurred.
 	for {
 		changed := false
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
 		case u := <-s.localC:
 			// Ignore spurious updates.
 			if nodes[nodeID].address == u {
 				break
 			}
 			supervisor.Logger(ctx).Infof("Got new local address: %s", u)
 			nodes[nodeID] = nodeInfo{
 				address: u,
 				local:   true,
 			}
 			changed = true
 		case u := <-s.clusterC:
 			// Loop through the nodeMap from the cluster subrunnable, making note of what
 			// changed. By design we don't care about any nodes disappearing from the
 			// nodeMap: we'd rather keep stale data about nodes that don't exist any more,
 			// as these might either be spurious or have a long tail of effectively still
 			// being used by the local node for communications while the node gets fully
 			// drained/disowned.
 			//
 			// MVP: we should at least log removed nodes.
 			for id, info := range u {
 				// We're not interested in what the cluster thinks about our local node, as that
 				// might be outdated (eg. when we haven't yet reported a new local address to
 				// the cluster).
 				if id == nodeID {
 					continue
 				}
 				if nodes[id].address == info.address {
 					continue
 				}
 				supervisor.Logger(ctx).Infof("Got new cluster address: %s is %s", id, info.address)
 				nodes[id] = info
 				changed = true
 			}
 		}

 		if !changed {
 			continue
 		}

 		supervisor.Logger(ctx).Infof("Updating hosts file: %d nodes", len(nodes))
 		if err := s.Ephemeral.Hosts.Write(nodes.hosts(ctx), 0644); err != nil {
 			return fmt.Errorf("failed to write %s: %w", s.Ephemeral.Hosts.FullPath(), err)
 		}

 		// Check that we are self-resolvable.
 		if _, err := net.ResolveIPAddr("ip", nodeID); err != nil {
 			supervisor.Logger(ctx).Errorf("Failed to self-resolve %q: %v", nodeID, err)
 		}

 		// Update this node's ClusterDirectory.
 		supervisor.Logger(ctx).Info("Updating ClusterDirectory.")
 		cd := nodes.clusterDirectory(ctx)
 		cdirRaw, err := proto.Marshal(cd)
 		if err != nil {
 			return fmt.Errorf("couldn't marshal ClusterDirectory: %w", err)
 		}
 		if err = s.ESP.Metropolis.ClusterDirectory.Write(cdirRaw, 0644); err != nil {
 			return err
 		}
 		unix.Sync()
 	}
 }

 // runLocal updates s.localC with the IP address of the local node, as retrieved
 // from the network service.
 func (s *Service) runLocal(ctx context.Context) error {
 	nw := s.Network.Watch()
 	for {
 		ns, err := nw.Get(ctx)
 		if err != nil {
 			return err
 		}
 		addr := ns.ExternalAddress.String()
 		if addr != "" {
 			s.localC <- addr
 		}
 	}
 }

 // runCluster updates s.clusterC with the IP addresses of cluster nodes, as
 // retrieved from a Curator client from the ClusterDialer. The returned map
 // reflects the up-to-date view of the cluster returned from the Curator Watch
 // call, including any node deletions.
 func (s *Service) runCluster(ctx context.Context) error {
 	cmw := s.Roleserver.ClusterMembership.Watch()
 	defer cmw.Close()

 	supervisor.Logger(ctx).Infof("Waiting for cluster membership...")
 	cm, err := cmw.Get(ctx, roleserve.FilterHome())
 	if err != nil {
 		return err
 	}
 	supervisor.Logger(ctx).Infof("Got cluster membership, starting...")

 	con, err := cm.DialCurator()
 	if err != nil {
 		return err
 	}
 	defer con.Close()
 	cur := ipb.NewCuratorClient(con)

 	w, err := cur.Watch(ctx, &ipb.WatchRequest{
 		Kind: &ipb.WatchRequest_NodesInCluster_{
 			NodesInCluster: &ipb.WatchRequest_NodesInCluster{},
 		},
 	})
 	if err != nil {
 		return fmt.Errorf("curator watch failed: %w", err)
 	}

 	nodes := make(nodeMap)
 	for {
 		ev, err := w.Recv()
 		if err != nil {
 			return fmt.Errorf("receive failed: %w", err)
 		}
 		for _, n := range ev.Nodes {
 			if n.Status == nil || n.Status.ExternalAddress == "" {
 				continue
 			}
 			nodes[n.Id] = nodeInfo{
 				address: n.Status.ExternalAddress,
 				local:   false,
 			}
 		}
 		for _, t := range ev.NodeTombstones {
 			delete(nodes, t.NodeId)
 		}
 		s.clusterC <- nodes
 	}
 }
	// hostsfile implements a service which owns and writes all node-local
	// files/interfaces used by the system to resolve the local node's name and the
	// names of other nodes in the cluster:
	//
	// 1. All cluster node names are written into /etc/hosts for DNS resolution.
	// 2. The local node's name is written into /etc/machine-id.
	// 3. The local node's name is set as the UNIX hostname of the machine (via the
	// sethostname call).
	// 4. The local node's ClusterDirectory is updated with the same set of
	// addresses as the one used in /etc/hosts.
	//
	// The hostsfile Service can start up in two modes: with cluster connectivity
	// and without cluster connectivity. Without cluster connectivity, only
	// information about the current node (as retrieved from the network service)
	// will be used to populate local data. In cluster mode, information about other
	// nodes is also used.
	package hostsfile

	import (
	"bytes"
	"context"
	"fmt"
	"net"
	"sort"

	"golang.org/x/sys/unix"
	"google.golang.org/grpc"
	"google.golang.org/protobuf/proto"

	ipb "source.monogon.dev/metropolis/node/core/curator/proto/api"
	"source.monogon.dev/metropolis/node/core/localstorage"
	"source.monogon.dev/metropolis/node/core/network"
	"source.monogon.dev/metropolis/node/core/roleserve"
	"source.monogon.dev/metropolis/pkg/supervisor"
	cpb "source.monogon.dev/metropolis/proto/common"
	)

	type Config struct {
	// Network is a handle to the Network service, used to update the hostsfile
	// service with information about the local node's external IP address.
	Network *network.Service
	// Ephemeral is the root of the ephemeral storage of the node, into which the
	// service will write its managed files.
	Ephemeral *localstorage.EphemeralDirectory
	// ESP is the root of the node's EFI System Partition.
	ESP *localstorage.ESPDirectory

	// Roleserver is an instance of the roleserver service which will be queried for
	// ClusterMembership and a Curator client.
	Roleserver *roleserve.Service
	}

	// Service is the hostsfile service instance. See package-level documentation
	// for more information.
	type Service struct {
	Config

	// localC is a channel populated by the local sub-runnable with the newest
	// available information about the local node's address. It is automatically
	// created and closed by Run.
	localC chan string
	// clusterC is a channel populated by the cluster sub-runnable with the newest
	// available information about the cluster nodes. It is automatically created and
	// closed by Run.
	clusterC chan nodeMap
	}

	type ClusterDialer func(ctx context.Context) (*grpc.ClientConn, error)

	// nodeInfo contains all of a single node's data needed to build its entry in
	// either hostsfile or ClusterDirectory.
	type nodeInfo struct {
	// address is the node's IP address.
	address string
	// local is true if address belongs to the local node.
	local bool
	}

	// nodeMap is a map from node ID (effectively DNS name) to node IP address.
	type nodeMap map[string]nodeInfo

	// hosts generates a complete /etc/hosts file based on the contents of the
	// nodeMap. Apart from the addresses in the nodeMap, entries for localhost
	// pointing to 127.0.0.1 and ::1 will also be generated.
	func (m nodeMap) hosts(ctx context.Context) []byte {
	var nodeIdsSorted []string
	for k, _ := range m {
	nodeIdsSorted = append(nodeIdsSorted, k)
	}
	sort.Slice(nodeIdsSorted, func(i, j int) bool {
	return nodeIdsSorted[i] < nodeIdsSorted[j]
	})

	lines := [][]byte{
	[]byte("127.0.0.1 localhost"),
	[]byte("::1 localhost"),
	}
	for _, nid := range nodeIdsSorted {
	addr := m[nid].address
	line := fmt.Sprintf("%s %s", addr, nid)
	supervisor.Logger(ctx).Infof("Hosts entry: %s", line)
	lines = append(lines, []byte(line))
	}
	lines = append(lines, []byte(""))

	return bytes.Join(lines, []byte("\n"))
	}

	// clusterDirectory builds a ClusterDirectory based on nodeMap contents. If m
	// is empty, an empty ClusterDirectory is returned.
	func (m nodeMap) clusterDirectory(ctx context.Context) *cpb.ClusterDirectory {
	var directory cpb.ClusterDirectory
	for _, ni := range m {
	// Skip local addresses.
	if ni.local {
	continue
	}

	supervisor.Logger(ctx).Infof("ClusterDirectory entry: %s", ni.address)
	addresses := []*cpb.ClusterDirectory_Node_Address{
	{Host: ni.address},
	}
	node := &cpb.ClusterDirectory_Node{
	Addresses: addresses,
	}
	directory.Nodes = append(directory.Nodes, node)
	}
	return &directory
	}

	func (s *Service) Run(ctx context.Context) error {
	s.localC = make(chan string)
	defer close(s.localC)
	s.clusterC = make(chan nodeMap)
	defer close(s.clusterC)

	cmw := s.Roleserver.ClusterMembership.Watch()
	defer cmw.Close()
	supervisor.Logger(ctx).Infof("Waiting for node ID...")
	nodeID, err := roleserve.GetNodeID(ctx, cmw)
	if err != nil {
	return err
	}
	supervisor.Logger(ctx).Infof("Got node ID, starting...")

	if err := supervisor.Run(ctx, "local", s.runLocal); err != nil {
	return err
	}
	if err := supervisor.Run(ctx, "cluster", s.runCluster); err != nil {
	return err
	}

	// Immediately update machine-id and hostname, we don't need network addresses
	// for that.
	if err := s.Ephemeral.MachineID.Write([]byte(nodeID), 0644); err != nil {
	return fmt.Errorf("failed to write /ephemeral/machine-id: %w", err)
	}
	if err := unix.Sethostname([]byte(nodeID)); err != nil {
	return fmt.Errorf("failed to set runtime hostname: %w", err)
	}
	// Immediately write an /etc/hosts just containing localhost, even if we don't
	// yet have a network address.
	nodes := make(nodeMap)
	if err := s.Ephemeral.Hosts.Write(nodes.hosts(ctx), 0644); err != nil {
	return fmt.Errorf("failed to write %s: %w", s.Ephemeral.Hosts.FullPath(), err)
	}

	supervisor.Signal(ctx, supervisor.SignalHealthy)
	// Update nodeMap in a loop, issuing writes/updates when any change occurred.
	for {
	changed := false
	select {
	case <-ctx.Done():
	return ctx.Err()
	case u := <-s.localC:
	// Ignore spurious updates.
	if nodes[nodeID].address == u {
	break
	}
	supervisor.Logger(ctx).Infof("Got new local address: %s", u)
	nodes[nodeID] = nodeInfo{
	address: u,
	local: true,
	}
	changed = true
	case u := <-s.clusterC:
	// Loop through the nodeMap from the cluster subrunnable, making note of what
	// changed. By design we don't care about any nodes disappearing from the
	// nodeMap: we'd rather keep stale data about nodes that don't exist any more,
	// as these might either be spurious or have a long tail of effectively still
	// being used by the local node for communications while the node gets fully
	// drained/disowned.
	//
	// MVP: we should at least log removed nodes.
	for id, info := range u {
	// We're not interested in what the cluster thinks about our local node, as that
	// might be outdated (eg. when we haven't yet reported a new local address to
	// the cluster).
	if id == nodeID {
	continue
	}
	if nodes[id].address == info.address {
	continue
	}
	supervisor.Logger(ctx).Infof("Got new cluster address: %s is %s", id, info.address)
	nodes[id] = info
	changed = true
	}
	}

	if !changed {
	continue
	}

	supervisor.Logger(ctx).Infof("Updating hosts file: %d nodes", len(nodes))
	if err := s.Ephemeral.Hosts.Write(nodes.hosts(ctx), 0644); err != nil {
	return fmt.Errorf("failed to write %s: %w", s.Ephemeral.Hosts.FullPath(), err)
	}

	// Check that we are self-resolvable.
	if _, err := net.ResolveIPAddr("ip", nodeID); err != nil {
	supervisor.Logger(ctx).Errorf("Failed to self-resolve %q: %v", nodeID, err)
	}

	// Update this node's ClusterDirectory.
	supervisor.Logger(ctx).Info("Updating ClusterDirectory.")
	cd := nodes.clusterDirectory(ctx)
	cdirRaw, err := proto.Marshal(cd)
	if err != nil {
	return fmt.Errorf("couldn't marshal ClusterDirectory: %w", err)
	}
	if err = s.ESP.Metropolis.ClusterDirectory.Write(cdirRaw, 0644); err != nil {
	return err
	}
	unix.Sync()
	}
	}

	// runLocal updates s.localC with the IP address of the local node, as retrieved
	// from the network service.
	func (s *Service) runLocal(ctx context.Context) error {
	nw := s.Network.Watch()
	for {
	ns, err := nw.Get(ctx)
	if err != nil {
	return err
	}
	addr := ns.ExternalAddress.String()
	if addr != "" {
	s.localC <- addr
	}
	}
	}

	// runCluster updates s.clusterC with the IP addresses of cluster nodes, as
	// retrieved from a Curator client from the ClusterDialer. The returned map
	// reflects the up-to-date view of the cluster returned from the Curator Watch
	// call, including any node deletions.
	func (s *Service) runCluster(ctx context.Context) error {
	cmw := s.Roleserver.ClusterMembership.Watch()
	defer cmw.Close()

	supervisor.Logger(ctx).Infof("Waiting for cluster membership...")
	cm, err := cmw.Get(ctx, roleserve.FilterHome())
	if err != nil {
	return err
	}
	supervisor.Logger(ctx).Infof("Got cluster membership, starting...")

	con, err := cm.DialCurator()
	if err != nil {
	return err
	}
	defer con.Close()
	cur := ipb.NewCuratorClient(con)

	w, err := cur.Watch(ctx, &ipb.WatchRequest{
	Kind: &ipb.WatchRequest_NodesInCluster_{
	NodesInCluster: &ipb.WatchRequest_NodesInCluster{},
	},
	})
	if err != nil {
	return fmt.Errorf("curator watch failed: %w", err)
	}

	nodes := make(nodeMap)
	for {
	ev, err := w.Recv()
	if err != nil {
	return fmt.Errorf("receive failed: %w", err)
	}
	for _, n := range ev.Nodes {
	if n.Status == nil \|\| n.Status.ExternalAddress == "" {
	continue
	}
	nodes[n.Id] = nodeInfo{
	address: n.Status.ExternalAddress,
	local: false,
	}
	}
	for _, t := range ev.NodeTombstones {
	delete(nodes, t.NodeId)
	}
	s.clusterC <- nodes
	}
	}