diff --git a/pkg/manager/manager.go b/pkg/manager/manager.go index 87df7037..4d0ca6c4 100644 --- a/pkg/manager/manager.go +++ b/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, - sm.svcProcessor, &sm.mutex, sm.clientSet, sm.bgpServer, sm.electionMgr, - sm.leaseMgr, sm.routeMgr, sm.nodeLabelManager) + w := sm.newModeWorker() // 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 } } - - // Shutdown function that will wait on this signal, unless we call it ourselves + // Start the shutdown watcher before any slow startup API calls, so a + // 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") - sm.dumpConfiguration(ctx) - case syscall.SIGINT, syscall.SIGTERM: - sm.closing.Store(true) - log.Info("Received kube-vip termination, signaling shutdown") - if cpCluster != nil { - cpCluster.Stop() + if sm.dumping.CompareAndSwap(false, true) { + log.Info("Received SIGUSR1, dumping configuration") + dump := sm.dump + if dump == nil { + dump = sm.dumpConfiguration + } + 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 - cancel() + case syscall.SIGINT, syscall.SIGTERM: + 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 { diff --git a/pkg/manager/manager_dump.go b/pkg/manager/manager_dump.go index e78aa6b7..221f1d08 100644 --- a/pkg/manager/manager_dump.go +++ b/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 - fmt.Printf("Kube-vip Active Service Instances: %d\n", len(instances)) - for i, inst := range instances { - if inst.ServiceSnapshot != nil { - svc := inst.ServiceSnapshot - vipConfigs := "" - for j, cfg := range svc.Status.LoadBalancer.Ingress { - if j > 0 { - vipConfigs += ", " - } - vipConfigs += cfg.IP + services := sm.svcProcessor.ServiceSnapshots() + fmt.Printf("Kube-vip Active Service Instances: %d\n", len(services)) + for i, svc := range services { + vipConfigs := "" + for j, cfg := range svc.Status.LoadBalancer.Ingress { + if j > 0 { + vipConfigs += ", " } - fmt.Printf(" Service %d: %s/%s (Type: %s, VIPs: %s)\n", - i+1, svc.Namespace, svc.Name, svc.Spec.Type, vipConfigs) + vipConfigs += cfg.IP } + 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 { diff --git a/pkg/manager/manager_test.go b/pkg/manager/manager_test.go index adabdca2..92e29255 100644 --- a/pkg/manager/manager_test.go +++ b/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") + } +}