From 1638dde57c85796bc1308980173238aaa85902c5 Mon Sep 17 00:00:00 2001 From: Nikita Andriyanov Date: Tue, 26 May 2026 16:38:46 +0300 Subject: [PATCH 1/8] add port type --- api/v1alpha1/pipeline_types.go | 6 ++++++ config/crd/bases/filter.plainsight.ai_pipelines.yaml | 7 +++++++ .../crds/filter.plainsight.ai_pipelines.yaml | 7 +++++++ .../controller/pipelineinstance_controller_streaming.go | 2 +- 4 files changed, 21 insertions(+), 1 deletion(-) diff --git a/api/v1alpha1/pipeline_types.go b/api/v1alpha1/pipeline_types.go index b8ea150..846e803 100644 --- a/api/v1alpha1/pipeline_types.go +++ b/api/v1alpha1/pipeline_types.go @@ -158,6 +158,12 @@ type ServicePort struct { // +kubebuilder:default=TCP // +kubebuilder:validation:Enum=TCP;UDP Protocol corev1.Protocol `json:"protocol,omitempty"` + + // service type + // +optional + // +kubebuilder:default=TCP + // +kubebuilder:validation:Enum=corev1.ServiceTypeClusterIP;corev1.ServiceTypeLoadBalancer + Type corev1.ServiceType `json:"type,omitempty"` } // Filter defines a containerized processing step in the pipeline diff --git a/config/crd/bases/filter.plainsight.ai_pipelines.yaml b/config/crd/bases/filter.plainsight.ai_pipelines.yaml index 8357a30..a6ddb40 100644 --- a/config/crd/bases/filter.plainsight.ai_pipelines.yaml +++ b/config/crd/bases/filter.plainsight.ai_pipelines.yaml @@ -395,6 +395,13 @@ spec: maximum: 65535 minimum: 1 type: integer + type: + default: TCP + description: service type + enum: + - corev1.ServiceTypeClusterIP + - corev1.ServiceTypeLoadBalancer + type: string required: - name - port diff --git a/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml b/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml index 8357a30..a6ddb40 100644 --- a/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml +++ b/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml @@ -395,6 +395,13 @@ spec: maximum: 65535 minimum: 1 type: integer + type: + default: TCP + description: service type + enum: + - corev1.ServiceTypeClusterIP + - corev1.ServiceTypeLoadBalancer + type: string required: - name - port diff --git a/internal/controller/pipelineinstance_controller_streaming.go b/internal/controller/pipelineinstance_controller_streaming.go index df74ac0..809b166 100644 --- a/internal/controller/pipelineinstance_controller_streaming.go +++ b/internal/controller/pipelineinstance_controller_streaming.go @@ -638,7 +638,7 @@ func (r *PipelineInstanceReconciler) ensureFilterServices(ctx context.Context, p Protocol: protocol, }, }, - Type: corev1.ServiceTypeClusterIP, + Type: svcPort.Type, }, } From d251bbc6d3c316e79ca7672eb80747c7f56420e9 Mon Sep 17 00:00:00 2001 From: Nikita Andriyanov Date: Wed, 27 May 2026 08:34:14 +0300 Subject: [PATCH 2/8] fix --- api/v1alpha1/pipeline_types.go | 2 +- config/crd/bases/filter.plainsight.ai_pipelines.yaml | 2 +- .../crds/filter.plainsight.ai_pipelines.yaml | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/api/v1alpha1/pipeline_types.go b/api/v1alpha1/pipeline_types.go index 846e803..758ab4a 100644 --- a/api/v1alpha1/pipeline_types.go +++ b/api/v1alpha1/pipeline_types.go @@ -161,7 +161,7 @@ type ServicePort struct { // service type // +optional - // +kubebuilder:default=TCP + // +kubebuilder:default=corev1.ServiceTypeClusterIP // +kubebuilder:validation:Enum=corev1.ServiceTypeClusterIP;corev1.ServiceTypeLoadBalancer Type corev1.ServiceType `json:"type,omitempty"` } diff --git a/config/crd/bases/filter.plainsight.ai_pipelines.yaml b/config/crd/bases/filter.plainsight.ai_pipelines.yaml index a6ddb40..5b648cf 100644 --- a/config/crd/bases/filter.plainsight.ai_pipelines.yaml +++ b/config/crd/bases/filter.plainsight.ai_pipelines.yaml @@ -396,7 +396,7 @@ spec: minimum: 1 type: integer type: - default: TCP + default: corev1.ServiceTypeClusterIP description: service type enum: - corev1.ServiceTypeClusterIP diff --git a/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml b/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml index a6ddb40..5b648cf 100644 --- a/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml +++ b/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml @@ -396,7 +396,7 @@ spec: minimum: 1 type: integer type: - default: TCP + default: corev1.ServiceTypeClusterIP description: service type enum: - corev1.ServiceTypeClusterIP From 02bc1b13e79ba07c0534cfebd0aeb819826f5888 Mon Sep 17 00:00:00 2001 From: Nikita Andriyanov Date: Wed, 27 May 2026 09:41:04 +0300 Subject: [PATCH 3/8] fix default --- api/v1alpha1/pipeline_types.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/api/v1alpha1/pipeline_types.go b/api/v1alpha1/pipeline_types.go index 758ab4a..883f525 100644 --- a/api/v1alpha1/pipeline_types.go +++ b/api/v1alpha1/pipeline_types.go @@ -161,8 +161,8 @@ type ServicePort struct { // service type // +optional - // +kubebuilder:default=corev1.ServiceTypeClusterIP - // +kubebuilder:validation:Enum=corev1.ServiceTypeClusterIP;corev1.ServiceTypeLoadBalancer + // +kubebuilder:default=ClusterIP + // +kubebuilder:validation:Enum=ClusterIP;LoadBalancer Type corev1.ServiceType `json:"type,omitempty"` } From 1a2c13ba1092b6ade1acd524aeec0088750b4067 Mon Sep 17 00:00:00 2001 From: Nikita Andriyanov Date: Wed, 27 May 2026 09:41:31 +0300 Subject: [PATCH 4/8] up crds --- config/crd/bases/filter.plainsight.ai_pipelines.yaml | 6 +++--- .../crds/filter.plainsight.ai_pipelines.yaml | 6 +++--- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/config/crd/bases/filter.plainsight.ai_pipelines.yaml b/config/crd/bases/filter.plainsight.ai_pipelines.yaml index 5b648cf..077fac4 100644 --- a/config/crd/bases/filter.plainsight.ai_pipelines.yaml +++ b/config/crd/bases/filter.plainsight.ai_pipelines.yaml @@ -396,11 +396,11 @@ spec: minimum: 1 type: integer type: - default: corev1.ServiceTypeClusterIP + default: ClusterIP description: service type enum: - - corev1.ServiceTypeClusterIP - - corev1.ServiceTypeLoadBalancer + - ClusterIP + - LoadBalancer type: string required: - name diff --git a/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml b/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml index 5b648cf..077fac4 100644 --- a/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml +++ b/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml @@ -396,11 +396,11 @@ spec: minimum: 1 type: integer type: - default: corev1.ServiceTypeClusterIP + default: ClusterIP description: service type enum: - - corev1.ServiceTypeClusterIP - - corev1.ServiceTypeLoadBalancer + - ClusterIP + - LoadBalancer type: string required: - name From 115b54cc4692b345a7364ef0636fb42dd5d08705 Mon Sep 17 00:00:00 2001 From: Nikita Andriyanov Date: Wed, 27 May 2026 10:01:10 +0300 Subject: [PATCH 5/8] fix description --- api/v1alpha1/pipeline_types.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/api/v1alpha1/pipeline_types.go b/api/v1alpha1/pipeline_types.go index 883f525..58c3c26 100644 --- a/api/v1alpha1/pipeline_types.go +++ b/api/v1alpha1/pipeline_types.go @@ -159,7 +159,7 @@ type ServicePort struct { // +kubebuilder:validation:Enum=TCP;UDP Protocol corev1.Protocol `json:"protocol,omitempty"` - // service type + // type is the type of Kubernetes Service to create (ClusterIP or LoadBalancer) // +optional // +kubebuilder:default=ClusterIP // +kubebuilder:validation:Enum=ClusterIP;LoadBalancer From a70301c99368055607d7c52022636467b71f327e Mon Sep 17 00:00:00 2001 From: Nikita Andriyanov Date: Wed, 27 May 2026 10:04:49 +0300 Subject: [PATCH 6/8] fix crds --- config/crd/bases/filter.plainsight.ai_pipelines.yaml | 3 ++- .../crds/filter.plainsight.ai_pipelines.yaml | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/config/crd/bases/filter.plainsight.ai_pipelines.yaml b/config/crd/bases/filter.plainsight.ai_pipelines.yaml index 077fac4..fbadafe 100644 --- a/config/crd/bases/filter.plainsight.ai_pipelines.yaml +++ b/config/crd/bases/filter.plainsight.ai_pipelines.yaml @@ -397,7 +397,8 @@ spec: type: integer type: default: ClusterIP - description: service type + description: type is the type of Kubernetes Service to create + (ClusterIP or LoadBalancer) enum: - ClusterIP - LoadBalancer diff --git a/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml b/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml index 077fac4..fbadafe 100644 --- a/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml +++ b/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml @@ -397,7 +397,8 @@ spec: type: integer type: default: ClusterIP - description: service type + description: type is the type of Kubernetes Service to create + (ClusterIP or LoadBalancer) enum: - ClusterIP - LoadBalancer From fa5af349b7e132c92d63aa7073f9a91a8907ed8e Mon Sep 17 00:00:00 2001 From: Nikita Andriyanov Date: Thu, 28 May 2026 17:47:19 +0300 Subject: [PATCH 7/8] fix docstring --- api/v1alpha1/pipeline_types.go | 4 ++-- config/crd/bases/filter.plainsight.ai_pipelines.yaml | 7 +++++-- .../crds/filter.plainsight.ai_pipelines.yaml | 7 +++++-- 3 files changed, 12 insertions(+), 6 deletions(-) diff --git a/api/v1alpha1/pipeline_types.go b/api/v1alpha1/pipeline_types.go index 58c3c26..b6e1544 100644 --- a/api/v1alpha1/pipeline_types.go +++ b/api/v1alpha1/pipeline_types.go @@ -159,10 +159,10 @@ type ServicePort struct { // +kubebuilder:validation:Enum=TCP;UDP Protocol corev1.Protocol `json:"protocol,omitempty"` - // type is the type of Kubernetes Service to create (ClusterIP or LoadBalancer) + // type defines the Kubernetes Service type to create for this port. Defaults to ClusterIP. Use LoadBalancer to expose the Service outside the cluster (note: cluster operators may restrict external types via ResourceQuota). // +optional // +kubebuilder:default=ClusterIP - // +kubebuilder:validation:Enum=ClusterIP;LoadBalancer + // +kubebuilder:validation:Enum=ClusterIP;LoadBalancer;NodePort Type corev1.ServiceType `json:"type,omitempty"` } diff --git a/config/crd/bases/filter.plainsight.ai_pipelines.yaml b/config/crd/bases/filter.plainsight.ai_pipelines.yaml index fbadafe..d36a611 100644 --- a/config/crd/bases/filter.plainsight.ai_pipelines.yaml +++ b/config/crd/bases/filter.plainsight.ai_pipelines.yaml @@ -397,11 +397,14 @@ spec: type: integer type: default: ClusterIP - description: type is the type of Kubernetes Service to create - (ClusterIP or LoadBalancer) + description: 'type defines the Kubernetes Service type to create + for this port. Defaults to ClusterIP. Use LoadBalancer to + expose the Service outside the cluster (note: cluster operators + may restrict external types via ResourceQuota).' enum: - ClusterIP - LoadBalancer + - NodePort type: string required: - name diff --git a/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml b/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml index fbadafe..d36a611 100644 --- a/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml +++ b/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml @@ -397,11 +397,14 @@ spec: type: integer type: default: ClusterIP - description: type is the type of Kubernetes Service to create - (ClusterIP or LoadBalancer) + description: 'type defines the Kubernetes Service type to create + for this port. Defaults to ClusterIP. Use LoadBalancer to + expose the Service outside the cluster (note: cluster operators + may restrict external types via ResourceQuota).' enum: - ClusterIP - LoadBalancer + - NodePort type: string required: - name From 9874243d8341f98b92faaea85b31bd347f399710 Mon Sep 17 00:00:00 2001 From: Nikita Andriyanov Date: Thu, 28 May 2026 17:47:38 +0300 Subject: [PATCH 8/8] add tests --- .../pipelineinstance_controller_streaming.go | 8 +- ...elineinstance_controller_streaming_test.go | 385 ++++++++++++++++++ 2 files changed, 392 insertions(+), 1 deletion(-) diff --git a/internal/controller/pipelineinstance_controller_streaming.go b/internal/controller/pipelineinstance_controller_streaming.go index 809b166..90680cb 100644 --- a/internal/controller/pipelineinstance_controller_streaming.go +++ b/internal/controller/pipelineinstance_controller_streaming.go @@ -615,6 +615,12 @@ func (r *PipelineInstanceReconciler) ensureFilterServices(ctx context.Context, p protocol = svcPort.Protocol } + // Apply ClusterIP as the default service type if not specified + serviceType := svcPort.Type + if serviceType == "" { + serviceType = corev1.ServiceTypeClusterIP + } + desiredService := &corev1.Service{ ObjectMeta: metav1.ObjectMeta{ Name: serviceName, @@ -638,7 +644,7 @@ func (r *PipelineInstanceReconciler) ensureFilterServices(ctx context.Context, p Protocol: protocol, }, }, - Type: svcPort.Type, + Type: serviceType, }, } diff --git a/internal/controller/pipelineinstance_controller_streaming_test.go b/internal/controller/pipelineinstance_controller_streaming_test.go index 856c549..cf8c36a 100644 --- a/internal/controller/pipelineinstance_controller_streaming_test.go +++ b/internal/controller/pipelineinstance_controller_streaming_test.go @@ -8,6 +8,8 @@ import ( "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes/scheme" + "sigs.k8s.io/controller-runtime/pkg/client/fake" pipelinesv1alpha1 "github.com/PlainsightAI/openfilter-pipelines-controller/api/v1alpha1" ) @@ -923,3 +925,386 @@ func TestBuildRTSPURLWithCredentials(t *testing.T) { }) } } + +// Test helper to create a reconciler with a fake k8s client +func makeReconcilerWithFakeClient() (*PipelineInstanceReconciler, error) { + // Add the api scheme to the fake client + if err := pipelinesv1alpha1.AddToScheme(scheme.Scheme); err != nil { + return nil, err + } + + fakeClient := fake.NewClientBuilder(). + WithScheme(scheme.Scheme). + Build() + + return &PipelineInstanceReconciler{ + Client: fakeClient, + Scheme: scheme.Scheme, + }, nil + +} + +// TestEnsureFilterServices_TypeUnset verifies that Service type defaults to ClusterIP when unset +// This is a regression guard for the default behavior +func TestEnsureFilterServices_TypeUnset(t *testing.T) { + ctx := context.Background() + r, err := makeReconcilerWithFakeClient() + if err != nil { + t.Fatalf("failed to create reconciler with fake client: %v", err) + } + + pi := makeMinimalStreamingPipelineInstance() + pipeline := &pipelinesv1alpha1.Pipeline{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-pipeline", + Namespace: "default", + }, + Spec: pipelinesv1alpha1.PipelineSpec{ + Services: []pipelinesv1alpha1.ServicePort{ + { + Name: "filter1", + Port: 8080, + // Type is not set - should default to ClusterIP + }, + }, + }, + } + + // Create the PipelineInstance so it can be used as controller reference + if err := r.Create(ctx, pi); err != nil { + t.Fatalf("failed to create PipelineInstance: %v", err) + } + + // Call ensureFilterServices + if err := r.ensureFilterServices(ctx, pi, pipeline); err != nil { + t.Fatalf("ensureFilterServices() failed: %v", err) + } + + // Verify the Service was created with ClusterIP type + service := &corev1.Service{} + serviceKey := types.NamespacedName{Name: "stream-instance-filter1-0", Namespace: "default"} + if err := r.Get(ctx, serviceKey, service); err != nil { + t.Fatalf("failed to get created Service: %v", err) + } + + if service.Spec.Type != corev1.ServiceTypeClusterIP { + t.Errorf("expected Service type ClusterIP (default), got %v", service.Spec.Type) + } +} + +// TestEnsureFilterServices_TypeClusterIPExplicit verifies that Service type is ClusterIP when explicitly set +func TestEnsureFilterServices_TypeClusterIPExplicit(t *testing.T) { + ctx := context.Background() + r, err := makeReconcilerWithFakeClient() + if err != nil { + t.Fatalf("failed to create reconciler with fake client: %v", err) + } + + pi := makeMinimalStreamingPipelineInstance() + clusterIPType := corev1.ServiceTypeClusterIP + pipeline := &pipelinesv1alpha1.Pipeline{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-pipeline", + Namespace: "default", + }, + Spec: pipelinesv1alpha1.PipelineSpec{ + Services: []pipelinesv1alpha1.ServicePort{ + { + Name: "filter1", + Port: 8080, + Type: clusterIPType, + }, + }, + }, + } + + // Create the PipelineInstance so it can be used as controller reference + if err := r.Create(ctx, pi); err != nil { + t.Fatalf("failed to create PipelineInstance: %v", err) + } + + // Call ensureFilterServices + if err := r.ensureFilterServices(ctx, pi, pipeline); err != nil { + t.Fatalf("ensureFilterServices() failed: %v", err) + } + + // Verify the Service was created with ClusterIP type + service := &corev1.Service{} + serviceKey := types.NamespacedName{Name: "stream-instance-filter1-0", Namespace: "default"} + if err := r.Get(ctx, serviceKey, service); err != nil { + t.Fatalf("failed to get created Service: %v", err) + } + + if service.Spec.Type != corev1.ServiceTypeClusterIP { + t.Errorf("expected Service type ClusterIP, got %v", service.Spec.Type) + } +} + +// TestEnsureFilterServices_TypeLoadBalancer verifies that Service type is LoadBalancer when explicitly set +func TestEnsureFilterServices_TypeLoadBalancer(t *testing.T) { + ctx := context.Background() + r, err := makeReconcilerWithFakeClient() + if err != nil { + t.Fatalf("failed to create reconciler with fake client: %v", err) + } + + pi := makeMinimalStreamingPipelineInstance() + loadBalancerType := corev1.ServiceTypeLoadBalancer + pipeline := &pipelinesv1alpha1.Pipeline{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-pipeline", + Namespace: "default", + }, + Spec: pipelinesv1alpha1.PipelineSpec{ + Services: []pipelinesv1alpha1.ServicePort{ + { + Name: "filter1", + Port: 8080, + Type: loadBalancerType, + }, + }, + }, + } + + // Create the PipelineInstance so it can be used as controller reference + if err := r.Create(ctx, pi); err != nil { + t.Fatalf("failed to create PipelineInstance: %v", err) + } + + // Call ensureFilterServices + if err := r.ensureFilterServices(ctx, pi, pipeline); err != nil { + t.Fatalf("ensureFilterServices() failed: %v", err) + } + + // Verify the Service was created with LoadBalancer type + service := &corev1.Service{} + serviceKey := types.NamespacedName{Name: "stream-instance-filter1-0", Namespace: "default"} + if err := r.Get(ctx, serviceKey, service); err != nil { + t.Fatalf("failed to get created Service: %v", err) + } + + if service.Spec.Type != corev1.ServiceTypeLoadBalancer { + t.Errorf("expected Service type LoadBalancer, got %v", service.Spec.Type) + } +} + +// TestEnsureFilterServices_TypeNodePort verifies that Service type is NodePort when explicitly set +func TestEnsureFilterServices_TypeNodePort(t *testing.T) { + ctx := context.Background() + r, err := makeReconcilerWithFakeClient() + if err != nil { + t.Fatalf("failed to create reconciler with fake client: %v", err) + } + + pi := makeMinimalStreamingPipelineInstance() + nodePortType := corev1.ServiceTypeNodePort + pipeline := &pipelinesv1alpha1.Pipeline{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-pipeline", + Namespace: "default", + }, + Spec: pipelinesv1alpha1.PipelineSpec{ + Services: []pipelinesv1alpha1.ServicePort{ + { + Name: "filter1", + Port: 8080, + Type: nodePortType, + }, + }, + }, + } + + // Create the PipelineInstance so it can be used as controller reference + if err := r.Create(ctx, pi); err != nil { + t.Fatalf("failed to create PipelineInstance: %v", err) + } + + // Call ensureFilterServices + if err := r.ensureFilterServices(ctx, pi, pipeline); err != nil { + t.Fatalf("ensureFilterServices() failed: %v", err) + } + + // Verify the Service was created with NodePort type + service := &corev1.Service{} + serviceKey := types.NamespacedName{Name: "stream-instance-filter1-0", Namespace: "default"} + if err := r.Get(ctx, serviceKey, service); err != nil { + t.Fatalf("failed to get created Service: %v", err) + } + + if service.Spec.Type != corev1.ServiceTypeNodePort { + t.Errorf("expected Service type NodePort, got %v", service.Spec.Type) + } +} + +// TestEnsureFilterServices_NoServices verifies that the function handles empty service list gracefully +func TestEnsureFilterServices_NoServices(t *testing.T) { + ctx := context.Background() + r, err := makeReconcilerWithFakeClient() + if err != nil { + t.Fatalf("failed to create reconciler with fake client: %v", err) + } + + pi := makeMinimalStreamingPipelineInstance() + pipeline := &pipelinesv1alpha1.Pipeline{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-pipeline", + Namespace: "default", + }, + Spec: pipelinesv1alpha1.PipelineSpec{ + Services: []pipelinesv1alpha1.ServicePort{}, + }, + } + + // Create the PipelineInstance + if err := r.Create(ctx, pi); err != nil { + t.Fatalf("failed to create PipelineInstance: %v", err) + } + + // Call ensureFilterServices with empty services - should return nil without creating anything + if err := r.ensureFilterServices(ctx, pi, pipeline); err != nil { + t.Fatalf("ensureFilterServices() failed: %v", err) + } + + // Verify no services were created + serviceList := &corev1.ServiceList{} + if err := r.List(ctx, serviceList); err != nil { + t.Fatalf("failed to list services: %v", err) + } + + if len(serviceList.Items) != 0 { + t.Errorf("expected no services to be created, got %d", len(serviceList.Items)) + } +} + +// TestEnsureFilterServices_UpdateExistingService verifies that existing services are updated with new type +func TestEnsureFilterServices_UpdateExistingService(t *testing.T) { + ctx := context.Background() + r, err := makeReconcilerWithFakeClient() + if err != nil { + t.Fatalf("failed to create reconciler with fake client: %v", err) + } + + pi := makeMinimalStreamingPipelineInstance() + loadBalancerType := corev1.ServiceTypeLoadBalancer + pipeline := &pipelinesv1alpha1.Pipeline{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-pipeline", + Namespace: "default", + }, + Spec: pipelinesv1alpha1.PipelineSpec{ + Services: []pipelinesv1alpha1.ServicePort{ + { + Name: "filter1", + Port: 8080, + Type: loadBalancerType, + }, + }, + }, + } + + // Create the PipelineInstance + if err := r.Create(ctx, pi); err != nil { + t.Fatalf("failed to create PipelineInstance: %v", err) + } + + // Create initial service with ClusterIP type + existingService := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: "stream-instance-filter1-0", + Namespace: "default", + }, + Spec: corev1.ServiceSpec{ + Type: corev1.ServiceTypeClusterIP, + Selector: map[string]string{ + "app": "pipeline-stream", + "pipelineinstance": "stream-instance", + }, + Ports: []corev1.ServicePort{ + { + Name: "filter1", + Port: 8080, + }, + }, + }, + } + if err := r.Create(ctx, existingService); err != nil { + t.Fatalf("failed to create initial Service: %v", err) + } + + // Call ensureFilterServices - should update the service to LoadBalancer + if err := r.ensureFilterServices(ctx, pi, pipeline); err != nil { + t.Fatalf("ensureFilterServices() failed: %v", err) + } + + // Verify the Service was updated to LoadBalancer type + service := &corev1.Service{} + serviceKey := types.NamespacedName{Name: "stream-instance-filter1-0", Namespace: "default"} + if err := r.Get(ctx, serviceKey, service); err != nil { + t.Fatalf("failed to get updated Service: %v", err) + } + + if service.Spec.Type != corev1.ServiceTypeLoadBalancer { + t.Errorf("expected Service type to be updated to LoadBalancer, got %v", service.Spec.Type) + } +} + +// TestEnsureFilterServices_MultipleServicesPerFilter verifies multiple services can be created for a single filter +func TestEnsureFilterServices_MultipleServicesPerFilter(t *testing.T) { + ctx := context.Background() + r, err := makeReconcilerWithFakeClient() + if err != nil { + t.Fatalf("failed to create reconciler with fake client: %v", err) + } + + pi := makeMinimalStreamingPipelineInstance() + pipeline := &pipelinesv1alpha1.Pipeline{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-pipeline", + Namespace: "default", + }, + Spec: pipelinesv1alpha1.PipelineSpec{ + Services: []pipelinesv1alpha1.ServicePort{ + { + Name: "filter1", + Port: 8080, + Type: corev1.ServiceTypeClusterIP, + }, + { + Name: "filter1", + Port: 9090, + Type: corev1.ServiceTypeLoadBalancer, + }, + }, + }, + } + + // Create the PipelineInstance + if err := r.Create(ctx, pi); err != nil { + t.Fatalf("failed to create PipelineInstance: %v", err) + } + + // Call ensureFilterServices + if err := r.ensureFilterServices(ctx, pi, pipeline); err != nil { + t.Fatalf("ensureFilterServices() failed: %v", err) + } + + // Verify first service (index 0) has ClusterIP type + service1 := &corev1.Service{} + serviceKey1 := types.NamespacedName{Name: "stream-instance-filter1-0", Namespace: "default"} + if err := r.Get(ctx, serviceKey1, service1); err != nil { + t.Fatalf("failed to get first Service: %v", err) + } + if service1.Spec.Type != corev1.ServiceTypeClusterIP { + t.Errorf("expected first Service type ClusterIP, got %v", service1.Spec.Type) + } + + // Verify second service (index 1) has LoadBalancer type + service2 := &corev1.Service{} + serviceKey2 := types.NamespacedName{Name: "stream-instance-filter1-1", Namespace: "default"} + if err := r.Get(ctx, serviceKey2, service2); err != nil { + t.Fatalf("failed to get second Service: %v", err) + } + if service2.Spec.Type != corev1.ServiceTypeLoadBalancer { + t.Errorf("expected second Service type LoadBalancer, got %v", service2.Spec.Type) + } +}