diff --git a/api/v1alpha1/pipeline_types.go b/api/v1alpha1/pipeline_types.go index b8ea150..6afb283 100644 --- a/api/v1alpha1/pipeline_types.go +++ b/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 diff --git a/config/crd/bases/filter.plainsight.ai_pipelines.yaml b/config/crd/bases/filter.plainsight.ai_pipelines.yaml index 8357a30..69109c1 100644 --- a/config/crd/bases/filter.plainsight.ai_pipelines.yaml +++ b/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 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..69109c1 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,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 diff --git a/internal/controller/pipelineinstance_controller_streaming.go b/internal/controller/pipelineinstance_controller_streaming.go index df74ac0..aabc40b 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.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) diff --git a/internal/controller/pipelineinstance_controller_streaming_test.go b/internal/controller/pipelineinstance_controller_streaming_test.go index 856c549..8f00a9e 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,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) + } +}