Address flaking tests (#21)

Author: george-zubrienko

app/app_dependencies.go

@@ -55,7 +55,7 @@ func (appServices *ApplicationServices) WithKubeClient(ctx context.Context, kube
 func (appServices *ApplicationServices) WithSupervisor(ctx context.Context, resourceNamespace string) *ApplicationServices {
 	if appServices.supervisor == nil {
 		logger := klog.FromContext(ctx)
-		appServices.supervisor = services.NewSupervisor(appServices.kubeClient, resourceNamespace, appServices.cqlStore, logger, nil)
+		appServices.supervisor = services.NewSupervisor(appServices.kubeClient, resourceNamespace, appServices.cqlStore, logger, nil, nil)
 	}
 
 	return appServices

services/supervisor.go

@@ -21,12 +21,17 @@ import (
 )
 
 type Supervisor struct {
-	logger               klog.Logger
-	factory              kubeinformers.SharedInformerFactory
-	eventInformer        cache.SharedIndexInformer
-	podInformer          cache.SharedIndexInformer
-	jobInformer          cache.SharedIndexInformer
-	informers            map[string]cache.SharedIndexInformer
+	logger        klog.Logger
+	factory       kubeinformers.SharedInformerFactory
+	eventInformer cache.SharedIndexInformer
+	podInformer   cache.SharedIndexInformer
+	jobInformer   cache.SharedIndexInformer
+	informers     map[string]cache.SharedIndexInformer
+
+	eventInformerSynced cache.InformerSynced
+	podInformerSynced   cache.InformerSynced
+	jobInformerSynced   cache.InformerSynced
+
 	kubeClient           kubernetes.Interface
 	resourceNamespace    string
 	cqlStore             *request.CqlStore
@@ -61,22 +66,37 @@ type RunStatusAnalysisResult struct {
 }
 
 // NewSupervisor creates a new cache + resource watcher for pod and job resources
-func NewSupervisor(client kubernetes.Interface, resourceNamespace string, cqlStore *request.CqlStore, logger klog.Logger, resyncPeriod *time.Duration) *Supervisor {
+func NewSupervisor(client kubernetes.Interface, resourceNamespace string, cqlStore *request.CqlStore, logger klog.Logger, resyncPeriod *time.Duration, syncState *func() bool) *Supervisor {
 	defaultResyncPeriod := time.Second * 30
 	factory := kubeinformers.NewSharedInformerFactoryWithOptions(client, *util.CoalescePointer(resyncPeriod, &defaultResyncPeriod), kubeinformers.WithNamespace(resourceNamespace))
 
 	eventInformer := factory.Core().V1().Events().Informer()
 	podInformer := factory.Core().V1().Pods().Informer()
 	jobInformer := factory.Batch().V1().Jobs().Informer()
 
+	eventInformerSynced := eventInformer.HasSynced
+	podInformerSynced := podInformer.HasSynced
+	jobInformerSynced := jobInformer.HasSynced
+
+	if syncState != nil {
+		eventInformerSynced = *syncState
+		podInformerSynced = *syncState
+		jobInformerSynced = *syncState
+	}
+
 	return &Supervisor{
-		logger:               logger,
-		factory:              factory,
-		kubeClient:           client,
-		resourceNamespace:    resourceNamespace,
-		eventInformer:        eventInformer,
-		podInformer:          podInformer,
-		jobInformer:          jobInformer,
+		logger:            logger,
+		factory:           factory,
+		kubeClient:        client,
+		resourceNamespace: resourceNamespace,
+		eventInformer:     eventInformer,
+		podInformer:       podInformer,
+		jobInformer:       jobInformer,
+
+		eventInformerSynced: eventInformerSynced,
+		podInformerSynced:   podInformerSynced,
+		jobInformerSynced:   jobInformerSynced,
+
 		cqlStore:             cqlStore,
 		elementReceiverActor: nil,
 	}
@@ -101,8 +121,10 @@ func (c *Supervisor) Init(_ context.Context, config *ProcessingConfig) error {
 		"Pod": c.podInformer,
 	}
 
-	_, eventErr := c.eventInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+	_, eventErr := c.eventInformer.AddEventHandlerWithOptions(cache.ResourceEventHandlerFuncs{
 		AddFunc: c.onEvent,
+	}, cache.HandlerOptions{
+		Logger: &c.logger,
 	})
 
 	if eventErr != nil {
@@ -113,6 +135,7 @@ func (c *Supervisor) Init(_ context.Context, config *ProcessingConfig) error {
 }
 
 func (c *Supervisor) onEvent(obj interface{}) {
+	c.logger.V(4).Info("event received", "object", obj)
 	_, err := cache.ObjectToName(obj)
 	if err != nil { // coverage-ignore
 		utilruntime.HandleError(err)
@@ -354,7 +377,7 @@ func (c *Supervisor) Start(ctx context.Context) {
 	c.elementReceiverActor.Start(ctx, pipeline.NewActorPostStart(func(ctx context.Context) error {
 		c.factory.Start(ctx.Done())
 
-		if ok := cache.WaitForCacheSync(ctx.Done(), c.eventInformer.HasSynced, c.podInformer.HasSynced, c.jobInformer.HasSynced); !ok { // coverage-ignore
+		if ok := cache.WaitForCacheSync(ctx.Done(), c.podInformerSynced, c.jobInformerSynced, c.eventInformerSynced); !ok { // coverage-ignore
 			return fmt.Errorf("failed to wait for pod informer caches to sync")
 		}
 

services/supervisor_test.go

@@ -18,6 +18,7 @@ import (
 )
 
 var noResyncPeriod = time.Second * 0
+var alwaysReady = func() bool { return true }
 
 type fixture struct {
 	supervisor *Supervisor
@@ -37,10 +38,7 @@ func newFixture(t *testing.T, k8sObjects []runtime.Object) *fixture {
 			Hosts: []string{"127.0.0.1"},
 		})
 	f.kubeClient = fake.NewClientset(k8sObjects...)
-
-	time.Sleep(time.Second)
-
-	f.supervisor = NewSupervisor(f.kubeClient, "nexus", f.cqlStore, klog.FromContext(f.ctx), &noResyncPeriod)
+	f.supervisor = NewSupervisor(f.kubeClient, "nexus", f.cqlStore, klog.FromContext(f.ctx), &noResyncPeriod, &alwaysReady)
 
 	return f
 }
@@ -553,11 +551,11 @@ func TestSupervisor(t *testing.T) {
 
 	f := newFixture(t, k8sObjects)
 	err := f.supervisor.Init(f.ctx, &ProcessingConfig{
-		FailureRateBaseDelay:       time.Second,
-		FailureRateMaxDelay:        time.Second * 2,
-		RateLimitElementsPerSecond: 1000,
-		RateLimitElementsBurst:     1000,
-		Workers:                    1,
+		FailureRateBaseDelay:       time.Millisecond * 100,
+		FailureRateMaxDelay:        time.Second,
+		RateLimitElementsPerSecond: 10,
+		RateLimitElementsBurst:     10,
+		Workers:                    4,
 	})
 
 	time.Sleep(time.Second)