client: add network reload and delete functionality to client mode.

internal/agent/agent.go

@@ -26,6 +26,7 @@ type Agent struct {
 	httpServer *http.Server
 	start      func() error
 	stop       func() error
+	reload     func()
 }
 
 // AgentReq is the configuration used to construct a new Agent.
@@ -39,6 +40,10 @@ type AgentReq struct {
 
 	// Stop is called by Agent.Stop after the HTTP server has been shut down.
 	Stop func() error
+
+	// Reload is an optional function called when SIGHUP is received. It
+	// should trigger a non-blocking reload of dynamic configuration.
+	Reload func()
 }
 
 // New constructs an Agent from the provided AgentReq. If HTTP is enabled in
@@ -53,6 +58,7 @@ func New(req *AgentReq) (*Agent, error) {
 		logger: req.Logger.Named(log.ComponentNameAgent),
 		start:  req.Start,
 		stop:   req.Stop,
+		reload: req.Reload,
 	}
 
 	if req.HTTPConfig != nil && req.HTTPConfig.Enabled != nil && *req.HTTPConfig.Enabled {
@@ -134,7 +140,10 @@ func (a *Agent) WaitForSignal() {
 
 		switch sig {
 		case syscall.SIGHUP:
-			a.logger.Info("SIGHUP received, configuration reload not yet implemented")
+			a.logger.Info("SIGHUP received, reloading agent")
+			if a.reload != nil {
+				a.reload()
+			}
 		default:
 			a.logger.Info("shutting down")
 			if err := a.Stop(); err != nil {

internal/agent/client.go

@@ -44,5 +44,6 @@ func NewClientAgent(cfg *config.ClientAgentConfig) (*Agent, error) {
 		HTTPConfig: cfg.HTTP,
 		Start:      cl.Start,
 		Stop:       cl.Stop,
+		Reload:     cl.Reload,
 	})
 }

internal/client/client.go

@@ -44,11 +44,23 @@ type Client struct {
 	networkManager *network.Manager
 
 	// networks tracks the networks that this Smuggle client is aware of and
-	// should configure on the host.
-	networks []*types.Network
+	// should configure on the host. This is used to compare against updates
+	// from the store to determine when networks are added or removed. All
+	// access to this slice must be synchronized using the networksLock mutex.
+	networks     []*types.Network
+	networksLock sync.RWMutex
+
+	subnetWatchers     map[string]*subnetWatcher
+	subnetWatchersLock sync.Mutex
+
+	// subnetHeartbeaters tracks the currently running subnet heartbeaters. It
+	// is keyed by subnet ID and all access must be synchronized using the
+	// subnetHeartbeatersLock mutex.
+	subnetHeartbeaters     map[string]*heartbeater
+	subnetHeartbeatersLock sync.Mutex
 
 	//
-	subnets []*types.Subnet
+	reloadCh chan struct{}
 
 	// shutdownCh is used to signal to all client processes that the agent is
 	// shutting down. All long-running processes should monitor this channel and
@@ -73,31 +85,44 @@ func New(req *ClientReq) (*Client, error) {
 	}
 
 	return &Client{
-		cfg:            req.Config,
-		logger:         req.Logger.Named(log.ComponentNameClient),
-		networks:       []*types.Network{},
-		store:          req.Store,
-		cniStore:       req.CNIStore,
-		networkManager: netManager,
-		shutdownCh:     make(chan struct{}),
+		cfg:                req.Config,
+		logger:             req.Logger.Named(log.ComponentNameClient),
+		networks:           []*types.Network{},
+		store:              req.Store,
+		cniStore:           req.CNIStore,
+		networkManager:     netManager,
+		reloadCh:           make(chan struct{}, 1),
+		shutdownCh:         make(chan struct{}),
+		subnetWatchers:     make(map[string]*subnetWatcher),
+		subnetHeartbeaters: make(map[string]*heartbeater),
 	}, nil
 }
 
+// Reload signals the client to immediately re-read the network list from the
+// store. If a reload is already pending, the signal is dropped to avoid
+// queuing up redundant work.
+func (c *Client) Reload() {
+	select {
+	case c.reloadCh <- struct{}{}:
+	default:
+		// If the channel is full, do not block. This means a reload is already
+		// pending, so we do not need to send another signal.
+	}
+}
+
 func (c *Client) Start() error {
 
 	if err := c.generateID(); err != nil {
 		return fmt.Errorf("failed to get client ID: %w", err)
 	}
 
-	if err := c.Init(); err != nil {
-		return fmt.Errorf("failed to initialize client: %w", err)
-	}
-
-	if err := c.startSubnetUpdateHandler(); err != nil {
-		return fmt.Errorf("failed to start remote subnet handler: %w", err)
-	}
+	// Perform the initial network setup synchronously so that all local subnets
+	// and remote subnet routes are in place before Start returns. The
+	// background monitor handles periodic refreshes and SIGHUP reloads after
+	// this point.
+	c.triggerNetworksRead()
 
-	c.startHeartbeaters()
+	go c.monitorNetworks()
 
 	return nil
 }
@@ -131,107 +156,7 @@ func (c *Client) Stop() error {
 	return nil
 }
 
-func (c *Client) Init() error {
-
-	// Read all network configurations from the store that we are able to see
-	// and therefore should configure on this host.
-	listResp, err := c.store.ListNetworks(nil)
-	if err != nil {
-		return fmt.Errorf("failed to get network configs: %w", err)
-	}
-
-	if len(listResp.Networks) == 0 {
-		return errors.New("no networks configurations found")
-	}
-
-	for _, networkConfig := range listResp.Networks {
-
-		// Validate the network configuration.
-		if err := networkConfig.Validate(); err != nil {
-			return fmt.Errorf("invalid network: %w", err)
-		}
-
-		c.networks = append(c.networks, networkConfig)
-
-		clientSubnetResp, err := c.store.GetSubnet(&types.StoreGetSubnetReq{
-			ID:          c.id.Load().(string),
-			NetworkName: networkConfig.Name,
-		})
-		if err != nil {
-			return fmt.Errorf("failed to get client subnet config: %w", err)
-		}
-
-		// Perform the canonicalization, so we have all fields set correctly
-		// set. It would be possible to write this back to the data store, but
-		// seeing as this happens on the client, if more than one started at the
-		// same time, they would all race to write it back.
-		networkConfig.Canonicalize()
-
-		subnet := clientSubnetResp.Subnet
-
-		if subnet == nil {
-
-			//
-			subnetListReq := types.StoreListSubnetsReq{Network: networkConfig.Name}
-
-			subnetListResp, err := c.store.ListSubnets(&subnetListReq)
-			if err != nil {
-				return fmt.Errorf("failed to list existing client subnets: %w", err)
-			}
-
-			subnet, err = c.networkManager.GenerateIPv4Subnet(c.getID(), networkConfig, subnetListResp.Subnets)
-			if err != nil {
-				return fmt.Errorf("failed to generate IPv4 subnet: %w", err)
-			}
-		}
-
-		c.logger.Info("initializing local host subnet", networkConfig.LoggingPairs()...)
-
-		if err := c.initSubnet(networkConfig, subnet); err != nil {
-			return fmt.Errorf("failed to initialize subnet: %w", err)
-		}
-
-		c.subnets = append(c.subnets, subnet)
-
-		if networkConfig.IPMasq != nil && *networkConfig.IPMasq {
-			if err := c.networkManager.Firewall.SetupMasqRules(networkConfig, subnet); err != nil {
-				return fmt.Errorf("failed to set up firewall masquerade rules: %w", err)
-			}
-		}
-
-		if err := c.networkManager.Firewall.SetupForwardRules(networkConfig); err != nil {
-			return fmt.Errorf("failed to set up firewall forward rules: %w", err)
-		}
-
-		c.logger.Info("successfully initialized local host subnet", subnet.LoggingPairs()...)
-	}
-
-	if err := c.networkManager.Firewall.EnsureIsolation(c.networks); err != nil {
-		return fmt.Errorf("failed to ensure network isolation: %w", err)
-	}
-
-	return nil
-}
-
-func (c *Client) initSubnet(netCfg *types.Network, cfg *types.Subnet) error {
-
-	providerResp, err := c.networkManager.SetLocal(&types.NetworkProviderSetReq{Client: cfg})
-	if err != nil {
-		return fmt.Errorf("failed to set up local subnet: %w", err)
-	}
-
-	if _, err := c.store.SetSubnet(&types.StoreSetSubnetReq{
-		Subnet: providerResp.Network,
-	}); err != nil {
-		return fmt.Errorf("failed to store client subnet: %w", err)
-	}
-
-	if err := c.cniStore.Set(types.GenerateCNIConfig(netCfg, cfg)); err != nil {
-		return fmt.Errorf("failed to write CNI config: %w", err)
-	}
-
-	return nil
-}
+func (c *Client) shutdownGroupDecrement() { c.shutdownGroup.Done() }
 
 // generateID attempts to read the client ID from disk. If the file does not exist,
 // it generates a new UUID, saves it to disk, and returns it.

internal/client/heartbeat.go

@@ -1,22 +1,43 @@
 package client
 
 import (
+	"sync"
 	"time"
 
 	"go.uber.org/zap"
 
 	"github.com/rasorp/smuggle/internal/types"
 )
 
-func (c *Client) startHeartbeaters() {
-	for _, subnet := range c.subnets {
-		go c.startSubnetHeartbeat(subnet)
-	}
+// heartbeater is responsible for periodically updating the expiration time
+// of a subnet in the store to indicate that the client is still active and
+// using that subnet.
+type heartbeater struct {
+	logger *zap.Logger
+	store  types.Store
+	subnet *types.Subnet
+
+	// shutdownCh is the client shutdown channel used to indicate that the agent
+	// is shutting down and all long-running processes should exit. This is a
+	// coarse-grained signal.
+	shutdownCh chan struct{}
+
+	// stopCh is used to signal to this specific heartbeater that it should
+	// stop. This allows for more fine-grained control, such as when a subnet is
+	// removed.
+	stopCh chan struct{}
+
+	// stopWGFn is a callback function that should be called when the
+	// heartbeater has fully stopped. This allows for proper coordination of
+	// shutdown processes across the client.
+	stopWGFn func()
+
+	// stopOnce ensures that the stop process is only initiated once, preventing
+	// potential issues from multiple stop signals.
+	stopOnce sync.Once
 }
 
-func (c *Client) startSubnetHeartbeat(subnet *types.Subnet) {
-	c.shutdownGroup.Add(1)
-	defer c.shutdownGroup.Done()
+func (h *heartbeater) start() {
 
 	// Calculate the heartbeat interval as half of the TTL to ensure we update
 	// before expiration. This provides a safety margin.
@@ -25,29 +46,25 @@ func (c *Client) startSubnetHeartbeat(subnet *types.Subnet) {
 	ticker := time.NewTicker(heartbeatInterval)
 	defer ticker.Stop()
 
-	// We may log more than one message, so caputre the pairs here to avoid
-	// multiple calls to the function and slice allocations. All fields are
-	// static.
-	logPairs := subnet.LoggingPairs()
-
-	c.logger.Info("starting subnet heartbeat",
-		append(logPairs, zap.String("interval", heartbeatInterval.String()))...,
+	h.logger.Info("starting subnet heartbeat",
+		append(h.subnet.LoggingPairs(), zap.String("interval", heartbeatInterval.String()))...,
 	)
 
+	// This is a small codebase currently and we known this cannot be nil. In
+	// the future, if this code is refactored or reused in other contexts, we
+	// may want to add some additional safety checks or validation.
+	defer h.stopWGFn()
+
 	for {
 		select {
 		case <-ticker.C:
 			// Create a copy of the subnet config to update the expiration time
 			// without modifying the original reference. Then write this update
 			// back to the store.
-			subnetCopy := subnet.Copy()
+			subnetCopy := h.subnet.Copy()
 			subnetCopy.Expiration = time.Now().Add(types.DefaultSubnetTTL)
 
-			c.logger.Debug("updating subnet expiration",
-				append(logPairs, zap.Time("expiration", subnetCopy.Expiration))...,
-			)
-
-			_, err := c.store.SetSubnet(&types.StoreSetSubnetReq{Subnet: subnetCopy})
+			_, err := h.store.SetSubnet(&types.StoreSetSubnetReq{Subnet: subnetCopy})
 
 			// Adjust the ticker interval based on success or failure. On
 			// success, we maintain the regular interval. On failure, we shorten
@@ -60,19 +77,28 @@ func (c *Client) startSubnetHeartbeat(subnet *types.Subnet) {
 			case nil:
 				ticker.Reset(types.DefaultSubnetTTL / 3)
 
-				c.logger.Info("successfully updated subnet expiration",
-					append(logPairs, zap.Time("expiration", subnetCopy.Expiration))...,
+				h.subnet = subnetCopy
+
+				h.logger.Debug("updated subnet expiration",
+					zap.String("network", subnetCopy.NetworkName),
+					zap.Time("new_expiration", subnetCopy.Expiration),
 				)
 			default:
 				ticker.Reset(10 * time.Second)
 
-				c.logger.Error("failed to update subnet expiration",
-					append(logPairs, zap.Error(err))...,
+				h.logger.Error("failed to update subnet expiration",
+					zap.String("network", subnetCopy.NetworkName),
+					zap.Error(err),
 				)
 			}
-		case <-c.shutdownCh:
-			c.logger.Info("shutting down subnet heartbeat", logPairs...)
+		case <-h.shutdownCh:
+			h.logger.Info("shutting down subnet heartbeat", zap.String("network", h.subnet.NetworkName))
+			return
+		case <-h.stopCh:
+			h.logger.Info("stopping subnet heartbeat", zap.String("network", h.subnet.NetworkName))
 			return
 		}
 	}
 }
+
+func (h *heartbeater) stop() { h.stopOnce.Do(func() { close(h.stopCh) }) }