cancel and stop shard watchers when servers are stopped

On-behalf-of: @SAP [email protected]

Author: xrstf

pkg/authentication/authentication_controller.go

@@ -215,7 +215,7 @@ func (c *Controller) process(ctx context.Context, key string) error {
 			return err
 		}
 
-		watcher := NewShardWatcher(shard.Name, client, &c.authIndex)
+		watcher := NewShardWatcher(ctx, shard.Name, client, &c.authIndex)
 		c.shardWatchers[shard.Name] = watcher
 	}
 
@@ -242,10 +242,11 @@ type shardWatcher struct {
 	state                       *state
 	workspaceTypeInformer       cache.SharedIndexInformer
 	workspaceAuthConfigInformer cache.SharedIndexInformer
-	stopCh                      chan struct{}
+	cancel                      context.CancelFunc
 }
 
 func NewShardWatcher(
+	ctx context.Context,
 	shardName string,
 	shardClient kcpclientset.ClusterInterface,
 	state *state,
@@ -288,23 +289,28 @@ func NewShardWatcher(
 		},
 	})
 
+	ctx, cancel := context.WithCancel(ctx)
+
 	watcher := &shardWatcher{
 		state:                       state,
 		workspaceTypeInformer:       wtInformer,
 		workspaceAuthConfigInformer: wacInformer,
-		stopCh:                      make(chan struct{}),
+		cancel:                      cancel,
 	}
 
-	go wacInformer.Run(watcher.stopCh)
-	go wtInformer.Run(watcher.stopCh)
+	go wacInformer.Run(ctx.Done())
+	go wtInformer.Run(ctx.Done())
 
 	// no need to wait. We only care about events and they arrive when they arrive.
 
 	return watcher
 }
 
 func (w *shardWatcher) Stop() {
-	close(w.stopCh)
+	if w.cancel != nil {
+		w.cancel()
+		w.cancel = nil
+	}
 }
 
 func (w *shardWatcher) Lookup(wsType logicalcluster.Path) (authenticator.Request, bool) {

pkg/authentication/index.go

@@ -207,6 +207,9 @@ func (c *state) Lookup(wsType logicalcluster.Path) (authenticator.Request, bool)
 		authenticatorKeys []authenticatorKey
 	)
 
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+
 	for shardKey, authenticatorsMap := range c.workspaceTypeAuthenticators {
 		var found bool
 		authenticatorKeys, found = authenticatorsMap[wsType]

pkg/server/config.go

@@ -466,7 +466,7 @@ func NewConfig(ctx context.Context, opts kcpserveroptions.CompletedOptions) (*Co
 		var authIndex authentication.AuthenticatorIndex
 		if kcpfeatures.DefaultFeatureGate.Enabled(kcpfeatures.WorkspaceAuthentication) {
 			authIndexState := authentication.NewIndex(ctx, genericConfig.Authentication.APIAudiences)
-			authentication.NewShardWatcher(c.Options.Extra.ShardName, c.KcpClusterClient, authIndexState)
+			authentication.NewShardWatcher(ctx, c.Options.Extra.ShardName, c.KcpClusterClient, authIndexState)
 
 			genericConfig.Authentication.Authenticator = authenticatorunion.New(
 				genericConfig.Authentication.Authenticator,