feat: machine Reset endpoint with asynchronous resource and data cleanup

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
@@ -41,6 +42,7 @@ type clusterController struct {
 	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...)
+}

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 {

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")
+}

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...)
+}

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")
+}

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}

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.")
-
-	// 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() {
+	// 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: 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
 }
 

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")
+}

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
+}