Telemetry Job

cmd/nginx-ingress/flags.go

@@ -3,20 +3,22 @@
 import (
 	"flag"
 	"fmt"
+	"github.com/golang/glog"
+	api_v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/util/validation"
 	"net"
 	"os"
 	"regexp"
 	"strconv"
 	"strings"
-
-	"github.com/golang/glog"
-	api_v1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/util/validation"
 )
 
 const (
-	dynamicSSLReloadParam = "ssl-dynamic-reload"
+	dynamicSSLReloadParam           = "ssl-dynamic-reload"
+	enableTelemetryReportingParam   = "enable-telemetry-reporting"
+	telemetryReportingPeriodParam   = "telemetry-reporting-period"
+	defaultTelemetryReportingPeriod = "24h"
 )
 
 var (
@@ -201,6 +203,10 @@
 
 	enableDynamicSSLReload = flag.Bool(dynamicSSLReloadParam, true, "Enable reloading of SSL Certificates without restarting the NGINX process.")
 
+	enableTelemetryReporting = flag.Bool(enableTelemetryReportingParam, true, "Enable gathering and reporting of product related telemetry.")
+
+	telemetryReportingPeriod = flag.String(telemetryReportingPeriodParam, defaultTelemetryReportingPeriod, "Period at which product telemetry is reported.")
+
 	startupCheckFn func() error
 )
 

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++ {

deployments/deployment/nginx-plus-ingress.yaml

@@ -4,7 +4,7 @@ metadata:
   name: nginx-ingress
   namespace: nginx-ingress
 spec:
-  replicas: 1
+  replicas: 2
   selector:
     matchLabels:
       app: nginx-ingress
@@ -33,7 +33,7 @@ spec:
 #      - name: nginx-log
 #        emptyDir: {}
       containers:
-      - image: nginx-plus-ingress:3.4.0
+      - image: nginx-plus-ingress:telemetry
         imagePullPolicy: IfNotPresent
         name: nginx-plus-ingress
         ports:
@@ -96,7 +96,7 @@ spec:
          #- -enable-external-dns
          #- -enable-app-protect
          #- -enable-app-protect-dos
-         #- -v=3 # Enables extensive logging. Useful for troubleshooting.
+          - -v=1 # Enables extensive logging. Useful for troubleshooting.
          #- -report-ingress-status
          #- -external-service=nginx-ingress
          #- -enable-prometheus-metrics

internal/k8s/controller.go

@@ -19,6 +19,7 @@
 import (
 	"context"
 	"fmt"
+	"github.com/nginxinc/kubernetes-ingress/internal/telemetry"
 	"net"
 	"strconv"
 	"strings"
@@ -161,6 +162,8 @@
 	enableBatchReload             bool
 	isIPV6Disabled                bool
 	namespaceWatcherController    cache.Controller
+	telemetryReporter             telemetry.Reporter
+	telemetryChan                 chan struct{}
 }
 
 var keyFunc = cache.DeletionHandlingMetaNamespaceKeyFunc
@@ -206,6 +209,8 @@
 	ExternalDNSEnabled           bool
 	IsIPV6Disabled               bool
 	WatchNamespaceLabel          string
+	TelemetryReportPeriod        string
+	EnableTelemetryReporting     bool
 }
 
 // NewLoadBalancerController creates a controller
@@ -271,6 +276,24 @@
 		lbc.externalDNSController = ed_controller.NewController(ed_controller.BuildOpts(context.TODO(), lbc.namespaceList, lbc.recorder, lbc.confClient, input.ResyncPeriod, isDynamicNs))
 	}
 
+	if input.EnableTelemetryReporting {
+		// Placeholder exporter.
+		exporter := &telemetry.StdOutExporter{}
+		lbc.telemetryChan = make(chan struct{})
+		period, err := time.ParseDuration(input.TelemetryReportPeriod)
+
+		if err != nil {
+			glog.Fatalf("Error parsing duration for telemetry: %v", err)
+		}
+
+		config := telemetry.TraceTelemetryReporterConfig{
+			Exporter:        exporter,
+			ReportingPeriod: period,
+			Data:            telemetry.Data{},
+		}
+		lbc.telemetryReporter = telemetry.NewTelemetryReporter(config)
+	}
+
 	glog.V(3).Infof("Nginx Ingress Controller has class: %v", input.IngressClass)
 
 	lbc.namespacedInformers = make(map[string]*namespacedInformer)
@@ -683,10 +706,24 @@
 	if lbc.externalDNSController != nil {
 		go lbc.externalDNSController.Run(lbc.ctx.Done())
 	}
+
 	if lbc.leaderElector != nil {
 		go lbc.leaderElector.Run(lbc.ctx)
 	}
 
+	if lbc.telemetryReporter != nil {
+		go func(ctx context.Context) {
+			glog.V(1).Info("-- Checking if leader is set --")
+			select {
+			case <-lbc.telemetryChan:
+				lbc.telemetryReporter.Start(lbc.ctx)
+			case <-ctx.Done():
+				glog.V(1).Info("-- DONE Reporting Telemetry --")
+				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,9 @@
 // Version provides a fake implementation of Version.
 func (*FakeManager) Version() Version {
 	glog.V(3).Info("Printing nginx version")
-	return Version{}
+	return Version{
+		raw: "nginx/1.25.1 (nginx-plus-r30-p1)",
+	}
 }
 
 // Start provides a fake implementation of Start.

internal/telemetry/exporter.go

@@ -0,0 +1,25 @@
+package telemetry
+
+import (
+	"context"
+	"github.com/golang/glog"
+)
+
+type Data struct {
+}
+
+type Exporter interface {
+	Export(ctx context.Context, data Data) error
+}
+
+type StdOutExporter struct {
+}
+
+func NewStdOutExporter() *StdOutExporter {
+	return &StdOutExporter{}
+}
+
+func (s *StdOutExporter) Export(_ context.Context, data Data) error {
+	glog.V(1).Infof("Exporting data %v", data)
+	return nil
+}

internal/telemetry/exporter_test.go

@@ -0,0 +1,18 @@
+package telemetry
+
+import (
+	"context"
+	"testing"
+)
+
+func TestExportData(t *testing.T) {
+	t.Parallel()
+
+	exporter := NewStdOutExporter()
+
+	err := exporter.Export(context.Background(), Data{})
+
+	if err != nil {
+		t.Fatalf("Expeceted no error, but got %s", err.Error())
+	}
+}

internal/telemetry/telemetry.go

@@ -0,0 +1,55 @@
+package telemetry
+
+import (
+	"context"
+	"github.com/golang/glog"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"time"
+)
+
+const (
+	jitterFactor = 0.1
+	sliding      = true
+)
+
+type Reporter interface {
+	Start(ctx context.Context)
+}
+
+type TraceTelemetryReporterConfig struct {
+	Data            Data
+	Exporter        Exporter
+	ReportingPeriod time.Duration
+}
+
+type TraceTelemetryReporter struct {
+	config TraceTelemetryReporterConfig
+}
+
+func NewTelemetryReporter(config TraceTelemetryReporterConfig) *TraceTelemetryReporter {
+	return &TraceTelemetryReporter{
+		config: config,
+	}
+}
+
+func (t *TraceTelemetryReporter) Start(ctx context.Context) {
+	wait.JitterUntilWithContext(ctx, t.report, t.config.ReportingPeriod, jitterFactor, sliding)
+}
+
+func (t *TraceTelemetryReporter) report(ctx context.Context) {
+	// Gather data here
+	t.setVirtualServerCount()
+	t.setTransportServerCount()
+
+	if err := t.config.Exporter.Export(ctx, t.config.Data); err != nil {
+		glog.Errorf("Error exporting telemetry data: %v", err)
+	}
+}
+
+func (t *TraceTelemetryReporter) setVirtualServerCount() {
+	// Placeholder function
+}
+
+func (t *TraceTelemetryReporter) setTransportServerCount() {
+	// Placeholder function
+}

internal/telemetry/telemetry_test.go

@@ -0,0 +1,30 @@
+package telemetry
+
+import (
+	"context"
+	"reflect"
+	"testing"
+)
+
+type MockTelemetryReport struct {
+	data Data
+}
+
+func (m *MockTelemetryReport) Start(_ context.Context) {
+	m.data = Data{}
+}
+
+func TestCollectData(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	mtr := &MockTelemetryReport{
+		Data{},
+	}
+	expectedData := Data{}
+	mtr.Start(ctx)
+
+	if !reflect.DeepEqual(mtr.data, expectedData) {
+		t.Fatalf("expected %v, but got %v", expectedData, mtr.data)
+	}
+}