BREAKING CHANGE: move Uncloud API socket to /run/uncloud/api/uncloud.sock, activate it by systemd socket unit

internal/cli/cli.go

@@ -603,7 +603,7 @@ func provisionOrConnectRemoteMachine(
 		if remoteMachine.User != rootUser {
 			// provisionMachine has just added the user to the uncloud group. Any SSH ControlMaster left over from
 			// a previous uc invocation (e.g. a failed uc command against the uninitialised machine) still holds
-			// the old user groups and would deny access to /run/uncloud/uncloud.sock. Close the current session
+			// the old user groups and would deny access to /run/uncloud/api/uncloud.sock. Close the current session
 			// if it exists so the next session picks up the new groups.
 			conn.CloseControlMaster(ctx)
 		}

internal/cli/config/connection_test.go

@@ -70,9 +70,9 @@ func TestMachineConnection_String(t *testing.T) {
 		{
 			name: "unix connection",
 			conn: MachineConnection{
-				Unix: "/run/uncloud/uncloud.sock",
+				Unix: "/run/uncloud/api/uncloud.sock",
 			},
-			want: "unix:///run/uncloud/uncloud.sock",
+			want: "unix:///run/uncloud/api/uncloud.sock",
 		},
 		{
 			name: "no connection",

internal/daemon/daemon.go

@@ -2,24 +2,50 @@ package daemon
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
+	"net"
 
+	"github.com/coreos/go-systemd/activation"
 	systemd "github.com/coreos/go-systemd/daemon"
 	"github.com/psviderski/uncloud/internal/machine"
 )
 
+const systemdSocketUnit = "uncloud.socket"
+
 type Daemon struct {
 	machine *machine.Machine
 }
 
 func New(dataDir string) (*Daemon, error) {
+	listeners, err := activation.ListenersWithNames()
+	if err != nil {
+		return nil, fmt.Errorf("get systemd-activated sockets: %w", err)
+	}
+
+	listener, err := selectActivatedListener(listeners)
+	if err != nil {
+		return nil, err
+	}
+	if listener != nil {
+		slog.Info("Using systemd-activated API socket.", "addr", listener.Addr().String())
+	}
+
 	config := &machine.Config{
-		DataDir: dataDir,
+		DataDir:            dataDir,
+		ClusterAPISockPath: machine.DefaultClusterAPISockPath,
+		ClusterAPIListener: listener,
 	}
 	mach, err := machine.NewMachine(config)
 	if err != nil {
-		return nil, fmt.Errorf("init machine: %w", err)
+		initErr := fmt.Errorf("init machine: %w", err)
+		if listener != nil {
+			if closeErr := listener.Close(); closeErr != nil {
+				initErr = errors.Join(initErr, fmt.Errorf("close systemd-activated socket: %w", closeErr))
+			}
+		}
+		return nil, initErr
 	}
 
 	return &Daemon{
@@ -27,6 +53,40 @@ func New(dataDir string) (*Daemon, error) {
 	}, nil
 }
 
+// selectActivatedListener returns the systemd activated listener. If systemd supplied multiple listeners,
+// it returns the one associated with systemdSocketUnit.
+func selectActivatedListener(listeners map[string][]net.Listener) (net.Listener, error) {
+	var activated []net.Listener
+	for _, named := range listeners {
+		activated = append(activated, named...)
+	}
+
+	if len(activated) == 0 {
+		return nil, nil
+	}
+	if len(activated) == 1 {
+		return activated[0], nil
+	}
+
+	named, ok := listeners[systemdSocketUnit]
+	if !ok || len(named) != 1 {
+		return nil, fmt.Errorf("expected one systemd-activated socket named '%s', received %d",
+			systemdSocketUnit, len(named))
+	}
+
+	// Close all other unused listeners.
+	for name, unused := range listeners {
+		if name == systemdSocketUnit {
+			continue
+		}
+		for _, l := range unused {
+			_ = l.Close()
+		}
+	}
+
+	return named[0], nil
+}
+
 func (d *Daemon) Run(ctx context.Context) error {
 	slog.Info("Starting machine.")
 

internal/daemon/daemon_test.go

@@ -0,0 +1,71 @@
+package daemon
+
+import (
+	"net"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSelectActivatedListener(t *testing.T) {
+	t.Run("no activated listeners", func(t *testing.T) {
+		listener, err := selectActivatedListener(nil)
+		require.NoError(t, err)
+		assert.Nil(t, listener)
+	})
+
+	t.Run("only listener regardless of name", func(t *testing.T) {
+		activated := testListener(t)
+
+		listener, err := selectActivatedListener(map[string][]net.Listener{"another.socket": {activated}})
+		require.NoError(t, err)
+		assert.Same(t, activated, listener)
+	})
+
+	t.Run("named listener when multiple", func(t *testing.T) {
+		activated := testListener(t)
+		extra := testListener(t)
+
+		listener, err := selectActivatedListener(map[string][]net.Listener{
+			systemdSocketUnit: {activated},
+			"another.socket":  {extra},
+		})
+		require.NoError(t, err)
+		assert.Same(t, activated, listener)
+		assert.Error(t, extra.Close())
+	})
+
+	t.Run("multiple listeners without expected name", func(t *testing.T) {
+		first := testListener(t)
+		second := testListener(t)
+
+		listener, err := selectActivatedListener(map[string][]net.Listener{
+			"first.socket":  {first},
+			"second.socket": {second},
+		})
+		require.ErrorContains(t, err, "expected one systemd-activated socket")
+		assert.Nil(t, listener)
+	})
+
+	t.Run("multiple listeners under expected name", func(t *testing.T) {
+		first := testListener(t)
+		second := testListener(t)
+
+		listener, err := selectActivatedListener(map[string][]net.Listener{
+			systemdSocketUnit: {first, second},
+		})
+		require.ErrorContains(t, err, "expected one systemd-activated socket")
+		assert.Nil(t, listener)
+	})
+}
+
+func testListener(t *testing.T) net.Listener {
+	t.Helper()
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = listener.Close() })
+
+	return listener
+}

internal/machine/cluster.go

@@ -176,7 +176,7 @@ func (cc *clusterController) Run(ctx context.Context) error {
 	// Start the network API server before waiting for the store sync so the machine is reachable on the mesh
 	// during the sync and can serve requests that don't depend on the store.
 	// Assume the management IP can't be changed when the network is running.
-	apiAddr := net.JoinHostPort(cc.state.Network.ManagementIP.String(), strconv.Itoa(constants.MachineAPIPort))
+	apiAddr := net.JoinHostPort(cc.state.Network.ManagementIP.String(), strconv.Itoa(constants.UncloudAPIPort))
 	listener, err := net.Listen("tcp", apiAddr)
 	if err != nil {
 		return fmt.Errorf("listen API port: %w", err)

internal/machine/constants/constants.go

@@ -1,8 +1,9 @@
 package constants
 
 const (
-	// MachineAPIPort is the port for the Machine API service on the management WireGuard network.
-	MachineAPIPort = 51000
-	// UnregistryPort is the port for the embedded container registry listening on the machine IP.
+	// UncloudAPIPort is the TCP port on which each machine serves the Uncloud API over the management WireGuard
+	// network.
+	UncloudAPIPort = 51000
+	// UnregistryPort is the TCP port for the embedded container registry listening on the machine IP.
 	UnregistryPort = 51500
 )

internal/machine/firewall/iptables_linux.go

@@ -48,12 +48,12 @@ func ConfigureIptablesChains(machineIP netip.Addr, wgPort int) error {
 		}
 	}
 
-	// Allow cluster machines to access Machine API via the management IPv6 WireGuard network.
+	// Allow cluster machines to access the Uncloud API via the management IPv6 WireGuard network.
 	acceptMachineAPIRule := []string{
 		"-i", network.WireGuardInterfaceName,
 		"-s", "fdcc::/16",
 		"-p", "tcp",
-		"--dport", strconv.Itoa(constants.MachineAPIPort),
+		"--dport", strconv.Itoa(constants.UncloudAPIPort),
 		"-j", "ACCEPT",
 	}
 	// Allow Corrosion gossip traffic from cluster machines via the management IPv6 WireGuard network.

internal/machine/machine.go

@@ -58,9 +58,13 @@ import (
 )
 
 const (
-	DefaultMachineSockPath = "/run/uncloud/machine.sock"
-	DefaultUncloudSockPath = "/run/uncloud/uncloud.sock"
-	DefaultSockGroup       = "uncloud"
+	// DefaultMachineAPISockPath is the default path to the Unix socket for API requests handled directly by this
+	// machine.
+	DefaultMachineAPISockPath = "/run/uncloud/machine.sock"
+	// DefaultClusterAPISockPath is the default path to the Unix socket for the client-facing API that routes requests
+	// across the cluster.
+	DefaultClusterAPISockPath = "/run/uncloud/api/uncloud.sock"
+	DefaultSockGroup          = "uncloud"
 	// DefaultCaddyAdminSockPath is the default path to the Caddy admin socket for validating the generated Caddy
 	// reverse proxy configuration.
 	DefaultCaddyAdminSockPath = "/run/uncloud/caddy/admin.sock"
@@ -70,9 +74,15 @@ const (
 
 type Config struct {
 	// DataDir is the directory where the machine stores its persistent state. Default is /var/lib/uncloud.
-	DataDir         string
-	MachineSockPath string
-	UncloudSockPath string
+	DataDir string
+	// MachineAPISockPath is the path to the Unix socket for API requests handled directly by this machine.
+	MachineAPISockPath string
+	// ClusterAPISockPath is the path to the Unix socket for the client-facing API that routes requests across the
+	// cluster.
+	ClusterAPISockPath string
+	// ClusterAPIListener is an optional pre-bound listener for the client-facing API that routes requests across the
+	// cluster. If set, Run takes ownership of it and closes it on shutdown or startup failure.
+	ClusterAPIListener net.Listener
 
 	CorrosionDataDir string
 	// CorrosionRunDir is the runtime directory for the corrosion service.
@@ -104,11 +114,11 @@ func (c *Config) SetDefaults() (*Config, error) {
 	if cfg.DataDir == "" {
 		cfg.DataDir = "/var/lib/uncloud"
 	}
-	if cfg.MachineSockPath == "" {
-		cfg.MachineSockPath = DefaultMachineSockPath
+	if cfg.MachineAPISockPath == "" {
+		cfg.MachineAPISockPath = DefaultMachineAPISockPath
 	}
-	if cfg.UncloudSockPath == "" {
-		cfg.UncloudSockPath = DefaultUncloudSockPath
+	if cfg.ClusterAPISockPath == "" && cfg.ClusterAPIListener == nil {
+		cfg.ClusterAPISockPath = DefaultClusterAPISockPath
 	}
 
 	if cfg.DockerClient == nil {
@@ -165,7 +175,7 @@ type Machine struct {
 
 	config Config
 	state  *State
-	// started is closed when the machine is ready to serve requests on the local API server.
+	// started is closed when the machine and cluster API servers are ready to serve requests.
 	started chan struct{}
 	// initialised is closed when the machine is configured as a member of a cluster.
 	initialised chan struct{}
@@ -186,15 +196,14 @@ type Machine struct {
 	// dockerService provides high-level operations for managing Docker containers.
 	dockerService *machinedocker.Service
 	dockerServer  *machinedocker.Server
-	// localMachineServer is the gRPC server for the machine API listening on the local Unix socket.
-	localMachineServer *grpc.Server
+	// machineAPIServer handles API requests directly on this machine.
+	machineAPIServer *grpc.Server
 
-	// proxyDirector manages routing of gRPC requests between local and remote machine API servers.
+	// proxyDirector routes API requests to local or remote machines.
 	proxyDirector *apiproxy.Director
-	// localProxyServer is the gRPC proxy server for the machine API listening on the local Unix socket.
-	// It proxies requests to the local or remote machine API servers depending on the request targets
-	// and aggregates responses.
-	localProxyServer *grpc.Server
+	// clusterAPIServer is the client-facing gRPC server that routes API requests across the cluster. It sends requests
+	// to one or more local or remote machines and aggregates their responses.
+	clusterAPIServer *grpc.Server
 
 	// mu protects the Machine from concurrent reads and writes.
 	mu sync.RWMutex
@@ -269,10 +278,10 @@ func NewMachine(config *Config) (*Machine, error) {
 	}
 	dockerService := machinedocker.NewService(config.DockerClient, db)
 
-	// Init a local gRPC proxy server that proxies requests to the local or remote machine API servers.
+	// Init the client-facing API server that routes requests to local or remote machines.
 	mapper := apiproxy.NewCorrosionMapper(corroStore)
-	proxyDirector := apiproxy.NewDirector(config.MachineSockPath, constants.MachineAPIPort, mapper)
-	localProxyServer := grpc.NewServer(
+	proxyDirector := apiproxy.NewDirector(config.MachineAPISockPath, constants.UncloudAPIPort, mapper)
+	clusterAPIServer := grpc.NewServer(
 		grpc.ForceServerCodecV2(proxy.Codec()),
 		grpc.UnaryInterceptor(grpcversion.ServerUnaryInterceptor),
 		grpc.StreamInterceptor(grpcversion.ServerStreamInterceptor),
@@ -291,7 +300,7 @@ func NewMachine(config *Config) (*Machine, error) {
 		store:            corroStore,
 		cluster:          c,
 		dockerService:    dockerService,
-		localProxyServer: localProxyServer,
+		clusterAPIServer: clusterAPIServer,
 		proxyDirector:    proxyDirector,
 	}
 
@@ -316,7 +325,7 @@ func NewMachine(config *Config) (*Machine, error) {
 	caddyStorageServer := caddystorage.NewServer(caddyStore)
 
 	leaseServer := distlockgrpc.NewServer(distlock.NewMemoryStore())
-	m.localMachineServer = newGRPCServer(m, c, m.dockerServer, caddyServer, caddyStorageServer, leaseServer)
+	m.machineAPIServer = newGRPCServer(m, c, m.dockerServer, caddyServer, caddyStorageServer, leaseServer)
 
 	if m.Initialised() {
 		close(m.initialised)
@@ -343,7 +352,7 @@ func newGRPCServer(
 	return s
 }
 
-// Started returns a channel that is closed when the machine is ready to serve requests on the local API server.
+// Started returns a channel that is closed when the machine and cluster API servers are ready to serve requests.
 func (m *Machine) Started() <-chan struct{} {
 	return m.started
 }
@@ -395,21 +404,54 @@ func (m *Machine) Run(ctx context.Context) error {
 	// Create a cancellable context for the Run method to allow stopping the machine gracefully.
 	ctx, m.stop = context.WithCancel(ctx)
 
+	// Take over the systemd-activated socket for the cluster API if provided, otherwise bind to the configured
+	// Unix socket path.
+	clusterAPIListener := m.config.ClusterAPIListener
+	clusterAPISockPath := m.config.ClusterAPISockPath
+	if clusterAPIListener != nil {
+		if clusterAPIListener.Addr().Network() == "unix" {
+			clusterAPISockPath = clusterAPIListener.Addr().String()
+		} else {
+			clusterAPISockPath = ""
+		}
+		defer clusterAPIListener.Close()
+	}
+
+	sockGID, err := socketGID()
+	if err != nil {
+		return err
+	}
+	if err = prepareUnixSocketDirectory(m.config.MachineAPISockPath, sockGID); err != nil {
+		return fmt.Errorf("prepare machine API Unix socket directory: %w", err)
+	}
+	// The cluster API socket activated by systemd may have been created with root group ownership.
+	// Apply the intended access permissions.
+	if clusterAPISockPath != "" {
+		if err = prepareUnixSocketDirectory(clusterAPISockPath, sockGID); err != nil {
+			return fmt.Errorf("prepare cluster API Unix socket directory: %w", err)
+		}
+	}
+
 	// Docker dependency is essential for the machine to function. Block until it's ready.
 	if err := docker.WaitDaemonReady(ctx, m.config.DockerClient); err != nil {
 		return fmt.Errorf("wait for Docker daemon: %w", err)
 	}
 	defer m.config.DockerClient.Close()
 
-	// Bind the local API listeners before starting the dependencies (e.g. corrosion) to not deal with the teardown
+	// Bind the API listeners before starting the dependencies (e.g. corrosion) to not deal with the teardown
 	// on failure.
-	machineListener, err := listenUnixSocket(m.config.MachineSockPath)
+	machineAPIListener, err := sockets.NewUnixSocket(m.config.MachineAPISockPath, sockGID)
 	if err != nil {
-		return fmt.Errorf("listen machine API unix socket %q: %w", m.config.MachineSockPath, err)
+		return fmt.Errorf("listen machine API Unix socket %q: %w", m.config.MachineAPISockPath, err)
 	}
-	proxyListener, err := listenUnixSocket(m.config.UncloudSockPath)
-	if err != nil {
-		return fmt.Errorf("listen API proxy unix socket %q: %w", m.config.UncloudSockPath, err)
+	defer machineAPIListener.Close()
+
+	if clusterAPIListener == nil {
+		clusterAPIListener, err = sockets.NewUnixSocket(clusterAPISockPath, sockGID)
+		if err != nil {
+			return fmt.Errorf("listen cluster API Unix socket %q: %w", clusterAPISockPath, err)
+		}
+		defer clusterAPIListener.Close()
 	}
 
 	// Configure and start the corrosion service on the loopback if the machine is not initialised as a cluster
@@ -437,20 +479,20 @@ func (m *Machine) Run(ctx context.Context) error {
 	// Use an errgroup to coordinate error handling and graceful shutdown of multiple machine components.
 	errGroup, ctx := errgroup.WithContext(ctx)
 
-	// Start the local machine API server.
+	// Start the API server that handles requests directly on this machine.
 	errGroup.Go(func() error {
-		slog.Info("Starting local machine API server.", "path", m.config.MachineSockPath)
-		if err := m.localMachineServer.Serve(machineListener); err != nil {
-			return fmt.Errorf("local machine API server failed: %w", err)
+		slog.Info("Starting machine API server.", "path", m.config.MachineAPISockPath)
+		if err := m.machineAPIServer.Serve(machineAPIListener); err != nil {
+			return fmt.Errorf("machine API server failed: %w", err)
 		}
 		return nil
 	})
 
-	// Start the local API proxy server.
+	// Start the client-facing API server that routes requests across the cluster.
 	errGroup.Go(func() error {
-		slog.Info("Starting local API proxy server.", "path", m.config.UncloudSockPath)
-		if err := m.localProxyServer.Serve(proxyListener); err != nil {
-			return fmt.Errorf("local API proxy server failed: %w", err)
+		slog.Info("Starting cluster API server.", "path", clusterAPIListener.Addr().String())
+		if err := m.clusterAPIServer.Serve(clusterAPIListener); err != nil {
+			return fmt.Errorf("cluster API server failed: %w", err)
 		}
 		return nil
 	})
@@ -478,7 +520,7 @@ func (m *Machine) Run(ctx context.Context) error {
 
 			slog.Info("Starting cluster controller.")
 			// Update the proxy director's local address to the machine's management IP address, allowing
-			// the proxy to identify which requests should be proxied to the local machine API server.
+			// the proxy to identify which requests should be handled by the local API server.
 			m.proxyDirector.UpdateLocalAddress(m.state.Network.ManagementIP.String())
 			proxyServer := grpc.NewServer(
 				grpc.ForceServerCodecV2(proxy.Codec()),
@@ -577,17 +619,17 @@ func (m *Machine) Run(ctx context.Context) error {
 	// Shutdown goroutine.
 	errGroup.Go(func() error {
 		<-ctx.Done()
-		slog.Info("Stopping local machine API server.")
+		slog.Info("Stopping machine API server.")
 		// TODO: implement timeout for graceful shutdown.
-		m.localMachineServer.GracefulStop()
-		slog.Info("Local machine API server stopped.")
+		m.machineAPIServer.GracefulStop()
+		slog.Info("Machine API server stopped.")
 
-		slog.Info("Stopping local API proxy server.")
+		slog.Info("Stopping cluster API server.")
 		// TODO: implement timeout for graceful shutdown.
-		m.localProxyServer.GracefulStop()
+		m.clusterAPIServer.GracefulStop()
 		// Close the proxy director to close all backend connections.
 		m.proxyDirector.Close()
-		slog.Info("Local API proxy server stopped.")
+		slog.Info("Cluster API server stopped.")
 
 		return nil
 	})
@@ -616,38 +658,46 @@ func (m *Machine) Run(ctx context.Context) error {
 	return err
 }
 
-// listenUnixSocket creates a new Unix socket listener with the specified path. The socket file is created with 0660
-// access mode and uncloud group if the group is found, otherwise it falls back to the root group.
-func listenUnixSocket(path string) (net.Listener, error) {
+// socketGID returns the uncloud group ID, or the root group ID if the uncloud group does not exist.
+func socketGID() (int, error) {
 	gid := 0 // Fall back to the root group if the uncloud group is not found.
 	group, err := user.LookupGroup(DefaultSockGroup)
 	if err != nil {
 		//goland:noinspection GoTypeAssertionOnErrors
 		if _, ok := err.(user.UnknownGroupError); ok {
 			slog.Info(
-				"Specified group not found, using root group for the API socket.",
-				"group", DefaultSockGroup, "path", path,
+				"Specified group not found, using root group for API sockets.",
+				"group", DefaultSockGroup,
 			)
 		} else {
-			return nil, fmt.Errorf("lookup %q group ID (GID): %w", DefaultSockGroup, err)
+			return 0, fmt.Errorf("lookup %q group ID (GID): %w", DefaultSockGroup, err)
 		}
 	} else {
 		gid, err = strconv.Atoi(group.Gid)
 		if err != nil {
-			return nil, fmt.Errorf("parse %q group ID (GID) %q: %w", DefaultSockGroup, group.Gid, err)
+			return 0, fmt.Errorf("parse %q group ID (GID) %q: %w", DefaultSockGroup, group.Gid, err)
 		}
 	}
 
-	// Ensure the parent directory exists and has the correct group permissions.
+	return gid, nil
+}
+
+// prepareUnixSocketDirectory ensures the socket's parent directory has the intended group and permissions.
+func prepareUnixSocketDirectory(path string, gid int) error {
 	parent, _ := filepath.Split(path)
-	if err = os.MkdirAll(parent, 0o750); err != nil {
-		return nil, fmt.Errorf("create directory %q: %w", parent, err)
+	if err := os.MkdirAll(parent, 0o750); err != nil {
+		return fmt.Errorf("create directory %q: %w", parent, err)
 	}
-	if err = os.Chown(parent, -1, gid); err != nil {
-		return nil, fmt.Errorf("chown directory %q: %w", parent, err)
+	if err := os.Chown(parent, -1, gid); err != nil {
+		return fmt.Errorf("chown directory %q: %w", parent, err)
+	}
+	// MkdirAll preserves the mode of an existing directory. A Docker container may have created the bind-mount source
+	// first, so apply the intended mode explicitly.
+	if err := os.Chmod(parent, 0o750); err != nil {
+		return fmt.Errorf("chmod directory %q: %w", parent, err)
 	}
 
-	return sockets.NewUnixSocket(path, gid)
+	return nil
 }
 
 func (m *Machine) configureCorrosion() error {