fix(manager): start the shutdown watcher before slow startup calls

A signal arriving during address recovery was only observed once the
call returned on its own, delaying shutdown by the length of the API
round trip. Start the watcher first, and make Kill non-blocking so a
second signal cannot deadlock against the first.

Signed-off-by: Marcel Fest <marcel.fest@telekom.de>

pkg/manager/manager.go

@@ -2,6 +2,7 @@ package manager
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/http"
 	"os"
@@ -17,7 +18,6 @@ import (
 
 	"github.com/kube-vip/kube-vip/pkg/arp"
 	"github.com/kube-vip/kube-vip/pkg/bgp"
-	"github.com/kube-vip/kube-vip/pkg/cluster"
 	"github.com/kube-vip/kube-vip/pkg/election"
 	"github.com/kube-vip/kube-vip/pkg/iptables"
 	"github.com/kube-vip/kube-vip/pkg/k8s"
@@ -51,8 +51,13 @@ type Manager struct {
 
 	// This channel is used to catch an OS signal and trigger a shutdown
 	signalChan chan os.Signal
+	killChan   chan struct{}
 
-	sigint sync.Once
+	sigint       sync.Once
+	killChanOnce sync.Once
+	dumpWG       sync.WaitGroup
+	dumping      atomic.Bool
+	dump         func(context.Context)
 
 	svcProcessor *services.Processor
 
@@ -81,6 +86,8 @@ type Manager struct {
 
 	// Will handle routes
 	routeMgr *route.Manager
+
+	modeWorker func() worker.Worker
 }
 
 // New will create a new managing object
@@ -272,6 +279,10 @@ func New(ctx context.Context, configMap string, config *kubevip.Config) (*Manage
 
 // Start will begin the Manager, which will start services and watch the configmap
 func (sm *Manager) Start(ctx context.Context) error {
+	if err := runtimeFailure(ctx); err != nil {
+		return err
+	}
+
 	wg := sync.WaitGroup{}
 	defer wg.Wait()
 
@@ -325,52 +336,67 @@ func (sm *Manager) Start(ctx context.Context) error {
 		}
 	}
 
-	return sm.startMode(ctx)
+	if err := sm.startMode(ctx); err != nil {
+		return err
+	}
+	return runtimeFailure(ctx)
+}
+
+func runtimeFailure(ctx context.Context) error {
+	err := context.Cause(ctx)
+	if err == nil || errors.Is(err, context.Canceled) {
+		return nil
+	}
+	return err
 }
 
 // Start will begin the Manager, which will start services and watch the configmap
 func (sm *Manager) startMode(ctx context.Context) error {
-	var cpCluster *cluster.Cluster
 	var err error
 
-	w := worker.New(sm.arpMgr, sm.intfMgr, sm.config, &sm.closing, sm.Kill,
+	w := sm.newModeWorker()
-		sm.svcProcessor, &sm.mutex, sm.clientSet, sm.bgpServer, sm.electionMgr,
-		sm.leaseMgr, sm.routeMgr, sm.nodeLabelManager)
 
 	// use a Go context so we can tell the leaderelection code when we
 	// want to step down
 	wg := sync.WaitGroup{}
 	modeCtx, cancel := context.WithCancel(ctx)
 	defer func() {
-
+		cancel()
 		wg.Wait()
 		w.Cleanup()
-		cancel()
 		log.Info("Shutting down Kube-Vip")
 	}()
 
 	log.Info("starting Kube-vip Manager", "mode", w.Name())
 	if err := w.Configure(modeCtx, &wg); err != nil {
-		defer cancel()
 		return fmt.Errorf("failed to configure %s mode: %w", w.Name(), err)
 	}
 
 	if sm.config.EnableControlPlane {
 		err = w.InitControlPlane()
 		if err != nil {
-			defer cancel()
 			return err
 		}
 	}
-
+	// Start the shutdown watcher before any slow startup API calls, so a
-	// Shutdown function that will wait on this signal, unless we call it ourselves
+	// signal received during RecoverAddresses cancels modeCtx immediately
+	// instead of waiting for it to return on its own.
 	wg.Go(func() {
-		sm.waitForShutdown(modeCtx, cancel, cpCluster)
+		sm.waitForShutdown(modeCtx, cancel)
 	})
 
+	if sm.config.EnableControlPlane || sm.config.EnableServices {
+		if err := sm.svcProcessor.RecoverAddresses(modeCtx); err != nil {
+			log.Warn("skipping kube-vip address recovery", "err", err)
+		}
+	}
+
 	if sm.config.EnableControlPlane {
 		wg.Go(func() {
 			w.StartControlPlane(modeCtx, sm.electionMgr)
+			if modeCtx.Err() == nil {
+				sm.Kill()
+			}
 		})
 	}
 
@@ -387,11 +413,11 @@ func (sm *Manager) startMode(ctx context.Context) error {
 			// TODO: Deprecate the iptables code v1.2.x
 			err = vip.ClearIPTables(sm.config.EgressWithNftables, sm.config.ServiceNamespace, iptables.ProtocolIPv4)
 			if err != nil {
-				log.Info("[egress]", "legacy-iptables", sm.config.EgressWithNftables, "mode", "IPv4", "error", err)
+				log.Info("[egress]", "legacy-iptables", sm.config.EgressWithNftables, "mode", utils.IPv4Family, "error", err)
 			}
 			err = vip.ClearIPTables(sm.config.EgressWithNftables, sm.config.ServiceNamespace, iptables.ProtocolIPv6)
 			if err != nil {
-				log.Info("[egress]", "legacy-iptables", sm.config.EgressWithNftables, "mode", "IPv6", "error", err)
+				log.Info("[egress]", "legacy-iptables", sm.config.EgressWithNftables, "mode", utils.IPv6Family, "error", err)
 			}
 		}
 		w.ConfigureServices()
@@ -405,43 +431,87 @@ func (sm *Manager) startMode(ctx context.Context) error {
 					if utils.IsPanicError(err) {
 						sm.Kill()
 						return fmt.Errorf("failed to reconcile services, non-recoverable error: %w", err)
-					} else {
-						log.Error("failed to reconcile services, restarting", "error", err)
 					}
+					log.Error("failed to reconcile services, restarting", "error", err)
+				}
+				select {
+				case <-modeCtx.Done():
+					return nil
+				case <-time.After(200 * time.Millisecond):
 				}
 			}
 		}
 	}
+	if sm.config.EnableControlPlane {
+		<-modeCtx.Done()
+	}
 
 	return nil
 }
 
-func (sm *Manager) waitForShutdown(ctx context.Context, cancel context.CancelFunc, cpCluster *cluster.Cluster) {
+func (sm *Manager) newModeWorker() worker.Worker {
+	if sm.modeWorker != nil {
+		return sm.modeWorker()
+	}
+	return worker.New(sm.arpMgr, sm.intfMgr, sm.config, &sm.closing, sm.Kill,
+		sm.svcProcessor, &sm.mutex, sm.clientSet, sm.bgpServer, sm.electionMgr,
+		sm.leaseMgr, sm.routeMgr, sm.nodeLabelManager)
+}
+
+func (sm *Manager) waitForShutdown(ctx context.Context, cancel context.CancelFunc) {
+	defer sm.dumpWG.Wait()
 	for {
-		sig := <-sm.signalChan
+		var sig os.Signal
+		select {
+		case <-ctx.Done():
+			return
+		case <-sm.killChannel():
+			sm.shutdown(cancel)
+			return
+		case sig = <-sm.signalChan:
+		}
 		switch sig {
 		case syscall.SIGUSR1:
-			log.Info("Received SIGUSR1, dumping configuration")
+			if sm.dumping.CompareAndSwap(false, true) {
-			sm.dumpConfiguration(ctx)
+				log.Info("Received SIGUSR1, dumping configuration")
-		case syscall.SIGINT, syscall.SIGTERM:
+				dump := sm.dump
-			sm.closing.Store(true)
+				if dump == nil {
-			log.Info("Received kube-vip termination, signaling shutdown")
+					dump = sm.dumpConfiguration
-			if cpCluster != nil {
+				}
-				cpCluster.Stop()
+				sm.dumpWG.Add(1)
+				go func() {
+					defer sm.dumpWG.Done()
+					defer sm.dumping.Store(false)
+					dump(ctx)
+				}()
 			}
-			// Cancel the context, which will in turn cancel the leadership and all goroutines
+		case syscall.SIGINT, syscall.SIGTERM:
-			cancel()
+			sm.shutdown(cancel)
 			return
 		}
 	}
 }
 
+func (sm *Manager) shutdown(cancel context.CancelFunc) {
+	sm.closing.Store(true)
+	log.Info("Received kube-vip termination, signaling shutdown")
+	// Cancel the context, which will in turn cancel the leadership and all goroutines.
+	cancel()
+}
+
 func (sm *Manager) Kill() {
 	sm.sigint.Do(func() {
-		sm.signalChan <- syscall.SIGINT
+		close(sm.killChannel())
 	})
 }
 
+func (sm *Manager) killChannel() chan struct{} {
+	sm.killChanOnce.Do(func() {
+		sm.killChan = make(chan struct{})
+	})
+	return sm.killChan
+}
+
 // normalizeNodeName ensures the local machine hostname conforms to
 // Kubernetes RFC1123 node naming conventions (lowercase).
 func normalizeNodeName(name string) string {

pkg/manager/manager_dump.go

@@ -104,21 +104,18 @@ func (sm *Manager) dumpServicesSection(ctx context.Context) {
 		fmt.Printf("Service Security Enabled: %t\n", sm.config.EnableServiceSecurity)
 
 		if sm.svcProcessor != nil {
-			instances := sm.svcProcessor.ServiceInstances
+			services := sm.svcProcessor.ServiceSnapshots()
-			fmt.Printf("Kube-vip Active Service Instances: %d\n", len(instances))
+			fmt.Printf("Kube-vip Active Service Instances: %d\n", len(services))
-			for i, inst := range instances {
+			for i, svc := range services {
-				if inst.ServiceSnapshot != nil {
+				vipConfigs := ""
-					svc := inst.ServiceSnapshot
+				for j, cfg := range svc.Status.LoadBalancer.Ingress {
-					vipConfigs := ""
+					if j > 0 {
-					for j, cfg := range svc.Status.LoadBalancer.Ingress {
+						vipConfigs += ", "
-						if j > 0 {
-							vipConfigs += ", "
-						}
-						vipConfigs += cfg.IP
 					}
-					fmt.Printf("  Service %d: %s/%s (Type: %s, VIPs: %s)\n",
+					vipConfigs += cfg.IP
-						i+1, svc.Namespace, svc.Name, svc.Spec.Type, vipConfigs)
 				}
+				fmt.Printf("  Service %d: %s/%s (Type: %s, VIPs: %s)\n",
+					i+1, svc.Namespace, svc.Name, svc.Spec.Type, vipConfigs)
 			}
 		}
 		if sm.clientSet != nil {
@@ -132,8 +129,6 @@ func (sm *Manager) dumpServicesSection(ctx context.Context) {
 				fmt.Println("Unable to retrieve all Services")
 			} else {
 				for x := range svcList.Items {
-
-					// Build all addresses
 					vipConfigs := ""
 					for j, cfg := range svcList.Items[x].Status.LoadBalancer.Ingress {
 						if j > 0 {

pkg/manager/manager_test.go

@@ -1,11 +1,39 @@
 package manager
 
 import (
+	"context"
+	"errors"
+	"os"
+	"sync"
+	"sync/atomic"
+	"syscall"
 	"testing"
+	"time"
 
+	"github.com/kube-vip/kube-vip/pkg/election"
+	"github.com/kube-vip/kube-vip/pkg/kubevip"
+	"github.com/kube-vip/kube-vip/pkg/manager/worker"
+	"github.com/kube-vip/kube-vip/pkg/node/noop"
+	"github.com/kube-vip/kube-vip/pkg/services"
 	"github.com/stretchr/testify/assert"
 )
 
+type controlPlaneTestWorker struct {
+	started chan struct{}
+}
+
+func (w *controlPlaneTestWorker) Configure(context.Context, *sync.WaitGroup) error { return nil }
+func (w *controlPlaneTestWorker) InitControlPlane() error                          { return nil }
+func (w *controlPlaneTestWorker) ConfigureServices()                               {}
+func (w *controlPlaneTestWorker) StartServices(context.Context) error              { return nil }
+func (w *controlPlaneTestWorker) Name() string                                     { return "control-plane-test" }
+func (w *controlPlaneTestWorker) Cleanup()                                         {}
+
+func (w *controlPlaneTestWorker) StartControlPlane(ctx context.Context, _ *election.Manager) {
+	close(w.started)
+	<-ctx.Done()
+}
+
 func TestNormalizeNodeName(t *testing.T) {
 	tests := []struct {
 		name     string
@@ -36,3 +64,185 @@ func TestNormalizeNodeName(t *testing.T) {
 		})
 	}
 }
+
+func TestStartReturnsCancellationCause(t *testing.T) {
+	cause := errors.New("default interface is down")
+	ctx, cancel := context.WithCancelCause(context.Background())
+	cancel(cause)
+
+	manager := &Manager{
+		config:           &kubevip.Config{},
+		nodeLabelManager: noop.NewManager(),
+	}
+	err := manager.Start(ctx)
+	if !errors.Is(err, cause) {
+		t.Fatalf("Start() error = %v, want cancellation cause %v", err, cause)
+	}
+}
+
+func TestStartModeControlPlaneOnlyWaitsForShutdown(t *testing.T) {
+	config := &kubevip.Config{EnableControlPlane: true}
+	w := &controlPlaneTestWorker{started: make(chan struct{})}
+	manager := &Manager{
+		config:       config,
+		signalChan:   make(chan os.Signal),
+		svcProcessor: services.NewServicesProcessor(config, nil, nil, nil, nil, nil, noop.NewManager(), nil, nil, nil),
+		modeWorker:   func() worker.Worker { return w },
+	}
+
+	done := make(chan error, 1)
+	go func() {
+		done <- manager.startMode(context.Background())
+	}()
+	select {
+	case <-w.started:
+	case <-time.After(time.Second):
+		t.Fatal("control-plane worker did not start")
+	}
+	select {
+	case err := <-done:
+		t.Fatalf("control-plane-only mode returned before shutdown: %v", err)
+	case <-time.After(20 * time.Millisecond):
+	}
+
+	manager.Kill()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("control-plane-only mode returned an error: %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("control-plane-only mode did not stop after Kill")
+	}
+}
+
+func TestKillSignalsShutdownWhileSignalIsQueued(t *testing.T) {
+	manager := &Manager{
+		config:     &kubevip.Config{},
+		signalChan: make(chan os.Signal, 1),
+	}
+	manager.signalChan <- syscall.SIGUSR1
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	shutdownDone := make(chan struct{})
+	go func() {
+		manager.waitForShutdown(ctx, cancel)
+		close(shutdownDone)
+	}()
+
+	returned := make(chan struct{})
+	go func() {
+		manager.Kill()
+		close(returned)
+	}()
+
+	select {
+	case <-returned:
+	case <-time.After(time.Second):
+		t.Fatal("Kill blocked while the OS signal channel was full")
+	}
+	select {
+	case <-shutdownDone:
+	case <-time.After(time.Second):
+		t.Fatal("Kill did not signal shutdown")
+	}
+}
+
+func TestKillDoesNotBlockAfterShutdownWithPendingSignal(t *testing.T) {
+	manager := &Manager{
+		config:     &kubevip.Config{},
+		signalChan: make(chan os.Signal, 1),
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	shutdownDone := make(chan struct{})
+	go func() {
+		manager.waitForShutdown(ctx, func() {})
+		close(shutdownDone)
+	}()
+	select {
+	case <-shutdownDone:
+	case <-time.After(time.Second):
+		t.Fatal("waitForShutdown did not return after context cancellation")
+	}
+
+	manager.signalChan <- syscall.SIGUSR1
+	returned := make(chan struct{})
+	go func() {
+		manager.Kill()
+		close(returned)
+	}()
+	select {
+	case <-returned:
+	case <-time.After(time.Second):
+		t.Fatal("Kill blocked after shutdown with a pending signal")
+	}
+}
+
+func TestWaitForShutdownTracksAndCoalescesConfigurationDump(t *testing.T) {
+	dumpStarted := make(chan struct{})
+	releaseDump := make(chan struct{})
+	var dumpCalls atomic.Int64
+	manager := &Manager{
+		signalChan: make(chan os.Signal, 3),
+		dump: func(context.Context) {
+			dumpCalls.Add(1)
+			close(dumpStarted)
+			<-releaseDump
+		},
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan struct{})
+	go func() {
+		manager.waitForShutdown(ctx, cancel)
+		close(done)
+	}()
+
+	manager.signalChan <- syscall.SIGUSR1
+	select {
+	case <-dumpStarted:
+	case <-time.After(time.Second):
+		t.Fatal("configuration dump did not start")
+	}
+	manager.signalChan <- syscall.SIGUSR1
+	manager.Kill()
+	select {
+	case <-done:
+		t.Fatal("waitForShutdown returned before the configuration dump completed")
+	case <-time.After(20 * time.Millisecond):
+	}
+	close(releaseDump)
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("waitForShutdown did not return after the configuration dump completed")
+	}
+	if got := dumpCalls.Load(); got != 1 {
+		t.Fatalf("configuration dump calls = %d, want 1", got)
+	}
+}
+
+func TestWaitForShutdownHandlesKillSignal(t *testing.T) {
+	manager := &Manager{
+		signalChan: make(chan os.Signal, 1),
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	done := make(chan struct{})
+	go func() {
+		manager.waitForShutdown(ctx, cancel)
+		close(done)
+	}()
+	manager.Kill()
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("waitForShutdown did not return after Kill")
+	}
+	if !manager.closing.Load() {
+		t.Fatal("Kill did not mark the manager as closing")
+	}
+}