Repository: numaproj/numaflow (go)

Issue:
UX server code refactor
# Summary

- [ ] Combine the [route groups for dex auth and local auth](https://github.com/numaproj/numaflow/blob/41b8dffc46a5c46e8d8412c69ef291a8be3821da/server/routes/routes.go#L85-L107) into one. 
- [ ] Use same endpoint for logout
- [ ] Use one logger for all the UX server



---
<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->
**Message from the maintainers**:

If you wish to see this enhancement implemented please add a 👍 reaction to this issue! We often sort issues this way to know what to prioritize.


Pull request:
chore: ux server refactor
Fixes #1420 

- Merged handlers for dex authn and localUser authn
- Enabled a common logger for the ux server

Patch:
diff --git a/cmd/commands/server.go b/cmd/commands/server.go
index 794c6788..11dfe4b2 100644
--- a/cmd/commands/server.go
+++ b/cmd/commands/server.go
@@ -20,7 +20,9 @@ import (
 	"strings"
 
 	"github.com/spf13/cobra"
+	"sigs.k8s.io/controller-runtime/pkg/manager/signals"
 
+	"github.com/numaproj/numaflow/pkg/shared/logging"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 	svrcmd "github.com/numaproj/numaflow/server/cmd/server"
 )
@@ -58,7 +60,8 @@ func NewServerCommand() *cobra.Command {
 				ServerAddr:       serverAddr,
 			}
 			server := svrcmd.NewServer(opts)
-			server.Start()
+			log := logging.NewLogger().Named("server")
+			server.Start(logging.WithLogger(signals.SetupSignalHandler(), log))
 		},
 	}
 	command.Flags().BoolVar(&insecure, "insecure", sharedutil.LookupEnvBoolOr("NUMAFLOW_SERVER_INSECURE", false), "Whether to disable TLS, defaults to false.")
diff --git a/server/apis/v1/handler.go b/server/apis/v1/handler.go
index a726468c..c276fa45 100644
--- a/server/apis/v1/handler.go
+++ b/server/apis/v1/handler.go
@@ -69,7 +69,7 @@ type handler struct {
 }
 
 // NewHandler is used to provide a new instance of the handler type
-func NewHandler(dexObj *DexObject, localUsersAuthObject *LocalUsersAuthObject) (*handler, error) {
+func NewHandler(ctx context.Context, dexObj *DexObject, localUsersAuthObject *LocalUsersAuthObject) (*handler, error) {
 	var (
 		k8sRestConfig *rest.Config
 		err           error
@@ -94,7 +94,7 @@ func NewHandler(dexObj *DexObject, localUsersAuthObject *LocalUsersAuthObject) (
 		daemonClientsCache:   daemonClientsCache,
 		dexObj:               dexObj,
 		localUsersAuthObject: localUsersAuthObject,
-		healthChecker:        NewHealthChecker(),
+		healthChecker:        NewHealthChecker(ctx),
 	}, nil
 }
 
diff --git a/server/apis/v1/health.go b/server/apis/v1/health.go
index f52c45f3..5a28338e 100644
--- a/server/apis/v1/health.go
+++ b/server/apis/v1/health.go
@@ -5,6 +5,7 @@ import (
 	"time"
 
 	evictCache "github.com/hashicorp/golang-lru/v2/expirable"
+	"go.uber.org/zap"
 	"golang.org/x/net/context"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
@@ -30,13 +31,15 @@ type resourceHealthResponse struct {
 // HealthChecker is the struct to hold the resource status cache for the pipeline
 type HealthChecker struct {
 	resourceStatusCache *evictCache.LRU[string, *resourceHealthResponse]
+	log                 *zap.SugaredLogger
 }
 
 // NewHealthChecker is used to create a new health checker
-func NewHealthChecker() *HealthChecker {
+func NewHealthChecker(ctx context.Context) *HealthChecker {
 	c := evictCache.NewLRU[string, *resourceHealthResponse](500, nil, resourceCacheRefreshDuration)
 	return &HealthChecker{
 		resourceStatusCache: c,
+		log:                 logging.FromContext(ctx),
 	}
 }
 
@@ -45,8 +48,6 @@ func NewHealthChecker() *HealthChecker {
 // and caches it.
 func (hc *HealthChecker) getPipelineResourceHealth(h *handler, ns string,
 	pipeline string) (*resourceHealthResponse, error) {
-	ctx := context.Background()
-	log := logging.FromContext(ctx)
 
 	// create a cache key for the pipeline
 	// It is a combination of namespace and pipeline name
@@ -55,11 +56,11 @@ func (hc *HealthChecker) getPipelineResourceHealth(h *handler, ns string,
 
 	// check if the pipeline status is cached
 	if status, ok := hc.resourceStatusCache.Get(cacheKey); ok {
-		log.Info("Pipeline status from cache: ", status)
+		hc.log.Info("Pipeline status from cache: ", status)
 		return status, nil
 	}
 	// if not present in cache, check for the current pipeline status
-	status, err := checkVertexLevelHealth(ctx, h, ns, pipeline)
+	status, err := checkVertexLevelHealth(h, ns, pipeline, hc.log)
 	if err != nil {
 		return status, err
 	}
@@ -81,9 +82,8 @@ func (hc *HealthChecker) getPipelineResourceHealth(h *handler, ns string,
 // are equal to the number of desired replicas and the pods are in running state
 // 2) If all the containers in the pod are in running state
 // if any of the above conditions are not met, the vertex is unhealthy
-func checkVertexLevelHealth(ctx context.Context, h *handler, ns string,
-	pipeline string) (*resourceHealthResponse, error) {
-	log := logging.FromContext(ctx)
+func checkVertexLevelHealth(h *handler, ns string,
+	pipeline string, log *zap.SugaredLogger) (*resourceHealthResponse, error) {
 	// get the pipeline object
 	pl, err := h.numaflowClient.Pipelines(ns).Get(context.Background(), pipeline, metav1.GetOptions{})
 	// if error return unknown status
diff --git a/server/apis/v1/local_user_authn.go b/server/apis/v1/local_user_authn.go
index e9f6274d..c5e97ea0 100644
--- a/server/apis/v1/local_user_authn.go
+++ b/server/apis/v1/local_user_authn.go
@@ -26,7 +26,9 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/golang-jwt/jwt/v5"
+	"go.uber.org/zap"
 
+	"github.com/numaproj/numaflow/pkg/shared/logging"
 	"github.com/numaproj/numaflow/pkg/shared/util"
 	"github.com/numaproj/numaflow/server/authn"
 	"github.com/numaproj/numaflow/server/common"
@@ -35,10 +37,11 @@ import (
 type LocalUsersAuthObject struct {
 	kubeClient   kubernetes.Interface
 	authDisabled bool
+	log          *zap.SugaredLogger
 }
 
 // NewLocalUsersAuthObject is used to provide a new LocalUsersAuthObject
-func NewLocalUsersAuthObject(authDisabled bool) (*LocalUsersAuthObject, error) {
+func NewLocalUsersAuthObject(ctx context.Context, authDisabled bool) (*LocalUsersAuthObject, error) {
 	var (
 		k8sRestConfig *rest.Config
 		err           error
@@ -55,6 +58,7 @@ func NewLocalUsersAuthObject(authDisabled bool) (*LocalUsersAuthObject, error) {
 	return &LocalUsersAuthObject{
 		kubeClient:   kubeClient,
 		authDisabled: authDisabled,
+		log:          logging.FromContext(ctx),
 	}, nil
 }
 
@@ -102,7 +106,8 @@ func (l *LocalUsersAuthObject) VerifyUser(c *gin.Context, username string, passw
 		return fmt.Errorf("username or password cannot be empty")
 	}
 
-	account, err := authn.GetAccount(c.Request.Context(), username, l.kubeClient)
+	ctx := logging.WithLogger(c.Request.Context(), l.log)
+	account, err := authn.GetAccount(ctx, username, l.kubeClient)
 	if err != nil {
 		return err
 	}
diff --git a/server/apis/v1/local_user_handler.go b/server/apis/v1/local_user_handler.go
deleted file mode 100644
index ced1ea71..00000000
--- a/server/apis/v1/local_user_handler.go
+++ /dev/null
@@ -1,126 +0,0 @@
-/*
-Copyright 2022 The Numaproj Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package v1
-
-import (
-	"fmt"
-	"net/http"
-
-	"github.com/gin-gonic/gin"
-
-	"github.com/numaproj/numaflow/server/authn"
-	"github.com/numaproj/numaflow/server/common"
-)
-
-type localUsersHandler struct {
-	localUsersAuthObject *LocalUsersAuthObject
-}
-
-// NewLocalUsersHandler is used to provide a new instance of the handler type
-func NewLocalUsersHandler(localUsersAuthObject *LocalUsersAuthObject) (*localUsersHandler, error) {
-	return &localUsersHandler{
-		localUsersAuthObject: localUsersAuthObject,
-	}, nil
-}
-
-type LoginInput struct {
-	Username string `json:"username" binding:"required"`
-	Password string `json:"password" binding:"required"`
-}
-
-// Login is used to generate the jwt token and return it as part of the return payload.
-// The jwt token and login type are also set as a cookie.
-func (h *localUsersHandler) Login(c *gin.Context) {
-	if h.localUsersAuthObject.authDisabled {
-		errMsg := "Auth is disabled"
-		c.JSON(http.StatusOK, NewNumaflowAPIResponse(&errMsg, nil))
-		return
-	}
-
-	var (
-		input LoginInput
-		err   error
-	)
-
-	if err = c.ShouldBindJSON(&input); err != nil {
-		errMsg := "Missing input fields"
-		c.JSON(http.StatusOK, NewNumaflowAPIResponse(&errMsg, nil))
-		return
-	}
-
-	user := authn.LoginCredentials{}
-	user.Username = input.Username
-	user.Password = input.Password
-
-	// Verify user for the given username and password
-	if err = h.localUsersAuthObject.VerifyUser(c, user.Username, user.Password); err != nil {
-		errMsg := fmt.Sprintf("%v", err)
-		c.JSON(http.StatusOK, NewNumaflowAPIResponse(&errMsg, nil))
-		return
-	}
-
-	//GenerateToken
-	token, err := h.localUsersAuthObject.GenerateToken(c, user.Username)
-	if err != nil {
-		errMsg := fmt.Sprintf("Error while generating user identity token: %v", err)
-		c.JSON(http.StatusOK, NewNumaflowAPIResponse(&errMsg, nil))
-		return
-	}
-	if token == "" {
-		errMsg := "Failed to get token: empty user identity Token"
-		c.JSON(http.StatusOK, NewNumaflowAPIResponse(&errMsg, nil))
-		return
-	}
-
-	// Set the login type as a cookie
-	c.SetCookie(common.LoginCookieName, "local", common.JWTCookieMaxAge, "/", "", true, true)
-	// Set the JWT token as a cookie
-	c.SetCookie(common.JWTCookieName, token, common.JWTCookieMaxAge, "/", "", true, true)
-	c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, token))
-}
-
-// Logout is used to remove auth cookies ending a user's session.
-func (h *localUsersHandler) Logout(c *gin.Context) {
-	tokenString, err := c.Cookie(common.JWTCookieName)
-	if err != nil {
-		errMsg := fmt.Sprintf("Failed to retrieve user identity token: %v", err)
-		c.JSON(http.StatusBadRequest, NewNumaflowAPIResponse(&errMsg, nil))
-		return
-	}
-	if tokenString == "" {
-		// no jwt found, return directly
-		c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, nil))
-		return
-	}
-	// clearing jwt cookie
-	c.SetCookie(common.JWTCookieName, "", -1, "/", "", true, true)
-
-	loginString, err := c.Cookie(common.LoginCookieName)
-	if err != nil {
-		errMsg := fmt.Sprintf("Failed to retrieve login type cookie: %v", err)
-		c.JSON(http.StatusBadRequest, NewNumaflowAPIResponse(&errMsg, nil))
-	}
-	if loginString == "" {
-		// no numaflow-login found, return directly
-		c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, nil))
-		return
-	}
-	// clearing numaflow-login cookie
-	c.SetCookie(common.LoginCookieName, "", -1, "/", "", true, true)
-
-	c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, nil))
-}
diff --git a/server/apis/v1/noauthhandler.go b/server/apis/v1/noauthhandler.go
index 9ae50842..8044814f 100644
--- a/server/apis/v1/noauthhandler.go
+++ b/server/apis/v1/noauthhandler.go
@@ -24,17 +24,25 @@ import (
 
 	"github.com/gin-gonic/gin"
 
+	"github.com/numaproj/numaflow/server/authn"
 	"github.com/numaproj/numaflow/server/common"
 )
 
 type noAuthHandler struct {
-	dexObj *DexObject
+	dexObj               *DexObject
+	localUsersAuthObject *LocalUsersAuthObject
+}
+
+type LoginInput struct {
+	Username string `json:"username" binding:"required"`
+	Password string `json:"password" binding:"required"`
 }
 
 // NewNoAuthHandler is used to provide a new instance of the handler type
-func NewNoAuthHandler(dexObj *DexObject) (*noAuthHandler, error) {
+func NewNoAuthHandler(dexObj *DexObject, localUsersAuthObject *LocalUsersAuthObject) (*noAuthHandler, error) {
 	return &noAuthHandler{
-		dexObj: dexObj,
+		dexObj:               dexObj,
+		localUsersAuthObject: localUsersAuthObject,
 	}, nil
 }
 
@@ -43,34 +51,64 @@ func (h *noAuthHandler) Login(c *gin.Context) {
 	h.dexObj.handleLogin(c)
 }
 
-// Callback is used to extract user authentication information from the Dex Server returned payload.
-func (h *noAuthHandler) Callback(c *gin.Context) {
-	h.dexObj.handleCallback(c)
-}
+// LoginLocalUsers is used to generate the jwt token and return it as part of the return payload.
+// The jwt token and login type are also set as a cookie.
+func (h *noAuthHandler) LoginLocalUsers(c *gin.Context) {
+	if h.localUsersAuthObject.authDisabled {
+		errMsg := "Auth is disabled"
+		c.JSON(http.StatusOK, NewNumaflowAPIResponse(&errMsg, nil))
+		return
+	}
 
-// Logout is used to remove auth cookie ending a user's session.
-func (h *noAuthHandler) Logout(c *gin.Context) {
-	cookies := c.Request.Cookies()
-	tokenString, err := common.JoinCookies(common.UserIdentityCookieName, cookies)
-	var invalidCookieE *common.InvalidCookieError
-	if errors.As(err, &invalidCookieE) {
-		errMsg := fmt.Sprintf("Failed to retrieve user identity token: %v", err)
-		c.JSON(http.StatusBadRequest, NewNumaflowAPIResponse(&errMsg, nil))
+	var (
+		input LoginInput
+		err   error
+	)
+
+	if err = c.ShouldBindJSON(&input); err != nil {
+		errMsg := "Missing input fields"
+		c.JSON(http.StatusOK, NewNumaflowAPIResponse(&errMsg, nil))
 		return
 	}
-	if tokenString == "" {
-		// no numaflow.token found, return directly
-		c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, nil))
+
+	user := authn.LoginCredentials{}
+	user.Username = input.Username
+	user.Password = input.Password
+
+	// Verify user for the given username and password
+	if err = h.localUsersAuthObject.VerifyUser(c, user.Username, user.Password); err != nil {
+		errMsg := fmt.Sprintf("%v", err)
+		c.JSON(http.StatusOK, NewNumaflowAPIResponse(&errMsg, nil))
 		return
 	}
-	// if there's any numaflow.token cookie, delete
-	for _, cookie := range cookies {
-		if !strings.HasPrefix(cookie.Name, common.UserIdentityCookieName) {
-			continue
-		}
-		c.SetCookie(cookie.Name, "", -1, "/", "", true, true)
+
+	//GenerateToken
+	token, err := h.localUsersAuthObject.GenerateToken(c, user.Username)
+	if err != nil {
+		errMsg := fmt.Sprintf("Error while generating user identity token: %v", err)
+		c.JSON(http.StatusOK, NewNumaflowAPIResponse(&errMsg, nil))
+		return
 	}
+	if token == "" {
+		errMsg := "Failed to get token: empty user identity Token"
+		c.JSON(http.StatusOK, NewNumaflowAPIResponse(&errMsg, nil))
+		return
+	}
+
+	// Set the login type as a cookie
+	c.SetCookie(common.LoginCookieName, "local", common.JWTCookieMaxAge, "/", "", true, true)
+	// Set the JWT token as a cookie
+	c.SetCookie(common.JWTCookieName, token, common.JWTCookieMaxAge, "/", "", true, true)
+	c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, token))
+}
+
+// Callback is used to extract user authentication information from the Dex Server returned payload.
+func (h *noAuthHandler) Callback(c *gin.Context) {
+	h.dexObj.handleCallback(c)
+}
 
+// Logout is used to remove auth cookie ending a user's session.
+func (h *noAuthHandler) Logout(c *gin.Context) {
 	loginString, err := c.Cookie(common.LoginCookieName)
 	if err != nil {
 		errMsg := fmt.Sprintf("Failed to retrieve login type cookie: %v", err)
@@ -81,7 +119,45 @@ func (h *noAuthHandler) Logout(c *gin.Context) {
 		c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, nil))
 		return
 	}
+	// clearing numaflow-login cookie
 	c.SetCookie(common.LoginCookieName, "", -1, "/", "", true, true)
 
+	if loginString == "dex" {
+		cookies := c.Request.Cookies()
+		tokenString, err := common.JoinCookies(common.UserIdentityCookieName, cookies)
+		var invalidCookieE *common.InvalidCookieError
+		if errors.As(err, &invalidCookieE) {
+			errMsg := fmt.Sprintf("Failed to retrieve user identity token: %v", err)
+			c.JSON(http.StatusBadRequest, NewNumaflowAPIResponse(&errMsg, nil))
+			return
+		}
+		if tokenString == "" {
+			// no numaflow.token found, return directly
+			c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, nil))
+			return
+		}
+		// if there's any numaflow.token cookie, delete
+		for _, cookie := range cookies {
+			if !strings.HasPrefix(cookie.Name, common.UserIdentityCookieName) {
+				continue
+			}
+			c.SetCookie(cookie.Name, "", -1, "/", "", true, true)
+		}
+	} else if loginString == "local" {
+		tokenString, err := c.Cookie(common.JWTCookieName)
+		if err != nil {
+			errMsg := fmt.Sprintf("Failed to retrieve user identity token: %v", err)
+			c.JSON(http.StatusBadRequest, NewNumaflowAPIResponse(&errMsg, nil))
+			return
+		}
+		if tokenString == "" {
+			// no jwt found, return directly
+			c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, nil))
+			return
+		}
+		// clearing jwt cookie
+		c.SetCookie(common.JWTCookieName, "", -1, "/", "", true, true)
+	}
+
 	c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, nil))
 }
diff --git a/server/authn/local_accounts.go b/server/authn/local_accounts.go
index 034b323d..60fd496b 100644
--- a/server/authn/local_accounts.go
+++ b/server/authn/local_accounts.go
@@ -46,9 +46,6 @@ type Account struct {
 	Enabled             bool
 }
 
-// TODO: refactor to use a logger from server
-var logger = logging.NewLogger().Named("accounts")
-
 // GetAccount return an account info by the specified name.
 func GetAccount(ctx context.Context, name string, kubeClient kubernetes.Interface) (*Account, error) {
 	accounts, err := GetAccounts(ctx, kubeClient)
@@ -73,7 +70,7 @@ func GetAccounts(ctx context.Context, kubeClient kubernetes.Interface) (map[stri
 	if err != nil {
 		return nil, err
 	}
-	return parseAccounts(secret, cm)
+	return parseAccounts(ctx, secret, cm)
 }
 
 // parseAdminAccount parses the admin account from the secret and configMap
@@ -100,7 +97,8 @@ func parseAdminAccount(secret *k8sv1.Secret, cm *k8sv1.ConfigMap) (*Account, err
 }
 
 // parseAccounts parses all the accounts from the secret and configMap
-func parseAccounts(secret *k8sv1.Secret, cm *k8sv1.ConfigMap) (map[string]Account, error) {
+func parseAccounts(ctx context.Context, secret *k8sv1.Secret, cm *k8sv1.ConfigMap) (map[string]Account, error) {
+	log := logging.FromContext(ctx)
 	adminAccount, err := parseAdminAccount(secret, cm)
 	if err != nil {
 		return nil, err
@@ -122,7 +120,7 @@ func parseAccounts(secret *k8sv1.Secret, cm *k8sv1.ConfigMap) (map[string]Accoun
 		if len(parts) == 2 {
 			accountName = parts[0]
 		} else {
-			logger.Warnf("Unexpected key %s in ConfigMap '%s'", key, cm.Name)
+			log.Warnf("Unexpected key %s in ConfigMap '%s'", key, cm.Name)
 			continue
 		}
 
diff --git a/server/authz/rbac.go b/server/authz/rbac.go
index d786c220..afd9ccea 100644
--- a/server/authz/rbac.go
+++ b/server/authz/rbac.go
@@ -17,6 +17,7 @@ limitations under the License.
 package authz
 
 import (
+	"context"
 	_ "embed"
 	"fmt"
 	"path"
@@ -29,6 +30,7 @@ import (
 	"github.com/fsnotify/fsnotify"
 	"github.com/gin-gonic/gin"
 	"github.com/spf13/viper"
+	"go.uber.org/zap"
 	"k8s.io/utils/strings/slices"
 
 	"github.com/numaproj/numaflow/pkg/shared/logging"
@@ -38,7 +40,6 @@ import (
 var (
 	//go:embed rbac-model.conf
 	rbacModel string
-	logger    = logging.NewLogger()
 )
 
 const (
@@ -63,11 +64,12 @@ type CasbinObject struct {
 	authRouteMap  RouteMap
 	opts          *options
 	rwMutex       *sync.RWMutex
+	log           *zap.SugaredLogger
 }
 
 // NewCasbinObject returns a new CasbinObject. It initializes the Casbin Enforcer with the model and policy.
 // It also initializes the config reader to watch for changes in the config file.
-func NewCasbinObject(authRouteMap RouteMap, inputOptions ...Option) (*CasbinObject, error) {
+func NewCasbinObject(ctx context.Context, authRouteMap RouteMap, inputOptions ...Option) (*CasbinObject, error) {
 	// Set the default options.
 	var opts = DefaultOptions()
 	// Apply the input options.
@@ -85,6 +87,7 @@ func NewCasbinObject(authRouteMap RouteMap, inputOptions ...Option) (*CasbinObje
 		return nil, err
 	}
 	currentScopes := getRBACScopes(configReader)
+	logger := logging.FromContext(ctx)
 	logger.Infow("Auth Scopes", "scopes", currentScopes)
 	// Set the default policy for authorization.
 	policyDefault := getDefaultPolicy(configReader)
@@ -98,6 +101,7 @@ func NewCasbinObject(authRouteMap RouteMap, inputOptions ...Option) (*CasbinObje
 		authRouteMap:  authRouteMap,
 		opts:          opts,
 		rwMutex:       &sync.RWMutex{},
+		log:           logger,
 	}
 
 	// Watch for changes in the config file.
@@ -133,7 +137,7 @@ func (cas *CasbinObject) Authorize(c *gin.Context, userInfo *authn.UserInfo) boo
 	// If the user does not have any policy defined, allocate a default policy for the user.
 	if !userHasPolicies {
 		defaultPolicy := cas.getDefaultPolicy()
-		logger.Debugw("No policy defined for the user, allocating default policy",
+		cas.log.Debugw("No policy defined for the user, allocating default policy",
 			"DefaultPolicy", defaultPolicy)
 		ok := enforceCheck(cas.enforcer, defaultPolicy, resource, object, action)
 		if ok {
@@ -292,7 +296,7 @@ func enforceCheck(enforcer *casbin.Enforcer, user, resource, object, action stri
 // configFileReload is used to reload the config file when it is changed. This is used to reload the policy without
 // restarting the server. The config file is in the format of yaml. The config file is read by viper.
 func (cas *CasbinObject) configFileReload(e fsnotify.Event) {
-	logger.Infow("RBAC conf file updated:", "fileName", e.Name)
+	cas.log.Infow("RBAC conf file updated:", "fileName", e.Name)
 	err := cas.configReader.ReadInConfig()
 	if err != nil {
 		return
@@ -306,7 +310,7 @@ func (cas *CasbinObject) configFileReload(e fsnotify.Event) {
 	// clear the userPermCount cache
 	cas.userPermCount = &sync.Map{}
 
-	logger.Infow("Auth Scopes Updated", "scopes", cas.getCurrentScopes())
+	cas.log.Infow("Auth Scopes Updated", "scopes", cas.getCurrentScopes())
 }
 
 // getDefaultPolicy returns the default policy from the rbac properties file. The default policy is used when the
@@ -335,7 +339,7 @@ func (cas *CasbinObject) hasPermissionsDefined(user string) bool {
 	// get the permissions for the user
 	cnt, err := cas.enforcer.GetImplicitPermissionsForUser(user)
 	if err != nil {
-		logger.Errorw("Failed to get permissions for user", "user", user, "error", err)
+		cas.log.Errorw("Failed to get permissions for user", "user", user, "error", err)
 		return false
 	}
 	// If the user has permissions in the policy, len(cnt) > 0
diff --git a/server/cmd/server/start.go b/server/cmd/server/start.go
index 56d3e48e..b43627df 100644
--- a/server/cmd/server/start.go
+++ b/server/cmd/server/start.go
@@ -17,6 +17,7 @@ limitations under the License.
 package cmd
 
 import (
+	"context"
 	"crypto/tls"
 	"fmt"
 	"net/http"
@@ -62,8 +63,8 @@ func NewServer(opts ServerOptions) *server {
 	}
 }
 
-func (s *server) Start() {
-	logger := logging.NewLogger().Named("server")
+func (s *server) Start(ctx context.Context) {
+	log := logging.FromContext(ctx)
 	router := gin.New()
 	router.Use(gin.LoggerWithConfig(gin.LoggerConfig{SkipPaths: []string{"/livez"}}))
 	router.RedirectTrailingSlash = true
@@ -77,6 +78,7 @@ func (s *server) Start() {
 	}
 	router.Any("/dex/*name", v1.NewDexReverseProxy(s.options.DexServerAddr))
 	routes.Routes(
+		ctx,
 		router,
 		routes.SystemInfo{
 			ManagedNamespace: s.options.ManagedNamespace,
@@ -97,7 +99,7 @@ func (s *server) Start() {
 	}
 
 	if s.options.Insecure {
-		logger.Infow(
+		log.Infow(
 			"Starting server (TLS disabled) on "+server.Addr,
 			"version", numaflow.GetVersion(),
 			"disable-auth", s.options.DisableAuth,
@@ -112,7 +114,7 @@ func (s *server) Start() {
 			panic(err)
 		}
 		server.TLSConfig = &tls.Config{Certificates: []tls.Certificate{*cert}, MinVersion: tls.VersionTLS12}
-		logger.Infow(
+		log.Infow(
 			"Starting server on "+server.Addr,
 			"version", numaflow.GetVersion(),
 			"disable-auth", s.options.DisableAuth,
diff --git a/server/routes/routes.go b/server/routes/routes.go
index 8926781f..b9dae1d4 100644
--- a/server/routes/routes.go
+++ b/server/routes/routes.go
@@ -17,6 +17,7 @@ limitations under the License.
 package routes
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 
@@ -41,9 +42,7 @@ type AuthInfo struct {
 	ServerAddr    string `json:"serverAddr"`
 }
 
-var logger = logging.NewLogger().Named("server")
-
-func Routes(r *gin.Engine, sysInfo SystemInfo, authInfo AuthInfo, baseHref string, authRouteMap authz.RouteMap) {
+func Routes(ctx context.Context, r *gin.Engine, sysInfo SystemInfo, authInfo AuthInfo, baseHref string, authRouteMap authz.RouteMap) {
 	r.GET("/livez", func(c *gin.Context) {
 		c.Status(http.StatusOK)
 	})
@@ -51,65 +50,54 @@ func Routes(r *gin.Engine, sysInfo SystemInfo, authInfo AuthInfo, baseHref strin
 	if err != nil {
 		panic(err)
 	}
-	// noAuthGroup is a group of routes that do not require AuthN/AuthZ no matter whether auth is enabled.
-	noAuthGroup := r.Group(baseHref + "auth/v1")
-	v1RoutesNoAuth(noAuthGroup, dexObj)
 
-	localUsersAuthObj, err := v1.NewLocalUsersAuthObject(authInfo.DisableAuth)
+	localUsersAuthObj, err := v1.NewLocalUsersAuthObject(ctx, authInfo.DisableAuth)
 	if err != nil {
 		panic(err)
 	}
-	// localUsersGroup is a group of routes that do not require AuthN/AuthZ no matter whether auth is enabled.
-	localUsersGroup := r.Group(baseHref + "/auth/local/v1")
-	v1RoutesLocalUser(localUsersGroup, localUsersAuthObj)
+
+	// noAuthGroup is a group of routes that do not require AuthN/AuthZ no matter whether auth is enabled.
+	noAuthGroup := r.Group(baseHref + "auth/v1")
+	v1RoutesNoAuth(noAuthGroup, dexObj, localUsersAuthObj)
 
 	// r1Group is a group of routes that require AuthN/AuthZ when auth is enabled.
 	// they share the AuthN/AuthZ middleware.
 	r1Group := r.Group(baseHref + "api/v1")
 	if !authInfo.DisableAuth {
-		authorizer, err := authz.NewCasbinObject(authRouteMap)
+		authorizer, err := authz.NewCasbinObject(ctx, authRouteMap)
 		if err != nil {
 			panic(err)
 		}
 		// Add the AuthN/AuthZ middleware to the group.
-		r1Group.Use(authMiddleware(authorizer, dexObj, localUsersAuthObj, authRouteMap))
-		v1Routes(r1Group, dexObj, localUsersAuthObj)
+		r1Group.Use(authMiddleware(ctx, authorizer, dexObj, localUsersAuthObj, authRouteMap))
+		v1Routes(ctx, r1Group, dexObj, localUsersAuthObj)
 	} else {
-		v1Routes(r1Group, nil, nil)
+		v1Routes(ctx, r1Group, nil, nil)
 	}
 	r1Group.GET("/sysinfo", func(c *gin.Context) {
 		c.JSON(http.StatusOK, v1.NewNumaflowAPIResponse(nil, sysInfo))
 	})
 }
 
-func v1RoutesNoAuth(r gin.IRouter, dexObj *v1.DexObject) {
-	handler, err := v1.NewNoAuthHandler(dexObj)
+func v1RoutesNoAuth(r gin.IRouter, dexObj *v1.DexObject, localUsersAuthObject *v1.LocalUsersAuthObject) {
+	handler, err := v1.NewNoAuthHandler(dexObj, localUsersAuthObject)
 	if err != nil {
 		panic(err)
 	}
 	// Handle the login request.
 	r.GET("/login", handler.Login)
+	// Handle the login request for local users.
+	r.POST("/login", handler.LoginLocalUsers)
 	// Handle the logout request.
 	r.GET("/logout", handler.Logout)
 	// Handle the callback request.
 	r.GET("/callback", handler.Callback)
 }
 
-func v1RoutesLocalUser(r gin.IRouter, localUsersAuthObject *v1.LocalUsersAuthObject) {
-	handler, err := v1.NewLocalUsersHandler(localUsersAuthObject)
-	if err != nil {
-		panic(err)
-	}
-	// Handle the login request.
-	r.POST("/login", handler.Login)
-	// Handle the logout request.
-	r.GET("/logout", handler.Logout)
-}
-
 // v1Routes defines the routes for the v1 API. For adding a new route, add a new handler function
 // for the route along with an entry in the RouteMap in auth/route_map.go.
-func v1Routes(r gin.IRouter, dexObj *v1.DexObject, localUsersAuthObject *v1.LocalUsersAuthObject) {
-	handler, err := v1.NewHandler(dexObj, localUsersAuthObject)
+func v1Routes(ctx context.Context, r gin.IRouter, dexObj *v1.DexObject, localUsersAuthObject *v1.LocalUsersAuthObject) {
+	handler, err := v1.NewHandler(ctx, dexObj, localUsersAuthObject)
 	if err != nil {
 		panic(err)
 	}
@@ -164,10 +152,11 @@ func v1Routes(r gin.IRouter, dexObj *v1.DexObject, localUsersAuthObject *v1.Loca
 // authMiddleware is the middleware for AuthN/AuthZ.
 // it ensures the user is authenticated and authorized
 // to execute the requested action before sending the request to the api handler.
-func authMiddleware(authorizer authz.Authorizer, dexAuthenticator authn.Authenticator, localUsersAuthenticator authn.Authenticator, authRouteMap authz.RouteMap) gin.HandlerFunc {
+func authMiddleware(ctx context.Context, authorizer authz.Authorizer, dexAuthenticator authn.Authenticator, localUsersAuthenticator authn.Authenticator, authRouteMap authz.RouteMap) gin.HandlerFunc {
 
 	return func(c *gin.Context) {
 
+		log := logging.FromContext(ctx)
 		var userInfo *authn.UserInfo
 
 		loginType, err := c.Cookie(common.LoginCookieName)
@@ -214,7 +203,7 @@ func authMiddleware(authorizer authz.Authorizer, dexAuthenticator authn.Authenti
 			c.Next()
 		} else {
 			// If the route is not present in the route map, return an error.
-			logger.Errorw("route not present in routeMap", "route", authz.GetRouteMapKey(c))
+			log.Errorw("route not present in routeMap", "route", authz.GetRouteMapKey(c))
 			errMsg := "Invalid route"
 			c.JSON(http.StatusForbidden, v1.NewNumaflowAPIResponse(&errMsg, nil))
 			c.Abort()
diff --git a/ui/src/components/common/AccountMenu/index.tsx b/ui/src/components/common/AccountMenu/index.tsx
index 220fe008..aa6291f0 100644
--- a/ui/src/components/common/AccountMenu/index.tsx
+++ b/ui/src/components/common/AccountMenu/index.tsx
@@ -11,10 +11,7 @@ export default function AccountMenu() {
   const navigate = useNavigate();
   const { userInfo } = useContext<AppContextProps>(AppContext);
 
-  const url =
-    userInfo?.email === ""
-      ? `${getBaseHref()}/auth/local/v1/logout`
-      : `${getBaseHref()}/auth/v1/logout`;
+  const url = `${getBaseHref()}/auth/v1/logout`;
 
   const handleLogout = useCallback(async () => {
     try {
diff --git a/ui/src/components/pages/Login/index.tsx b/ui/src/components/pages/Login/index.tsx
index 6e0c9cc4..b74c8e56 100644
--- a/ui/src/components/pages/Login/index.tsx
+++ b/ui/src/components/pages/Login/index.tsx
@@ -144,7 +144,7 @@ export function Login() {
     };
 
     try {
-      const response = await fetch(`${getBaseHref()}/auth/local/v1/login`, {
+      const response = await fetch(`${getBaseHref()}/auth/v1/login`, {
         method: "POST",
         body: JSON.stringify(data),
       });


Test patch:
diff --git a/server/authz/rbac_test.go b/server/authz/rbac_test.go
index 242bfa05..5a41a951 100644
--- a/server/authz/rbac_test.go
+++ b/server/authz/rbac_test.go
@@ -33,7 +33,7 @@ var (
 // TestCreateAuthorizer is a test implementation of the NewCasbinObject function.
 // It checks that the authorizer is created correctly and the policies, configs are loaded correctly.
 func TestCreateAuthorizer(t *testing.T) {
-	authorizer, err := NewCasbinObject(RouteMap{}, WithPolicyMap(testPolicyMapPath), WithPropertyFile(testPropertyFilePath))
+	authorizer, err := NewCasbinObject(context.Background(), RouteMap{}, WithPolicyMap(testPolicyMapPath), WithPropertyFile(testPropertyFilePath))
 	assert.NoError(t, err)
 
 	// Test that the authorizer is not nil
@@ -56,7 +56,7 @@ func TestCreateAuthorizer(t *testing.T) {
 // TestAuthorize is a test implementation of the Authorize functionality.
 // It tests that the user is authorized correctly for the given request.
 func TestAuthorize(t *testing.T) {
-	authorizer, err := NewCasbinObject(RouteMap{}, WithPolicyMap(testPolicyMapPath), WithPropertyFile(testPropertyFilePath))
+	authorizer, err := NewCasbinObject(context.Background(), RouteMap{}, WithPolicyMap(testPolicyMapPath), WithPropertyFile(testPropertyFilePath))
 	assert.NoError(t, err)
 
 	// Test that the authorizer is not nil
@@ -92,7 +92,7 @@ func TestAuthorize(t *testing.T) {
 // Additionally, it tests that the default policy is applied correctly when a user is not found in the policy map.
 // The default policy is set to "role:readonly" in the test data.
 func TestDefaultPolicy(t *testing.T) {
-	authorizer, err := NewCasbinObject(RouteMap{}, WithPolicyMap(testPolicyMapPath), WithPropertyFile(testPropertyFilePath))
+	authorizer, err := NewCasbinObject(context.Background(), RouteMap{}, WithPolicyMap(testPolicyMapPath), WithPropertyFile(testPropertyFilePath))
 	assert.NoError(t, err)
 
 	// Test that the authorizer is not nil
@@ -119,7 +119,7 @@ func TestDefaultPolicy(t *testing.T) {
 // It tests that the scopes are set correctly when the authorizer is created.
 // Additionally, it tests that the required scopes are tested for the user.
 func TestScopes(t *testing.T) {
-	authorizer, err := NewCasbinObject(RouteMap{}, WithPolicyMap(testPolicyMapPath), WithPropertyFile(testPropertyFilePath))
+	authorizer, err := NewCasbinObject(context.Background(), RouteMap{}, WithPolicyMap(testPolicyMapPath), WithPropertyFile(testPropertyFilePath))
 	assert.NoError(t, err)
 
 	// Test that the authorizer is not nil
@@ -161,7 +161,7 @@ func TestScopes(t *testing.T) {
 // TestNamespaces is a test implementation of the Namespaces based access.
 // It tests that a user can access a namespace that is in the policy map.
 func TestNamespaces(t *testing.T) {
-	authorizer, err := NewCasbinObject(RouteMap{}, WithPolicyMap(testPolicyMapPath), WithPropertyFile(testPropertyFilePath))
+	authorizer, err := NewCasbinObject(context.Background(), RouteMap{}, WithPolicyMap(testPolicyMapPath), WithPropertyFile(testPropertyFilePath))
 	assert.NoError(t, err)
 
 	// Test that the authorizer is not nil
@@ -186,7 +186,7 @@ func TestNamespaces(t *testing.T) {
 func TestConfigFileReload(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
-	authorizer, err := NewCasbinObject(RouteMap{}, WithPolicyMap(testPolicyMapPath), WithPropertyFile(testPropertyReloadFilePath))
+	authorizer, err := NewCasbinObject(context.Background(), RouteMap{}, WithPolicyMap(testPolicyMapPath), WithPropertyFile(testPropertyReloadFilePath))
 	assert.NoError(t, err)
 
 	// Test that the authorizer is not nil
diff --git a/server/routes/routes_test.go b/server/routes/routes_test.go
index 97e1b25f..fd5a1bbb 100644
--- a/server/routes/routes_test.go
+++ b/server/routes/routes_test.go
@@ -17,12 +17,15 @@ limitations under the License.
 package routes
 
 import (
-	"github.com/numaproj/numaflow/server/authz"
 	"net/http"
 	"net/http/httptest"
 	"testing"
 
+	"sigs.k8s.io/controller-runtime/pkg/manager/signals"
+
 	"github.com/gin-gonic/gin"
+	"github.com/numaproj/numaflow/pkg/shared/logging"
+	"github.com/numaproj/numaflow/server/authz"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -30,6 +33,7 @@ import (
 func TestRoutes(t *testing.T) {
 	// skipping this test for the time being
 	t.Skip()
+	log := logging.NewLogger().Named("server")
 	router := gin.Default()
 	managedNamespace := "numaflow-system"
 	namespaced := false
@@ -43,7 +47,7 @@ func TestRoutes(t *testing.T) {
 		DexServerAddr: "test-dex-server-addr",
 	}
 	authRouteMap := authz.RouteMap{}
-	Routes(router, sysInfo, authInfo, "/", authRouteMap)
+	Routes(logging.WithLogger(signals.SetupSignalHandler(), log), router, sysInfo, authInfo, "/", authRouteMap)
 	t.Run("/404", func(t *testing.T) {
 		w := httptest.NewRecorder()
 		req, err := http.NewRequest(http.MethodGet, "/404", nil)