feat(filter): expose openfilter reply port via appProtocol on ServicePort

api/v1alpha1/pipeline_types.go

@@ -132,7 +132,20 @@ const (
 	PipelineModeStream PipelineMode = "stream"
 )
 
+// AppProtocol declares the application protocol carried over this port
+// +kubebuilder:validation:Enum=tcp;openfilter
+type AppProtocol string
+
+const (
+	// AppProtocolTCP uses a single port with no special handling.
+	AppProtocolTCP AppProtocol = "tcp"
+	// AppProtocolOpenFilter uses the OpenFilter ZMQ stream — also exposes port+1
+	// (PUSH/PULL reply channel paired with the PUB/SUB data port).
+	AppProtocolOpenFilter AppProtocol = "openfilter"
+)
+
 // 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
@@ -158,6 +171,15 @@ type ServicePort struct {
 	// +kubebuilder:default=TCP
 	// +kubebuilder:validation:Enum=TCP;UDP
 	Protocol corev1.Protocol `json:"protocol,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
+	AppProtocol AppProtocol `json:"appProtocol,omitempty"`
 }
 
 // Filter defines a containerized processing step in the pipeline

config/crd/bases/filter.plainsight.ai_pipelines.yaml

@@ -367,6 +367,18 @@ spec:
                   description: ServicePort defines a port to expose as a Kubernetes
                     Service for a filter
                   properties:
+                    appProtocol:
+                      default: tcp
+                      description: |-
+                        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)
+                      enum:
+                      - tcp
+                      - openfilter
+                      type: string
                     name:
                       description: |-
                         name is the name of the filter to expose
@@ -399,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

deployment/openfilter-pipelines-controller/crds/filter.plainsight.ai_pipelines.yaml

@@ -367,6 +367,18 @@ spec:
                   description: ServicePort defines a port to expose as a Kubernetes
                     Service for a filter
                   properties:
+                    appProtocol:
+                      default: tcp
+                      description: |-
+                        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)
+                      enum:
+                      - tcp
+                      - openfilter
+                      type: string
                     name:
                       description: |-
                         name is the name of the filter to expose
@@ -399,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

internal/controller/pipelineinstance_controller_streaming.go

@@ -642,6 +642,15 @@ func (r *PipelineInstanceReconciler) ensureFilterServices(ctx context.Context, p
 				},
 			}
 
+			if svcPort.AppProtocol == pipelinesv1alpha1.AppProtocolOpenFilter {
+				desiredService.Spec.Ports = append(desiredService.Spec.Ports, corev1.ServicePort{
+					Name:       filterName + "-reply",
+					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)

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,198 @@ 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]
+	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)
+	}
+	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)
+	}
+}
+
+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)
+	}
+}