Telemetry Job

cmd/nginx-ingress/flags.go

@@ -1,13 +1,15 @@
 package main
 
 import (
+	"errors"
 	"flag"
 	"fmt"
 	"net"
 	"os"
 	"regexp"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/golang/glog"
 	api_v1 "k8s.io/api/core/v1"
@@ -201,6 +203,9 @@
 
 	enableDynamicSSLReload = flag.Bool(dynamicSSLReloadParam, true, "Enable reloading of SSL Certificates without restarting the NGINX process.")
 
+	enableTelemetryReporting = flag.Bool("enable-telemetry-reporting", true, "Enable gathering and reporting of product related telemetry.")
+	telemetryReportingPeriod = flag.String("telemetry-reporting-period", "24h", "Period at which product telemetry is reported.")
+
 	startupCheckFn func() error
 )
 
@@ -384,6 +389,12 @@
 			glog.Fatalf("Invalid value for app-protect-log-level: %v", *appProtectLogLevel)
 		}
 	}
+
+	if telemetryReportingPeriod != nil {
+		if err := validateReportingPeriod(*telemetryReportingPeriod); err != nil {
+			glog.Fatalf("Invalid value for telemetry-reporting-period: %v", err)
+		}
+	}
 }
 
 // validateNamespaceNames validates the namespaces are in the correct format
@@ -489,3 +500,15 @@
 	}
 	return nil
 }
+
+// validateReportingPeriod checks if the reporting period parameter can be parsed.
+func validateReportingPeriod(period string) error {
+	duration, err := time.ParseDuration(period)
+	if err != nil {
+		return err
+	}
+	if duration.Minutes() < 60 {
+		return errors.New("invalid reporting period, expected minimum 1h")
+	}
+	return nil
+}

cmd/nginx-ingress/main.go

@@ -41,7 +41,9 @@
 )
 
 // Injected during build
-var version string
+var (
+	version string
+)
 
 const (
 	nginxVersionLabel      = "app.nginx.org/version"
@@ -199,6 +201,8 @@
 		ExternalDNSEnabled:           *enableExternalDNS,
 		IsIPV6Disabled:               *disableIPV6,
 		WatchNamespaceLabel:          *watchNamespaceLabel,
+		TelemetryReportPeriod:        *telemetryReportingPeriod,
+		EnableTelemetryReporting:     *enableTelemetryReporting,
 	}
 
 	lbc := k8s.NewLoadBalancerController(lbcInput)
@@ -790,9 +794,12 @@
 }
 
 func updateSelfWithVersionInfo(kubeClient *kubernetes.Clientset, version, nginxVersion, appProtectVersion string, maxRetries int, waitTime time.Duration) {
-	nginxVer := strings.TrimSuffix(strings.Split(nginxVersion, "/")[1], "\n")
-	replacer := strings.NewReplacer(" ", "-", "(", "", ")", "")
-	nginxVer = replacer.Replace(nginxVer)
+	var nginxVer string
+	if nginxVersion != "" {
+		nginxVer = strings.TrimSuffix(strings.Split(nginxVersion, "/")[1], "\n")
+		replacer := strings.NewReplacer(" ", "-", "(", "", ")", "")
+		nginxVer = replacer.Replace(nginxVer)
+	}
 	podUpdated := false
 
 	for i := 0; (i < maxRetries || maxRetries == 0) && !podUpdated; i++ {

cmd/nginx-ingress/main_test.go

@@ -172,3 +172,27 @@ func TestValidateNamespaces(t *testing.T) {
 		}
 	}
 }
+
+func TestValidateReportingPeriodWithInvalidInput(t *testing.T) {
+	t.Parallel()
+
+	periods := []string{"", "-1", "1x", "abc", "-", "30s", "10ms", "30m", "59m", "0h"}
+	for _, p := range periods {
+		err := validateReportingPeriod(p)
+		if err == nil {
+			t.Errorf("want error on invalid period %s, got nil", p)
+		}
+	}
+}
+
+func TestValidateReportingPeriodWithValidInput(t *testing.T) {
+	t.Parallel()
+
+	periods := []string{"1h", "24h"}
+	for _, p := range periods {
+		err := validateReportingPeriod(p)
+		if err != nil {
+			t.Error(err)
+		}
+	}
+}

internal/k8s/controller.go

@@ -25,6 +25,8 @@
 	"sync"
 	"time"
 
+	"github.com/nginxinc/kubernetes-ingress/internal/telemetry"
+
 	"github.com/nginxinc/kubernetes-ingress/pkg/apis/dos/v1beta1"
 	"golang.org/x/exp/maps"
 
@@ -161,6 +163,8 @@
 	enableBatchReload             bool
 	isIPV6Disabled                bool
 	namespaceWatcherController    cache.Controller
+	telemetryCollector            *telemetry.Collector
+	telemetryChan                 chan struct{}
 }
 
 var keyFunc = cache.DeletionHandlingMetaNamespaceKeyFunc
@@ -206,6 +210,8 @@
 	ExternalDNSEnabled           bool
 	IsIPV6Disabled               bool
 	WatchNamespaceLabel          string
+	TelemetryReportPeriod        string
+	EnableTelemetryReporting     bool
 }
 
 // NewLoadBalancerController creates a controller
@@ -271,6 +277,18 @@
 		lbc.externalDNSController = ed_controller.NewController(ed_controller.BuildOpts(context.TODO(), lbc.namespaceList, lbc.recorder, lbc.confClient, input.ResyncPeriod, isDynamicNs))
 	}
 
+	// NIC Telemetry Reporting
+	if input.EnableTelemetryReporting {
+		lbc.telemetryChan = make(chan struct{})
+		collector, err := telemetry.NewCollector(
+			telemetry.WithTimePeriod(input.TelemetryReportPeriod),
+		)
+		if err != nil {
+			glog.Fatalf("failed to initialize telemetry collector: %v", err)
+		}
+		lbc.telemetryCollector = collector
+	}
+
 	glog.V(3).Infof("Nginx Ingress Controller has class: %v", input.IngressClass)
 
 	lbc.namespacedInformers = make(map[string]*namespacedInformer)
@@ -683,10 +701,22 @@
 	if lbc.externalDNSController != nil {
 		go lbc.externalDNSController.Run(lbc.ctx.Done())
 	}
+
 	if lbc.leaderElector != nil {
 		go lbc.leaderElector.Run(lbc.ctx)
 	}
 
+	if lbc.telemetryCollector != nil {
+		go func(ctx context.Context) {
+			select {
+			case <-lbc.telemetryChan:
+				lbc.telemetryCollector.Start(lbc.ctx)
+			case <-ctx.Done():
+				return
+			}
+		}(lbc.ctx)
+	}
+
 	for _, nif := range lbc.namespacedInformers {
 		nif.start()
 	}

internal/k8s/leader.go

@@ -58,6 +58,10 @@
 	return leaderelection.LeaderCallbacks{
 		OnStartedLeading: func(ctx context.Context) {
 			glog.V(3).Info("started leading")
+			// Closing this channel allows the leader to start the telemetry reporting process
+			if lbc.telemetryChan != nil {
+				close(lbc.telemetryChan)
+			}
 			if lbc.reportIngressStatus {
 				ingresses := lbc.configuration.GetResourcesWithFilter(resourceFilter{Ingresses: true})
 

internal/nginx/fake_manager.go

@@ -103,7 +103,7 @@
 // Version provides a fake implementation of Version.
 func (*FakeManager) Version() Version {
 	glog.V(3).Info("Printing nginx version")
-	return Version{}
+	return NewVersion("nginx version: nginx/1.25.3 (nginx-plus-r31)")
 }
 
 // Start provides a fake implementation of Start.

internal/telemetry/telemetry.go

@@ -0,0 +1,135 @@
+// Package telemetry provides functionality for collecting and exporting NIC telemetry data.
+package telemetry
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/golang/glog"
+	"k8s.io/apimachinery/pkg/util/wait"
+)
+
+// DiscardExporter is a temporary exporter
+// for discarding collected telemetry data.
+var DiscardExporter = Exporter{Endpoint: io.Discard}
+
+// Exporter represents a temporary telemetry data exporter.
+type Exporter struct {
+	Endpoint io.Writer
+}
+
+// Export takes context and trace data and writes to the endpoint.
+func (e *Exporter) Export(_ context.Context, td TraceData) error {
+	// Note: exporting functionality will be implemented in a separate module.
+	fmt.Fprintf(e.Endpoint, "%+v", td)
+	return nil
+}
+
+// TraceData holds collected NIC telemetry data.
+type TraceData struct {
+	// Numer of VirtualServers
+	VSCount int
+	// Number of TransportServers
+	TSCount int
+
+	// TODO
+	// Add more fields for NIC data points
+}
+
+// Option is a functional option used for configuring TraceReporter.
+type Option func(*Collector) error
+
+// WithTimePeriod configures reporting time on TraceReporter.
+func WithTimePeriod(period string) Option {
+	return func(c *Collector) error {
+		d, err := time.ParseDuration(period)
+		if err != nil {
+			return err
+		}
+		c.Period = d
+		return nil
+	}
+}
+
+// WithExporter configures telemetry collector to use given exporter.
+//
+// This may change in the future when we use exporter implemented
+// in the external module.
+func WithExporter(e Exporter) Option {
+	return func(c *Collector) error {
+		c.Exporter = e
+		return nil
+	}
+}
+
+// Collector is NIC telemetry data collector.
+type Collector struct {
+	Period time.Duration
+
+	// Exporter is a temp exporter for exporting telemetry data.
+	// The concrete implementation will be implemented in a separate module.
+	Exporter Exporter
+}
+
+// NewCollector takes 0 or more options and creates a new TraceReporter.
+// If no options are provided, NewReporter returns TraceReporter
+// configured to gather data every 24h.
+func NewCollector(opts ...Option) (*Collector, error) {
+	c := Collector{
+		Period:   24 * time.Hour,
+		Exporter: DiscardExporter, // Use DiscardExporter until the real exporter is available.
+	}
+	for _, o := range opts {
+		if err := o(&c); err != nil {
+			return nil, err
+		}
+	}
+	return &c, nil
+}
+
+// BuildReport takes context and builds report from gathered telemetry data.
+func (c *Collector) BuildReport(context.Context) (TraceData, error) {
+	dt := TraceData{}
+
+	// TODO: Implement handling and logging errors for each collected data point
+
+	return dt, nil
+}
+
+// Collect collects and exports telemetry data.
+// It exports data using provided exporter.
+func (c *Collector) Collect(ctx context.Context) {
+	glog.V(3).Info("Collecting telemetry data")
+	traceData, err := c.BuildReport(ctx)
+	if err != nil {
+		glog.Errorf("Error collecting telemetry data: %v", err)
+	}
+	err = c.Exporter.Export(ctx, traceData)
+	if err != nil {
+		glog.Errorf("Error exporting telemetry data: %v", err)
+	}
+	glog.V(3).Info("Exported telemetry data")
+}
+
+// Start starts running NIC Telemetry Collector.
+func (c *Collector) Start(ctx context.Context) {
+	wait.JitterUntilWithContext(ctx, c.Collect, c.Period, 0.1, true)
+}
+
+// GetVSCount returns number of VirtualServers in watched namespaces.
+//
+// Note: this is a placeholder function.
+func (c *Collector) GetVSCount() int {
+	// Placeholder function
+	return 0
+}
+
+// GetTSCount returns number of TransportServers in watched namespaces.
+//
+// Note: this is a placeholder function.
+func (c *Collector) GetTSCount() int {
+	// Placeholder function
+	return 0
+}