replace API server with transparent gRPC proxy that routes requests to local machine API

internal/machine/api/proxy/director.go

@@ -0,0 +1,37 @@
+package proxy
+
+import (
+	"context"
+	"github.com/siderolabs/grpc-proxy/proxy"
+	"sync"
+)
+
+// Director manages routing of gRPC requests between local and remote backends.
+type Director struct {
+	localTarget    string
+	localBackend   proxy.Backend
+	remoteBackends sync.Map
+
+	mu sync.RWMutex
+}
+
+func NewDirector(localSockPath string) *Director {
+	return &Director{
+		localBackend: NewLocalBackend(localSockPath),
+	}
+}
+
+// UpdateLocalAddress updates the local machine address used to identify which requests should be proxied
+// to the local gRPC server.
+func (d *Director) UpdateLocalAddress(target string) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	d.localTarget = target
+}
+
+// Director implements proxy.StreamDirector for grpc-proxy, routing requests to local or remote backends based
+// on gRPC metadata in the context.
+func (d *Director) Director(ctx context.Context, fullMethodName string) (proxy.Mode, []proxy.Backend, error) {
+	return proxy.One2One, []proxy.Backend{d.localBackend}, nil
+}

internal/machine/api/proxy/local.go

@@ -0,0 +1,68 @@
+package proxy
+
+import (
+	"context"
+	"github.com/siderolabs/grpc-proxy/proxy"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/metadata"
+	"sync"
+)
+
+// LocalBackend is a proxy.Backend implementation that proxies to a local gRPC server listening on a Unix socket.
+type LocalBackend struct {
+	sockPath string
+
+	mu   sync.RWMutex
+	conn *grpc.ClientConn
+}
+
+var _ proxy.Backend = (*LocalBackend)(nil)
+
+// NewLocalBackend returns a new LocalBackend for the given Unix socket path.
+func NewLocalBackend(sockPath string) *LocalBackend {
+	return &LocalBackend{
+		sockPath: sockPath,
+	}
+}
+
+func (l *LocalBackend) String() string {
+	return "local"
+}
+
+// GetConnection returns a gRPC connection to the local server listening on the Unix socket.
+func (l *LocalBackend) GetConnection(ctx context.Context, _ string) (context.Context, *grpc.ClientConn, error) {
+	md, _ := metadata.FromIncomingContext(ctx)
+	outCtx := metadata.NewOutgoingContext(ctx, md)
+
+	l.mu.RLock()
+	if l.conn != nil {
+		l.mu.RUnlock()
+		return outCtx, l.conn, nil
+	}
+	l.mu.RUnlock()
+
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	var err error
+	l.conn, err = grpc.NewClient(
+		"unix://"+l.sockPath,
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithDefaultCallOptions(
+			grpc.ForceCodecV2(proxy.Codec()),
+		),
+	)
+
+	return outCtx, l.conn, err
+}
+
+// AppendInfo is called to enhance response from the backend with additional data.
+func (l *LocalBackend) AppendInfo(_ bool, resp []byte) ([]byte, error) {
+	return resp, nil
+}
+
+// BuildError is called to convert error from upstream into response field.
+func (l *LocalBackend) BuildError(bool, error) ([]byte, error) {
+	return nil, nil
+}

internal/machine/machine.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"github.com/docker/docker/client"
 	"github.com/docker/go-connections/sockets"
+	"github.com/siderolabs/grpc-proxy/proxy"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -20,6 +21,7 @@ import (
 	"strconv"
 	"uncloud/internal/corrosion"
 	"uncloud/internal/machine/api/pb"
+	apiproxy "uncloud/internal/machine/api/proxy"
 	"uncloud/internal/machine/cluster"
 	"uncloud/internal/machine/corroservice"
 	"uncloud/internal/machine/docker"
@@ -87,10 +89,18 @@ type Machine struct {
 	initialised chan struct{}
 
 	// store is the cluster store backed by a distributed Corrosion database.
-	store       *store.Store
+	store   *store.Store
-	cluster     *cluster.Cluster
+	cluster *cluster.Cluster
-	docker      *docker.Server
+	docker  *docker.Server
-	localServer *grpc.Server
+	// localMachineServer is the gRPC server for the machine API listening on the local Unix socket.
+	localMachineServer *grpc.Server
+
+	// proxyDirector manages routing of gRPC requests between local and remote machine API servers.
+	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
 }
 
 func NewMachine(config *Config) (*Machine, error) {
@@ -137,16 +147,27 @@ func NewMachine(config *Config) (*Machine, error) {
 	}
 	dockerServer := docker.NewServer(dockerCli)
 
+	// Init a local gRPC proxy server that proxies requests to the local or remote machine API servers.
+	proxyDirector := apiproxy.NewDirector(config.MachineSockPath)
+	localProxyServer := grpc.NewServer(
+		grpc.ForceServerCodecV2(proxy.Codec()),
+		grpc.UnknownServiceHandler(
+			proxy.TransparentHandler(proxyDirector.Director),
+		),
+	)
+
 	m := &Machine{
-		config:      *config,
+		config:           *config,
-		state:       state,
+		state:            state,
-		started:     make(chan struct{}),
+		started:          make(chan struct{}),
-		initialised: make(chan struct{}, 1),
+		initialised:      make(chan struct{}, 1),
-		store:       corroStore,
+		store:            corroStore,
-		cluster:     c,
+		cluster:          c,
-		docker:      dockerServer,
+		docker:           dockerServer,
+		localProxyServer: localProxyServer,
+		proxyDirector:    proxyDirector,
 	}
-	m.localServer = newGRPCServer(m, c, dockerServer)
+	m.localMachineServer = newGRPCServer(m, c, dockerServer)
 
 	if m.Initialised() {
 		m.initialised <- struct{}{}
@@ -195,17 +216,31 @@ 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 machine local API server.
+	// Start the local machine API server.
-	// TODO: start this on machine.sock and start grpc-proxy on uncloud.sock. Use 700 mode for machine.sock.
+	machineListener, err := listenUnixSocket(m.config.MachineSockPath)
-	localListener, err := listenUnixSocket(m.config.UncloudSockPath)
 	if err != nil {
-		return fmt.Errorf("listen API unix socket %q: %w", m.config.UncloudSockPath, err)
+		return fmt.Errorf("listen machine API unix socket %q: %w", m.config.MachineSockPath, err)
 	}
 	errGroup.Go(
 		func() error {
-			slog.Info("Starting local API server.", "path", m.config.UncloudSockPath)
+			slog.Info("Starting local machine API server.", "path", m.config.MachineSockPath)
-			if err := m.localServer.Serve(localListener); err != nil {
+			if err := m.localMachineServer.Serve(machineListener); err != nil {
-				return fmt.Errorf("local API server failed: %w", err)
+				return fmt.Errorf("local machine API server failed: %w", err)
+			}
+			return nil
+		},
+	)
+
+	// Start the local API proxy server.
+	proxyListener, err := listenUnixSocket(m.config.UncloudSockPath)
+	if err != nil {
+		return fmt.Errorf("listen API proxy unix socket %q: %w", m.config.UncloudSockPath, err)
+	}
+	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)
 			}
 			return nil
 		},
@@ -241,8 +276,17 @@ func (m *Machine) Run(ctx context.Context) error {
 					slog.Info("Configured corrosion service.", "dir", m.config.CorrosionDir)
 
 					slog.Info("Starting network controller.")
-					networkServer := newGRPCServer(m, m.cluster, m.docker)
+					// Update the proxy director's local address to the machine's management IP address, allowing
-					ctrl, err = newNetworkController(m.state, m.store, networkServer, m.config.CorrosionService)
+					// the proxy to identify which requests should be proxied to the local machine API server.
+					m.proxyDirector.UpdateLocalAddress(m.state.Network.ManagementIP.String())
+					proxyServer := grpc.NewServer(
+						grpc.ForceServerCodecV2(proxy.Codec()),
+						grpc.UnknownServiceHandler(
+							proxy.TransparentHandler(m.proxyDirector.Director),
+						),
+					)
+
+					ctrl, err = newNetworkController(m.state, m.store, proxyServer, m.config.CorrosionService)
 					if err != nil {
 						return fmt.Errorf("initialise network controller: %w", err)
 					}
@@ -277,10 +321,15 @@ func (m *Machine) Run(ctx context.Context) error {
 	errGroup.Go(
 		func() error {
 			<-ctx.Done()
-			slog.Info("Stopping local API server.")
+			slog.Info("Stopping local machine API server.")
 			// TODO: implement timeout for graceful shutdown.
-			m.localServer.GracefulStop()
+			m.localMachineServer.GracefulStop()
-			slog.Info("Local API server stopped.")
+			slog.Info("Local machine API server stopped.")
+
+			slog.Info("Stopping local API proxy server.")
+			// TODO: implement timeout for graceful shutdown.
+			m.localProxyServer.GracefulStop()
+			slog.Info("Local API proxy server stopped.")
 			return nil
 		},
 	)