From 50c1006b8199c7a594e3d40f68042b614b4ba8d9 Mon Sep 17 00:00:00 2001 From: Nikita Andriyanov Date: Tue, 26 May 2026 17:10:35 +0300 Subject: [PATCH 1/6] expose two ports for filter --- api/v1alpha1/pipeline_types.go | 6 ++++++ config/crd/bases/filter.plainsight.ai_pipelines.yaml | 4 ++++ .../crds/filter.plainsight.ai_pipelines.yaml | 4 ++++ .../controller/pipelineinstance_controller_streaming.go | 9 +++++++++ 4 files changed, 23 insertions(+) diff --git a/api/v1alpha1/pipeline_types.go b/api/v1alpha1/pipeline_types.go index b8ea150..57f73d0 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"` + + // if isFilter expose port +1 + // +optional + // +kubebuilder:default=false + // +kubebuilder:validation:Bool=true;false + IsFilter bool `json:"isFilter,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..7a796ae 100644 --- a/config/crd/bases/filter.plainsight.ai_pipelines.yaml +++ b/config/crd/bases/filter.plainsight.ai_pipelines.yaml @@ -367,6 +367,10 @@ spec: description: ServicePort defines a port to expose as a Kubernetes Service for a filter properties: + isFilter: + default: false + description: if isFilter expose port +1 + type: boolean name: description: |- name is the name of the filter to expose 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..7a796ae 100644 --- a/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml +++ b/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml @@ -367,6 +367,10 @@ spec: description: ServicePort defines a port to expose as a Kubernetes Service for a filter properties: + isFilter: + default: false + description: if isFilter expose port +1 + type: boolean name: description: |- name is the name of the filter to expose diff --git a/internal/controller/pipelineinstance_controller_streaming.go b/internal/controller/pipelineinstance_controller_streaming.go index df74ac0..f77bed2 100644 --- a/internal/controller/pipelineinstance_controller_streaming.go +++ b/internal/controller/pipelineinstance_controller_streaming.go @@ -642,6 +642,15 @@ func (r *PipelineInstanceReconciler) ensureFilterServices(ctx context.Context, p }, } + if svcPort.IsFilter { + desiredService.Spec.Ports = append(desiredService.Spec.Ports, corev1.ServicePort{ + Name: filterName + "1", + Port: svcPort.Port + 1, + TargetPort: intstr.FromInt32(targetPort + 1), + Protocol: protocol, + }) + } + if apierrors.IsNotFound(err) { // Create the Service log.Info("Creating filter service", "service", serviceName, "filter", filterName, "port", svcPort.Port) From e4f7a6bda55dd8af65495fff4a3d74c00ab28e95 Mon Sep 17 00:00:00 2001 From: Nikita Andriyanov Date: Wed, 27 May 2026 10:04:16 +0300 Subject: [PATCH 2/6] fix comment --- api/v1alpha1/pipeline_types.go | 5 ++--- config/crd/bases/filter.plainsight.ai_pipelines.yaml | 5 +++-- .../crds/filter.plainsight.ai_pipelines.yaml | 5 +++-- 3 files changed, 8 insertions(+), 7 deletions(-) diff --git a/api/v1alpha1/pipeline_types.go b/api/v1alpha1/pipeline_types.go index 57f73d0..e0cd443 100644 --- a/api/v1alpha1/pipeline_types.go +++ b/api/v1alpha1/pipeline_types.go @@ -159,9 +159,8 @@ type ServicePort struct { // +kubebuilder:validation:Enum=TCP;UDP Protocol corev1.Protocol `json:"protocol,omitempty"` - // if isFilter expose port +1 - // +optional - // +kubebuilder:default=false + // isFilter indicates that this service port is for a filter + // and additional port should be exposed // +kubebuilder:validation:Bool=true;false IsFilter bool `json:"isFilter,omitempty"` } diff --git a/config/crd/bases/filter.plainsight.ai_pipelines.yaml b/config/crd/bases/filter.plainsight.ai_pipelines.yaml index 7a796ae..477d007 100644 --- a/config/crd/bases/filter.plainsight.ai_pipelines.yaml +++ b/config/crd/bases/filter.plainsight.ai_pipelines.yaml @@ -368,8 +368,9 @@ spec: Service for a filter properties: isFilter: - default: false - description: if isFilter expose port +1 + description: |- + isFilter indicates that this service port is for a filter + and additional port should be exposed type: boolean name: description: |- 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 7a796ae..477d007 100644 --- a/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml +++ b/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml @@ -368,8 +368,9 @@ spec: Service for a filter properties: isFilter: - default: false - description: if isFilter expose port +1 + description: |- + isFilter indicates that this service port is for a filter + and additional port should be exposed type: boolean name: description: |- From 3925e9030d3b36d0a3da92ea819096b8f5c7fd79 Mon Sep 17 00:00:00 2001 From: Nikita Andriyanov Date: Fri, 29 May 2026 10:45:29 +0300 Subject: [PATCH 3/6] test XValidation and AppProtocol --- api/v1alpha1/pipeline_types.go | 25 +++- .../bases/filter.plainsight.ai_pipelines.yaml | 24 +++- .../crds/filter.plainsight.ai_pipelines.yaml | 24 +++- .../pipelineinstance_controller_streaming.go | 4 +- ...elineinstance_controller_streaming_test.go | 130 ++++++++++++++++++ 5 files changed, 193 insertions(+), 14 deletions(-) diff --git a/api/v1alpha1/pipeline_types.go b/api/v1alpha1/pipeline_types.go index e0cd443..531ab83 100644 --- a/api/v1alpha1/pipeline_types.go +++ b/api/v1alpha1/pipeline_types.go @@ -132,6 +132,17 @@ const ( PipelineModeStream PipelineMode = "stream" ) +// AppProtocol declares the application protocol carried over this port +// +kubebuilder:validation:Enum=tcp;openfilter +type AppProtocol string + +const ( + // AppProtocol use the single port, no special handling + AppProtocolTCP AppProtocol = "tcp" + // AppProtocol use OpenFilter ZMQ stream — also exposes `port + 1` + AppProtocolOpenFilter AppProtocol = "openfilter" +) + // ServicePort defines a port to expose as a Kubernetes Service for a filter type ServicePort struct { // name is the name of the filter to expose @@ -144,6 +155,7 @@ type ServicePort struct { // +kubebuilder:validation:Required // +kubebuilder:validation:Minimum=1 // +kubebuilder:validation:Maximum=65535 + // +kubebuilder:validation:XValidation:rule="!(has(self.appProtocol) && self.appProtocol == 'openfilter') || (self.port <= 65534)", message="When appProtocol='openfilter', port must be <= 65534 to leave room for the reply channel on port+1" Port int32 `json:"port"` // targetPort is the port on the container to forward to @@ -159,10 +171,15 @@ type ServicePort struct { // +kubebuilder:validation:Enum=TCP;UDP Protocol corev1.Protocol `json:"protocol,omitempty"` - // isFilter indicates that this service port is for a filter - // and additional port should be exposed - // +kubebuilder:validation:Bool=true;false - IsFilter bool `json:"isFilter,omitempty"` + // appProtocol declares the application protocol carried over this port. + // The controller uses this to apply protocol-specific exposure logic. + // - "tcp" (default): single port, no special handling + // - "openfilter" : OpenFilter ZMQ stream — also exposes `port + 1` + // (PUSH/PULL reply channel paired with the PUB/SUB data port) + // +optional + // +kubebuilder:default=tcp + // +kubebuilder:validation:Enum=tcp;openfilter + AppProtocol AppProtocol `json:"appProtocol,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 477d007..c59b84a 100644 --- a/config/crd/bases/filter.plainsight.ai_pipelines.yaml +++ b/config/crd/bases/filter.plainsight.ai_pipelines.yaml @@ -367,11 +367,22 @@ spec: description: ServicePort defines a port to expose as a Kubernetes Service for a filter properties: - isFilter: + appProtocol: + allOf: + - enum: + - tcp + - openfilter + - enum: + - tcp + - openfilter + default: tcp description: |- - isFilter indicates that this service port is for a filter - and additional port should be exposed - type: boolean + appProtocol declares the application protocol carried over this port. + The controller uses this to apply protocol-specific exposure logic. + - "tcp" (default): single port, no special handling + - "openfilter" : OpenFilter ZMQ stream — also exposes `port + 1` + (PUSH/PULL reply channel paired with the PUB/SUB data port) + type: string name: description: |- name is the name of the filter to expose @@ -384,6 +395,11 @@ spec: maximum: 65535 minimum: 1 type: integer + x-kubernetes-validations: + - message: When appProtocol='openfilter', port must be <= 65534 + to leave room for the reply channel on port+1 + rule: '!(has(self.appProtocol) && self.appProtocol == ''openfilter'') + || (self.port <= 65534)' protocol: default: TCP description: protocol is the network protocol for this 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 477d007..c59b84a 100644 --- a/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml +++ b/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml @@ -367,11 +367,22 @@ spec: description: ServicePort defines a port to expose as a Kubernetes Service for a filter properties: - isFilter: + appProtocol: + allOf: + - enum: + - tcp + - openfilter + - enum: + - tcp + - openfilter + default: tcp description: |- - isFilter indicates that this service port is for a filter - and additional port should be exposed - type: boolean + appProtocol declares the application protocol carried over this port. + The controller uses this to apply protocol-specific exposure logic. + - "tcp" (default): single port, no special handling + - "openfilter" : OpenFilter ZMQ stream — also exposes `port + 1` + (PUSH/PULL reply channel paired with the PUB/SUB data port) + type: string name: description: |- name is the name of the filter to expose @@ -384,6 +395,11 @@ spec: maximum: 65535 minimum: 1 type: integer + x-kubernetes-validations: + - message: When appProtocol='openfilter', port must be <= 65534 + to leave room for the reply channel on port+1 + rule: '!(has(self.appProtocol) && self.appProtocol == ''openfilter'') + || (self.port <= 65534)' protocol: default: TCP description: protocol is the network protocol for this port diff --git a/internal/controller/pipelineinstance_controller_streaming.go b/internal/controller/pipelineinstance_controller_streaming.go index f77bed2..aabc40b 100644 --- a/internal/controller/pipelineinstance_controller_streaming.go +++ b/internal/controller/pipelineinstance_controller_streaming.go @@ -642,9 +642,9 @@ func (r *PipelineInstanceReconciler) ensureFilterServices(ctx context.Context, p }, } - if svcPort.IsFilter { + if svcPort.AppProtocol == pipelinesv1alpha1.AppProtocolOpenFilter { desiredService.Spec.Ports = append(desiredService.Spec.Ports, corev1.ServicePort{ - Name: filterName + "1", + Name: filterName + "-reply", Port: svcPort.Port + 1, TargetPort: intstr.FromInt32(targetPort + 1), Protocol: protocol, diff --git a/internal/controller/pipelineinstance_controller_streaming_test.go b/internal/controller/pipelineinstance_controller_streaming_test.go index 856c549..bc5b0f3 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,131 @@ func TestBuildRTSPURLWithCredentials(t *testing.T) { }) } } + +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 + +} + +func TestEnsureFilterServices_OpenFilterAddsReplyPortWithExplicitTargetPort(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() + targetPort := int32(9100) + pipeline := &pipelinesv1alpha1.Pipeline{ + Spec: pipelinesv1alpha1.PipelineSpec{ + Services: []pipelinesv1alpha1.ServicePort{ + { + Name: "filter", + Port: 9000, + TargetPort: &targetPort, + AppProtocol: pipelinesv1alpha1.AppProtocolOpenFilter, + }, + }, + }, + } + + if err := r.ensureFilterServices(ctx, pi, pipeline); err != nil { + t.Fatalf("ensureFilterServices returned error: %v", err) + } + + service := &corev1.Service{} + if err := r.Get(ctx, types.NamespacedName{Name: "stream-instance-filter-0", Namespace: "default"}, service); err != nil { + t.Fatalf("expected Service created: %v", err) + } + + if got, want := len(service.Spec.Ports), 2; got != want { + t.Fatalf("expected 2 service ports, got %d", got) + } + + first := service.Spec.Ports[0] + if first.Name != "filter" { + t.Errorf("first port name = %q, want %q", first.Name, "filter") + } + if first.Port != 9000 { + t.Errorf("first port = %d, want %d", first.Port, 9000) + } + if first.TargetPort.IntVal != 9100 { + t.Errorf("first targetPort = %d, want %d", first.TargetPort.IntVal, 9100) + } + if first.Protocol != corev1.ProtocolTCP { + t.Errorf("first protocol = %q, want %q", first.Protocol, corev1.ProtocolTCP) + } + + second := service.Spec.Ports[1] + if second.Name != "filter-reply" { + t.Errorf("second port name = %q, want %q", second.Name, "filter-reply") + } + if second.Port != 9001 { + t.Errorf("second port = %d, want %d", second.Port, 9001) + } + if second.TargetPort.IntVal != 9101 { + t.Errorf("second targetPort = %d, want %d", second.TargetPort.IntVal, 9101) + } + if second.Protocol != corev1.ProtocolTCP { + t.Errorf("second protocol = %q, want %q", second.Protocol, corev1.ProtocolTCP) + } +} + +func TestEnsureFilterServices_DefaultProtocolDoesNotAddReplyPort(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() + targetPort := int32(9100) + pipeline := &pipelinesv1alpha1.Pipeline{ + Spec: pipelinesv1alpha1.PipelineSpec{ + Services: []pipelinesv1alpha1.ServicePort{ + { + Name: "filter", + Port: 9000, + TargetPort: &targetPort, + }, + }, + }, + } + + if err := r.ensureFilterServices(ctx, pi, pipeline); err != nil { + t.Fatalf("ensureFilterServices returned error: %v", err) + } + + service := &corev1.Service{} + if err := r.Get(ctx, types.NamespacedName{Name: "stream-instance-filter-0", Namespace: "default"}, service); err != nil { + t.Fatalf("expected Service created: %v", err) + } + + if got, want := len(service.Spec.Ports), 1; got != want { + t.Fatalf("expected 1 service port, got %d", got) + } + + only := service.Spec.Ports[0] + if only.Name != "filter" { + t.Errorf("port name = %q, want %q", only.Name, "filter") + } + if only.Port != 9000 { + t.Errorf("port = %d, want %d", only.Port, 9000) + } + if only.TargetPort.IntVal != 9100 { + t.Errorf("targetPort = %d, want %d", only.TargetPort.IntVal, 9100) + } + if only.Protocol != corev1.ProtocolTCP { + t.Errorf("protocol = %q, want %q", only.Protocol, corev1.ProtocolTCP) + } +} From bbf366f0953d988a85ad2476c4e93c8f1dba87c1 Mon Sep 17 00:00:00 2001 From: Nikita Andriyanov Date: Fri, 29 May 2026 18:08:25 +0300 Subject: [PATCH 4/6] style fixes --- api/v1alpha1/pipeline_types.go | 6 +++--- config/crd/bases/filter.plainsight.ai_pipelines.yaml | 10 +++------- .../crds/filter.plainsight.ai_pipelines.yaml | 10 +++------- .../pipelineinstance_controller_streaming_test.go | 4 ++-- 4 files changed, 11 insertions(+), 19 deletions(-) diff --git a/api/v1alpha1/pipeline_types.go b/api/v1alpha1/pipeline_types.go index 531ab83..ca768a7 100644 --- a/api/v1alpha1/pipeline_types.go +++ b/api/v1alpha1/pipeline_types.go @@ -137,9 +137,10 @@ const ( type AppProtocol string const ( - // AppProtocol use the single port, no special handling + // AppProtocolTCP uses a single port with no special handling. AppProtocolTCP AppProtocol = "tcp" - // AppProtocol use OpenFilter ZMQ stream — also exposes `port + 1` + // AppProtocolOpenFilter uses the OpenFilter ZMQ stream — also exposes port+1 + // (PUSH/PULL reply channel paired with the PUB/SUB data port). AppProtocolOpenFilter AppProtocol = "openfilter" ) @@ -178,7 +179,6 @@ type ServicePort struct { // (PUSH/PULL reply channel paired with the PUB/SUB data port) // +optional // +kubebuilder:default=tcp - // +kubebuilder:validation:Enum=tcp;openfilter AppProtocol AppProtocol `json:"appProtocol,omitempty"` } diff --git a/config/crd/bases/filter.plainsight.ai_pipelines.yaml b/config/crd/bases/filter.plainsight.ai_pipelines.yaml index c59b84a..78f6cfe 100644 --- a/config/crd/bases/filter.plainsight.ai_pipelines.yaml +++ b/config/crd/bases/filter.plainsight.ai_pipelines.yaml @@ -368,13 +368,6 @@ spec: Service for a filter properties: appProtocol: - allOf: - - enum: - - tcp - - openfilter - - enum: - - tcp - - openfilter default: tcp description: |- appProtocol declares the application protocol carried over this port. @@ -382,6 +375,9 @@ spec: - "tcp" (default): single port, no special handling - "openfilter" : OpenFilter ZMQ stream — also exposes `port + 1` (PUSH/PULL reply channel paired with the PUB/SUB data port) + enum: + - tcp + - openfilter type: string name: description: |- 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 c59b84a..78f6cfe 100644 --- a/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml +++ b/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml @@ -368,13 +368,6 @@ spec: Service for a filter properties: appProtocol: - allOf: - - enum: - - tcp - - openfilter - - enum: - - tcp - - openfilter default: tcp description: |- appProtocol declares the application protocol carried over this port. @@ -382,6 +375,9 @@ spec: - "tcp" (default): single port, no special handling - "openfilter" : OpenFilter ZMQ stream — also exposes `port + 1` (PUSH/PULL reply channel paired with the PUB/SUB data port) + enum: + - tcp + - openfilter type: string name: description: |- diff --git a/internal/controller/pipelineinstance_controller_streaming_test.go b/internal/controller/pipelineinstance_controller_streaming_test.go index bc5b0f3..8afbbbf 100644 --- a/internal/controller/pipelineinstance_controller_streaming_test.go +++ b/internal/controller/pipelineinstance_controller_streaming_test.go @@ -944,7 +944,7 @@ func makeReconcilerWithFakeClient() (*PipelineInstanceReconciler, error) { } func TestEnsureFilterServices_OpenFilterAddsReplyPortWithExplicitTargetPort(t *testing.T) { - ctx = context.Background() + ctx := context.Background() r, err := makeReconcilerWithFakeClient() if err != nil { t.Fatalf("failed to create reconciler with fake client: %v", err) @@ -1007,7 +1007,7 @@ func TestEnsureFilterServices_OpenFilterAddsReplyPortWithExplicitTargetPort(t *t } func TestEnsureFilterServices_DefaultProtocolDoesNotAddReplyPort(t *testing.T) { - ctx = context.Background() + ctx := context.Background() r, err := makeReconcilerWithFakeClient() if err != nil { t.Fatalf("failed to create reconciler with fake client: %v", err) From bb2dd6ab5ec08a74d2cdcb44a23e6be4ac80e7cb Mon Sep 17 00:00:00 2001 From: Nikita Andriyanov Date: Mon, 8 Jun 2026 18:25:36 +0300 Subject: [PATCH 5/6] some fixes --- api/v1alpha1/pipeline_types.go | 2 +- .../bases/filter.plainsight.ai_pipelines.yaml | 9 ++- .../crds/filter.plainsight.ai_pipelines.yaml | 9 ++- ...elineinstance_controller_streaming_test.go | 66 +++++++++++++++++++ 4 files changed, 75 insertions(+), 11 deletions(-) diff --git a/api/v1alpha1/pipeline_types.go b/api/v1alpha1/pipeline_types.go index ca768a7..6afb283 100644 --- a/api/v1alpha1/pipeline_types.go +++ b/api/v1alpha1/pipeline_types.go @@ -145,6 +145,7 @@ const ( ) // ServicePort defines a port to expose as a Kubernetes Service for a filter +// +kubebuilder:validation:XValidation:rule="self.appProtocol != 'openfilter' || self.port <= 65534", message="When appProtocol='openfilter', port must be <= 65534 to leave room for the reply channel on port+1" type ServicePort struct { // name is the name of the filter to expose // Must match one of the filter names in the pipeline @@ -156,7 +157,6 @@ type ServicePort struct { // +kubebuilder:validation:Required // +kubebuilder:validation:Minimum=1 // +kubebuilder:validation:Maximum=65535 - // +kubebuilder:validation:XValidation:rule="!(has(self.appProtocol) && self.appProtocol == 'openfilter') || (self.port <= 65534)", message="When appProtocol='openfilter', port must be <= 65534 to leave room for the reply channel on port+1" Port int32 `json:"port"` // targetPort is the port on the container to forward to diff --git a/config/crd/bases/filter.plainsight.ai_pipelines.yaml b/config/crd/bases/filter.plainsight.ai_pipelines.yaml index 78f6cfe..69109c1 100644 --- a/config/crd/bases/filter.plainsight.ai_pipelines.yaml +++ b/config/crd/bases/filter.plainsight.ai_pipelines.yaml @@ -391,11 +391,6 @@ spec: maximum: 65535 minimum: 1 type: integer - x-kubernetes-validations: - - message: When appProtocol='openfilter', port must be <= 65534 - to leave room for the reply channel on port+1 - rule: '!(has(self.appProtocol) && self.appProtocol == ''openfilter'') - || (self.port <= 65534)' protocol: default: TCP description: protocol is the network protocol for this port @@ -416,6 +411,10 @@ spec: - name - port type: object + x-kubernetes-validations: + - message: When appProtocol='openfilter', port must be <= 65534 + to leave room for the reply channel on port+1 + rule: self.appProtocol != 'openfilter' || self.port <= 65534 type: array videoInputPath: default: /ws/input.mp4 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 78f6cfe..69109c1 100644 --- a/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml +++ b/deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml @@ -391,11 +391,6 @@ spec: maximum: 65535 minimum: 1 type: integer - x-kubernetes-validations: - - message: When appProtocol='openfilter', port must be <= 65534 - to leave room for the reply channel on port+1 - rule: '!(has(self.appProtocol) && self.appProtocol == ''openfilter'') - || (self.port <= 65534)' protocol: default: TCP description: protocol is the network protocol for this port @@ -416,6 +411,10 @@ spec: - name - port type: object + x-kubernetes-validations: + - message: When appProtocol='openfilter', port must be <= 65534 + to leave room for the reply channel on port+1 + rule: self.appProtocol != 'openfilter' || self.port <= 65534 type: array videoInputPath: default: /ws/input.mp4 diff --git a/internal/controller/pipelineinstance_controller_streaming_test.go b/internal/controller/pipelineinstance_controller_streaming_test.go index 8afbbbf..580025b 100644 --- a/internal/controller/pipelineinstance_controller_streaming_test.go +++ b/internal/controller/pipelineinstance_controller_streaming_test.go @@ -1053,3 +1053,69 @@ func TestEnsureFilterServices_DefaultProtocolDoesNotAddReplyPort(t *testing.T) { t.Errorf("protocol = %q, want %q", only.Protocol, corev1.ProtocolTCP) } } + +func TestEnsureFilterServices_OpenFilterWithoutExplicitTargetPort(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{ + Spec: pipelinesv1alpha1.PipelineSpec{ + Services: []pipelinesv1alpha1.ServicePort{ + { + Name: "filter", + Port: 9000, + TargetPort: nil, // No explicit TargetPort, should fall back to Port + AppProtocol: pipelinesv1alpha1.AppProtocolOpenFilter, + }, + }, + }, + } + + if err := r.ensureFilterServices(ctx, pi, pipeline); err != nil { + t.Fatalf("ensureFilterServices returned error: %v", err) + } + + service := &corev1.Service{} + if err := r.Get(ctx, types.NamespacedName{Name: "stream-instance-filter-0", Namespace: "default"}, service); err != nil { + t.Fatalf("expected Service created: %v", err) + } + + if got, want := len(service.Spec.Ports), 2; got != want { + t.Fatalf("expected 2 service ports (main + reply), got %d", got) + } + + // First port: main data port + first := service.Spec.Ports[0] + if first.Name != "filter" { + t.Errorf("first port name = %q, want %q", first.Name, "filter") + } + if first.Port != 9000 { + t.Errorf("first port = %d, want %d", first.Port, 9000) + } + // When TargetPort is nil, should default to Port value + if first.TargetPort.IntVal != 9000 { + t.Errorf("first targetPort = %d, want %d (should default to Port when nil)", first.TargetPort.IntVal, 9000) + } + if first.Protocol != corev1.ProtocolTCP { + t.Errorf("first protocol = %q, want %q", first.Protocol, corev1.ProtocolTCP) + } + + // Second port: reply channel (port + 1) + second := service.Spec.Ports[1] + if second.Name != "filter-reply" { + t.Errorf("second port name = %q, want %q", second.Name, "filter-reply") + } + if second.Port != 9001 { + t.Errorf("second port = %d, want %d", second.Port, 9001) + } + // Reply channel should also default to Port+1 + if second.TargetPort.IntVal != 9001 { + t.Errorf("second targetPort = %d, want %d (should default to Port+1 when nil)", second.TargetPort.IntVal, 9001) + } + if second.Protocol != corev1.ProtocolTCP { + t.Errorf("second protocol = %q, want %q", second.Protocol, corev1.ProtocolTCP) + } +} From 6d733b27bbb3a26f249ece016f1d37e145c924ea Mon Sep 17 00:00:00 2001 From: Andriyanov Nikita Date: Wed, 1 Jul 2026 20:32:42 +0300 Subject: [PATCH 6/6] Update pipelineinstance_controller_streaming_test.go --- .../controller/pipelineinstance_controller_streaming_test.go | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/internal/controller/pipelineinstance_controller_streaming_test.go b/internal/controller/pipelineinstance_controller_streaming_test.go index 580025b..8f00a9e 100644 --- a/internal/controller/pipelineinstance_controller_streaming_test.go +++ b/internal/controller/pipelineinstance_controller_streaming_test.go @@ -978,8 +978,9 @@ func TestEnsureFilterServices_OpenFilterAddsReplyPortWithExplicitTargetPort(t *t } first := service.Spec.Ports[0] - if first.Name != "filter" { - t.Errorf("first port name = %q, want %q", first.Name, "filter") + const filterName = "filter" + if first.Name != filterName { + t.Errorf("first port name = %q, want %q", first.Name, filterName) } if first.Port != 9000 { t.Errorf("first port = %d, want %d", first.Port, 9000)