From d36429e6e578404e56747b1f8f1abe0a42caffe5 Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Tue, 5 Nov 2024 20:05:31 +1000 Subject: [PATCH] replace API server with transparent gRPC proxy that routes requests to local machine API --- internal/machine/api/proxy/director.go | 37 ++++++++++ internal/machine/api/proxy/local.go | 68 ++++++++++++++++++ internal/machine/machine.go | 97 +++++++++++++++++++------- 3 files changed, 178 insertions(+), 24 deletions(-) create mode 100644 internal/machine/api/proxy/director.go create mode 100644 internal/machine/api/proxy/local.go diff --git a/internal/machine/api/proxy/director.go b/internal/machine/api/proxy/director.go new file mode 100644 index 00000000..dc9e13ab --- /dev/null +++ b/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 +} diff --git a/internal/machine/api/proxy/local.go b/internal/machine/api/proxy/local.go new file mode 100644 index 00000000..e057f669 --- /dev/null +++ b/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 +} diff --git a/internal/machine/machine.go b/internal/machine/machine.go index 75b35213..ccccf1df 100644 --- a/internal/machine/machine.go +++ b/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 - cluster *cluster.Cluster - docker *docker.Server - localServer *grpc.Server + store *store.Store + cluster *cluster.Cluster + docker *docker.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, - state: state, - started: make(chan struct{}), - initialised: make(chan struct{}, 1), - store: corroStore, - cluster: c, - docker: dockerServer, + config: *config, + state: state, + started: make(chan struct{}), + initialised: make(chan struct{}, 1), + store: corroStore, + cluster: c, + 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. - // TODO: start this on machine.sock and start grpc-proxy on uncloud.sock. Use 700 mode for machine.sock. - localListener, err := listenUnixSocket(m.config.UncloudSockPath) + // Start the local machine API server. + machineListener, err := listenUnixSocket(m.config.MachineSockPath) 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) - if err := m.localServer.Serve(localListener); err != nil { - return fmt.Errorf("local API server failed: %w", err) + 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) + } + 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) - ctrl, err = newNetworkController(m.state, m.store, networkServer, m.config.CorrosionService) + // 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. + 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() - slog.Info("Local API server stopped.") + m.localMachineServer.GracefulStop() + 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 }, )