NO-JIRA: Refactor connection checker to fix redundant looping

pkg/cmd/checkendpoints/controller/connection_checker.go

@@ -30,18 +30,41 @@ type ConnectionChecker interface {
 
 type GetCheckFunc func() *operatorcontrolplanev1alpha1.PodNetworkConnectivityCheck
 
+// ConnectionCheckerConfig holds the configuration for creating a ConnectionChecker.
+type ConnectionCheckerConfig struct {
+	Name             string
+	PodName          string
+	PodNamespace     string
+	GetCheck         GetCheckFunc
+	Client           v1alpha1helpers.PodNetworkConnectivityCheckClient
+	ClientCertGetter CertificatesGetter
+	Recorder         Recorder
+	CheckPeriod      time.Duration
+	CheckTimeout     time.Duration
+}
+
 // NewConnectionChecker returns a ConnectionChecker.
-func NewConnectionChecker(name, podName, podNamespace string, getCheck GetCheckFunc, client v1alpha1helpers.PodNetworkConnectivityCheckClient, clientCertGetter CertificatesGetter, recorder Recorder) ConnectionChecker {
+func NewConnectionChecker(config ConnectionCheckerConfig) ConnectionChecker {
+	checkPeriodToUse := config.CheckPeriod
+	if checkPeriodToUse == 0 {
+		checkPeriodToUse = checkPeriod
+	}
+
+	checkTimeoutToUse := config.CheckTimeout
+	if checkTimeoutToUse == 0 {
+		checkTimeoutToUse = checkTimeout
+	}
+
 	return &connectionChecker{
-		name:             name,
-		podName:          podName,
-		getCheck:         getCheck,
-		client:           client,
-		clientCertGetter: clientCertGetter,
-		recorder:         recorder,
-		updates:          NewUpdatesManager(checkPeriod, checkTimeout, newUpdatesProcessor(client, name)),
+		name:             config.Name,
+		podName:          config.PodName,
+		getCheck:         config.GetCheck,
+		checkPeriod:      checkPeriodToUse,
+		clientCertGetter: config.ClientCertGetter,
+		recorder:         config.Recorder,
+		updates:          NewUpdatesManager(checkPeriodToUse, checkTimeoutToUse, newUpdatesProcessor(config.Client, config.Name)),
 		stop:             make(chan any),
-		metrics:          NewMetricsContext(podNamespace, name),
+		metrics:          NewMetricsContext(config.PodNamespace, config.Name),
 	}
 }
 
@@ -55,60 +78,45 @@ func newUpdatesProcessor(client v1alpha1helpers.PodNetworkConnectivityCheckClien
 type CertificatesGetter func() []tls.Certificate
 
 type connectionChecker struct {
-	name     string
-	podName  string
-	getCheck GetCheckFunc
+	name        string
+	podName     string
+	getCheck    GetCheckFunc
+	checkPeriod time.Duration
 
-	client           v1alpha1helpers.PodNetworkConnectivityCheckClient
 	clientCertGetter CertificatesGetter
 	recorder         Recorder
 	updates          UpdatesManager
 	stop             chan any
 	metrics          MetricsContext
 }
 
-// checkConnection checks the connection periodically, updating status as needed
+// checkConnection checks the connection once, updating status as needed
 func (c *connectionChecker) checkConnection(ctx context.Context) {
-	ticker := time.NewTicker(checkPeriod)
-	defer ticker.Stop()
-	defer klog.V(1).Infof("Stopped connectivity check %s.", c.name)
-	for {
-		select {
-		case <-c.stop:
-			return
-		case <-ctx.Done():
-			return
-
-		case <-ticker.C:
-			go func() {
-				currCheck := c.getCheck()
-				// if we have no check or the check isn't for us or the check has no target, report status if needed, but nothing else
-				if currCheck == nil || currCheck.Spec.SourcePod != c.podName || len(currCheck.Spec.TargetEndpoint) == 0 {
-					c.updateStatus(ctx, false)
-					return
-				}
-				c.checkEndpoint(ctx, currCheck)
-				c.updateStatus(ctx, false)
-			}()
-		}
+	currCheck := c.getCheck()
+	// if we have no check or the check isn't for us or the check has no target, report status if needed, but nothing else
+	if currCheck == nil || currCheck.Spec.SourcePod != c.podName || len(currCheck.Spec.TargetEndpoint) == 0 {
+		c.updateStatus(ctx, false)
+		return
 	}
+	c.checkEndpoint(ctx, currCheck)
+	c.updateStatus(ctx, false)
 }
 
 // Run starts the connection checker.
 func (c *connectionChecker) Run(ctx context.Context) {
 	ctx2, cancel := context.WithCancel(ctx)
+	defer cancel()
 	go func() {
 		select {
 		case <-c.stop:
 			cancel()
 		case <-ctx2.Done():
 		}
 	}()
-	go wait.UntilWithContext(ctx2, func(ctx context.Context) {
-		c.checkConnection(ctx2)
-	}, checkPeriod)
+
 	klog.V(1).Infof("Started connectivity check %s.", c.name)
-	<-ctx2.Done()
+	wait.UntilWithContext(ctx2, c.checkConnection, c.checkPeriod)
+	klog.V(1).Infof("Stopped connectivity check %s.", c.name)
 }
 
 // Stop

pkg/cmd/checkendpoints/controller/connection_checker_test.go

@@ -1,12 +1,16 @@
 package controller
 
 import (
+	"context"
+	"crypto/tls"
 	"errors"
 	"fmt"
 	"net"
+	"sync/atomic"
 	"testing"
 	"time"
 
+	. "github.com/onsi/gomega"
 	"github.com/openshift/api/operatorcontrolplane/v1alpha1"
 	"github.com/openshift/library-go/pkg/operator/events"
 	"github.com/stretchr/testify/assert"
@@ -535,3 +539,85 @@ func logEntry(success bool, start int, reason, message string, options ...func(e
 	}
 	return entry
 }
+
+func TestConnectionCheckerRun(t *testing.T) {
+	newConnectionChecker := func(getCheck GetCheckFunc) ConnectionChecker {
+		return NewConnectionChecker(ConnectionCheckerConfig{
+			Name:             "test-check",
+			PodName:          "test-pod",
+			PodNamespace:     "test-namespace",
+			GetCheck:         getCheck,
+			Client:           &mockClient{},
+			ClientCertGetter: func() []tls.Certificate { return nil },
+			Recorder:         events.NewInMemoryRecorder("test", clock.RealClock{}),
+			CheckPeriod:      50 * time.Millisecond,
+		})
+	}
+
+	runChecker := func(ctx context.Context) (ConnectionChecker, *atomic.Int32, chan struct{}) {
+		var checkCount atomic.Int32
+
+		checker := newConnectionChecker(func() *v1alpha1.PodNetworkConnectivityCheck {
+			checkCount.Add(1)
+			return nil
+		})
+
+		done := make(chan struct{})
+		go func() {
+			checker.Run(ctx)
+			close(done)
+		}()
+
+		return checker, &checkCount, done
+	}
+
+	t.Run("should stop when Stop is called", func(t *testing.T) {
+		g := NewGomegaWithT(t)
+
+		checker, checkCount, done := runChecker(context.Background())
+
+		// Wait for a few checks to run.
+		g.Eventually(func() int32 {
+			return checkCount.Load()
+		}).Within(time.Second).Should(BeNumerically(">", 2))
+
+		// Call Stop
+		stopCtx, stopCancel := context.WithTimeout(context.Background(), 1*time.Second)
+		defer stopCancel()
+		checker.Stop(stopCtx)
+
+		g.Eventually(done).Within(500*time.Millisecond).Should(BeClosed(), "Run did not exit after Stop was called")
+	})
+
+	t.Run("should stop when the context is cancelled", func(t *testing.T) {
+		g := NewGomegaWithT(t)
+
+		ctx, cancel := context.WithCancel(context.Background())
+
+		_, checkCount, done := runChecker(ctx)
+
+		// Wait for it to start.
+		g.Eventually(func() int32 {
+			return checkCount.Load()
+		}).Within(time.Second).Should(BeNumerically(">", 0))
+
+		// Cancel the context
+		cancel()
+
+		g.Eventually(done).Within(500*time.Millisecond).Should(BeClosed(), "Run did not exit after context was cancelled")
+	})
+}
+
+// mockClient is a mock implementation of PodNetworkConnectivityCheckClient
+type mockClient struct {
+}
+
+func (m *mockClient) UpdateStatus(ctx context.Context, check *v1alpha1.PodNetworkConnectivityCheck, opts metav1.UpdateOptions) (*v1alpha1.PodNetworkConnectivityCheck, error) {
+	return check, nil
+}
+
+func (m *mockClient) Get(name string) (*v1alpha1.PodNetworkConnectivityCheck, error) {
+	return &v1alpha1.PodNetworkConnectivityCheck{
+		ObjectMeta: metav1.ObjectMeta{Name: name},
+	}, nil
+}

pkg/cmd/checkendpoints/controller/pod_network_connectivity_check_controller.go

@@ -83,7 +83,15 @@ func (c *controller) Sync(ctx context.Context, syncContext factory.SyncContext)
 	// create & start status updaters if needed
 	for _, check := range checks {
 		if updater := c.updaters[check.Name]; updater == nil {
-			c.updaters[check.Name] = NewConnectionChecker(check.Name, c.podName, c.podNamespace, c.newCheckFunc(check.Name), c, c.getClientCerts(check), c.recorder)
+			c.updaters[check.Name] = NewConnectionChecker(ConnectionCheckerConfig{
+				Name:             check.Name,
+				PodName:          c.podName,
+				PodNamespace:     c.podNamespace,
+				GetCheck:         c.newCheckFunc(check.Name),
+				Client:           c,
+				ClientCertGetter: c.getClientCerts(check),
+				Recorder:         c.recorder,
+			})
 			go c.updaters[check.Name].Run(ctx)
 		}
 	}