Make telemetry job a generic runnable (nginx#1529)

Problem: Multiple reporting systems need to use the telemetry job, and duplicated code would like to be avoided.

Solution: Turn the telemetry job into a generic job that runs a "worker" function, provided by the caller.

Author: miledxz

internal/framework/runnables/cronjob.go

@@ -0,0 +1,50 @@
+package runnables
+
+import (
+	"context"
+	"time"
+
+	"github.com/go-logr/logr"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+)
+
+// CronJobConfig is the configuration for a cronjob.
+type CronJobConfig struct {
+	// Worker is the function that will be run for every cronjob iteration.
+	Worker func(context.Context)
+	// Logger is the logger.
+	Logger logr.Logger
+	// Period defines the period of the cronjob. The cronjob will run every Period.
+	Period time.Duration
+	// JitterFactor sets the jitter for the cronjob. If positive, the period is jittered before every
+	// run of the worker. If jitterFactor is not positive, the period is unchanged and not jittered.
+	JitterFactor float64
+}
+
+// CronJob periodically runs a worker function.
+type CronJob struct {
+	cfg CronJobConfig
+}
+
+// NewCronJob creates a new cronjob.
+func NewCronJob(cfg CronJobConfig) *CronJob {
+	return &CronJob{
+		cfg: cfg,
+	}
+}
+
+// Start starts the cronjob.
+// Implements controller-runtime manager.Runnable
+func (j *CronJob) Start(ctx context.Context) error {
+	j.cfg.Logger.Info("Starting cronjob")
+
+	sliding := true // This means the period with jitter will be calculated after each worker call.
+
+	wait.JitterUntilWithContext(ctx, j.cfg.Worker, j.cfg.Period, j.cfg.JitterFactor, sliding)
+
+	j.cfg.Logger.Info("Stopping cronjob")
+	return nil
+}
+
+var _ manager.Runnable = &CronJob{}

internal/framework/runnables/cronjob_test.go

@@ -0,0 +1,46 @@
+package runnables
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	. "github.com/onsi/gomega"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+)
+
+func TestCronJob(t *testing.T) {
+	g := NewWithT(t)
+
+	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
+	}
+	job := NewCronJob(cfg)
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+
+	errCh := make(chan error)
+	go func() {
+		errCh <- job.Start(ctx)
+		close(errCh)
+	}()
+
+	minReports := 2 // ensure that the CronJob reports more than once: it doesn't exit after the first run
+
+	g.Eventually(valCh).Should(Receive(BeNumerically(">=", minReports)))
+
+	cancel()
+	g.Eventually(errCh).Should(Receive(BeNil()))
+	g.Eventually(errCh).Should(BeClosed())
+}

internal/mode/static/manager.go

@@ -213,17 +213,7 @@ func StartManager(cfg config.Config) error {
 		return fmt.Errorf("cannot register status updater: %w", err)
 	}
 
-	telemetryJob := &runnables.Leader{
-		Runnable: telemetry.NewJob(
-			telemetry.JobConfig{
-				Exporter: telemetry.NewLoggingExporter(cfg.Logger.WithName("telemetryExporter").V(1 /* debug */)),
-				Logger:   cfg.Logger.WithName("telemetryJob"),
-				Period:   cfg.TelemetryReportPeriod,
-			},
-		),
-	}
-
-	if err = mgr.Add(telemetryJob); err != nil {
+	if err = mgr.Add(createTelemetryJob(cfg)); err != nil {
 		return fmt.Errorf("cannot register telemetry job: %w", err)
 	}
 
@@ -448,6 +438,40 @@ func prepareFirstEventBatchPreparerArgs(
 	return objects, objectLists
 }
 
+func createTelemetryJob(cfg config.Config) *runnables.Leader {
+	logger := cfg.Logger.WithName("telemetryJob")
+	exporter := telemetry.NewLoggingExporter(cfg.Logger.WithName("telemetryExporter").V(1 /* debug */))
+
+	worker := func(ctx context.Context) {
+		// Gather telemetry
+		logger.V(1).Info("Gathering telemetry")
+
+		// We will need to gather data as defined in https://github.com/nginxinc/nginx-gateway-fabric/issues/793
+		data := telemetry.Data{}
+
+		// Export telemetry
+		logger.V(1).Info("Exporting telemetry")
+
+		if err := exporter.Export(ctx, data); err != nil {
+			logger.Error(err, "Failed to export telemetry")
+		}
+	}
+	// 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
+	jitterFactor := 10.0 / (24 * 60) // added jitter is bound by jitterFactor * period
+
+	return &runnables.Leader{
+		Runnable: runnables.NewCronJob(
+			runnables.CronJobConfig{
+				Worker:       worker,
+				Logger:       logger,
+				Period:       cfg.TelemetryReportPeriod,
+				JitterFactor: jitterFactor,
+			},
+		),
+	}
+}
+
 func setInitialConfig(
 	reader client.Reader,
 	logger logr.Logger,