feat: Support running UX server with namespace scope. Fixes #248 (#249)

Signed-off-by: David Seapy <dseapy@gmail.com>

cmd/commands/server.go

@@ -1,14 +1,17 @@
 package commands
 
 import (
+	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 	svrcmd "github.com/numaproj/numaflow/server/cmd"
 	"github.com/spf13/cobra"
 )
 
 func NewServerCommand() *cobra.Command {
 	var (
-		insecure bool
-		port     int
+		insecure         bool
+		port             int
+		namespaced       bool
+		managedNamespace string
 	)
 
 	command := &cobra.Command{
@@ -18,10 +21,12 @@ func NewServerCommand() *cobra.Command {
 			if !cmd.Flags().Changed("port") && insecure {
 				port = 8080
 			}
-			svrcmd.Start(insecure, port)
+			svrcmd.Start(insecure, port, namespaced, managedNamespace)
 		},
 	}
 	command.Flags().BoolVar(&insecure, "insecure", false, "Whether to disable TLS, defaults to false.")
 	command.Flags().IntVarP(&port, "port", "p", 8443, "Port to listen on, defaults to 8443 or 8080 if insecure is set")
+	command.Flags().BoolVar(&namespaced, "namespaced", false, "Whether to run in namespaced scope, defaults to false.")
+	command.Flags().StringVar(&managedNamespace, "managed-namespace", sharedutil.LookupEnvStringOr("NAMESPACE", "numaflow-system"), "The namespace that the server watches when \"--namespaced\" is \"true\".")
 	return command
 }

config/namespace-install.yaml

@@ -12904,6 +12904,7 @@ spec:
       containers:
       - args:
         - server
+        - --namespaced
         env:
         - name: NAMESPACE
           valueFrom:

config/namespace-install/kustomization.yaml

@@ -13,3 +13,10 @@ patches:
     target:
       kind: Deployment
       name: numaflow-controller
+  - patch: |-
+      - op: add
+        path: /spec/template/spec/containers/0/args/-
+        value: --namespaced
+    target:
+      kind: Deployment
+      name: numaflow-server

server/apis/v1/handler.go

@@ -85,7 +85,8 @@ func (h *handler) GetPipeline(c *gin.Context) {
 
 // ListNamespaces is used to provide all the namespaces that have numaflow pipelines running
 func (h *handler) ListNamespaces(c *gin.Context) {
-	l, err := h.numaflowClient.Pipelines("").List(context.Background(), metav1.ListOptions{})
+	ns := c.GetString("namespace")
+	l, err := h.numaflowClient.Pipelines(ns).List(context.Background(), metav1.ListOptions{})
 	if err != nil {
 		c.JSON(http.StatusInternalServerError, err.Error())
 		return

server/cmd/start.go

@@ -20,12 +20,15 @@ var (
 	}
 )
 
-func Start(insecure bool, port int) {
+func Start(insecure bool, port int, namespaced bool, managedNamespace string) {
 	logger := logging.NewLogger().Named("server")
 	router := gin.New()
 	router.Use(gin.Logger())
 	router.RedirectTrailingSlash = true
 	router.Use(static.Serve("/", static.LocalFile("./ui/build", true)))
+	if namespaced {
+		router.Use(Namespace(managedNamespace))
+	}
 	routes.Routes(router)
 	router.Use(UrlRewrite(router))
 	server := http.Server{
@@ -70,3 +73,10 @@ func UrlRewrite(r *gin.Engine) gin.HandlerFunc {
 		c.Next()
 	}
 }
+
+func Namespace(ns string) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		c.Set("namespace", ns)
+		c.Next()
+	}
+}