Add generic ReadyChannel to cronjob

internal/framework/runnables/cronjob.go

@@ -13,6 +13,8 @@ import (
 type CronJobConfig struct {
 	// Worker is the function that will be run for every cronjob iteration.
 	Worker func(context.Context)
+	// ReadyCh represents if the cronjob is ready to start.
+	ReadyCh <-chan struct{}
 	// Logger is the logger.
 	Logger logr.Logger
 	// Period defines the period of the cronjob. The cronjob will run every Period.
@@ -37,6 +39,13 @@ func NewCronJob(cfg CronJobConfig) *CronJob {
 // Start starts the cronjob.
 // Implements controller-runtime manager.Runnable
 func (j *CronJob) Start(ctx context.Context) error {
+	select {
+	case <-j.cfg.ReadyCh:
+	case <-ctx.Done():
+		j.cfg.Logger.Info("Context canceled, failed to start cronjob")
+		return ctx.Err()
+	}
+
 	j.cfg.Logger.Info("Starting cronjob")
 
 	sliding := true // This means the period with jitter will be calculated after each worker call.

internal/framework/runnables/cronjob_test.go

@@ -7,11 +7,18 @@ import (
 
 	. "github.com/onsi/gomega"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+
+	"github.com/nginxinc/nginx-gateway-fabric/internal/mode/static/telemetry/telemetryfakes"
 )
 
 func TestCronJob(t *testing.T) {
 	g := NewWithT(t)
 
+	healthCollector := &telemetryfakes.FakeHealthChecker{}
+
+	readyChannel := make(chan struct{})
+	healthCollector.GetReadyChReturns(readyChannel)
+
 	timeout := 10 * time.Second
 	var callCount int
 
@@ -22,9 +29,10 @@ func TestCronJob(t *testing.T) {
 	}
 
 	cfg := CronJobConfig{
-		Worker: worker,
-		Logger: zap.New(),
-		Period: 1 * time.Millisecond, // 1ms is much smaller than timeout so the CronJob should run a few times
+		Worker:  worker,
+		Logger:  zap.New(),
+		Period:  1 * time.Millisecond, // 1ms is much smaller than timeout so the CronJob should run a few times
+		ReadyCh: healthCollector.GetReadyCh(),
 	}
 	job := NewCronJob(cfg)
 
@@ -35,6 +43,7 @@ func TestCronJob(t *testing.T) {
 		errCh <- job.Start(ctx)
 		close(errCh)
 	}()
+	close(readyChannel)
 
 	minReports := 2 // ensure that the CronJob reports more than once: it doesn't exit after the first run
 
@@ -44,3 +53,41 @@ func TestCronJob(t *testing.T) {
 	g.Eventually(errCh).Should(Receive(BeNil()))
 	g.Eventually(errCh).Should(BeClosed())
 }
+
+func TestCronJob_ContextCanceled(t *testing.T) {
+	g := NewWithT(t)
+
+	healthCollector := &telemetryfakes.FakeHealthChecker{}
+
+	readyChannel := make(chan struct{})
+	healthCollector.GetReadyChReturns(readyChannel)
+
+	timeout := 10 * time.Second
+	var callCount int
+
+	valCh := make(chan int, 128)
+	worker := func(context.Context) {
+		callCount++
+		valCh <- callCount
+	}
+
+	cfg := CronJobConfig{
+		Worker:  worker,
+		Logger:  zap.New(),
+		Period:  1 * time.Millisecond, // 1ms is much smaller than timeout so the CronJob should run a few times
+		ReadyCh: healthCollector.GetReadyCh(),
+	}
+	job := NewCronJob(cfg)
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+
+	errCh := make(chan error)
+	go func() {
+		errCh <- job.Start(ctx)
+		close(errCh)
+	}()
+	cancel()
+
+	g.Eventually(errCh).Should(Receive())
+	g.Eventually(errCh).Should(BeClosed())
+}

internal/mode/static/handler_test.go

@@ -429,7 +429,7 @@ var _ = Describe("eventHandler", func() {
 	It("should set the health checker status properly when there are changes", func() {
 		e := &events.UpsertEvent{Resource: &gatewayv1.HTTPRoute{}}
 		batch := []interface{}{e}
-		readyChannel := handler.cfg.healthChecker.GetReadyIfClosedChannel()
+		readyChannel := handler.cfg.healthChecker.GetReadyCh()
 
 		fakeProcessor.ProcessReturns(state.ClusterStateChange, &graph.Graph{})
 
@@ -446,7 +446,7 @@ var _ = Describe("eventHandler", func() {
 	It("should set the health checker status properly when there are no changes or errors", func() {
 		e := &events.UpsertEvent{Resource: &gatewayv1.HTTPRoute{}}
 		batch := []interface{}{e}
-		readyChannel := handler.cfg.healthChecker.GetReadyIfClosedChannel()
+		readyChannel := handler.cfg.healthChecker.GetReadyCh()
 
 		Expect(handler.cfg.healthChecker.readyCheck(nil)).ToNot(Succeed())
 		handler.HandleEventBatch(context.Background(), ctlrZap.New(), batch)
@@ -461,7 +461,7 @@ var _ = Describe("eventHandler", func() {
 	It("should set the health checker status properly when there is an error", func() {
 		e := &events.UpsertEvent{Resource: &gatewayv1.HTTPRoute{}}
 		batch := []interface{}{e}
-		readyChannel := handler.cfg.healthChecker.GetReadyIfClosedChannel()
+		readyChannel := handler.cfg.healthChecker.GetReadyCh()
 
 		fakeProcessor.ProcessReturns(state.ClusterStateChange, &graph.Graph{})
 		fakeNginxRuntimeMgr.ReloadReturns(errors.New("reload error"))

internal/mode/static/health.go

@@ -9,7 +9,7 @@ import (
 // newHealthCheckerImpl creates a new healthCheckerImpl.
 func newHealthCheckerImpl() *healthCheckerImpl {
 	return &healthCheckerImpl{
-		readyIfClosedChan: make(chan struct{}),
+		readyCh: make(chan struct{}),
 	}
 }
 
@@ -18,10 +18,10 @@ type healthCheckerImpl struct {
 	// firstBatchError is set when the first batch fails to configure nginx
 	// and we don't want to set ourselves as ready on the next batch if nothing changes
 	firstBatchError error
-	// readyIfClosedChan is a channel that is initialized in NewHealthCheckerImpl and represents if the NGF Pod is ready.
-	readyIfClosedChan chan struct{}
-	lock              sync.RWMutex
-	ready             bool
+	// readyCh is a channel that is initialized in NewHealthCheckerImpl and represents if the NGF Pod is ready.
+	readyCh chan struct{}
+	lock    sync.RWMutex
+	ready   bool
 }
 
 // readyCheck returns the ready-state of the Pod. It satisfies the controller-runtime Checker type.
@@ -45,10 +45,10 @@ func (h *healthCheckerImpl) setAsReady() {
 
 	h.ready = true
 	h.firstBatchError = nil
-	close(h.readyIfClosedChan)
+	close(h.readyCh)
 }
 
-// GetReadyIfClosedChannel returns a read-only channel, which determines if the NGF Pod is ready.
-func (h *healthCheckerImpl) GetReadyIfClosedChannel() <-chan struct{} {
-	return h.readyIfClosedChan
+// GetReadyCh returns a read-only channel, which determines if the NGF Pod is ready.
+func (h *healthCheckerImpl) GetReadyCh() <-chan struct{} {
+	return h.readyCh
 }

internal/mode/static/manager.go

@@ -413,7 +413,7 @@ func createTelemetryJob(
 	logger := cfg.Logger.WithName("telemetryJob")
 	exporter := telemetry.NewLoggingExporter(cfg.Logger.WithName("telemetryExporter").V(1 /* debug */))
 
-	worker := telemetry.CreateTelemetryJobWorker(logger, exporter, dataCollector, hc)
+	worker := telemetry.CreateTelemetryJobWorker(logger, exporter, dataCollector)
 
 	// 10 min jitter is enough per telemetry destination recommendation
 	// For the default period of 24 hours, jitter will be 10min /(24*60)min  = 0.0069
@@ -426,6 +426,7 @@ func createTelemetryJob(
 				Logger:       logger,
 				Period:       cfg.TelemetryReportPeriod,
 				JitterFactor: jitterFactor,
+				ReadyCh:      hc.GetReadyCh(),
 			},
 		),
 	}

internal/mode/static/telemetry/job_worker.go

@@ -18,25 +18,16 @@ type DataCollector interface {
 
 // HealthChecker checks if the NGF Pod is ready.
 type HealthChecker interface {
-	// GetReadyIfClosedChannel returns a channel which determines if the NGF Pod is ready.
-	GetReadyIfClosedChannel() <-chan struct{}
+	// GetReadyCh returns a channel which determines if the NGF Pod is ready.
+	GetReadyCh() <-chan struct{}
 }
 
 func CreateTelemetryJobWorker(
 	logger logr.Logger,
 	exporter Exporter,
 	dataCollector DataCollector,
-	healthChecker HealthChecker,
 ) func(ctx context.Context) {
 	return func(ctx context.Context) {
-		readyChannel := healthChecker.GetReadyIfClosedChannel()
-		select {
-		case <-readyChannel:
-		case <-ctx.Done():
-			logger.Info("Context canceled, failed to start telemetry job")
-			return
-		}
-
 		// Gather telemetry
 		logger.V(1).Info("Gathering telemetry data")
 

internal/mode/static/telemetry/job_worker_test.go

@@ -20,9 +20,9 @@ func TestCreateTelemetryJobWorker(t *testing.T) {
 	healthCollector := &telemetryfakes.FakeHealthChecker{}
 
 	readyChannel := make(chan struct{})
-	healthCollector.GetReadyIfClosedChannelReturns(readyChannel)
+	healthCollector.GetReadyChReturns(readyChannel)
 
-	worker := telemetry.CreateTelemetryJobWorker(zap.New(), exporter, dataCollector, healthCollector)
+	worker := telemetry.CreateTelemetryJobWorker(zap.New(), exporter, dataCollector)
 
 	expData := telemetry.Data{
 		ProjectMetadata: telemetry.ProjectMetadata{Name: "NGF", Version: "1.1"},
@@ -51,28 +51,3 @@ func TestCreateTelemetryJobWorker(t *testing.T) {
 	g.Expect(data).To(Equal(expData))
 	cancel()
 }
-
-func TestCreateTelemetryJobWorker_ContextCanceled(t *testing.T) {
-	g := NewWithT(t)
-
-	exporter := &telemetryfakes.FakeExporter{}
-	dataCollector := &telemetryfakes.FakeDataCollector{}
-	healthCollector := &telemetryfakes.FakeHealthChecker{}
-
-	readyChannel := make(chan struct{})
-	healthCollector.GetReadyIfClosedChannelReturns(readyChannel)
-
-	worker := telemetry.CreateTelemetryJobWorker(zap.New(), exporter, dataCollector, healthCollector)
-
-	timeout := 10 * time.Second
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
-
-	go func() {
-		worker(ctx)
-	}()
-
-	cancel()
-
-	g.Expect(dataCollector.CollectCallCount()).To(BeZero())
-	g.Expect(exporter.ExportCallCount()).To(BeZero())
-}