fix: Consider the existence of Istio when applying and deleting resources for Log Gateway

controllers/telemetry/logpipeline_controller.go

@@ -204,12 +204,18 @@ func configureOtelReconciler(client client.Client, config LogPipelineControllerC
 		// TODO: Add validators
 	}
 
+	discoveryClient, err := discovery.NewDiscoveryClientForConfig(config.RestConfig)
+	if err != nil {
+		return nil, err
+	}
+
 	otelReconciler := logpipelineotel.New(
 		client,
 		config.TelemetryNamespace,
 		otelcollector.NewLogGatewayApplierDeleter(config.OTelCollectorImage, config.TelemetryNamespace, config.LogGatewayPriorityClassName),
 		&gateway.Builder{Reader: client},
 		&workloadstatus.DeploymentProber{Client: client},
+		istiostatus.NewChecker(discoveryClient),
 		pipelineValidator,
 		&conditions.ErrorToMessageConverter{})
 

internal/reconciler/logpipeline/otel/reconciler.go

@@ -14,6 +14,7 @@ import (
 	"github.com/kyma-project/telemetry-manager/internal/errortypes"
 	"github.com/kyma-project/telemetry-manager/internal/otelcollector/config/log/gateway"
 	"github.com/kyma-project/telemetry-manager/internal/otelcollector/config/otlpexporter"
+	"github.com/kyma-project/telemetry-manager/internal/otelcollector/ports"
 	"github.com/kyma-project/telemetry-manager/internal/reconciler/commonstatus"
 	"github.com/kyma-project/telemetry-manager/internal/reconciler/logpipeline"
 	"github.com/kyma-project/telemetry-manager/internal/resources/otelcollector"
@@ -38,6 +39,10 @@ type FlowHealthProber interface {
 	Probe(ctx context.Context, pipelineName string) (prober.OTelPipelineProbeResult, error)
 }
 
+type IstioStatusChecker interface {
+	IsIstioActive(ctx context.Context) bool
+}
+
 var _ logpipeline.LogPipelineReconciler = &Reconciler{}
 
 type Reconciler struct {
@@ -49,6 +54,7 @@ type Reconciler struct {
 	gatewayApplierDeleter GatewayApplierDeleter
 	gatewayConfigBuilder  GatewayConfigBuilder
 	gatewayProber         commonstatus.Prober
+	istioStatusChecker    IstioStatusChecker
 	pipelineValidator     *Validator
 	errToMessageConverter commonstatus.ErrorToMessageConverter
 }
@@ -59,6 +65,7 @@ func New(
 	gatewayApplierDeleter GatewayApplierDeleter,
 	gatewayConfigBuilder GatewayConfigBuilder,
 	gatewayProber commonstatus.Prober,
+	istioStatusChecker IstioStatusChecker,
 	pipelineValidator *Validator,
 	errToMessageConverter commonstatus.ErrorToMessageConverter,
 ) *Reconciler {
@@ -68,6 +75,7 @@ func New(
 		gatewayApplierDeleter: gatewayApplierDeleter,
 		gatewayConfigBuilder:  gatewayConfigBuilder,
 		gatewayProber:         gatewayProber,
+		istioStatusChecker:    istioStatusChecker,
 		pipelineValidator:     pipelineValidator,
 		errToMessageConverter: errToMessageConverter,
 	}
@@ -107,8 +115,7 @@ func (r *Reconciler) doReconcile(ctx context.Context, pipeline *telemetryv1alpha
 	if len(reconcilablePipelines) == 0 {
 		logf.FromContext(ctx).V(1).Info("cleaning up log pipeline resources: all log pipelines are non-reconcilable")
 
-		// TODO: Set 'false' to 'r.istioStatusChecker.IsIstioActive(ctx)' istio is implemented for this type of pipeline
-		if err = r.gatewayApplierDeleter.DeleteResources(ctx, r.Client, false); err != nil {
+		if err = r.gatewayApplierDeleter.DeleteResources(ctx, r.Client, r.istioStatusChecker.IsIstioActive(ctx)); err != nil {
 			return fmt.Errorf("failed to delete gateway resources: %w", err)
 		}
 
@@ -174,9 +181,19 @@ func (r *Reconciler) reconcileLogGateway(ctx context.Context, pipeline *telemetr
 		return fmt.Errorf("failed to marshal collector config: %w", err)
 	}
 
+	isIstioActive := r.istioStatusChecker.IsIstioActive(ctx)
+
+	allowedPorts := getGatewayPorts()
+	if isIstioActive {
+		allowedPorts = append(allowedPorts, ports.IstioEnvoy)
+	}
+
 	opts := otelcollector.GatewayApplyOptions{
+		AllowedPorts:                   allowedPorts,
 		CollectorConfigYAML:            string(collectorConfigYAML),
 		CollectorEnvVars:               collectorEnvVars,
+		IstioEnabled:                   isIstioActive,
+		IstioExcludePorts:              []int32{ports.Metrics},
 		Replicas:                       r.getReplicaCountFromTelemetry(ctx),
 		ResourceRequirementsMultiplier: len(allPipelines),
 	}
@@ -218,3 +235,12 @@ func (r *Reconciler) getReplicaCountFromTelemetry(ctx context.Context) int32 {
 
 	return defaultReplicaCount
 }
+
+func getGatewayPorts() []int32 {
+	return []int32{
+		ports.Metrics,
+		ports.HealthCheck,
+		ports.OTLPHTTP,
+		ports.OTLPGRPC,
+	}
+}

internal/reconciler/logpipeline/otel/reconciler_test.go

@@ -20,6 +20,7 @@ import (
 	commonStatusStubs "github.com/kyma-project/telemetry-manager/internal/reconciler/commonstatus/stubs"
 	logpipelinemocks "github.com/kyma-project/telemetry-manager/internal/reconciler/logpipeline/mocks"
 	"github.com/kyma-project/telemetry-manager/internal/reconciler/logpipeline/otel/mocks"
+	"github.com/kyma-project/telemetry-manager/internal/reconciler/logpipeline/stubs"
 	testutils "github.com/kyma-project/telemetry-manager/internal/utils/test"
 	"github.com/kyma-project/telemetry-manager/internal/workloadstatus"
 )
@@ -32,6 +33,8 @@ func TestReconcile(t *testing.T) {
 	overridesHandlerStub := &logpipelinemocks.OverridesHandler{}
 	overridesHandlerStub.On("LoadOverrides", context.Background()).Return(&overrides.Config{}, nil)
 
+	istioStatusCheckerStub := &stubs.IstioStatusChecker{IsActive: false}
+
 	telemetryNamespace := "default"
 
 	t.Run("log gateway probing failed", func(t *testing.T) {
@@ -63,6 +66,7 @@ func TestReconcile(t *testing.T) {
 			gatewayApplierDeleterMock,
 			gatewayConfigBuilderMock,
 			gatewayProberStub,
+			istioStatusCheckerStub,
 			pipelineValidatorWithStubs,
 			errToMsg)
 		err := sut.Reconcile(context.Background(), &pipeline)
@@ -111,6 +115,7 @@ func TestReconcile(t *testing.T) {
 			gatewayApplierDeleterMock,
 			gatewayConfigBuilderMock,
 			gatewayProberStub,
+			istioStatusCheckerStub,
 			pipelineValidatorWithStubs,
 			errToMsg)
 		err := sut.Reconcile(context.Background(), &pipeline)
@@ -159,6 +164,7 @@ func TestReconcile(t *testing.T) {
 			gatewayApplierDeleterMock,
 			gatewayConfigBuilderMock,
 			gatewayProberStub,
+			istioStatusCheckerStub,
 			pipelineValidatorWithStubs,
 			errToMsg)
 		err := sut.Reconcile(context.Background(), &pipeline)

internal/resources/otelcollector/gateway.go

@@ -77,8 +77,9 @@ var (
 
 func NewLogGatewayApplierDeleter(image, namespace, priorityClassName string) *GatewayApplierDeleter {
 	extraLabels := map[string]string{
-		logGatewayIngestKey: "true",
-		logGatewayExportKey: "true",
+		logGatewayIngestKey:   "true",
+		logGatewayExportKey:   "true",
+		istioSidecarInjectKey: "true", // inject istio sidecar
 	}
 
 	return &GatewayApplierDeleter{

internal/resources/otelcollector/testdata/log-gateway.yaml

@@ -99,6 +99,7 @@ spec:
       creationTimestamp: null
       labels:
         app.kubernetes.io/name: telemetry-log-gateway
+        sidecar.istio.io/inject: "true"
         telemetry.kyma-project.io/log-export: "true"
         telemetry.kyma-project.io/log-ingest: "true"
     spec: