From 7a6c5bf6d7322fadcbd2985879bb0ee874470deb Mon Sep 17 00:00:00 2001 From: Pasha Sviderski Date: Thu, 24 Jul 2025 19:50:25 +1000 Subject: [PATCH] feat: machine Reset endpoint with asynchronous resource and data cleanup --- internal/machine/cluster.go | 47 +++++++++--- internal/machine/docker/manager.go | 1 + internal/machine/docker/manager_darwin.go | 5 ++ internal/machine/docker/manager_linux.go | 75 ++++++++++++++++++++ internal/machine/firewall/iptables_darwin.go | 5 ++ internal/machine/firewall/iptables_linux.go | 5 +- internal/machine/machine.go | 75 ++++++++++++++------ internal/machine/network/wireguard_darwin.go | 4 ++ internal/machine/network/wireguard_linux.go | 35 +++++++++ 9 files changed, 217 insertions(+), 35 deletions(-) diff --git a/internal/machine/cluster.go b/internal/machine/cluster.go index a67eecb0..ed681d2e 100644 --- a/internal/machine/cluster.go +++ b/internal/machine/cluster.go @@ -30,7 +30,8 @@ const ( ) // clusterController is the main controller for the machine that is a cluster member. It manages components such as -// the WireGuard network, Corrosion service, Docker network and containers, and embedded DNS server. +// the WireGuard network, API server listening the WireGuard network, Corrosion service, Docker network and containers, +// and others. type clusterController struct { state *State store *store.Store @@ -38,9 +39,10 @@ type clusterController struct { wgnet *network.WireGuardNetwork endpointChanges <-chan network.EndpointChangeEvent - server *grpc.Server - corroService corroservice.Service - dockerCli *client.Client + server *grpc.Server + corroService corroservice.Service + dockerCli *client.Client + dockerManager *docker.Manager // dockerReady is signalled when Docker is configured and ready for containers. dockerReady chan<- struct{} caddyconfigCtrl *caddyconfig.Controller @@ -48,6 +50,9 @@ type clusterController struct { // dnsServer is the embedded internal DNS server for the cluster listening on the machine IP. dnsServer *dns.Server dnsResolver *dns.ClusterResolver + + // stopped is a channel that is closed when the controller is stopped. + stopped chan struct{} } func newClusterController( @@ -76,14 +81,18 @@ func newClusterController( server: server, corroService: corroService, dockerCli: dockerCli, + dockerManager: docker.NewManager(dockerCli, state.ID, store), dockerReady: dockerReady, caddyconfigCtrl: caddyfileCtrl, dnsServer: dnsServer, dnsResolver: dnsResolver, + stopped: make(chan struct{}), }, nil } func (cc *clusterController) Run(ctx context.Context) error { + defer close(cc.stopped) + if err := firewall.ConfigureIptablesChains(); err != nil { return fmt.Errorf("configure iptables chains: %w", err) } @@ -210,7 +219,6 @@ func (cc *clusterController) Run(ctx context.Context) error { // It's safe to stop the Corrosion service after the controllers depending on it and API server are stopped. if corroErr := cc.corroService.Stop(ctx); corroErr != nil { - slog.Error("Failed to stop corrosion service.", "err", corroErr) err = errors.Join(err, fmt.Errorf("stop corrosion service: %w", corroErr)) } else { slog.Info("Corrosion service stopped.") @@ -222,12 +230,15 @@ func (cc *clusterController) Run(ctx context.Context) error { // prepareAndWatchDocker configures the Docker network and watches local Docker containers to sync them // to the cluster store. func (cc *clusterController) prepareAndWatchDocker(ctx context.Context) error { - manager := docker.NewManager(cc.dockerCli, cc.state.ID, cc.store) - if err := manager.WaitDaemonReady(ctx); err != nil { + if err := cc.dockerManager.WaitDaemonReady(ctx); err != nil { return fmt.Errorf("wait for Docker daemon: %w", err) } - if err := manager.EnsureUncloudNetwork(ctx, cc.state.Network.Subnet, cc.dnsServer.ListenAddr()); err != nil { + if err := cc.dockerManager.EnsureUncloudNetwork( + ctx, + cc.state.Network.Subnet, + cc.dnsServer.ListenAddr(), + ); err != nil { return fmt.Errorf("ensure Docker network: %w", err) } slog.Info("Docker network configured.") @@ -243,7 +254,7 @@ func (cc *clusterController) prepareAndWatchDocker(ctx context.Context) error { backoff.WithMaxElapsedTime(0), ), ctx) watchAndSync := func() error { - if wErr := manager.WatchAndSyncContainers(ctx); wErr != nil { + if wErr := cc.dockerManager.WatchAndSyncContainers(ctx); wErr != nil { slog.Error("Failed to watch and sync containers to cluster store, retrying.", "err", wErr) return wErr } @@ -385,4 +396,20 @@ func (cc *clusterController) configurePeers(machines []*pb.MachineInfo) error { return nil } -// TODO: method to shutdown network when leaving a cluster. Regular context cancellation shouldn't bring it down. +// Cleanup cleans up the cluster resources such as the WireGuard network, iptables rules, Docker network and containers. +func (cc *clusterController) Cleanup() error { + // Wait for the controller to stop before cleaning up. + <-cc.stopped + + var errs []error + if err := cc.dockerManager.Cleanup(); err != nil { + errs = append(errs, fmt.Errorf("cleanup Docker resources: %w", err)) + } + if err := cc.wgnet.Cleanup(); err != nil { + errs = append(errs, fmt.Errorf("cleanup WireGuard network: %w", err)) + } + // TODO: cleanup custom iptables chains. They're flushed when the machine is initialised again but it would + // cleaner to delete them here. + + return errors.Join(errs...) +} diff --git a/internal/machine/docker/manager.go b/internal/machine/docker/manager.go index 19d9a7da..d572b6e2 100644 --- a/internal/machine/docker/manager.go +++ b/internal/machine/docker/manager.go @@ -155,6 +155,7 @@ func (m *Manager) syncContainersToStore(ctx context.Context) error { Filters: filters.NewArgs( filters.Arg("label", api.LabelServiceID), filters.Arg("label", api.LabelServiceName), + filters.Arg("label", api.LabelManaged), ), }) if err != nil { diff --git a/internal/machine/docker/manager_darwin.go b/internal/machine/docker/manager_darwin.go index 7f65b1c7..e6ce1381 100644 --- a/internal/machine/docker/manager_darwin.go +++ b/internal/machine/docker/manager_darwin.go @@ -12,3 +12,8 @@ import ( func (m *Manager) EnsureUncloudNetwork(ctx context.Context, subnet netip.Prefix, dnsServer netip.Addr) error { return fmt.Errorf("not supported on Darwin") } + +// Cleanup is a stub for Darwin. +func (m *Manager) Cleanup() error { + return fmt.Errorf("not supported on Darwin") +} diff --git a/internal/machine/docker/manager_linux.go b/internal/machine/docker/manager_linux.go index 8c38bca4..08c20d7f 100644 --- a/internal/machine/docker/manager_linux.go +++ b/internal/machine/docker/manager_linux.go @@ -2,11 +2,14 @@ package docker import ( "context" + "errors" "fmt" "log/slog" "net/netip" "strconv" + dockercontainer "github.com/docker/docker/api/types/container" + "github.com/docker/docker/api/types/filters" dnetwork "github.com/docker/docker/api/types/network" "github.com/docker/docker/client" "github.com/docker/docker/libnetwork/iptables" @@ -114,3 +117,75 @@ func configureIptables(bridgeName string, dnsServer netip.Addr) error { return nil } + +// cleanupIptables deletes the iptables rules for the uncloud Docker network. +func cleanupIptables(bridgeName string) error { + ipt := iptables.GetIptable(iptables.IPv4) + // Delete the rule allowing traffic from the WireGuard network to the Docker bridge. + wgRule := []string{"--in-interface", network.WireGuardInterfaceName, "--out-interface", bridgeName, "-j", "ACCEPT"} + if err := ipt.ProgramRule(iptables.Filter, firewall.DockerUserChain, iptables.Delete, wgRule); err != nil { + return fmt.Errorf("delete iptables rule: %w", err) + } + // Rules in uncloud-owned chains will be automatically cleaned up by the machine cleanup. + + return nil +} + +// Cleanup removes all uncloud-managed containers and the uncloud Docker network. +func (m *Manager) Cleanup() error { + ctx := context.Background() + var errs []error + + // Remove uncloud-managed Docker containers. + containers, err := m.client.ContainerList(ctx, dockercontainer.ListOptions{ + All: true, // Include stopped containers. + Filters: filters.NewArgs( + filters.Arg("label", api.LabelManaged), + ), + }) + if err != nil { + errs = append(errs, fmt.Errorf("list uncloud-managed Docker containers: %w", err)) + } else if len(containers) > 0 { + slog.Info("Removing uncloud-managed Docker containers.", "count", len(containers)) + removed := 0 + + for _, ctr := range containers { + err = m.client.ContainerStop(ctx, ctr.ID, dockercontainer.StopOptions{}) + if err != nil && !client.IsErrNotFound(err) { + errs = append(errs, fmt.Errorf("stop container '%s': %w", ctr.ID, err)) + } + + err = m.client.ContainerRemove(ctx, ctr.ID, dockercontainer.RemoveOptions{ + // Remove anonymous volumes created by the container. + RemoveVolumes: true, + }) + if err == nil { + removed++ + } else if !client.IsErrNotFound(err) { + errs = append(errs, fmt.Errorf("remove container '%s': %w", ctr.ID, err)) + } + } + slog.Info("Removed uncloud-managed Docker containers.", "count", removed) + } + + // Remove the uncloud Docker network and related iptables rules. + nw, err := m.client.NetworkInspect(ctx, NetworkName, dnetwork.InspectOptions{}) + if err == nil { + bridgeName := "br-" + nw.ID[:12] + if err = cleanupIptables(bridgeName); err != nil { + errs = append(errs, fmt.Errorf("cleanup iptables for Docker network '%s': %w", NetworkName, err)) + } else { + slog.Info("Cleaned up iptables rules for Docker network.", "name", NetworkName, "bridge", bridgeName) + } + + if err = m.client.NetworkRemove(ctx, NetworkName); err == nil { + slog.Info("Docker network removed.", "name", NetworkName) + } else if !client.IsErrNotFound(err) { + errs = append(errs, fmt.Errorf("remove Docker network '%s': %w", NetworkName, err)) + } + } else if !client.IsErrNotFound(err) { + errs = append(errs, fmt.Errorf("inspect Docker network '%s': %w", NetworkName, err)) + } + + return errors.Join(errs...) +} diff --git a/internal/machine/firewall/iptables_darwin.go b/internal/machine/firewall/iptables_darwin.go index 5202f531..046ec6fc 100644 --- a/internal/machine/firewall/iptables_darwin.go +++ b/internal/machine/firewall/iptables_darwin.go @@ -6,3 +6,8 @@ import "fmt" func ConfigureIptablesChains() error { return fmt.Errorf("not supported on Darwin") } + +// CleanupIptablesChains is a stub for Darwin. +func CleanupIptablesChains() error { + return fmt.Errorf("not supported on Darwin") +} diff --git a/internal/machine/firewall/iptables_linux.go b/internal/machine/firewall/iptables_linux.go index 2ede6805..7928a71f 100644 --- a/internal/machine/firewall/iptables_linux.go +++ b/internal/machine/firewall/iptables_linux.go @@ -16,7 +16,7 @@ const ( // ConfigureIptablesChains sets up custom iptables chains and initial firewall rules for Uncloud networking. func ConfigureIptablesChains() error { - // Ensure iptables UNCLOUD-INPUT chain with a RETURN rule exists. All existing rules are flushed. + // Ensure iptables UNCLOUD-INPUT chain exists. All existing rules are flushed. ipt := iptables.GetIptable(iptables.IPv4) if _, err := ipt.NewChain(UncloudInputChain, iptables.Filter); err != nil { return fmt.Errorf("create iptables chain '%s': %w", UncloudInputChain, err) @@ -24,9 +24,6 @@ func ConfigureIptablesChains() error { if err := ipt.RawCombinedOutput("-t", string(iptables.Filter), "-F", UncloudInputChain); err != nil { return fmt.Errorf("flush iptables chain '%s': %w", UncloudInputChain, err) } - if err := ipt.AddReturnRule(UncloudInputChain); err != nil { - return fmt.Errorf("add the RETURN rule for iptables chain '%s': %w", UncloudInputChain, err) - } // Ensure the main iptables INPUT chain has a jump rule to the UNCLOUD-INPUT chain before any DROP/REJECT rules. jumpRule := []string{"-m", "comment", "--comment", "Uncloud-managed", "-j", UncloudInputChain} diff --git a/internal/machine/machine.go b/internal/machine/machine.go index bd93f690..7ee8f05d 100644 --- a/internal/machine/machine.go +++ b/internal/machine/machine.go @@ -152,6 +152,8 @@ type Machine struct { initialised chan struct{} // networkReady is signalled when the Docker network is configured and ready for containers. networkReady chan struct{} + // resetting is true when the machine is being reset. + resetting bool // stop cancels the Run method context to stop the machine gracefully. stop func() @@ -421,6 +423,8 @@ func (m *Machine) Run(ctx context.Context) error { // Shutdown goroutine. errGroup.Go(func() error { + var err error + <-ctx.Done() slog.Info("Stopping local machine API server.") // TODO: implement timeout for graceful shutdown. @@ -434,8 +438,19 @@ func (m *Machine) Run(ctx context.Context) error { m.proxyDirector.Close() slog.Info("Local API proxy server stopped.") + // Clean up the machine data and resources if the machine shutdown was initiated by a reset. + m.mu.RLock() + resetting := m.resetting + m.mu.RUnlock() + if resetting { + slog.Info("Cleaning up machine data and resources.") + if err = m.cleanup(); err != nil { + slog.Error("Failed to clean up machine data and resources.", "err", err) + } + } + m.config.DockerClient.Close() - return nil + return err }) return errGroup.Wait() @@ -526,6 +541,29 @@ func (m *Machine) configureCorrosion() error { return nil } +// cleanup removes the machine resources and persistent state. +func (m *Machine) cleanup() error { + var errs []error + + m.mu.RLock() + clusterCtrl := m.clusterCtrl + m.mu.RUnlock() + if clusterCtrl != nil { + if err := clusterCtrl.Cleanup(); err != nil { + errs = append(errs, fmt.Errorf("cleanup cluster resources: %w", err)) + } + } + + if err := os.RemoveAll(m.config.DataDir); err != nil { + errs = append(errs, + fmt.Errorf("remove data directory with persistent machine state '%s': %w", m.config.DataDir, err)) + } else { + slog.Info("Removed data directory storing persistent machine state.", "path", m.config.DataDir) + } + + return errors.Join(errs...) +} + // CheckPrerequisites verifies if the machine meets all necessary system requirements to participate in the cluster. func (m *Machine) CheckPrerequisites(ctx context.Context, _ *emptypb.Empty) (*pb.CheckPrerequisitesResponse, error) { // Check DNS port (UDP) availability. @@ -791,33 +829,28 @@ func (m *Machine) WaitForNetworkReady(ctx context.Context) error { } } -// Reset restores the machine to a clean state, removing all cluster-related configuration and data and scheduling -// a graceful shutdown. The uncloud daemon will restart the machine if managed by systemd. +// Reset restores the machine to a clean state, scheduling a graceful shutdown and removing all cluster-related +// configuration and resource. The uncloud daemon will restart the machine if managed by systemd. func (m *Machine) Reset(ctx context.Context, _ *pb.ResetRequest) (*emptypb.Empty, error) { if !m.Initialised() { return nil, nil } + // Check if the machine is already being reset to avoid concurrent resets. + m.mu.Lock() + if m.resetting { + m.mu.Unlock() + return nil, status.Error(codes.FailedPrecondition, "machine is already being reset") + } + m.resetting = true + m.mu.Unlock() + slog.Info("Resetting machine to a clean state.") + // Trigger the machine shutdown. The resetting boolean informs the machine to clean up its resources on shutdown. + // We can't clean up the resources synchronously here because this is an RPC call that depends on the running + // gRPC server and network. + m.stop() - // TODO: stop and remove all managed service containers. - // TODO: check if the request is coming from the unix or network socket. For the network socket, the reset should - // be called in a separate goroutine to avoid blocking the RPC response. - - // TODO: Stop the machine asynchronously. The gRPC servers will wait for this request to complete before stopping. - go func() { - m.stop() - // TODO: wait for the cluster controller to stop. - // TODO: Cleanup cluster controller (WG network, Docker network, iptables rules, etc.) - - // TODO: uncomment after testing all other cleanup steps. - //if err := os.RemoveAll(m.config.DataDir); err != nil { - // slog.Error("Failed to remove data directory storing persistent machine state.", - // "path", m.config.DataDir, "err", err) - // return nil, status.Errorf(codes.Internal, "remove data directory on machine '%s': %v", m.config.DataDir, err) - //} - //slog.Info("Removed data directory storing persistent machine state.", "path", m.config.DataDir) - }() return &emptypb.Empty{}, nil } diff --git a/internal/machine/network/wireguard_darwin.go b/internal/machine/network/wireguard_darwin.go index 99b15faf..595e28bf 100644 --- a/internal/machine/network/wireguard_darwin.go +++ b/internal/machine/network/wireguard_darwin.go @@ -24,3 +24,7 @@ func (n *WireGuardNetwork) Run(ctx context.Context) error { func (n *WireGuardNetwork) WatchEndpoints() <-chan EndpointChangeEvent { return nil } + +func (n *WireGuardNetwork) Cleanup() error { + return errors.New("not implemented on darwin") +} diff --git a/internal/machine/network/wireguard_linux.go b/internal/machine/network/wireguard_linux.go index f11857b7..111237fc 100644 --- a/internal/machine/network/wireguard_linux.go +++ b/internal/machine/network/wireguard_linux.go @@ -27,6 +27,8 @@ type WireGuardNetwork struct { peers map[string]*peer // watchers is a list of channels that are notified when the endpoints of the peers change. watchers []chan EndpointChangeEvent + // running indicates whether the network control loop (Run) is currently running. + running bool // mu synchronises concurrent network configuration changes. mu sync.Mutex } @@ -273,6 +275,14 @@ func (n *WireGuardNetwork) Run(ctx context.Context) error { } defer wg.Close() + n.mu.Lock() + if n.running { + n.mu.Unlock() + return errors.New("network is already running") + } + n.running = true + n.mu.Unlock() + ticker := time.NewTicker(1 * time.Second) for { select { @@ -291,6 +301,11 @@ func (n *WireGuardNetwork) Run(ctx context.Context) error { for _, ch := range n.watchers { close(ch) } + + n.mu.Lock() + n.running = false + n.mu.Unlock() + return nil } } @@ -428,3 +443,23 @@ func (n *WireGuardNetwork) notifyWatchers(ctx context.Context, events []Endpoint } return nil } + +// Cleanup deletes the WireGuard link. The network must not be running when this method is called. +func (n *WireGuardNetwork) Cleanup() error { + n.mu.Lock() + defer n.mu.Unlock() + + if n.running { + return errors.New("network is still running, stop it before cleanup") + } + + // Delete the WireGuard link. + name := n.link.Attrs().Name + if err := netlink.LinkDel(n.link); err != nil { + return fmt.Errorf("delete WireGuard link %q: %w", name, err) + } + n.link = nil + slog.Info("Deleted WireGuard interface.", "name", name) + + return nil +}