Collect count of VirtualServer, VirtualServerRoute and TransportServer resources

cmd/nginx-ingress/main.go

@@ -152,7 +152,12 @@ func main() {
 	controllerNamespace := os.Getenv("POD_NAMESPACE")
 
 	transportServerValidator := cr_validation.NewTransportServerValidator(*enableTLSPassthrough, *enableSnippets, *nginxPlus)
-	virtualServerValidator := cr_validation.NewVirtualServerValidator(cr_validation.IsPlus(*nginxPlus), cr_validation.IsDosEnabled(*appProtectDos), cr_validation.IsCertManagerEnabled(*enableCertManager), cr_validation.IsExternalDNSEnabled(*enableExternalDNS))
+	virtualServerValidator := cr_validation.NewVirtualServerValidator(
+		cr_validation.IsPlus(*nginxPlus),
+		cr_validation.IsDosEnabled(*appProtectDos),
+		cr_validation.IsCertManagerEnabled(*enableCertManager),
+		cr_validation.IsExternalDNSEnabled(*enableExternalDNS),
+	)
 
 	if *enableServiceInsight {
 		createHealthProbeEndpoint(kubeClient, plusClient, cnf)

go.mod

@@ -19,6 +19,7 @@ require (
 	github.com/prometheus/common v0.47.0
 	github.com/spiffe/go-spiffe/v2 v2.1.7
 	github.com/stretchr/testify v1.8.4
+	go.opentelemetry.io/otel v1.21.0
 	golang.org/x/exp v0.0.0-20231226003508-02704c960a9b
 	k8s.io/api v0.29.2
 	k8s.io/apimachinery v0.29.2
@@ -98,7 +99,6 @@ require (
 	go.etcd.io/etcd/client/v3 v3.5.11 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.46.1 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1 // indirect
-	go.opentelemetry.io/otel v1.21.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.21.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.21.0 // indirect
 	go.opentelemetry.io/otel/metric v1.21.0 // indirect

internal/configs/configurator.go

@@ -962,6 +962,7 @@ func (cnf *Configurator) deleteTransportServer(key string) error {
 	name := getFileNameForTransportServerFromKey(key)
 	cnf.nginxManager.DeleteStreamConfig(name)
 
+	delete(cnf.transportServers, name)
 	// update TLS Passthrough Hosts config in case we have a TLS Passthrough TransportServer
 	if _, exists := cnf.tlsPassthroughPairs[key]; exists {
 		delete(cnf.tlsPassthroughPairs, key)
@@ -1468,23 +1469,29 @@ func (cnf *Configurator) GetIngressCounts() map[string]int {
 		}
 	}
 
-	for _, min := range cnf.minions {
-		counters["minion"] += len(min)
+	for _, minion := range cnf.minions {
+		counters["minion"] += len(minion)
 	}
 
 	return counters
 }
 
-// GetVirtualServerCounts returns the total count of VS/VSR resources that are handled by the Ingress Controller
+// GetVirtualServerCounts returns the total count of
+// VirtualServer and VirtualServerRoute resources that are handled by the Ingress Controller
 func (cnf *Configurator) GetVirtualServerCounts() (vsCount int, vsrCount int) {
 	vsCount = len(cnf.virtualServers)
 	for _, vs := range cnf.virtualServers {
 		vsrCount += len(vs.VirtualServerRoutes)
 	}
-
 	return vsCount, vsrCount
 }
 
+// GetTransportServerCounts returns the total count of
+// TransportServer resources that are handled by the Ingress Controller
+func (cnf *Configurator) GetTransportServerCounts() (tsCount int) {
+	return len(cnf.transportServers)
+}
+
 // AddOrUpdateSpiffeCerts writes Spiffe certs and keys to disk and reloads NGINX
 func (cnf *Configurator) AddOrUpdateSpiffeCerts(svidResponse *workloadapi.X509Context) error {
 	svid := svidResponse.DefaultSVID()

internal/k8s/controller.go

@@ -245,7 +245,6 @@ func NewLoadBalancerController(input NewLoadBalancerControllerInput) *LoadBalanc
 		isLatencyMetricsEnabled:      input.IsLatencyMetricsEnabled,
 		isIPV6Disabled:               input.IsIPV6Disabled,
 	}
-
 	eventBroadcaster := record.NewBroadcaster()
 	eventBroadcaster.StartLogging(glog.Infof)
 	eventBroadcaster.StartRecordingToSink(&core_v1.EventSinkImpl{
@@ -277,18 +276,6 @@ func NewLoadBalancerController(input NewLoadBalancerControllerInput) *LoadBalanc
 		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.TelemetryReportingPeriod),
-		)
-		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)
@@ -357,6 +344,24 @@ func NewLoadBalancerController(input NewLoadBalancerControllerInput) *LoadBalanc
 
 	lbc.secretStore = secrets.NewLocalSecretStore(lbc.configurator)
 
+	// NIC Telemetry Reporting
+	if input.EnableTelemetryReporting {
+		collectorConfig := telemetry.CollectorConfig{
+			K8sClientReader:       input.KubeClient,
+			CustomK8sClientReader: input.ConfClient,
+			Period:                5 * time.Second,
+			Configurator:          lbc.configurator,
+		}
+		lbc.telemetryChan = make(chan struct{})
+		collector, err := telemetry.NewCollector(
+			collectorConfig,
+		)
+		if err != nil {
+			glog.Fatalf("failed to initialize telemetry collector: %v", err)
+		}
+		lbc.telemetryCollector = collector
+	}
+
 	return lbc
 }
 

internal/k8s/controller_test.go

@@ -3757,6 +3757,7 @@ func TestNewTelemetryCollector(t *testing.T) {
 	testCases := []struct {
 		testCase          string
 		input             NewLoadBalancerControllerInput
+		collectorConfig   telemetry.CollectorConfig
 		expectedCollector telemetry.Collector
 	}{
 		{
@@ -3767,8 +3768,10 @@ func TestNewTelemetryCollector(t *testing.T) {
 				TelemetryReportingPeriod: "24h",
 			},
 			expectedCollector: telemetry.Collector{
-				Period:   24 * time.Hour,
-				Exporter: telemetry.DiscardExporter,
+				Config: telemetry.CollectorConfig{
+					Period: 24 * time.Hour,
+				},
+				Exporter: &telemetry.StdoutExporter{},
 			},
 		},
 		{
@@ -3784,7 +3787,7 @@ func TestNewTelemetryCollector(t *testing.T) {
 	for _, tc := range testCases {
 		lbc := NewLoadBalancerController(tc.input)
 		if reflect.DeepEqual(tc.expectedCollector, lbc.telemetryCollector) {
-			t.Fatalf("Expected %x, but got %x", tc.expectedCollector, lbc.telemetryCollector)
+			t.Fatalf("Expected %v, but got %v", tc.expectedCollector, lbc.telemetryCollector)
 		}
 	}
 }

internal/telemetry/collector.go

@@ -0,0 +1,104 @@
+// Package telemetry provides functionality for collecting and exporting NIC telemetry data.
+package telemetry
+
+import (
+	"context"
+	"io"
+	"time"
+
+	"github.com/nginxinc/kubernetes-ingress/internal/configs"
+
+	k8s_nginx "github.com/nginxinc/kubernetes-ingress/pkg/client/clientset/versioned"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/kubernetes"
+
+	"github.com/golang/glog"
+)
+
+// Option is a functional option used for configuring TraceReporter.
+type Option func(*Collector) error
+
+// 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 {
+	// Exporter is a temp exporter for exporting telemetry data.
+	// The concrete implementation will be implemented in a separate module.
+	Exporter Exporter
+
+	// Configuration for the collector.
+	Config CollectorConfig
+}
+
+// CollectorConfig contains configuration options for a Collector
+type CollectorConfig struct {
+	// K8sClientReader is a kubernetes client.
+	K8sClientReader kubernetes.Interface
+
+	// CustomK8sClientReader is a kubernetes client for our CRDs.
+	// Note: May not need this client.
+	CustomK8sClientReader k8s_nginx.Interface
+
+	// Period to collect telemetry
+	Period time.Duration
+
+	Configurator *configs.Configurator
+}
+
+// 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(cfg CollectorConfig, opts ...Option) (*Collector, error) {
+	c := Collector{
+		Exporter: &StdoutExporter{Endpoint: io.Discard},
+		Config:   cfg,
+	}
+	for _, o := range opts {
+		if err := o(&c); err != nil {
+			return nil, err
+		}
+	}
+	return &c, nil
+}
+
+// Start starts running NIC Telemetry Collector.
+func (c *Collector) Start(ctx context.Context) {
+	wait.JitterUntilWithContext(ctx, c.Collect, c.Config.Period, 0.1, true)
+}
+
+// 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")
+	// TODO: Re-add ctx to BuildReport when collecting Node Count.
+	data, err := c.BuildReport()
+	if err != nil {
+		glog.Errorf("Error collecting telemetry data: %v", err)
+	}
+	err = c.Exporter.Export(ctx, data)
+	if err != nil {
+		glog.Errorf("Error exporting telemetry data: %v", err)
+	}
+	glog.V(3).Infof("Exported telemetry data: %+v", data)
+}
+
+// BuildReport takes context and builds report from gathered telemetry data.
+func (c *Collector) BuildReport() (Data, error) {
+	d := Data{}
+	var err error
+
+	if c.Config.Configurator != nil {
+		d.NICResourceCounts.VirtualServers, d.NICResourceCounts.VirtualServerRoutes = c.Config.Configurator.GetVirtualServerCounts()
+		d.NICResourceCounts.TransportServers = c.Config.Configurator.GetTransportServerCounts()
+	}
+	return d, err
+}