fix(cluster): make Cluster.Stop concurrency-safe and lock Processor.Stop

Signed-off-by: Maximilian Rink <maximilian.rink@telekom.de>

pkg/cluster/cluster.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net/http"
 	"os"
+	"sync"
 	"time"
 
 	log "log/slog"
@@ -22,6 +23,7 @@ import (
 // Cluster - The Cluster object manages the state of the cluster for a particular node
 type Cluster struct {
 	stop                  chan bool
+	stopMutex             sync.Mutex
 	Network               []vip.Network
 	arpMgr                *arp.Manager
 	routeMgr              *route.Manager
@@ -93,6 +95,9 @@ func startNetworking(c *kubevip.Config, intfMgr *networkinterface.Manager) ([]vi
 
 // Stop - Will stop the Cluster and release VIP if needed
 func (cluster *Cluster) Stop() {
+	cluster.stopMutex.Lock()
+	defer cluster.stopMutex.Unlock()
+
 	// Close the stop channel, which will shut down the VIP (if needed)
 	if cluster.stop != nil {
 		close(cluster.stop)

pkg/cluster/cluster_stop_test.go

@@ -0,0 +1,32 @@
+package cluster
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+)
+
+func TestStopConcurrentDoesNotRaceOrPanic(t *testing.T) {
+	c := &Cluster{stop: make(chan bool)}
+	start := make(chan struct{})
+	var wg sync.WaitGroup
+	var panics atomic.Int64
+
+	for range 128 {
+		wg.Go(func() {
+			<-start
+			defer func() {
+				if recover() != nil {
+					panics.Add(1)
+				}
+			}()
+			c.Stop()
+		})
+	}
+
+	close(start)
+	wg.Wait()
+	if got := panics.Load(); got != 0 {
+		t.Fatalf("concurrent Stop panicked %d time(s)", got)
+	}
+}

pkg/services/processor.go

@@ -352,6 +352,9 @@ func (p *Processor) deleteTrackedService(svc *v1.Service) error {
 }
 
 func (p *Processor) Stop() {
+	p.mutex.Lock()
+	defer p.mutex.Unlock()
+
 	for _, instance := range p.ServiceInstances {
 		for _, cluster := range instance.Clusters {
 			cluster.Stop()