feat(etcd): Version the etcd keys (#3458)

Signed-off-by: Graham King <grahamk@nvidia.com>

lib/bindings/python/rust/lib.rs

@@ -594,7 +594,7 @@ fn bind_tcp_port(port: u16) -> std::io::Result<socket2::Socket> {
 }
 
 fn make_port_key(namespace: &str, node_ip: IpAddr, port: u16) -> anyhow::Result<String> {
-    Ok(format!("dyn://{namespace}/ports/{node_ip}/{port}"))
+    Ok(format!("v1/{namespace}/ports/{node_ip}/{port}"))
 }
 
 fn local_ip() -> Result<IpAddr, local_ip_address::Error> {

lib/bindings/python/rust/planner.rs

@@ -102,7 +102,7 @@ impl VirtualConnectorCoordinator {
         let prefix = root_key(&self.0.namespace);
         let inner = self.0.clone();
         pyo3_async_runtimes::tokio::future_into_py(py, async move {
-            let kv_cache = KvCache::new(inner.etcd_client.clone(), prefix, HashMap::new())
+            let kv_cache = KvCache::new(inner.etcd_client.clone(), "v1", prefix, HashMap::new())
                 .await
                 .map_err(to_pyerr)?;
             *inner.kv_cache.lock() = Some(Arc::new(kv_cache));
@@ -497,5 +497,5 @@ fn load(a: &AtomicUsize) -> usize {
 }
 
 fn root_key(namespace: &str) -> String {
-    format!("/{namespace}/planner/")
+    format!("{namespace}/planner/")
 }

lib/llm/src/discovery.rs

@@ -8,4 +8,4 @@ mod watcher;
 pub use watcher::{ModelUpdate, ModelWatcher};
 
 /// The root etcd path for KV Router registrations
-pub const KV_ROUTERS_ROOT_PATH: &str = "kv_routers";
+pub const KV_ROUTERS_ROOT_PATH: &str = "v1/kv_routers";

lib/llm/src/discovery/watcher.rs

@@ -318,7 +318,6 @@ impl ModelWatcher {
                 namespace = endpoint_id.namespace,
                 "New endpoint for existing model"
             );
-            //self.notify_on_model.notify_waiters();
             return Ok(());
         }
 
@@ -413,7 +412,7 @@ impl ModelWatcher {
             .await?;
             let engine = Arc::new(push_router);
             self.manager
-                .add_embeddings_model(&model_entry.name, engine)?;
+                .add_embeddings_model(card.name(), checksum, engine)?;
         } else if card.model_input == ModelInput::Text && card.model_type.supports_chat() {
             // Case 3: Text + Chat
             let push_router = PushRouter::<
@@ -560,26 +559,20 @@ impl ModelWatcher {
 /// The ModelDeploymentCard is published in etcd with a key like "v1/mdc/dynamo/backend/generate/694d9981145a61ad".
 /// Extract the EndpointId and instance_id from that.
 fn etcd_key_extract(s: &str) -> anyhow::Result<(EndpointId, String)> {
+    if !s.starts_with(model_card::ROOT_PATH) {
+        anyhow::bail!("Invalid format: expected model card ROOT_PATH segment in {s}");
+    }
     let parts: Vec<&str> = s.split('/').collect();
-    let start_idx = if !parts.is_empty() && parts[0] == "v1" {
-        1
-    } else {
-        0
-    };
 
-    // Need at least prefix model_card::ROOT_PATH + 3 parts: namespace, component, name
-    if parts.len() <= start_idx + 3 {
+    // Need at least prefix model_card::ROOT_PATH (2 parts) + namespace, component, name (3 parts)
+    if parts.len() <= 5 {
         anyhow::bail!("Invalid format: not enough path segments in {s}");
     }
 
-    if parts.get(start_idx) != Some(&model_card::ROOT_PATH) {
-        anyhow::bail!("Invalid format: expected model card ROOT_PATH segment in {s}");
-    }
-
     let endpoint_id = EndpointId {
-        namespace: parts[start_idx + 1].to_string(),
-        component: parts[start_idx + 2].to_string(),
-        name: parts[start_idx + 3].to_string(),
+        namespace: parts[2].to_string(),
+        component: parts[3].to_string(),
+        name: parts[4].to_string(),
     };
     Ok((endpoint_id, parts[parts.len() - 1].to_string()))
 }
@@ -590,16 +583,6 @@ mod tests {
 
     #[test]
     fn test_etcd_key_extract() {
-        let input = format!(
-            "v1/{}/dynamo/backend/generate/694d9981145a61ad",
-            model_card::ROOT_PATH
-        );
-        let (endpoint_id, instance_id) = etcd_key_extract(&input).unwrap();
-        assert_eq!(endpoint_id.namespace, "dynamo");
-        assert_eq!(endpoint_id.component, "backend");
-        assert_eq!(endpoint_id.name, "generate");
-        assert_eq!(instance_id, "694d9981145a61ad");
-
         let input = format!(
             "{}/dynamo/backend/generate/694d9981145a61ad",
             model_card::ROOT_PATH

lib/llm/src/model_card.rs

@@ -34,7 +34,7 @@ use crate::gguf::{Content, ContentConfig, ModelConfigLike};
 use crate::protocols::TokenIdType;
 
 /// Identify model deployment cards in the key-value store
-pub const ROOT_PATH: &str = "mdc";
+pub const ROOT_PATH: &str = "v1/mdc";
 
 #[derive(Serialize, Deserialize, Clone, Debug)]
 #[serde(rename_all = "snake_case")]

lib/runtime/src/component.rs

@@ -34,7 +34,7 @@ use crate::{
     discovery::Lease,
     metrics::{MetricsRegistry, prometheus_names},
     service::ServiceSet,
-    transports::etcd::EtcdPath,
+    transports::etcd::{ETCD_ROOT_PATH, EtcdPath},
 };
 
 use super::{
@@ -72,10 +72,7 @@ pub use client::{Client, InstanceSource};
 
 /// The root etcd path where each instance registers itself in etcd.
 /// An instance is namespace+component+endpoint+lease_id and must be unique.
-pub const INSTANCE_ROOT_PATH: &str = "instances";
-
-/// The root etcd path where each namespace is registered in etcd.
-pub const ETCD_ROOT_PATH: &str = "dynamo://";
+pub const INSTANCE_ROOT_PATH: &str = "v1/instances";
 
 #[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
 #[serde(rename_all = "snake_case")]
@@ -601,7 +598,7 @@ impl Namespace {
     }
 
     pub fn etcd_path(&self) -> String {
-        format!("{}{}", ETCD_ROOT_PATH, self.name())
+        format!("{ETCD_ROOT_PATH}{}", self.name())
     }
 
     pub fn name(&self) -> String {

lib/runtime/src/storage/key_value_store.rs

@@ -252,7 +252,7 @@ mod tests {
     use super::*;
     use futures::{StreamExt, pin_mut};
 
-    const BUCKET_NAME: &str = "mdc";
+    const BUCKET_NAME: &str = "v1/mdc";
 
     /// Convert the value returned by `watch()` into a broadcast stream that multiple
     /// clients can listen to.

lib/runtime/src/storage/key_value_store/etcd.rs

@@ -5,7 +5,7 @@ use std::collections::HashMap;
 use std::pin::Pin;
 use std::time::Duration;
 
-use crate::{slug::Slug, storage::key_value_store::Key, transports::etcd::Client};
+use crate::{storage::key_value_store::Key, transports::etcd::Client};
 use async_stream::stream;
 use async_trait::async_trait;
 use etcd_client::{Compare, CompareOp, EventType, PutOptions, Txn, TxnOp, WatchOptions};
@@ -240,7 +240,7 @@ impl EtcdBucket {
 }
 
 fn make_key(bucket_name: &str, key: &Key) -> String {
-    [Slug::slugify(bucket_name).to_string(), key.to_string()].join("/")
+    [bucket_name.to_string(), key.to_string()].join("/")
 }
 
 #[cfg(feature = "integration")]

lib/runtime/src/transports/etcd.rs

@@ -28,8 +28,6 @@ pub use path::*;
 
 use super::utils::build_in_runtime;
 
-//pub use etcd::ConnectOptions as EtcdConnectOptions;
-
 /// ETCD Client
 #[derive(Clone)]
 pub struct Client {
@@ -517,11 +515,14 @@ impl KvCache {
     /// Create a new KV cache for the given prefix
     pub async fn new(
         client: Client,
+        version: &str,
         prefix: String,
         initial_values: HashMap<String, Vec<u8>>,
     ) -> Result<Self> {
         let mut cache = HashMap::new();
 
+        let prefix = format!("{version}/{prefix}");
+
         // First get all existing keys with this prefix
         let existing_kvs = client.kv_get_prefix(&prefix).await?;
         for kv in existing_kvs {
@@ -575,21 +576,21 @@ impl KvCache {
                             let key = String::from_utf8_lossy(kv.key()).to_string();
                             let value = kv.value().to_vec();
 
-                            tracing::debug!("KvCache update: {} = {:?}", key, value);
+                            tracing::trace!("KvCache update: {} = {:?}", key, value);
                             let mut cache_write = cache.write().await;
                             cache_write.insert(key, value);
                         }
                         WatchEvent::Delete(kv) => {
                             let key = String::from_utf8_lossy(kv.key()).to_string();
 
-                            tracing::debug!("KvCache delete: {}", key);
+                            tracing::trace!("KvCache delete: {}", key);
                             let mut cache_write = cache.write().await;
                             cache_write.remove(&key);
                         }
                     }
                 }
 
-                tracing::info!("KvCache watcher for prefix '{}' stopped", prefix);
+                tracing::debug!("KvCache watcher for prefix '{}' stopped", prefix);
             });
         }
 
@@ -719,7 +720,7 @@ mod tests {
         initial_values.insert("key2".to_string(), b"value2".to_vec());
 
         // Create the KV cache
-        let kv_cache = KvCache::new(client.clone(), prefix.clone(), initial_values).await?;
+        let kv_cache = KvCache::new(client.clone(), "v1", prefix.clone(), initial_values).await?;
 
         // Test get
         let value1 = kv_cache.get("key1").await;

lib/runtime/src/transports/etcd/path.rs

@@ -8,7 +8,7 @@ use std::str::FromStr;
 use validator::ValidationError;
 
 /// The root etcd path prefix
-pub const ETCD_ROOT_PATH: &str = "dynamo://";
+pub const ETCD_ROOT_PATH: &str = "v1/dynamo/";
 
 /// Reserved keyword for component paths (with underscores to prevent user conflicts)
 pub const COMPONENT_KEYWORD: &str = "_component_";
@@ -371,86 +371,28 @@ fn validate_allowed_chars(input: &str) -> Result<(), ValidationError> {
 mod tests {
     use super::*;
 
-    #[test]
-    fn test_namespace_only() {
-        let path = EtcdPath::parse("dynamo://ns1").unwrap();
-        assert_eq!(path.namespace, "ns1");
-        assert_eq!(path.component, None);
-        assert_eq!(path.endpoint, None);
-        assert_eq!(path.extra_path, None);
-        assert_eq!(path.to_string(), "dynamo://ns1");
-    }
-
-    #[test]
-    fn test_hierarchical_namespace() {
-        let path = EtcdPath::parse("dynamo://ns1.ns2.ns3").unwrap();
-        assert_eq!(path.namespace, "ns1.ns2.ns3");
-        assert_eq!(path.component, None);
-        assert_eq!(path.endpoint, None);
-        assert_eq!(path.extra_path, None);
-        assert_eq!(path.to_string(), "dynamo://ns1.ns2.ns3");
-    }
-
     #[test]
     fn test_namespace_and_component() {
-        let path = EtcdPath::parse("dynamo://ns1.ns2/_component_/my-component").unwrap();
+        let s = format!("{ETCD_ROOT_PATH}ns1.ns2/_component_/my-component");
+        let path = EtcdPath::parse(&s).unwrap();
         assert_eq!(path.namespace, "ns1.ns2");
         assert_eq!(path.component, Some("my-component".to_string()));
         assert_eq!(path.endpoint, None);
         assert_eq!(path.extra_path, None);
-        assert_eq!(
-            path.to_string(),
-            "dynamo://ns1.ns2/_component_/my-component"
-        );
+        assert_eq!(path.to_string(), s);
     }
 
     #[test]
     fn test_full_path_with_endpoint() {
-        let path = EtcdPath::parse(
-            "dynamo://ns1.ns2.ns3/_component_/component-name/_endpoint_/endpoint-name",
-        )
-        .unwrap();
+        let s = format!(
+            "{ETCD_ROOT_PATH}ns1.ns2.ns3/_component_/component-name/_endpoint_/endpoint-name"
+        );
+        let path = EtcdPath::parse(&s).unwrap();
         assert_eq!(path.namespace, "ns1.ns2.ns3");
         assert_eq!(path.component, Some("component-name".to_string()));
         assert_eq!(path.endpoint, Some("endpoint-name".to_string()));
         assert_eq!(path.extra_path, None);
-        assert_eq!(
-            path.to_string(),
-            "dynamo://ns1.ns2.ns3/_component_/component-name/_endpoint_/endpoint-name"
-        );
-    }
-
-    #[test]
-    fn test_with_extra_path() {
-        let path = EtcdPath::parse("dynamo://ns1/_component_/comp1/extra1/extra2").unwrap();
-        assert_eq!(path.namespace, "ns1");
-        assert_eq!(path.component, Some("comp1".to_string()));
-        assert_eq!(path.endpoint, None);
-        assert_eq!(
-            path.extra_path,
-            Some(vec!["extra1".to_string(), "extra2".to_string()])
-        );
-        assert_eq!(
-            path.to_string(),
-            "dynamo://ns1/_component_/comp1/extra1/extra2"
-        );
-    }
-
-    #[test]
-    fn test_endpoint_with_extra_path() {
-        let path =
-            EtcdPath::parse("dynamo://ns1/_component_/comp1/_endpoint_/ep1/path1/path2").unwrap();
-        assert_eq!(path.namespace, "ns1");
-        assert_eq!(path.component, Some("comp1".to_string()));
-        assert_eq!(path.endpoint, Some("ep1".to_string()));
-        assert_eq!(
-            path.extra_path,
-            Some(vec!["path1".to_string(), "path2".to_string()])
-        );
-        assert_eq!(
-            path.to_string(),
-            "dynamo://ns1/_component_/comp1/_endpoint_/ep1/path1/path2"
-        );
+        assert_eq!(path.to_string(), s);
     }
 
     #[test]
@@ -461,38 +403,25 @@ mod tests {
 
     #[test]
     fn test_invalid_characters() {
-        let result = EtcdPath::parse("dynamo://ns1!/_component_/comp1");
+        let result = EtcdPath::parse(&format!("{ETCD_ROOT_PATH}ns1!/_component_/comp1"));
         assert!(matches!(result, Err(EtcdPathError::InvalidNamespace(_))));
     }
 
-    #[test]
-    fn test_endpoint_without_component() {
-        let result = EtcdPath::parse("dynamo://ns1/_endpoint_/ep1");
-        assert!(matches!(
-            result,
-            Err(EtcdPathError::EndpointWithoutComponent)
-        ));
-    }
-
-    #[test]
-    fn test_from_str_trait() {
-        let path: EtcdPath = "dynamo://ns1.ns2/_component_/comp1".parse().unwrap();
-        assert_eq!(path.namespace, "ns1.ns2");
-        assert_eq!(path.component, Some("comp1".to_string()));
-    }
-
     #[test]
     fn test_constructor_methods() {
         let path = EtcdPath::new_namespace("ns1.ns2.ns3").unwrap();
-        assert_eq!(path.to_string(), "dynamo://ns1.ns2.ns3");
+        assert_eq!(path.to_string(), format!("{ETCD_ROOT_PATH}ns1.ns2.ns3"));
 
         let path = EtcdPath::new_component("ns1.ns2", "comp1").unwrap();
-        assert_eq!(path.to_string(), "dynamo://ns1.ns2/_component_/comp1");
+        assert_eq!(
+            path.to_string(),
+            format!("{ETCD_ROOT_PATH}ns1.ns2/_component_/comp1")
+        );
 
         let path = EtcdPath::new_endpoint("ns1", "comp1", "ep1").unwrap();
         assert_eq!(
             path.to_string(),
-            "dynamo://ns1/_component_/comp1/_endpoint_/ep1"
+            format!("{ETCD_ROOT_PATH}ns1/_component_/comp1/_endpoint_/ep1")
         );
     }
 
@@ -504,31 +433,10 @@ mod tests {
             .unwrap();
         assert_eq!(
             path.to_string(),
-            "dynamo://ns1/_component_/comp1/path1/path2"
+            format!("{ETCD_ROOT_PATH}ns1/_component_/comp1/path1/path2")
         );
     }
 
-    #[test]
-    fn test_reserved_keyword_in_extra_path() {
-        // Test that reserved keywords cannot be used in extra paths
-        let result = EtcdPath::parse("dynamo://ns1/_component_/comp1/extra/_component_");
-        assert!(matches!(result, Err(EtcdPathError::ReservedKeyword(_))));
-
-        let result = EtcdPath::parse("dynamo://ns1/_component_/comp1/extra/_endpoint_");
-        assert!(matches!(result, Err(EtcdPathError::ReservedKeyword(_))));
-
-        // Test that with_extra_path also validates reserved keywords
-        let result = EtcdPath::new_component("ns1", "comp1")
-            .unwrap()
-            .with_extra_path(vec!["_component_".to_string()]);
-        assert!(matches!(result, Err(EtcdPathError::ReservedKeyword(_))));
-
-        let result = EtcdPath::new_component("ns1", "comp1")
-            .unwrap()
-            .with_extra_path(vec!["_endpoint_".to_string()]);
-        assert!(matches!(result, Err(EtcdPathError::ReservedKeyword(_))));
-    }
-
     #[test]
     fn test_endpoint_with_lease_id() {
         // Test creating endpoint with lease ID
@@ -539,14 +447,17 @@ mod tests {
         assert_eq!(path.lease_id, Some(0xabc123));
         assert_eq!(
             path.to_string(),
-            "dynamo://ns1/_component_/comp1/_endpoint_/ep1:abc123"
+            format!("{ETCD_ROOT_PATH}ns1/_component_/comp1/_endpoint_/ep1:abc123")
         );
     }
 
     #[test]
     fn test_parse_endpoint_with_lease_id() {
         // Test parsing endpoint with lease ID
-        let path = EtcdPath::parse("dynamo://ns1/_component_/comp1/_endpoint_/ep1:abc123").unwrap();
+        let path = EtcdPath::parse(&format!(
+            "{ETCD_ROOT_PATH}ns1/_component_/comp1/_endpoint_/ep1:abc123"
+        ))
+        .unwrap();
         assert_eq!(path.namespace, "ns1");
         assert_eq!(path.component, Some("comp1".to_string()));
         assert_eq!(path.endpoint, Some("ep1".to_string()));
@@ -557,7 +468,10 @@ mod tests {
     #[test]
     fn test_parse_endpoint_without_lease_id() {
         // Test that endpoints without lease ID still work
-        let path = EtcdPath::parse("dynamo://ns1/_component_/comp1/_endpoint_/ep1").unwrap();
+        let path = EtcdPath::parse(&format!(
+            "{ETCD_ROOT_PATH}ns1/_component_/comp1/_endpoint_/ep1"
+        ))
+        .unwrap();
         assert_eq!(path.namespace, "ns1");
         assert_eq!(path.component, Some("comp1".to_string()));
         assert_eq!(path.endpoint, Some("ep1".to_string()));
@@ -568,7 +482,9 @@ mod tests {
     #[test]
     fn test_invalid_lease_id_format() {
         // Test invalid lease ID format
-        let result = EtcdPath::parse("dynamo://ns1/_component_/comp1/_endpoint_/ep1:invalid");
+        let result = EtcdPath::parse(&format!(
+            "{ETCD_ROOT_PATH}ns1/_component_/comp1/_endpoint_/ep1:invalid"
+        ));
         assert!(matches!(result, Err(EtcdPathError::InvalidEndpoint(_))));
     }
 
@@ -583,7 +499,7 @@ mod tests {
         let path_string = original_path.to_string();
         assert_eq!(
             path_string,
-            "dynamo://production/_component_/api-gateway/_endpoint_/http:deadbeef"
+            format!("{ETCD_ROOT_PATH}production/_component_/api-gateway/_endpoint_/http:deadbeef")
         );
 
         // Parse back from string
@@ -606,19 +522,21 @@ mod tests {
         let path = EtcdPath::new_endpoint_with_lease("ns", "comp", "ep", 0).unwrap();
         assert_eq!(
             path.to_string(),
-            "dynamo://ns/_component_/comp/_endpoint_/ep:0"
+            format!("{ETCD_ROOT_PATH}ns/_component_/comp/_endpoint_/ep:0")
         );
 
         // Test with maximum i64 value
         let path = EtcdPath::new_endpoint_with_lease("ns", "comp", "ep", i64::MAX).unwrap();
         assert_eq!(
             path.to_string(),
-            "dynamo://ns/_component_/comp/_endpoint_/ep:7fffffffffffffff"
+            format!("{ETCD_ROOT_PATH}ns/_component_/comp/_endpoint_/ep:7fffffffffffffff")
         );
 
         // Test parsing maximum value
-        let parsed =
-            EtcdPath::parse("dynamo://ns/_component_/comp/_endpoint_/ep:7fffffffffffffff").unwrap();
+        let parsed = EtcdPath::parse(&format!(
+            "{ETCD_ROOT_PATH}ns/_component_/comp/_endpoint_/ep:7fffffffffffffff"
+        ))
+        .unwrap();
         assert_eq!(parsed.lease_id, Some(i64::MAX));
     }
 }

lib/runtime/src/utils/pool.rs

@@ -668,6 +668,8 @@ mod tests {
         println!("1000 sync pool operations took {:?}", duration);
 
         // Should be fast (< 10ms on most systems)
-        assert!(duration < Duration::from_millis(50));
+        // Update(grahamk): Takes 144ms on my box which is much faster than CI, so something
+        // is odd about claim above.
+        assert!(duration < Duration::from_millis(200));
     }
 }

lib/runtime/src/utils/typed_prefix_watcher.rs

@@ -70,7 +70,7 @@ where
 /// // Watch for ModelDeploymentCard objects and extract runtime_config field
 /// let watcher = watch_prefix_with_extraction(
 ///     etcd_client,
-///     "mdc/",
+///     "v1/mdc/",
 ///     |kv| Some(kv.lease()),  // Use lease_id as key
 ///     |card: ModelDeploymentCard| card.runtime_config,  // Extract runtime_config field
 ///     cancellation_token,

lib/runtime/src/utils/worker_monitor.rs

@@ -94,7 +94,7 @@ impl WorkerMonitor {
         // That means we cannot use ModelDeploymentCard, so use serde_json::Value for now .
         let runtime_configs_watcher = watch_prefix_with_extraction(
             etcd_client,
-            "mdc/", // should be model_card::ROOT_PREFIX but wrong crate
+            "v1/mdc/", // should be model_card::ROOT_PREFIX but wrong crate
             key_extractors::lease_id,
             |card: serde_json::Value| {
                 card.get("runtime_config")