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) } }