feat(global-router): reserve POST /routers until router registration is designed

lib/global-discovery-service/src/backends/kubernetes.rs

@@ -46,6 +46,7 @@
 //! let store = DiscoveryStore {
 //!     pools: Arc::new(McsPoolDirectory::new(client.clone(), mcs, suffix, true, locations)),
 //!     routers: Arc::new(EndpointSliceReplicaDirectory::imported(client, &router, mcs)),
+//!     router_source: RouterSource::Mcs,
 //!     registrar: None,
 //!     authenticator: Arc::new(TrustedNetworkAuthenticator),
 //! };

lib/global-discovery-service/src/main.rs

@@ -22,7 +22,7 @@ use dynamo_global_discovery_service::backends::memory::{MemoryPoolStore, StaticR
 use dynamo_global_discovery_service::pool_discovery::{
     LeasePolicy, McsImplementation, PoolDirectory, PoolRegistrar, ReplicaEndpoint, ReplicaId,
 };
-use dynamo_global_discovery_service::service::{self, DiscoveryStore};
+use dynamo_global_discovery_service::service::{self, DiscoveryStore, RouterSource};
 use kube::Client;
 use tokio::net::TcpListener;
 use tokio_util::sync::CancellationToken;
@@ -136,6 +136,7 @@ fn memory_store(args: &Args) -> Result<DiscoveryStore> {
     Ok(DiscoveryStore {
         pools,
         routers: Arc::new(StaticReplicaDirectory::new(args.routers.clone())),
+        router_source: RouterSource::Configured,
         registrar: Some(registrar),
         authenticator: Arc::new(TrustedNetworkAuthenticator),
     })
@@ -172,6 +173,7 @@ async fn kubernetes_store(args: &Args, mcs: McsImplementation) -> Result<Discove
         routers: Arc::new(EndpointSliceReplicaDirectory::imported(
             client, &router, mcs,
         )),
+        router_source: RouterSource::Mcs,
         registrar: None,
         authenticator: Arc::new(TrustedNetworkAuthenticator),
     })

lib/global-discovery-service/src/service.rs

@@ -6,6 +6,8 @@
 //! - `POST /pools`: a PoolRelay registers its pool. It calls this on a timer,
 //!   and each call is also the pool's heartbeat.
 //! - `GET /routers`: the Global Router replicas a PoolRelay connects to.
+//! - `POST /routers`: `RegisterRouter` from the design, not available yet.
+//!   It answers `501` with the reason (see [`RouterSource`]).
 //! - `GET /pools`: the pools a Global Router routes to.
 //! - `GET /healthz`: OK while the service can answer from its store. Use it
 //!   for readiness; liveness needs only the TCP port.
@@ -45,16 +47,30 @@ use url::Url;
 pub struct DiscoveryStore {
     pub pools: Arc<PoolDirectory>,
     pub routers: Arc<ReplicaDirectory>,
+    pub router_source: RouterSource,
     /// `None` when pools are announced another way, as with MCS.
     pub registrar: Option<Arc<dyn PoolRegistrar>>,
     /// Names the caller of `POST /pools` from its connection.
     pub authenticator: Arc<dyn RelayAuthenticator>,
 }
 
+/// Where the service finds router replicas. Router replicas cannot register
+/// yet: `POST /routers` answers `501`. With MCS they need not. Without it,
+/// how a router proves it is a router, and what it sends, are not decided.
+/// Until then every caller would be trusted, and a fake router entry would
+/// make every PoolRelay stream its pool's state to it.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum RouterSource {
+    /// `--router` configuration, in the memory store.
+    Configured,
+    /// The router Service's EndpointSlices that MCS imports.
+    Mcs,
+}
+
 pub fn router(store: DiscoveryStore) -> Router {
     Router::new()
         .route("/pools", get(list_pools).post(register_pool))
-        .route("/routers", get(list_routers))
+        .route("/routers", get(list_routers).post(register_router))
         .route("/healthz", get(healthz))
         .with_state(Arc::new(store))
 }
@@ -125,6 +141,22 @@ async fn list_routers(State(store): State<Arc<DiscoveryStore>>) -> Response {
     }
 }
 
+/// `RegisterRouter` from the design. It reads no body: the answer is the
+/// same for any request.
+async fn register_router(State(store): State<Arc<DiscoveryStore>>) -> Response {
+    let reason = match store.router_source {
+        RouterSource::Mcs => {
+            "router replicas need not register with this store: it lists them from the router \
+             Service's EndpointSlices, which MCS imports"
+        }
+        RouterSource::Configured => {
+            "router registration is not implemented yet: how a router proves it is a router, \
+             and what it sends, are not decided. List router replicas with --router"
+        }
+    };
+    (StatusCode::NOT_IMPLEMENTED, reason).into_response()
+}
+
 /// Lists what `GET /pools` and `GET /routers` would serve. A Kubernetes store
 /// answers from its cache once that has synced, so an API server outage does
 /// not turn every replica unready at once. The first probe also starts the
@@ -392,6 +424,7 @@ mod tests {
         DiscoveryStore {
             pools: pools.clone(),
             routers: Arc::new(StaticReplicaDirectory::new(vec![router_replica()])),
+            router_source: RouterSource::Configured,
             registrar: Some(pools),
             authenticator: Arc::new(TrustedNetworkAuthenticator),
         }
@@ -473,6 +506,41 @@ mod tests {
         shutdown.cancel();
     }
 
+    #[tokio::test]
+    async fn routers_cannot_register_and_the_answer_says_why() {
+        let registration =
+            json!({"replica": "aws-west/global-router-0", "url": "http://10.43.0.5:50051"});
+        // The MCS answer must not depend on the body, so that case sends none.
+        for (source, body, reason) in [
+            (RouterSource::Mcs, String::new(), "need not register"),
+            (
+                RouterSource::Configured,
+                registration.to_string(),
+                "not decided",
+            ),
+        ] {
+            let mut store = memory_store();
+            store.router_source = source;
+            let (base, shutdown) = start(store).await;
+            let response = reqwest::Client::new()
+                .post(format!("{base}/routers"))
+                .body(body)
+                .send()
+                .await
+                .unwrap();
+            assert_eq!(
+                response.status(),
+                reqwest::StatusCode::NOT_IMPLEMENTED,
+                "{source:?}"
+            );
+            assert!(
+                response.text().await.unwrap().contains(reason),
+                "{source:?}"
+            );
+            shutdown.cancel();
+        }
+    }
+
     #[tokio::test]
     async fn invalid_key_is_a_bad_request() {
         let (base, shutdown) = start(memory_store()).await;
@@ -525,6 +593,7 @@ mod tests {
         let (base, shutdown) = start(DiscoveryStore {
             pools: Arc::new(Down),
             routers: Arc::new(Down),
+            router_source: RouterSource::Mcs,
             registrar: None,
             authenticator: Arc::new(TrustedNetworkAuthenticator),
         })

lib/global-ingress/tests/round_robin.rs

@@ -18,7 +18,7 @@ use dynamo_global_discovery_client::{
 use dynamo_global_discovery_service::backends::kubernetes::TrustedNetworkAuthenticator;
 use dynamo_global_discovery_service::backends::memory::{MemoryPoolStore, StaticReplicaDirectory};
 use dynamo_global_discovery_service::pool_discovery::LeasePolicy;
-use dynamo_global_discovery_service::service::{self, DiscoveryStore};
+use dynamo_global_discovery_service::service::{self, DiscoveryStore, RouterSource};
 use dynamo_global_router::{DiscoveryPoolCatalog, PoolCatalog};
 use reqwest::StatusCode;
 use tokio_util::sync::CancellationToken;
@@ -33,6 +33,7 @@ async fn discovery_service() -> (Server, GlobalDiscoveryClient) {
     let store = DiscoveryStore {
         pools: pools.clone(),
         routers: Arc::new(StaticReplicaDirectory::new(Vec::new())),
+        router_source: RouterSource::Configured,
         registrar: Some(pools),
         authenticator: Arc::new(TrustedNetworkAuthenticator),
     };