Reshard form: discover all destination shards and known external stores

controlplane/admin/reshard.go

@@ -7,11 +7,15 @@ import (
 	"errors"
 	"net/http"
 	"regexp"
+	"sort"
 	"strconv"
 	"strings"
 
 	"github.com/gin-gonic/gin"
 	"gorm.io/gorm"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
 
 	"github.com/posthog/duckgres/controlplane/configstore"
 )
@@ -37,6 +41,7 @@ type ReshardStore interface {
 	FinishPendingReshardOperation(id int64, state configstore.ReshardState, errMsg string) (bool, error)
 	AppendReshardLog(opID int64, level, message string) error
 	GetManagedWarehouse(orgID string) (*configstore.ManagedWarehouse, error)
+	ListExternalMetadataStores() ([]configstore.ExternalMetadataStoreInfo, error)
 }
 
 // ReshardPasswordStash hands the runner the ephemeral external password for
@@ -73,20 +78,97 @@ type startReshardRequest struct {
 
 // RegisterReshardAPI wires the reshard endpoints. lister may be nil (duckling
 // client unavailable) — starting a reshard then 503s; reads still work.
-// stash may be nil (no local runner) — external targets then 503.
-func RegisterReshardAPI(r *gin.RouterGroup, store ReshardStore, lister DucklingMetadataLister, stash ReshardPasswordStash) {
-	h := &reshardHandler{store: store, lister: lister, stash: stash}
+// stash may be nil (no local runner) — external targets then 503. cluster may
+// be nil (non-k8s) — shard discovery then falls back to the shards tenants
+// already occupy.
+func RegisterReshardAPI(r *gin.RouterGroup, store ReshardStore, lister DucklingMetadataLister, stash ReshardPasswordStash, cluster kubernetes.Interface) {
+	h := &reshardHandler{store: store, lister: lister, stash: stash, cluster: cluster}
 	r.POST("/orgs/:id/reshard", h.start)
 	r.GET("/orgs/:id/reshards", h.listForOrg)
 	r.GET("/reshards/:opid", h.get)
 	r.GET("/reshards/:opid/log", h.log)
+	r.GET("/reshards/targets", h.targets)
 	r.POST("/reshards/:opid/cancel", h.cancel)
 }
 
 type reshardHandler struct {
-	store  ReshardStore
-	lister DucklingMetadataLister
-	stash  ReshardPasswordStash
+	store   ReshardStore
+	lister  DucklingMetadataLister
+	stash   ReshardPasswordStash
+	cluster kubernetes.Interface
+}
+
+// cnpgShardsNamespace is where the shared CNPG metadata shards run; instance
+// pods carry the operator's cnpg.io/cluster=<shard> label.
+const cnpgShardsNamespace = "cnpg-shards"
+
+// targets returns everything the reshard form can offer as a destination:
+// every cnpg shard (including EMPTY ones no tenant occupies yet — discovered
+// from the CNPG instance pods via the cluster-topology read the Nodes view
+// already uses; an RBAC Forbidden degrades to the shards tenants occupy, read
+// from the duckling statuses) and every external metadata store a live
+// warehouse references (endpoint + SM secret NAME only — never a password).
+func (h *reshardHandler) targets(c *gin.Context) {
+	ctx, cancel := context.WithTimeout(c.Request.Context(), ducklingMetadataTimeout)
+	defer cancel()
+
+	shardSet := map[string]struct{}{}
+	clusterAvailable := false
+	if h.cluster != nil {
+		pods, err := h.cluster.CoreV1().Pods(cnpgShardsNamespace).List(ctx, metav1.ListOptions{
+			LabelSelector: "cnpg.io/cluster",
+		})
+		switch {
+		case err == nil:
+			clusterAvailable = true
+			for i := range pods.Items {
+				if shard := pods.Items[i].Labels["cnpg.io/cluster"]; shard != "" {
+					shardSet[shard] = struct{}{}
+				}
+			}
+		case apierrors.IsForbidden(err):
+			// Same degrade contract as the cluster topology endpoints: the
+			// e2e CP has no cluster-scoped RBAC; fall back to occupied shards.
+		default:
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+			return
+		}
+	}
+	if h.lister != nil {
+		stores, err := h.lister.CRMetadataStores(ctx)
+		if err == nil {
+			for _, ms := range stores {
+				if ms.Kind == configstore.MetadataStoreKindCnpgShard {
+					if shard := cnpgShardFromEndpoint(ms.Endpoint); shard != "" {
+						shardSet[shard] = struct{}{}
+					}
+				}
+			}
+		}
+	}
+	shards := make([]string, 0, len(shardSet))
+	for s := range shardSet {
+		shards = append(shards, s)
+	}
+	sort.Strings(shards)
+
+	external, err := h.store.ListExternalMetadataStores()
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	if external == nil {
+		external = []configstore.ExternalMetadataStoreInfo{}
+	}
+
+	c.JSON(http.StatusOK, gin.H{
+		"shards": shards,
+		// cluster_discovery=false means the shard list only contains shards
+		// tenants already occupy (RBAC degrade / non-k8s) — a brand-new empty
+		// shard would be missing and needs manual entry.
+		"cluster_discovery": clusterAvailable,
+		"external_stores":   external,
+	})
 }
 
 func (h *reshardHandler) start(c *gin.Context) {

controlplane/admin/reshard_test.go

@@ -13,6 +13,14 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"gorm.io/gorm"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/kubernetes"
+	k8sfake "k8s.io/client-go/kubernetes/fake"
+	k8stesting "k8s.io/client-go/testing"
 
 	"github.com/posthog/duckgres/controlplane/configstore"
 )
@@ -113,6 +121,12 @@ func (f *fakeReshardStore) GetManagedWarehouse(orgID string) (*configstore.Manag
 	return &cp, nil
 }
 
+func (f *fakeReshardStore) ListExternalMetadataStores() ([]configstore.ExternalMetadataStoreInfo, error) {
+	return []configstore.ExternalMetadataStoreInfo{
+		{Endpoint: "known.rds.example.com", PasswordAWSSecret: "known-secret", User: "postgres", Database: "postgres"},
+	}, nil
+}
+
 func (f *fakeReshardStore) StashExternalPassword(opID int64, password string) {
 	if f.stashed == nil {
 		f.stashed = map[int64]string{}
@@ -129,12 +143,16 @@ func (f fakeShardLister) CRMetadataStores(context.Context) (map[string]DucklingM
 }
 
 func reshardRouter(store *fakeReshardStore) *gin.Engine {
+	return reshardRouterWithCluster(store, nil)
+}
+
+func reshardRouterWithCluster(store *fakeReshardStore, cluster kubernetes.Interface) *gin.Engine {
 	gin.SetMode(gin.TestMode)
 	r := gin.New()
 	lister := fakeShardLister{stores: map[string]DucklingMetadataStore{
 		"acme": {Kind: "cnpg-shard", Endpoint: "shard-001-pooler.cnpg-shards.svc.cluster.local"},
 	}}
-	RegisterReshardAPI(r.Group("/api/v1"), store, lister, store)
+	RegisterReshardAPI(r.Group("/api/v1"), store, lister, store, cluster)
 	return r
 }
 
@@ -303,3 +321,70 @@ func TestReshardGetListLogCancel(t *testing.T) {
 func itoa(n int64) string {
 	return strconv.FormatInt(n, 10)
 }
+
+func cnpgPod(name, shard string) *corev1.Pod {
+	return &corev1.Pod{ObjectMeta: metav1.ObjectMeta{
+		Name:      name,
+		Namespace: cnpgShardsNamespace,
+		Labels:    map[string]string{"cnpg.io/cluster": shard},
+	}}
+}
+
+type reshardTargets struct {
+	Shards           []string                                `json:"shards"`
+	ClusterDiscovery bool                                    `json:"cluster_discovery"`
+	ExternalStores   []configstore.ExternalMetadataStoreInfo `json:"external_stores"`
+}
+
+// TestReshardTargetsClusterDiscovery pins the load-bearing property of the
+// endpoint: an EMPTY shard (no tenant on it yet) shows up, discovered from the
+// CNPG instance pods, unioned with the occupied shards from duckling statuses.
+func TestReshardTargetsClusterDiscovery(t *testing.T) {
+	cs := k8sfake.NewSimpleClientset(
+		cnpgPod("shard-001-1", "shard-001"),
+		cnpgPod("shard-002-1", "shard-002"), // empty shard: no duckling on it
+	)
+	w := doJSON(reshardRouterWithCluster(newFakeReshardStore(), cs), http.MethodGet, "/api/v1/reshards/targets", "")
+	if w.Code != http.StatusOK {
+		t.Fatalf("status = %d body %s", w.Code, w.Body.String())
+	}
+	var resp reshardTargets
+	if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+	if !resp.ClusterDiscovery {
+		t.Fatal("cluster_discovery = false, want true")
+	}
+	if strings.Join(resp.Shards, ",") != "shard-001,shard-002" {
+		t.Fatalf("shards = %v, want [shard-001 shard-002]", resp.Shards)
+	}
+	if len(resp.ExternalStores) != 1 || resp.ExternalStores[0].Endpoint != "known.rds.example.com" {
+		t.Fatalf("external stores = %+v", resp.ExternalStores)
+	}
+}
+
+// TestReshardTargetsDegrades pins the fallbacks: an RBAC Forbidden (the e2e
+// CP) degrades to the occupied shards with cluster_discovery=false, and a nil
+// cluster client behaves the same.
+func TestReshardTargetsDegrades(t *testing.T) {
+	cs := k8sfake.NewSimpleClientset()
+	cs.PrependReactor("list", "pods", func(k8stesting.Action) (bool, runtime.Object, error) {
+		return true, nil, apierrors.NewForbidden(schema.GroupResource{Resource: "pods"}, "", nil)
+	})
+	for _, cluster := range []kubernetes.Interface{cs, nil} {
+		w := doJSON(reshardRouterWithCluster(newFakeReshardStore(), cluster), http.MethodGet, "/api/v1/reshards/targets", "")
+		if w.Code != http.StatusOK {
+			t.Fatalf("status = %d body %s", w.Code, w.Body.String())
+		}
+		var resp reshardTargets
+		if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
+			t.Fatalf("unmarshal: %v", err)
+		}
+		if resp.ClusterDiscovery {
+			t.Fatal("cluster_discovery = true, want false on degrade")
+		}
+		if strings.Join(resp.Shards, ",") != "shard-001" {
+			t.Fatalf("shards = %v, want the occupied [shard-001] fallback", resp.Shards)
+		}
+	}
+}

controlplane/admin/ui/src/hooks/useApi.ts

@@ -38,6 +38,7 @@ import type {
   QueryDetail,
   ReshardLogEntry,
   ReshardOperation,
+  ReshardTargetsResponse,
   RunningQuery,
   SessionStatus,
   StartReshardBody,
@@ -527,6 +528,25 @@ export function useReshardLog(opId: number | null, opState: string | undefined)
   return entries;
 }
 
+const EMPTY_RESHARD_TARGETS: ReshardTargetsResponse = {
+  shards: [],
+  cluster_discovery: false,
+  external_stores: [],
+};
+
+// Destination discovery for the reshard form. Pre-rollout backends without
+// the endpoint degrade to empty (the form still allows manual entry).
+export function useReshardTargets() {
+  return useQuery<ReshardTargetsResponse>({
+    queryKey: ["reshardTargets"],
+    queryFn: () =>
+      tolerateStatus<ReshardTargetsResponse>(EMPTY_RESHARD_TARGETS, 403, 404, 503)(
+        api.getReshardTargets(),
+      ).catch(() => EMPTY_RESHARD_TARGETS),
+    refetchInterval: POLL.slow,
+  });
+}
+
 export function useCancelReshard() {
   const qc = useQueryClient();
   return useMutation({

controlplane/admin/ui/src/lib/api.ts

@@ -32,6 +32,7 @@ import type {
   QueryResult,
   ReshardLogEntry,
   ReshardOperation,
+  ReshardTargetsResponse,
   RunningQuery,
   SessionStatus,
   StartReshardBody,
@@ -237,4 +238,7 @@ export const api = {
     ),
   cancelReshard: (opId: number) =>
     post<{ cancel_requested?: boolean; state?: string }>(`/reshards/${opId}/cancel`, {}),
+  // Destination discovery for the reshard form: all cnpg shards (incl. empty
+  // ones) + known external stores.
+  getReshardTargets: () => get<ReshardTargetsResponse>("/reshards/targets"),
 };

controlplane/admin/ui/src/pages/ReshardForm.tsx

@@ -24,7 +24,7 @@ import {
   SelectValue,
 } from "@/components/ui/select";
 import { ducklingEntryFor } from "@/lib/format";
-import { useDucklingsMetadata, useStartReshard, useWarehouse } from "@/hooks/useApi";
+import { useDucklingsMetadata, useReshardTargets, useStartReshard, useWarehouse } from "@/hooks/useApi";
 import type { StartReshardBody } from "@/types/api";
 
 // Reshard start form: pick a target metadata store for the org, confirm, run.
@@ -55,15 +55,20 @@ export function ReshardForm() {
   const sourceKind = entry?.kind ?? warehouse.data?.metadata_store?.kind ?? "cnpg-shard";
   const currentShard = entry?.cnpg_shard ?? "";
 
-  // Known shards: the distinct set the fleet is already using. Free-text
-  // covers a brand-new shard no tenant landed on yet.
+  // Destination discovery: every cnpg shard in the cluster (including EMPTY
+  // ones no tenant occupies yet) + known external stores, from
+  // /reshards/targets. Unioned with the shards tenants occupy (duckling
+  // metadata) as a belt-and-suspenders fallback; free-text still covers a
+  // shard the server can't see (cluster_discovery=false on an RBAC degrade).
+  const targets = useReshardTargets();
   const knownShards = useMemo(() => {
-    const s = new Set<string>();
+    const s = new Set<string>(targets.data?.shards ?? []);
     for (const e of Object.values(metadata.data?.entries ?? {})) {
       if (e.kind === "cnpg-shard" && e.cnpg_shard) s.add(e.cnpg_shard);
     }
     return [...s].sort();
-  }, [metadata.data]);
+  }, [targets.data, metadata.data]);
+  const knownExternal = targets.data?.external_stores ?? [];
 
   const effectiveShard = shard === "__custom__" ? shardCustom.trim() : shard;
   const targetLabel =
@@ -180,6 +185,12 @@ export function ReshardForm() {
                     </SelectContent>
                   </Select>
                 </div>
+                {targets.data && !targets.data.cluster_discovery && (
+                  <p className="col-span-2 text-xs text-muted-foreground">
+                    Shard list limited to shards tenants already occupy (no cluster read access) —
+                    use "other…" for a new empty shard.
+                  </p>
+                )}
                 {shard === "__custom__" && (
                   <div className="space-y-1">
                     <Label>Shard name</Label>
@@ -194,6 +205,36 @@ export function ReshardForm() {
               </div>
             ) : (
               <div className="space-y-3">
+                {knownExternal.length > 0 && (
+                  <div className="space-y-1">
+                    <Label>Known external stores</Label>
+                    <Select
+                      onValueChange={(v) => {
+                        const st = knownExternal[Number(v)];
+                        if (!st) return;
+                        setEndpoint(st.endpoint);
+                        setPasswordSecret(st.password_aws_secret);
+                        setUser(st.user || "postgres");
+                        setDatabase(st.database || "postgres");
+                      }}
+                    >
+                      <SelectTrigger>
+                        <SelectValue placeholder="Prefill from a store already in use…" />
+                      </SelectTrigger>
+                      <SelectContent>
+                        {knownExternal.map((st, i) => (
+                          <SelectItem key={st.endpoint + st.database} value={String(i)}>
+                            {st.endpoint}/{st.database || "postgres"}
+                          </SelectItem>
+                        ))}
+                      </SelectContent>
+                    </Select>
+                    <p className="text-xs text-muted-foreground">
+                      Prefills endpoint/secret/user/database from an external store another
+                      warehouse already uses — the password below is still required.
+                    </p>
+                  </div>
+                )}
                 <div className="grid grid-cols-2 gap-3">
                   <div className="space-y-1">
                     <Label>Endpoint (RDS host)</Label>