Revert "chore: Refactor OTel resources (#1659)"

This reverts commit fbcb5b1.

.testcoverage.yml

@@ -35,7 +35,7 @@ override:
     path: ^internal/reconciler/telemetry$
   - threshold: 74
     path: ^internal/reconciler/tracepipeline$
-  - threshold: 83
+  - threshold: 82
     path: ^internal/resources/otelcollector$
   - threshold: 78
     path: ^internal/resources/selfmonitor$

controllers/telemetry/logpipeline_controller.go

@@ -52,6 +52,7 @@ import (
 )
 
 const (
+	// FluentBit
 	fbBaseName                = "telemetry-fluent-bit"
 	fbSectionsConfigMapName   = fbBaseName + "-sections"
 	fbFilesConfigMapName      = fbBaseName + "-files"
@@ -60,6 +61,9 @@ const (
 	fbEnvConfigSecretName     = fbBaseName + "-env"
 	fbTLSFileConfigSecretName = fbBaseName + "-output-tls-config"
 	fbDaemonSetName           = fbBaseName
+
+	// OTel
+	otelLogGatewayName = "telemetry-log-gateway"
 )
 
 var (
@@ -68,6 +72,17 @@ var (
 	fbMemoryLimit   = resource.MustParse("1Gi")
 	fbCPURequest    = resource.MustParse("100m")
 	fbMemoryRequest = resource.MustParse("50Mi")
+
+	// OTel
+	// TODO: Check if these values need to be adjusted
+	logGatewayBaseCPULimit         = resource.MustParse("700m")
+	logGatewayDynamicCPULimit      = resource.MustParse("500m")
+	logGatewayBaseMemoryLimit      = resource.MustParse("500Mi")
+	logGatewayDynamicMemoryLimit   = resource.MustParse("1500Mi")
+	logGatewayBaseCPURequest       = resource.MustParse("100m")
+	logGatewayDynamicCPURequest    = resource.MustParse("100m")
+	logGatewayBaseMemoryRequest    = resource.MustParse("32Mi")
+	logGatewayDynamicMemoryRequest = resource.MustParse("0")
 )
 
 // LogPipelineController reconciles a LogPipeline object
@@ -84,6 +99,7 @@ type LogPipelineControllerConfig struct {
 	OTelCollectorImage          string
 	FluentBitPriorityClassName  string
 	LogGatewayPriorityClassName string
+	LogGatewayServiceName       string
 	RestConfig                  *rest.Config
 	SelfMonitorName             string
 	TelemetryNamespace          string
@@ -200,14 +216,48 @@ func configureFluentBitReconciler(client client.Client, config LogPipelineContro
 
 //nolint:unparam // error is always nil: An error could be returned after implementing the IstioStatusChecker (TODO)
 func configureOtelReconciler(client client.Client, config LogPipelineControllerConfig, _ *prober.LogPipelineProber) (*logpipelineotel.Reconciler, error) {
+	otelConfig := logpipelineotel.Config{
+		LogGatewayName:     otelLogGatewayName,
+		TelemetryNamespace: config.TelemetryNamespace,
+	}
+
+	gatewayConfig := otelcollector.GatewayConfig{
+		Config: otelcollector.Config{
+			BaseName:  otelLogGatewayName,
+			Namespace: config.TelemetryNamespace,
+		},
+		Deployment: otelcollector.DeploymentConfig{
+			Image:                config.OTelCollectorImage,
+			PriorityClassName:    config.LogGatewayPriorityClassName,
+			BaseCPULimit:         logGatewayBaseCPULimit,
+			DynamicCPULimit:      logGatewayDynamicCPULimit,
+			BaseMemoryLimit:      logGatewayBaseMemoryLimit,
+			DynamicMemoryLimit:   logGatewayDynamicMemoryLimit,
+			BaseCPURequest:       logGatewayBaseCPURequest,
+			DynamicCPURequest:    logGatewayDynamicCPURequest,
+			BaseMemoryRequest:    logGatewayBaseMemoryRequest,
+			DynamicMemoryRequest: logGatewayDynamicMemoryRequest,
+		},
+		OTLPServiceName: config.LogGatewayServiceName,
+	}
+
 	pipelineValidator := &logpipelineotel.Validator{
 		// TODO: Add validators
 	}
 
+	rbac := otelcollector.MakeLogGatewayRBAC(
+		types.NamespacedName{
+			Name:      otelLogGatewayName,
+			Namespace: config.TelemetryNamespace,
+		})
+
 	otelReconciler := logpipelineotel.New(
 		client,
-		config.TelemetryNamespace,
-		otelcollector.NewLogGatewayApplierDeleter(config.OTelCollectorImage, config.TelemetryNamespace, config.LogGatewayPriorityClassName),
+		otelConfig,
+		&otelcollector.GatewayApplierDeleter{
+			Config: gatewayConfig,
+			RBAC:   rbac,
+		},
 		&gateway.Builder{Reader: client},
 		&workloadstatus.DeploymentProber{Client: client},
 		pipelineValidator,

controllers/telemetry/metricpipeline_controller.go

@@ -24,6 +24,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	networkingv1 "k8s.io/api/networking/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/rest"
@@ -55,7 +56,25 @@ import (
 )
 
 const (
-	maxMetricPipelines = 3
+	maxMetricPipelines    = 3
+	metricGatewayBaseName = "telemetry-metric-gateway"
+	metricAgentBaseName   = "telemetry-metric-agent"
+)
+
+var (
+	metricAgentCPULimit      = resource.MustParse("1")
+	metricAgentMemoryLimit   = resource.MustParse("1200Mi")
+	metricAgentCPURequest    = resource.MustParse("15m")
+	metricAgentMemoryRequest = resource.MustParse("50Mi")
+
+	metricGatewayBaseCPULimit         = resource.MustParse("900m")
+	metricGatewayDynamicCPULimit      = resource.MustParse("100m")
+	metricGatewayBaseMemoryLimit      = resource.MustParse("512Mi")
+	metricGatewayDynamicMemoryLimit   = resource.MustParse("512Mi")
+	metricGatewayBaseCPURequest       = resource.MustParse("25m")
+	metricGatewayDynamicCPURequest    = resource.MustParse("0")
+	metricGatewayBaseMemoryRequest    = resource.MustParse("32Mi")
+	metricGatewayDynamicMemoryRequest = resource.MustParse("0")
 )
 
 // MetricPipelineController reconciles a MetricPipeline object
@@ -69,6 +88,7 @@ type MetricPipelineController struct {
 type MetricPipelineControllerConfig struct {
 	MetricAgentPriorityClassName   string
 	MetricGatewayPriorityClassName string
+	MetricGatewayServiceName       string
 	ModuleVersion                  string
 	OTelCollectorImage             string
 	RestConfig                     *rest.Config
@@ -103,24 +123,25 @@ func NewMetricPipelineController(client client.Client, reconcileTriggerChan <-ch
 		return nil, err
 	}
 
-	agentConfigBuilder := &agent.Builder{
-		Config: agent.BuilderConfig{
-			GatewayOTLPServiceName: types.NamespacedName{Namespace: config.TelemetryNamespace, Name: otelcollector.MetricOTLPServiceName},
-		},
+	reconcilerConfig := metricpipeline.Config{
+		AgentName:          metricAgentBaseName,
+		GatewayName:        metricGatewayBaseName,
+		ModuleVersion:      config.ModuleVersion,
+		TelemetryNamespace: config.TelemetryNamespace,
 	}
-
-	gatewayConfigBuilder := &gateway.Builder{Reader: client}
-
 	reconciler := metricpipeline.New(
 		client,
-		config.TelemetryNamespace,
-		config.ModuleVersion,
-		otelcollector.NewMetricAgentApplierDeleter(config.OTelCollectorImage, config.TelemetryNamespace, config.MetricAgentPriorityClassName),
-		agentConfigBuilder,
+		reconcilerConfig,
+		newMetricAgentApplierDeleter(config),
+		&agent.Builder{
+			Config: agent.BuilderConfig{
+				GatewayOTLPServiceName: types.NamespacedName{Namespace: config.TelemetryNamespace, Name: config.MetricGatewayServiceName},
+			},
+		},
 		&workloadstatus.DaemonSetProber{Client: client},
 		flowHealthProber,
-		otelcollector.NewMetricGatewayApplierDeleter(config.OTelCollectorImage, config.TelemetryNamespace, config.MetricGatewayPriorityClassName),
-		gatewayConfigBuilder,
+		newMetricGatewayApplierDeleter(config),
+		&gateway.Builder{Reader: client},
 		&workloadstatus.DeploymentProber{Client: client},
 		istiostatus.NewChecker(discoveryClient),
 		overrides.New(client, overrides.HandlerConfig{SystemNamespace: config.TelemetryNamespace}),
@@ -136,6 +157,69 @@ func NewMetricPipelineController(client client.Client, reconcileTriggerChan <-ch
 	}, nil
 }
 
+func newMetricAgentApplierDeleter(config MetricPipelineControllerConfig) *otelcollector.AgentApplierDeleter {
+	rbac := otelcollector.MakeMetricAgentRBAC(
+		types.NamespacedName{
+			Name:      metricAgentBaseName,
+			Namespace: config.TelemetryNamespace,
+		},
+	)
+
+	agentConfig := otelcollector.AgentConfig{
+		Config: otelcollector.Config{
+			BaseName:  metricAgentBaseName,
+			Namespace: config.TelemetryNamespace,
+		},
+		DaemonSet: otelcollector.DaemonSetConfig{
+			Image:             config.OTelCollectorImage,
+			PriorityClassName: config.MetricAgentPriorityClassName,
+			CPULimit:          metricAgentCPULimit,
+			MemoryLimit:       metricAgentMemoryLimit,
+			CPURequest:        metricAgentCPURequest,
+			MemoryRequest:     metricAgentMemoryRequest,
+		},
+	}
+
+	return &otelcollector.AgentApplierDeleter{
+		Config: agentConfig,
+		RBAC:   rbac,
+	}
+}
+
+func newMetricGatewayApplierDeleter(config MetricPipelineControllerConfig) *otelcollector.GatewayApplierDeleter {
+	rbac := otelcollector.MakeMetricGatewayRBAC(
+		types.NamespacedName{
+			Name:      metricGatewayBaseName,
+			Namespace: config.TelemetryNamespace,
+		},
+	)
+
+	gatewayConfig := otelcollector.GatewayConfig{
+		Config: otelcollector.Config{
+			BaseName:  metricGatewayBaseName,
+			Namespace: config.TelemetryNamespace,
+		},
+		Deployment: otelcollector.DeploymentConfig{
+			Image:                config.OTelCollectorImage,
+			PriorityClassName:    config.MetricGatewayPriorityClassName,
+			BaseCPULimit:         metricGatewayBaseCPULimit,
+			DynamicCPULimit:      metricGatewayDynamicCPULimit,
+			BaseMemoryLimit:      metricGatewayBaseMemoryLimit,
+			DynamicMemoryLimit:   metricGatewayDynamicMemoryLimit,
+			BaseCPURequest:       metricGatewayBaseCPURequest,
+			DynamicCPURequest:    metricGatewayDynamicCPURequest,
+			BaseMemoryRequest:    metricGatewayBaseMemoryRequest,
+			DynamicMemoryRequest: metricGatewayDynamicMemoryRequest,
+		},
+		OTLPServiceName: config.MetricGatewayServiceName,
+	}
+
+	return &otelcollector.GatewayApplierDeleter{
+		Config: gatewayConfig,
+		RBAC:   rbac,
+	}
+}
+
 func (r *MetricPipelineController) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	return r.reconciler.Reconcile(ctx, req)
 }

controllers/telemetry/tracepipeline_controller.go

@@ -24,6 +24,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	networkingv1 "k8s.io/api/networking/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/rest"
@@ -54,7 +55,19 @@ import (
 )
 
 const (
-	maxTracePipelines = 3
+	maxTracePipelines    = 3
+	traceGatewayBaseName = "telemetry-trace-gateway"
+)
+
+var (
+	traceGatewayBaseCPULimit         = resource.MustParse("700m")
+	traceGatewayDynamicCPULimit      = resource.MustParse("500m")
+	traceGatewayBaseMemoryLimit      = resource.MustParse("500Mi")
+	traceGatewayDynamicMemoryLimit   = resource.MustParse("1500Mi")
+	traceGatewayBaseCPURequest       = resource.MustParse("100m")
+	traceGatewayDynamicCPURequest    = resource.MustParse("100m")
+	traceGatewayBaseMemoryRequest    = resource.MustParse("32Mi")
+	traceGatewayDynamicMemoryRequest = resource.MustParse("0")
 )
 
 // TracePipelineController reconciles a TracePipeline object
@@ -70,6 +83,7 @@ type TracePipelineControllerConfig struct {
 	TelemetryNamespace            string
 	OTelCollectorImage            string
 	TraceGatewayPriorityClassName string
+	TraceGatewayServiceName       string
 }
 
 func NewTracePipelineController(client client.Client, reconcileTriggerChan <-chan event.GenericEvent, config TracePipelineControllerConfig) (*TracePipelineController, error) {
@@ -99,11 +113,15 @@ func NewTracePipelineController(client client.Client, reconcileTriggerChan <-cha
 		return nil, err
 	}
 
+	reconcilerConfig := tracepipeline.Config{
+		TraceGatewayName:   traceGatewayBaseName,
+		TelemetryNamespace: config.TelemetryNamespace,
+	}
 	reconciler := tracepipeline.New(
 		client,
-		config.TelemetryNamespace,
+		reconcilerConfig,
 		flowHealthProber,
-		otelcollector.NewTraceGatewayApplierDeleter(config.OTelCollectorImage, config.TelemetryNamespace, config.TraceGatewayPriorityClassName),
+		newTraceGatewayApplierDeleter(config),
 		&gateway.Builder{Reader: client},
 		&workloadstatus.DeploymentProber{Client: client},
 		istiostatus.NewChecker(discoveryClient),
@@ -119,6 +137,39 @@ func NewTracePipelineController(client client.Client, reconcileTriggerChan <-cha
 	}, nil
 }
 
+func newTraceGatewayApplierDeleter(config TracePipelineControllerConfig) *otelcollector.GatewayApplierDeleter {
+	rbac := otelcollector.MakeTraceGatewayRBAC(
+		types.NamespacedName{
+			Name:      traceGatewayBaseName,
+			Namespace: config.TelemetryNamespace,
+		})
+
+	gatewayConfig := otelcollector.GatewayConfig{
+		Config: otelcollector.Config{
+			BaseName:  traceGatewayBaseName,
+			Namespace: config.TelemetryNamespace,
+		},
+		Deployment: otelcollector.DeploymentConfig{
+			Image:                config.OTelCollectorImage,
+			PriorityClassName:    config.TraceGatewayPriorityClassName,
+			BaseCPULimit:         traceGatewayBaseCPULimit,
+			DynamicCPULimit:      traceGatewayDynamicCPULimit,
+			BaseMemoryLimit:      traceGatewayBaseMemoryLimit,
+			DynamicMemoryLimit:   traceGatewayDynamicMemoryLimit,
+			BaseCPURequest:       traceGatewayBaseCPURequest,
+			DynamicCPURequest:    traceGatewayDynamicCPURequest,
+			BaseMemoryRequest:    traceGatewayBaseMemoryRequest,
+			DynamicMemoryRequest: traceGatewayDynamicMemoryRequest,
+		},
+		OTLPServiceName: config.TraceGatewayServiceName,
+	}
+
+	return &otelcollector.GatewayApplierDeleter{
+		Config: gatewayConfig,
+		RBAC:   rbac,
+	}
+}
+
 func (r *TracePipelineController) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	return r.reconciler.Reconcile(ctx, req)
 }