make sure that Start and Stop can be used concurrently

Signed-off-by: Tim Ramlot <[email protected]>

Author: inteon

pkg/internal/source/informer.go

@@ -54,12 +54,12 @@ func (is *Informer) Start(ctx context.Context, handler handler.EventHandler, que
 
 	is.mu.Lock()
 	defer is.mu.Unlock()
-	if is.canceled {
-		return nil
+	if is.started != nil {
+		return fmt.Errorf("Informer source already started")
 	}
 
-	if is.started != nil {
-		return errors.New("already started")
+	if is.canceled {
+		return nil
 	}
 
 	is.started = make(chan error, 1)
@@ -84,11 +84,11 @@ func (is *Informer) Start(ctx context.Context, handler handler.EventHandler, que
 			is.informer = informer
 
 			if is.WaitForInformerSync == nil {
-				is.WaitForInformerSync = func(ctx context.Context) error {
+				is.WaitForInformerSync = func(informerSyncCtx context.Context) error {
 					return wait.PollImmediateUntil(
 						syncedPollPeriod,
 						func() (bool, error) { return is.informer.HasSynced(), nil },
-						ctx.Done(),
+						informerSyncCtx.Done(),
 					)
 				}
 			}
@@ -161,8 +161,11 @@ func (is *Informer) Stop() error {
 	case <-is.started:
 	default:
 		is.startCancel()
+
+		is.mu.Unlock()
 		// Wait for starting abort
 		<-is.started
+		is.mu.Lock()
 	}
 
 	if is.eventHandlerRegistration != nil {

pkg/internal/source/kind.go

@@ -40,49 +40,58 @@ func (ks *Kind) Start(ctx context.Context, handler handler.EventHandler, queue w
 		return fmt.Errorf("must create Kind with a non-nil cache")
 	}
 
-	ks.informer = Informer{
-		GetInformer: func(ctx context.Context) (cache.Informer, error) {
-			var (
-				informer cache.Informer
-				lastErr  error
-			)
-
-			// Tries to get an informer until it returns true,
-			// an error or the specified context is cancelled or expired.
-			if err := wait.PollImmediateUntilWithContext(ctx, 10*time.Second, func(ctx context.Context) (bool, error) {
-				// Lookup the Informer from the Cache and add an EventHandler which populates the Queue
-				informer, lastErr = ks.Cache.GetInformer(ctx, ks.Type)
-				if lastErr != nil {
-					kindMatchErr := &meta.NoKindMatchError{}
-					switch {
-					case errors.As(lastErr, &kindMatchErr):
-						log.Error(lastErr, "if kind is a CRD, it should be installed before calling Start",
-							"kind", kindMatchErr.GroupKind)
-					case runtime.IsNotRegisteredError(lastErr):
-						log.Error(lastErr, "kind must be registered to the Scheme")
-					default:
-						log.Error(lastErr, "failed to get informer from cache")
-					}
-					return false, nil // Retry.
-				}
-				return true, nil
-			}); err != nil {
-				if lastErr != nil {
-					return nil, fmt.Errorf("failed to get informer from cache: %w", lastErr)
+	getInformer := func(ctx context.Context) (cache.Informer, error) {
+		var (
+			informer cache.Informer
+			lastErr  error
+		)
+
+		// Tries to get an informer until it returns true,
+		// an error or the specified context is cancelled or expired.
+		if err := wait.PollImmediateUntilWithContext(ctx, 10*time.Second, func(ctx context.Context) (bool, error) {
+			// Lookup the Informer from the Cache and add an EventHandler which populates the Queue
+			informer, lastErr = ks.Cache.GetInformer(ctx, ks.Type)
+			if lastErr != nil {
+				kindMatchErr := &meta.NoKindMatchError{}
+				switch {
+				case errors.As(lastErr, &kindMatchErr):
+					log.Error(lastErr, "if kind is a CRD, it should be installed before calling Start",
+						"kind", kindMatchErr.GroupKind)
+				case runtime.IsNotRegisteredError(lastErr):
+					log.Error(lastErr, "kind must be registered to the Scheme")
+				default:
+					log.Error(lastErr, "failed to get informer from cache")
 				}
-				return nil, err
+				return false, nil // Retry.
 			}
-
-			return informer, nil
-		},
-		WaitForInformerSync: func(ctx context.Context) error {
-			if ks.Cache.WaitForCacheSync(ctx) {
-				return nil
+			return true, nil
+		}); err != nil {
+			if lastErr != nil {
+				return nil, fmt.Errorf("failed to get informer from cache: %w", lastErr)
 			}
+			return nil, err
+		}
+
+		return informer, nil
+	}
+
+	waitForInformerSync := func(ctx context.Context) error {
+		if ks.Cache.WaitForCacheSync(ctx) {
+			return nil
+		}
+
+		return fmt.Errorf("cache did not sync")
+	}
 
-			return fmt.Errorf("cache did not sync")
-		},
+	ks.informer.mu.Lock()
+	// If the GetInformer function is set, the informer has already been started.
+	if ks.informer.GetInformer != nil || ks.informer.WaitForInformerSync != nil {
+		ks.informer.mu.Unlock()
+		return fmt.Errorf("Kind source already started")
 	}
+	ks.informer.GetInformer = getInformer
+	ks.informer.WaitForInformerSync = waitForInformerSync
+	ks.informer.mu.Unlock()
 
 	return ks.informer.Start(ctx, handler, queue, prct...)
 }

pkg/source/source.go

@@ -128,6 +128,11 @@ func (cs *Channel) Start(ctx context.Context, handler handler.EventHandler, queu
 	dst := make(chan event.GenericEvent, cs.DestBufferSize)
 
 	cs.mu.Lock()
+	if cs.canceled {
+		cs.mu.Unlock()
+		return nil
+	}
+
 	cs.destinations = append(cs.destinations, dst)
 
 	if cs.groupWait == nil {
@@ -208,6 +213,9 @@ func (cs *Channel) loop(ctx context.Context) {
 // Stop is internal and should be called only by the Controller to stop the Source. It should block until the
 // Source has stopped.
 func (cs *Channel) Stop() error {
+	groupCancel := func() {}
+	groupWait := func() error { return nil }
+
 	if func() bool {
 		cs.mu.Lock()
 		defer cs.mu.Unlock()
@@ -217,13 +225,18 @@ func (cs *Channel) Stop() error {
 		}
 		cs.canceled = true
 
+		if cs.groupCancel != nil {
+			groupCancel = cs.groupCancel
+			groupWait = cs.groupWait
+		}
+
 		return false
 	}() {
 		return nil
 	}
 
-	cs.groupCancel()
-	return cs.groupWait()
+	groupCancel()
+	return groupWait()
 }
 
 var _ Source = Func(nil)

pkg/source/source_test.go

@@ -226,6 +226,27 @@ var _ = Describe("Source", func() {
 			Expect(err.Error()).To(Equal("cache did not sync"))
 
 		})
+
+		Context("should not panic", func() {
+			It("when Stop is called before Start", func() {
+				instance := source.Kind(ic, &corev1.Pod{})
+				err := instance.Stop()
+				Expect(err).NotTo(HaveOccurred())
+			})
+
+			It("when Start is called twice", func() {
+				instance := source.Kind(ic, &corev1.Pod{})
+				Expect(instance.Start(ctx, nil, nil)).To(Succeed())
+				Expect(instance.Start(ctx, nil, nil)).To(MatchError("Kind source already started"))
+			})
+
+			It("when Stop is called twice", func() {
+				instance := source.Kind(ic, &corev1.Pod{})
+				Expect(instance.Start(ctx, nil, nil)).To(Succeed())
+				Expect(instance.Stop()).To(Succeed())
+				Expect(instance.Stop()).To(Succeed())
+			})
+		})
 	})
 
 	Describe("Func", func() {
@@ -528,5 +549,29 @@ var _ = Describe("Source", func() {
 				Expect(resEvent1).To(Equal(resEvent2))
 			})
 		})
+
+		Context("should not panic", func() {
+			It("when Stop is called before Start", func() {
+				ch := make(chan event.GenericEvent)
+				instance := &source.Channel{Source: ch}
+				err := instance.Stop()
+				Expect(err).NotTo(HaveOccurred())
+			})
+
+			It("when Start is called twice", func() {
+				ch := make(chan event.GenericEvent)
+				instance := &source.Channel{Source: ch}
+				Expect(instance.Start(ctx, nil, nil)).To(Succeed())
+				Expect(instance.Start(ctx, nil, nil)).To(Succeed())
+			})
+
+			It("when Stop is called twice", func() {
+				ch := make(chan event.GenericEvent)
+				instance := &source.Channel{Source: ch}
+				Expect(instance.Start(ctx, nil, nil)).To(Succeed())
+				Expect(instance.Stop()).To(Succeed())
+				Expect(instance.Stop()).To(Succeed())
+			})
+		})
 	})
 })