redis/valkey: get endpoints for any namespace/name

routesrv/redishandler.go

@@ -14,7 +14,9 @@ import (
 )
 
 type RedisHandler struct {
-	AddrUpdater func() ([]byte, error)
+	AddrUpdater      func(namespace, name string) ([]byte, error)
+	DefaultNamespace string
+	DefaultName      string
 }
 
 type RedisEndpoint struct {
@@ -30,8 +32,25 @@ func (rh *RedisHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(http.StatusMethodNotAllowed)
 		return
 	}
+
+	ns := r.PathValue("namespace")
+	name := r.PathValue("name")
+
+	// fall back to operator-configured defaults (backwards-compatible path)
+	if ns == "" {
+		ns = rh.DefaultNamespace
+	}
+	if name == "" {
+		name = rh.DefaultName
+	}
+
+	if ns == "" || name == "" {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
 	w.Header().Add("Content-Type", "application/json")
-	address, err := rh.AddrUpdater()
+	address, err := rh.AddrUpdater(ns, name)
 	if err != nil {
 		log.Errorf("Failed to update address for redis handler %v", err)
 		w.WriteHeader(http.StatusInternalServerError)
@@ -40,10 +59,10 @@ func (rh *RedisHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	w.Write(address)
 }
 
-func getRedisAddresses(opts *skipper.Options, kdc *kubernetes.Client, m metrics.Metrics) func() ([]byte, error) {
-	return func() ([]byte, error) {
-		a := kdc.GetEndpointAddresses("", opts.KubernetesRedisServiceNamespace, opts.KubernetesRedisServiceName)
-		log.Debugf("Redis updater called and found %d redis endpoints: %v", len(a), a)
+func getRedisAddresses(opts *skipper.Options, kdc *kubernetes.Client, m metrics.Metrics) func(namespace, name string) ([]byte, error) {
+	return func(namespace, name string) ([]byte, error) {
+		a := kdc.GetEndpointAddresses("", namespace, name)
+		log.Debugf("Redis updater called for %s/%s and found %d redis endpoints: %v", namespace, name, len(a), a)
 		m.UpdateGauge("redis_endpoints", float64(len(a)))
 
 		result := RedisEndpoints{

routesrv/routesrv.go

@@ -111,30 +111,40 @@ func New(opts skipper.Options) (*RouteServer, error) {
 		oauthConfig.CallbackPath = opts.OAuth2CallbackPath
 	}
 
-	var rh *RedisHandler
 	// in case we have kubernetes dataclient and we can detect redis instances, we patch redisOptions
 	if opts.KubernetesRedisServiceNamespace != "" && opts.KubernetesRedisServiceName != "" {
 		log.Infof("Use endpoints %s/%s to fetch updated redis shards", opts.KubernetesRedisServiceNamespace, opts.KubernetesRedisServiceName)
-		rh = &RedisHandler{}
 		_, err := dataclient.LoadAll()
 		if err != nil {
 			return nil, err
 		}
-		rh.AddrUpdater = getRedisAddresses(&opts, dataclient, m)
+		rh := &RedisHandler{
+			AddrUpdater:      getRedisAddresses(&opts, dataclient, m),
+			DefaultNamespace: opts.KubernetesRedisServiceNamespace,
+			DefaultName:      opts.KubernetesRedisServiceName,
+		}
+		// backwards-compatible static path
 		mux.Handle("/swarm/redis/shards", rh)
+		// dynamic path: any namespace/service reachable by this routesrv's service account
+		mux.Handle("/swarm/redis/shards/{namespace}/{name}", rh)
 	}
 
-	var vh *ValkeyHandler
 	// in case we have kubernetes dataclient and we can detect valkey instances, we patch valkeyOptions
 	if opts.KubernetesValkeyServiceNamespace != "" && opts.KubernetesValkeyServiceName != "" {
 		log.Infof("Use endpoints %s/%s to fetch updated valkey shards", opts.KubernetesValkeyServiceNamespace, opts.KubernetesValkeyServiceName)
-		vh = &ValkeyHandler{}
 		_, err := dataclient.LoadAll()
 		if err != nil {
 			return nil, err
 		}
-		vh.AddrUpdater = getValkeyAddresses(&opts, dataclient, m)
+		vh := &ValkeyHandler{
+			AddrUpdater:      getValkeyAddresses(&opts, dataclient, m),
+			DefaultNamespace: opts.KubernetesValkeyServiceNamespace,
+			DefaultName:      opts.KubernetesValkeyServiceName,
+		}
+		// backwards-compatible static path
 		mux.Handle("/swarm/valkey/shards", vh)
+		// dynamic path: any namespace/service reachable by this routesrv's service account
+		mux.Handle("/swarm/valkey/shards/{namespace}/{name}", vh)
 	}
 
 	rs.server = &http.Server{

routesrv/routesrv_test.go

@@ -170,6 +170,14 @@ func getRedisURLs(rs *routesrv.RouteServer) *httptest.ResponseRecorder {
 	return w
 }
 
+func getRedisURLsForNamespace(rs *routesrv.RouteServer, namespace, name string) *httptest.ResponseRecorder {
+	w := httptest.NewRecorder()
+	r := httptest.NewRequest("GET", "/swarm/redis/shards/"+namespace+"/"+name, nil)
+	rs.ServeHTTP(w, r)
+
+	return w
+}
+
 func getValkeyURLs(rs *routesrv.RouteServer) *httptest.ResponseRecorder {
 	w := httptest.NewRecorder()
 	r := httptest.NewRequest("GET", "/swarm/valkey/shards", nil)
@@ -178,6 +186,14 @@ func getValkeyURLs(rs *routesrv.RouteServer) *httptest.ResponseRecorder {
 	return w
 }
 
+func getValkeyURLsForNamespace(rs *routesrv.RouteServer, namespace, name string) *httptest.ResponseRecorder {
+	w := httptest.NewRecorder()
+	r := httptest.NewRequest("GET", "/swarm/valkey/shards/"+namespace+"/"+name, nil)
+	rs.ServeHTTP(w, r)
+
+	return w
+}
+
 func headRoutes(rs *routesrv.RouteServer) *httptest.ResponseRecorder {
 	w := httptest.NewRecorder()
 	r := httptest.NewRequest("HEAD", "/routes", nil)
@@ -1426,6 +1442,76 @@ func TestZoneAwareFallbackToAllRoutesWhenBelowThreshold(t *testing.T) {
 	assert.Equal(t, fullCount, gzipResp.Header().Get(routing.RoutesCountName))
 }
 
+func TestRedisEndpointsDynamicNamespace(t *testing.T) {
+	defer tl.Reset()
+	ks, _ := newKubeServer(t, loadKubeYAML(t, "testdata/redis-multi-namespace.yaml"))
+	ks.Start()
+	defer ks.Close()
+	rs := newRouteServerWithOptions(t, skipper.Options{
+		SourcePollTimeout:               pollInterval,
+		Kubernetes:                      true,
+		KubernetesURL:                   ks.URL,
+		KubernetesRedisServiceNamespace: "namespace1",
+		KubernetesRedisServiceName:      "service1",
+		KubernetesRedisServicePort:      6379,
+	})
+
+	t.Run("static path returns default namespace endpoints", func(t *testing.T) {
+		w := getRedisURLs(rs)
+		wantHTTPCode(t, w, http.StatusOK)
+		want := parseIP(t, "testdata/redis-ip.json")
+		assert.JSONEq(t, string(want), w.Body.String())
+	})
+
+	t.Run("dynamic path returns endpoints for requested namespace", func(t *testing.T) {
+		w := getRedisURLsForNamespace(rs, "namespace2", "service2")
+		wantHTTPCode(t, w, http.StatusOK)
+		want := parseIP(t, "testdata/redis-ip-namespace2.json")
+		assert.JSONEq(t, string(want), w.Body.String())
+	})
+
+	t.Run("dynamic path with unknown namespace returns empty endpoints", func(t *testing.T) {
+		w := getRedisURLsForNamespace(rs, "does-not-exist", "service1")
+		wantHTTPCode(t, w, http.StatusOK)
+		assert.JSONEq(t, `{"endpoints":[]}`, w.Body.String())
+	})
+}
+
+func TestValkeyEndpointsDynamicNamespace(t *testing.T) {
+	defer tl.Reset()
+	ks, _ := newKubeServer(t, loadKubeYAML(t, "testdata/valkey-multi-namespace.yaml"))
+	ks.Start()
+	defer ks.Close()
+	rs := newRouteServerWithOptions(t, skipper.Options{
+		SourcePollTimeout:                pollInterval,
+		Kubernetes:                       true,
+		KubernetesURL:                    ks.URL,
+		KubernetesValkeyServiceNamespace: "namespace1",
+		KubernetesValkeyServiceName:      "service1",
+		KubernetesValkeyServicePort:      6379,
+	})
+
+	t.Run("static path returns default namespace endpoints", func(t *testing.T) {
+		w := getValkeyURLs(rs)
+		wantHTTPCode(t, w, http.StatusOK)
+		want := parseIP(t, "testdata/valkey-ip.json")
+		assert.JSONEq(t, string(want), w.Body.String())
+	})
+
+	t.Run("dynamic path returns endpoints for requested namespace", func(t *testing.T) {
+		w := getValkeyURLsForNamespace(rs, "namespace2", "service2")
+		wantHTTPCode(t, w, http.StatusOK)
+		want := parseIP(t, "testdata/valkey-ip-namespace2.json")
+		assert.JSONEq(t, string(want), w.Body.String())
+	})
+
+	t.Run("dynamic path with unknown namespace returns empty endpoints", func(t *testing.T) {
+		w := getValkeyURLsForNamespace(rs, "does-not-exist", "service1")
+		wantHTTPCode(t, w, http.StatusOK)
+		assert.JSONEq(t, `{"endpoints":[]}`, w.Body.String())
+	})
+}
+
 func TestXCountShouldBeSameForZoneAwareAndZoneUnawareRoutes(t *testing.T) {
 	defer tl.Reset()
 

routesrv/testdata/redis-ip-namespace2.json

@@ -0,0 +1 @@
+{"endpoints":[{"address":"10.3.1.1:6379"}]}

routesrv/testdata/redis-multi-namespace.yaml

@@ -0,0 +1,51 @@
+apiVersion: v1
+kind: Service
+metadata:
+  namespace: namespace1
+  name: service1
+spec:
+  ports:
+  - port: 6379
+    protocol: TCP
+    targetPort: 6379
+  type: ClusterIP
+---
+apiVersion: v1
+kind: Endpoints
+metadata:
+  namespace: namespace1
+  name: service1
+subsets:
+- addresses:
+  - hostname: skipper-ingress-redis-1
+    ip: 10.2.5.7
+  - hostname: skipper-ingress-redis-0
+    ip: 10.2.55.7
+  ports:
+  - port: 6379
+    protocol: TCP
+---
+apiVersion: v1
+kind: Service
+metadata:
+  namespace: namespace2
+  name: service2
+spec:
+  ports:
+  - port: 6379
+    protocol: TCP
+    targetPort: 6379
+  type: ClusterIP
+---
+apiVersion: v1
+kind: Endpoints
+metadata:
+  namespace: namespace2
+  name: service2
+subsets:
+- addresses:
+  - hostname: skipper-ingress-redis-2
+    ip: 10.3.1.1
+  ports:
+  - port: 6379
+    protocol: TCP

routesrv/testdata/valkey-ip-namespace2.json

@@ -0,0 +1 @@
+{"endpoints":[{"address":"10.3.1.1:6379"}]}

routesrv/testdata/valkey-multi-namespace.yaml

@@ -0,0 +1,51 @@
+apiVersion: v1
+kind: Service
+metadata:
+  namespace: namespace1
+  name: service1
+spec:
+  ports:
+  - port: 6379
+    protocol: TCP
+    targetPort: 6379
+  type: ClusterIP
+---
+apiVersion: v1
+kind: Endpoints
+metadata:
+  namespace: namespace1
+  name: service1
+subsets:
+- addresses:
+  - hostname: skipper-ingress-valkey-1
+    ip: 10.2.5.7
+  - hostname: skipper-ingress-valkey-0
+    ip: 10.2.55.7
+  ports:
+  - port: 6379
+    protocol: TCP
+---
+apiVersion: v1
+kind: Service
+metadata:
+  namespace: namespace2
+  name: service2
+spec:
+  ports:
+  - port: 6379
+    protocol: TCP
+    targetPort: 6379
+  type: ClusterIP
+---
+apiVersion: v1
+kind: Endpoints
+metadata:
+  namespace: namespace2
+  name: service2
+subsets:
+- addresses:
+  - hostname: skipper-ingress-valkey-2
+    ip: 10.3.1.1
+  ports:
+  - port: 6379
+    protocol: TCP

routesrv/valkeyhandler.go

@@ -14,7 +14,9 @@ import (
 )
 
 type ValkeyHandler struct {
-	AddrUpdater func() ([]byte, error)
+	AddrUpdater      func(namespace, name string) ([]byte, error)
+	DefaultNamespace string
+	DefaultName      string
 }
 
 type ValkeyEndpoint struct {
@@ -30,8 +32,25 @@ func (vh *ValkeyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(http.StatusMethodNotAllowed)
 		return
 	}
+
+	ns := r.PathValue("namespace")
+	name := r.PathValue("name")
+
+	// fall back to operator-configured defaults (backwards-compatible path)
+	if ns == "" {
+		ns = vh.DefaultNamespace
+	}
+	if name == "" {
+		name = vh.DefaultName
+	}
+
+	if ns == "" || name == "" {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
 	w.Header().Add("Content-Type", "application/json")
-	address, err := vh.AddrUpdater()
+	address, err := vh.AddrUpdater(ns, name)
 	if err != nil {
 		log.Errorf("Failed to update address for valkey handler %v", err)
 		w.WriteHeader(http.StatusInternalServerError)
@@ -40,10 +59,10 @@ func (vh *ValkeyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	w.Write(address)
 }
 
-func getValkeyAddresses(opts *skipper.Options, kdc *kubernetes.Client, m metrics.Metrics) func() ([]byte, error) {
-	return func() ([]byte, error) {
-		a := kdc.GetEndpointAddresses("", opts.KubernetesValkeyServiceNamespace, opts.KubernetesValkeyServiceName)
-		log.Debugf("Valkey updater called and found %d valkey endpoints: %v", len(a), a)
+func getValkeyAddresses(opts *skipper.Options, kdc *kubernetes.Client, m metrics.Metrics) func(namespace, name string) ([]byte, error) {
+	return func(namespace, name string) ([]byte, error) {
+		a := kdc.GetEndpointAddresses("", namespace, name)
+		log.Debugf("Valkey updater called for %s/%s and found %d valkey endpoints: %v", namespace, name, len(a), a)
 		m.UpdateGauge("valkey_endpoints", float64(len(a)))
 
 		result := ValkeyEndpoints{