fix(servicecontext): synchronize access to shared context fields

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

pkg/endpoints/endpoints_bgp.go

@@ -66,9 +66,7 @@ func (b *BGP) clear(svcCtx *servicecontext.Context, lastKnownGoodEndpoint *strin
 
 	b.clearEgress(lastKnownGoodEndpoint, service)
 
-	if svcCtx.LeaderCancel != nil {
-		svcCtx.LeaderCancel()
-	}
+	svcCtx.CallLeaderCancel()
 }
 
 func (b *BGP) getEndpoints(service *v1.Service, id string) ([]string, error) {

pkg/endpoints/endpoints_generic.go

@@ -64,9 +64,7 @@ func (g *generic) processInstance(_ *servicecontext.Context, _ *v1.Service) erro
 
 func (g *generic) clear(svcCtx *servicecontext.Context, lastKnownGoodEndpoint *string, service *v1.Service) {
 	g.clearEgress(lastKnownGoodEndpoint, service)
-	if svcCtx.LeaderCancel != nil {
-		svcCtx.LeaderCancel()
-	}
+	svcCtx.CallLeaderCancel()
 }
 
 func (g *generic) clearEgress(lastKnownGoodEndpoint *string, service *v1.Service) {

pkg/endpoints/endpoints_routing_table.go

@@ -66,9 +66,7 @@ func (rt *RoutingTable) clear(svcCtx *servicecontext.Context, lastKnownGoodEndpo
 
 	rt.clearEgress(lastKnownGoodEndpoint, service)
 
-	if svcCtx.LeaderCancel != nil {
-		svcCtx.LeaderCancel()
-	}
+	svcCtx.CallLeaderCancel()
 }
 
 func (rt *RoutingTable) getEndpoints(service *v1.Service, id string) ([]string, error) {

pkg/endpoints/endpoints_wireguard.go

@@ -212,9 +212,7 @@ func (w *wireguardWorker) clear(svcCtx *servicecontext.Context, lastKnownGoodEnd
 		}
 	}
 
-	if svcCtx != nil && svcCtx.LeaderCancel != nil {
-		svcCtx.LeaderCancel()
-	}
+	svcCtx.CallLeaderCancel()
 }
 
 // getEndpoints retrieves the list of endpoints for a service

pkg/servicecontext/servicecontext.go

@@ -12,6 +12,7 @@ type Context struct {
 	IsWatched          bool
 	ConfiguredNetworks sync.Map
 	EndpointsReady     chan any
+	mu                 sync.Mutex
 	epReady            sync.Once
 	leaderElection     sync.Once
 	Signalled          atomic.Bool
@@ -51,6 +52,9 @@ func (ctx *Context) StartLeaderElectionOnce(f func()) {
 }
 
 func (ctx *Context) SignalReadiness() {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+
 	ctx.epReady.Do(func() {
 		close(ctx.EndpointsReady)
 		ctx.Signalled.Store(true)
@@ -58,9 +62,50 @@ func (ctx *Context) SignalReadiness() {
 }
 
 func (ctx *Context) ResetReadiness() {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+
 	if ctx.Signalled.Load() {
 		ctx.EndpointsReady = make(chan any)
 		ctx.epReady = sync.Once{}
 		ctx.Signalled.Store(false)
 	}
 }
+
+func (ctx *Context) GetEndpointsReady() chan any {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+
+	return ctx.EndpointsReady
+}
+
+func (ctx *Context) SetLeaderCancel(cancel context.CancelFunc) {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+
+	ctx.LeaderCancel = cancel
+}
+
+func (ctx *Context) CallLeaderCancel() {
+	ctx.mu.Lock()
+	cancel := ctx.LeaderCancel
+	ctx.mu.Unlock()
+
+	if cancel != nil {
+		cancel()
+	}
+}
+
+func (ctx *Context) SetWatched(watched bool) {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+
+	ctx.IsWatched = watched
+}
+
+func (ctx *Context) IsWatchedLocked() bool {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+
+	return ctx.IsWatched
+}

pkg/servicecontext/servicecontext_race_test.go

@@ -0,0 +1,56 @@
+package servicecontext
+
+import (
+	"context"
+	"sync"
+	"testing"
+)
+
+func TestReadinessResetConcurrentWithSignal(t *testing.T) {
+	ctx := New(context.Background())
+	start := make(chan struct{})
+	var wg sync.WaitGroup
+
+	wg.Go(func() {
+		<-start
+		for range 1000 {
+			ctx.SignalReadiness()
+			ctx.ResetReadiness()
+		}
+	})
+	wg.Go(func() {
+		<-start
+		for range 1000 {
+			ready := ctx.GetEndpointsReady()
+			select {
+			case <-ready:
+			default:
+			}
+		}
+	})
+
+	close(start)
+	wg.Wait()
+}
+
+func TestLeaderCancelConcurrentWithEndpointCleanup(t *testing.T) {
+	ctx := New(context.Background())
+	start := make(chan struct{})
+	var wg sync.WaitGroup
+
+	wg.Go(func() {
+		<-start
+		for range 1000 {
+			ctx.SetLeaderCancel(func() {})
+		}
+	})
+	wg.Go(func() {
+		<-start
+		for range 1000 {
+			ctx.CallLeaderCancel()
+		}
+	})
+
+	close(start)
+	wg.Wait()
+}

pkg/services/leader.go

@@ -99,7 +99,7 @@ func (p *Processor) StartServicesLeaderElection(svcCtx *servicecontext.Context,
 		return fmt.Errorf("service context cancelled before election start: %w", svcCtx.Ctx.Err())
 	case <-svcLease.Ctx.Done():
 		return fmt.Errorf("lease context cancelled before election start: %w", svcLease.Ctx.Err())
-	case <-svcCtx.EndpointsReady:
+	case <-svcCtx.GetEndpointsReady():
 	}
 
 	// this service is sharing lease with another service
@@ -135,7 +135,7 @@ func (p *Processor) StartServicesLeaderElection(svcCtx *servicecontext.Context,
 	log.Info("new leader election", "service", service.Name, "namespace", service.Namespace, "lock_name", serviceLease, "host_id", p.config.NodeName)
 
 	leaderCtx, leaderCancel := context.WithCancel(svcLease.Ctx)
-	svcCtx.LeaderCancel = leaderCancel
+	svcCtx.SetLeaderCancel(leaderCancel)
 
 	run := election.RunConfig{
 		Config:           p.config,

pkg/services/processor.go

@@ -223,13 +223,13 @@ func (p *Processor) AddOrModify(ctx context.Context, event watch.Event, serviceF
 	}
 
 	// this goroutine starts service handling function (with or without leaderelection)
-	if !svcCtx.IsWatched {
+	if !svcCtx.IsWatchedLocked() {
 		wg.Go(func() {
 			watchWg := sync.WaitGroup{}
 			defer func() {
 				// wait for the sub-goroutines and tag service as not watched
 				watchWg.Wait()
-				svcCtx.IsWatched = false
+				svcCtx.SetWatched(false)
 			}()
 
 			watchWg.Go(func() {
@@ -268,7 +268,7 @@ func (p *Processor) AddOrModify(ctx context.Context, event watch.Event, serviceF
 		})
 
 		// tag service as watched
-		svcCtx.IsWatched = true
+		svcCtx.SetWatched(true)
 	}
 
 	if !p.config.EnableServicesElection {

pkg/services/processor_race_test.go

@@ -0,0 +1,33 @@
+package services
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"github.com/kube-vip/kube-vip/pkg/servicecontext"
+)
+
+func TestWatchedFlagConcurrentWithWatcherTeardown(t *testing.T) {
+	svcCtx := servicecontext.New(context.Background())
+	start := make(chan struct{})
+	var wg sync.WaitGroup
+
+	wg.Go(func() {
+		<-start
+		for range 1000 {
+			svcCtx.SetWatched(false)
+		}
+	})
+	wg.Go(func() {
+		<-start
+		for range 1000 {
+			if !svcCtx.IsWatchedLocked() {
+				svcCtx.SetWatched(true)
+			}
+		}
+	})
+
+	close(start)
+	wg.Wait()
+}

pkg/services/services.go

@@ -65,7 +65,7 @@ func (p *Processor) SyncServices(ctx *servicecontext.Context, svc *v1.Service, w
 			select {
 			case <-ctx.Ctx.Done():
 				return nil
-			case <-ctx.EndpointsReady:
+			case <-ctx.GetEndpointsReady():
 			}
 		}