From 912060b98ba8383bb8330cbc2b3ef1bbcda32108 Mon Sep 17 00:00:00 2001 From: Tom Pantelis Date: Mon, 13 Jul 2026 13:03:45 -0400 Subject: [PATCH] Refactor connection checker to fix redundant looping The previous implementation had two layers of periodic execution - Run() used wait.UntilWithContext() to call checkConnection() periodically, but checkConnection() also had its own ticker and infinite loop. The code was simplified to have checkConnection() execute once per call, relying on wait.UntilWithContext() for periodic execution. Also, the previous Run() method launched wait.UntilWithContext() in a goroutine and then blocked on <-ctx2.Done(). This is redundant because wait.UntilWithContext() already blocks until the context is cancelled. The new implementation calls wait.UntilWithContext() directly, which properly blocks the Run() method until the context is cancelled or Stop() is called. Unit tests were added to cover the Run() functionality. To facilitate this, the check period was made configurable by introducing a ConnectionCheckerConfig struct to replace the long parameter list in NewConnectionChecker(). Signed-off-by: Tom Pantelis --- .../controller/connection_checker.go | 88 ++++++++++--------- .../controller/connection_checker_test.go | 86 ++++++++++++++++++ ...d_network_connectivity_check_controller.go | 10 ++- 3 files changed, 143 insertions(+), 41 deletions(-) diff --git a/pkg/cmd/checkendpoints/controller/connection_checker.go b/pkg/cmd/checkendpoints/controller/connection_checker.go index 7a7c1f5b3c..33b268af52 100644 --- a/pkg/cmd/checkendpoints/controller/connection_checker.go +++ b/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,11 +78,11 @@ 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 @@ -67,36 +90,22 @@ type connectionChecker struct { 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: @@ -104,11 +113,10 @@ func (c *connectionChecker) Run(ctx context.Context) { 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 diff --git a/pkg/cmd/checkendpoints/controller/connection_checker_test.go b/pkg/cmd/checkendpoints/controller/connection_checker_test.go index 3ffb54d311..0a37adbe65 100644 --- a/pkg/cmd/checkendpoints/controller/connection_checker_test.go +++ b/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 +} diff --git a/pkg/cmd/checkendpoints/controller/pod_network_connectivity_check_controller.go b/pkg/cmd/checkendpoints/controller/pod_network_connectivity_check_controller.go index abc3c1e303..307c9b2a90 100644 --- a/pkg/cmd/checkendpoints/controller/pod_network_connectivity_check_controller.go +++ b/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) } }