dynamo_runtime/transports/etcd/
path.rs

1// SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4//! EtcdPath - Parsing and validation for hierarchical etcd paths
5
6use once_cell::sync::Lazy;
7use std::str::FromStr;
8use validator::ValidationError;
9
10/// The root etcd path prefix
11pub const ETCD_ROOT_PATH: &str = "v1/dynamo/";
12
13/// Reserved keyword for component paths (with underscores to prevent user conflicts)
14pub const COMPONENT_KEYWORD: &str = "_component_";
15
16/// Reserved keyword for endpoint paths (with underscores to prevent user conflicts)
17pub const ENDPOINT_KEYWORD: &str = "_endpoint_";
18
19static ALLOWED_CHARS_REGEX: Lazy<regex::Regex> =
20    Lazy::new(|| regex::Regex::new(r"^[a-z0-9-_]+$").unwrap());
21
22// TODO(ryan): this was an initial implementation that inspired the DEP; we'll keep it asis for now
23// and update this impl with respect to the DEP.
24//
25// Notes:
26// - follow up on this comment: https://github.com/ai-dynamo/dynamo/pull/1459#discussion_r2140616397
27//   - we will be decoupling the "identifer" from the "extra path" bits as two separate objects
28//   - this issue above is a problem, but will be solved by the DEP
29
30/// Represents a parsed etcd path with hierarchical namespaces, components, endpoints, and extra paths
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct EtcdPath {
33    /// The hierarchical namespace (e.g., "ns1.ns2.ns3")
34    pub namespace: String,
35    /// Optional component name
36    pub component: Option<String>,
37    /// Optional endpoint name (requires component to be present)
38    pub endpoint: Option<String>,
39    /// Optional lease ID (only valid with endpoint, in hexadecimal format)
40    pub lease_id: Option<i64>,
41    /// Optional additional path segments beyond the standard structure
42    pub extra_path: Option<Vec<String>>,
43}
44
45/// Errors that can occur during etcd path parsing
46#[derive(Debug, thiserror::Error)]
47pub enum EtcdPathError {
48    #[error("Path must start with '{}'", ETCD_ROOT_PATH)]
49    InvalidPrefix,
50    #[error("Invalid namespace: {0}")]
51    InvalidNamespace(String),
52    #[error("Invalid component name: {0}")]
53    InvalidComponent(String),
54    #[error("Invalid endpoint name: {0}")]
55    InvalidEndpoint(String),
56    #[error("Invalid extra path segment: {0}")]
57    InvalidExtraPath(String),
58    #[error("Endpoint requires component to be present")]
59    EndpointWithoutComponent,
60    #[error("Expected '{}' keyword after namespace", COMPONENT_KEYWORD)]
61    ExpectedComponentKeyword,
62    #[error("Expected '{}' keyword after component", ENDPOINT_KEYWORD)]
63    ExpectedEndpointKeyword,
64    #[error("Reserved keyword '{0}' cannot be used in extra path")]
65    ReservedKeyword(String),
66    #[error("Empty namespace not allowed")]
67    EmptyNamespace,
68    #[error("Empty component name not allowed")]
69    EmptyComponent,
70    #[error("Empty endpoint name not allowed")]
71    EmptyEndpoint,
72}
73
74impl EtcdPath {
75    /// Create a new EtcdPath with just a namespace
76    pub fn new_namespace(namespace: &str) -> Result<Self, EtcdPathError> {
77        validate_namespace(namespace)?;
78        Ok(Self {
79            namespace: namespace.to_string(),
80            component: None,
81            endpoint: None,
82            lease_id: None,
83            extra_path: None,
84        })
85    }
86
87    /// Create a new EtcdPath with namespace and component
88    pub fn new_component(namespace: &str, component: &str) -> Result<Self, EtcdPathError> {
89        validate_namespace(namespace)?;
90        validate_component(component)?;
91        Ok(Self {
92            namespace: namespace.to_string(),
93            component: Some(component.to_string()),
94            endpoint: None,
95            lease_id: None,
96            extra_path: None,
97        })
98    }
99
100    /// Create a new EtcdPath with namespace, component, and endpoint
101    pub fn new_endpoint(
102        namespace: &str,
103        component: &str,
104        endpoint: &str,
105    ) -> Result<Self, EtcdPathError> {
106        validate_namespace(namespace)?;
107        validate_component(component)?;
108        validate_endpoint(endpoint)?;
109        Ok(Self {
110            namespace: namespace.to_string(),
111            component: Some(component.to_string()),
112            endpoint: Some(endpoint.to_string()),
113            lease_id: None,
114            extra_path: None,
115        })
116    }
117
118    /// Create a new EtcdPath for an endpoint with lease ID
119    pub fn new_endpoint_with_lease(
120        namespace: &str,
121        component: &str,
122        endpoint: &str,
123        lease_id: i64,
124    ) -> Result<Self, EtcdPathError> {
125        validate_namespace(namespace)?;
126        validate_component(component)?;
127        validate_endpoint(endpoint)?;
128
129        Ok(Self {
130            namespace: namespace.to_string(),
131            component: Some(component.to_string()),
132            endpoint: Some(endpoint.to_string()),
133            lease_id: Some(lease_id),
134            extra_path: None,
135        })
136    }
137
138    /// Add extra path segments to this EtcdPath
139    pub fn with_extra_path(mut self, extra_path: Vec<String>) -> Result<Self, EtcdPathError> {
140        for segment in &extra_path {
141            validate_extra_path_segment(segment)?;
142        }
143        self.extra_path = if extra_path.is_empty() {
144            None
145        } else {
146            Some(extra_path)
147        };
148        self.lease_id = None;
149        Ok(self)
150    }
151
152    /// Internal method to convert the EtcdPath back to a string representation
153    fn _to_string(&self) -> String {
154        let mut path = format!("{}{}", ETCD_ROOT_PATH, self.namespace);
155
156        if let Some(ref component) = self.component {
157            path.push('/');
158            path.push_str(COMPONENT_KEYWORD);
159            path.push('/');
160            path.push_str(component);
161
162            if let Some(ref endpoint) = self.endpoint {
163                path.push('/');
164                path.push_str(ENDPOINT_KEYWORD);
165                path.push('/');
166                path.push_str(endpoint);
167
168                // Add lease ID if present
169                if let Some(lease_id) = self.lease_id {
170                    path.push(':');
171                    path.push_str(&format!("{:x}", lease_id));
172                }
173            }
174        }
175
176        if let Some(ref extra_path) = self.extra_path {
177            for segment in extra_path {
178                path.push('/');
179                path.push_str(segment);
180            }
181        }
182
183        path
184    }
185
186    /// Parse an etcd path string into its components
187    pub fn parse(input: &str) -> Result<Self, EtcdPathError> {
188        // Check for required prefix
189        if !input.starts_with(ETCD_ROOT_PATH) {
190            return Err(EtcdPathError::InvalidPrefix);
191        }
192
193        // Remove the prefix and split into segments
194        let path_without_prefix = &input[ETCD_ROOT_PATH.len()..];
195        let segments: Vec<&str> = path_without_prefix.split('/').collect();
196
197        if segments.is_empty() || segments[0].is_empty() {
198            return Err(EtcdPathError::EmptyNamespace);
199        }
200
201        // First segment is always the namespace
202        let namespace = segments[0].to_string();
203        validate_namespace(&namespace)?;
204
205        let mut etcd_path = Self {
206            namespace,
207            component: None,
208            endpoint: None,
209            lease_id: None,
210            extra_path: None,
211        };
212
213        // Parse remaining segments
214        let mut i = 1;
215        while i < segments.len() {
216            match segments[i] {
217                COMPONENT_KEYWORD => {
218                    if i + 1 >= segments.len() {
219                        return Err(EtcdPathError::EmptyComponent);
220                    }
221                    let component_name = segments[i + 1].to_string();
222                    validate_component(&component_name)?;
223                    etcd_path.component = Some(component_name);
224                    i += 2;
225                }
226                ENDPOINT_KEYWORD => {
227                    if etcd_path.component.is_none() {
228                        return Err(EtcdPathError::EndpointWithoutComponent);
229                    }
230                    if i + 1 >= segments.len() {
231                        return Err(EtcdPathError::EmptyEndpoint);
232                    }
233                    let endpoint_segment = segments[i + 1];
234
235                    // Check if endpoint has a lease ID suffix (:lease_id)
236                    if let Some(colon_pos) = endpoint_segment.find(':') {
237                        let endpoint_name = endpoint_segment[..colon_pos].to_string();
238                        let lease_id_str = &endpoint_segment[colon_pos + 1..];
239
240                        validate_endpoint(&endpoint_name)?;
241
242                        // Parse lease ID as hexadecimal
243                        let lease_id = i64::from_str_radix(lease_id_str, 16).map_err(|_| {
244                            EtcdPathError::InvalidEndpoint(format!(
245                                "Invalid lease ID format: {}",
246                                lease_id_str
247                            ))
248                        })?;
249
250                        etcd_path.endpoint = Some(endpoint_name);
251                        etcd_path.lease_id = Some(lease_id);
252                    } else {
253                        let endpoint_name = endpoint_segment.to_string();
254                        validate_endpoint(&endpoint_name)?;
255                        etcd_path.endpoint = Some(endpoint_name);
256                    }
257                    i += 2;
258                }
259                _ => {
260                    // This is an extra path segment
261                    let mut extra_path = Vec::new();
262                    while i < segments.len() {
263                        validate_extra_path_segment(segments[i])?;
264                        extra_path.push(segments[i].to_string());
265                        i += 1;
266                    }
267                    etcd_path.extra_path = if extra_path.is_empty() {
268                        None
269                    } else {
270                        Some(extra_path)
271                    };
272                    break;
273                }
274            }
275        }
276
277        Ok(etcd_path)
278    }
279}
280
281impl FromStr for EtcdPath {
282    type Err = EtcdPathError;
283
284    fn from_str(s: &str) -> Result<Self, Self::Err> {
285        Self::parse(s)
286    }
287}
288
289impl EtcdPath {
290    /// Try to create an EtcdPath from a String
291    pub fn from_string(s: String) -> Result<Self, EtcdPathError> {
292        Self::parse(&s)
293    }
294}
295
296impl std::fmt::Display for EtcdPath {
297    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
298        write!(f, "{}", self._to_string())
299    }
300}
301
302/// Validate namespace using the existing validation function
303fn validate_namespace(namespace: &str) -> Result<(), EtcdPathError> {
304    if namespace.is_empty() {
305        return Err(EtcdPathError::EmptyNamespace);
306    }
307
308    // Split by dots and validate each part
309    for part in namespace.split('.') {
310        if part.is_empty() {
311            return Err(EtcdPathError::InvalidNamespace(format!(
312                "Empty namespace segment in '{}'",
313                namespace
314            )));
315        }
316        validate_allowed_chars(part).map_err(|_| {
317            EtcdPathError::InvalidNamespace(format!("Invalid characters in '{}'", part))
318        })?;
319    }
320    Ok(())
321}
322
323/// Validate component name
324fn validate_component(component: &str) -> Result<(), EtcdPathError> {
325    if component.is_empty() {
326        return Err(EtcdPathError::EmptyComponent);
327    }
328    validate_allowed_chars(component)
329        .map_err(|_| EtcdPathError::InvalidComponent(component.to_string()))
330}
331
332/// Validate endpoint name
333fn validate_endpoint(endpoint: &str) -> Result<(), EtcdPathError> {
334    if endpoint.is_empty() {
335        return Err(EtcdPathError::EmptyEndpoint);
336    }
337    validate_allowed_chars(endpoint)
338        .map_err(|_| EtcdPathError::InvalidEndpoint(endpoint.to_string()))
339}
340
341/// Validate extra path segment
342fn validate_extra_path_segment(segment: &str) -> Result<(), EtcdPathError> {
343    if segment.is_empty() {
344        return Err(EtcdPathError::InvalidExtraPath(
345            "Empty path segment".to_string(),
346        ));
347    }
348
349    // Check for reserved keywords
350    if segment == COMPONENT_KEYWORD {
351        return Err(EtcdPathError::ReservedKeyword(segment.to_string()));
352    }
353    if segment == ENDPOINT_KEYWORD {
354        return Err(EtcdPathError::ReservedKeyword(segment.to_string()));
355    }
356
357    validate_allowed_chars(segment)
358        .map_err(|_| EtcdPathError::InvalidExtraPath(segment.to_string()))
359}
360
361/// Custom validator function (same as in component.rs)
362fn validate_allowed_chars(input: &str) -> Result<(), ValidationError> {
363    if ALLOWED_CHARS_REGEX.is_match(input) {
364        Ok(())
365    } else {
366        Err(ValidationError::new("invalid_characters"))
367    }
368}
369
370#[cfg(test)]
371mod tests {
372    use super::*;
373
374    #[test]
375    fn test_namespace_and_component() {
376        let s = format!("{ETCD_ROOT_PATH}ns1.ns2/_component_/my-component");
377        let path = EtcdPath::parse(&s).unwrap();
378        assert_eq!(path.namespace, "ns1.ns2");
379        assert_eq!(path.component, Some("my-component".to_string()));
380        assert_eq!(path.endpoint, None);
381        assert_eq!(path.extra_path, None);
382        assert_eq!(path.to_string(), s);
383    }
384
385    #[test]
386    fn test_full_path_with_endpoint() {
387        let s = format!(
388            "{ETCD_ROOT_PATH}ns1.ns2.ns3/_component_/component-name/_endpoint_/endpoint-name"
389        );
390        let path = EtcdPath::parse(&s).unwrap();
391        assert_eq!(path.namespace, "ns1.ns2.ns3");
392        assert_eq!(path.component, Some("component-name".to_string()));
393        assert_eq!(path.endpoint, Some("endpoint-name".to_string()));
394        assert_eq!(path.extra_path, None);
395        assert_eq!(path.to_string(), s);
396    }
397
398    #[test]
399    fn test_invalid_prefix() {
400        let result = EtcdPath::parse("invalid://ns1");
401        assert!(matches!(result, Err(EtcdPathError::InvalidPrefix)));
402    }
403
404    #[test]
405    fn test_invalid_characters() {
406        let result = EtcdPath::parse(&format!("{ETCD_ROOT_PATH}ns1!/_component_/comp1"));
407        assert!(matches!(result, Err(EtcdPathError::InvalidNamespace(_))));
408    }
409
410    #[test]
411    fn test_constructor_methods() {
412        let path = EtcdPath::new_namespace("ns1.ns2.ns3").unwrap();
413        assert_eq!(path.to_string(), format!("{ETCD_ROOT_PATH}ns1.ns2.ns3"));
414
415        let path = EtcdPath::new_component("ns1.ns2", "comp1").unwrap();
416        assert_eq!(
417            path.to_string(),
418            format!("{ETCD_ROOT_PATH}ns1.ns2/_component_/comp1")
419        );
420
421        let path = EtcdPath::new_endpoint("ns1", "comp1", "ep1").unwrap();
422        assert_eq!(
423            path.to_string(),
424            format!("{ETCD_ROOT_PATH}ns1/_component_/comp1/_endpoint_/ep1")
425        );
426    }
427
428    #[test]
429    fn test_with_extra_path_method() {
430        let path = EtcdPath::new_component("ns1", "comp1")
431            .unwrap()
432            .with_extra_path(vec!["path1".to_string(), "path2".to_string()])
433            .unwrap();
434        assert_eq!(
435            path.to_string(),
436            format!("{ETCD_ROOT_PATH}ns1/_component_/comp1/path1/path2")
437        );
438    }
439
440    #[test]
441    fn test_endpoint_with_lease_id() {
442        // Test creating endpoint with lease ID
443        let path = EtcdPath::new_endpoint_with_lease("ns1", "comp1", "ep1", 0xabc123).unwrap();
444        assert_eq!(path.namespace, "ns1");
445        assert_eq!(path.component, Some("comp1".to_string()));
446        assert_eq!(path.endpoint, Some("ep1".to_string()));
447        assert_eq!(path.lease_id, Some(0xabc123));
448        assert_eq!(
449            path.to_string(),
450            format!("{ETCD_ROOT_PATH}ns1/_component_/comp1/_endpoint_/ep1:abc123")
451        );
452    }
453
454    #[test]
455    fn test_parse_endpoint_with_lease_id() {
456        // Test parsing endpoint with lease ID
457        let path = EtcdPath::parse(&format!(
458            "{ETCD_ROOT_PATH}ns1/_component_/comp1/_endpoint_/ep1:abc123"
459        ))
460        .unwrap();
461        assert_eq!(path.namespace, "ns1");
462        assert_eq!(path.component, Some("comp1".to_string()));
463        assert_eq!(path.endpoint, Some("ep1".to_string()));
464        assert_eq!(path.lease_id, Some(0xabc123));
465        assert_eq!(path.extra_path, None);
466    }
467
468    #[test]
469    fn test_parse_endpoint_without_lease_id() {
470        // Test that endpoints without lease ID still work
471        let path = EtcdPath::parse(&format!(
472            "{ETCD_ROOT_PATH}ns1/_component_/comp1/_endpoint_/ep1"
473        ))
474        .unwrap();
475        assert_eq!(path.namespace, "ns1");
476        assert_eq!(path.component, Some("comp1".to_string()));
477        assert_eq!(path.endpoint, Some("ep1".to_string()));
478        assert_eq!(path.lease_id, None);
479        assert_eq!(path.extra_path, None);
480    }
481
482    #[test]
483    fn test_invalid_lease_id_format() {
484        // Test invalid lease ID format
485        let result = EtcdPath::parse(&format!(
486            "{ETCD_ROOT_PATH}ns1/_component_/comp1/_endpoint_/ep1:invalid"
487        ));
488        assert!(matches!(result, Err(EtcdPathError::InvalidEndpoint(_))));
489    }
490
491    #[test]
492    fn test_lease_id_round_trip() {
493        // Test round-trip: create -> to_string -> parse -> verify
494        let original_path =
495            EtcdPath::new_endpoint_with_lease("production", "api-gateway", "http", 0xdeadbeef)
496                .unwrap();
497
498        // Convert to string
499        let path_string = original_path.to_string();
500        assert_eq!(
501            path_string,
502            format!("{ETCD_ROOT_PATH}production/_component_/api-gateway/_endpoint_/http:deadbeef")
503        );
504
505        // Parse back from string
506        let parsed_path = EtcdPath::parse(&path_string).unwrap();
507
508        // Verify all fields match
509        assert_eq!(parsed_path.namespace, "production");
510        assert_eq!(parsed_path.component, Some("api-gateway".to_string()));
511        assert_eq!(parsed_path.endpoint, Some("http".to_string()));
512        assert_eq!(parsed_path.lease_id, Some(0xdeadbeef));
513        assert_eq!(parsed_path.extra_path, None);
514
515        // Verify the parsed path equals the original
516        assert_eq!(parsed_path, original_path);
517    }
518
519    #[test]
520    fn test_lease_id_edge_cases() {
521        // Test with lease ID 0
522        let path = EtcdPath::new_endpoint_with_lease("ns", "comp", "ep", 0).unwrap();
523        assert_eq!(
524            path.to_string(),
525            format!("{ETCD_ROOT_PATH}ns/_component_/comp/_endpoint_/ep:0")
526        );
527
528        // Test with maximum i64 value
529        let path = EtcdPath::new_endpoint_with_lease("ns", "comp", "ep", i64::MAX).unwrap();
530        assert_eq!(
531            path.to_string(),
532            format!("{ETCD_ROOT_PATH}ns/_component_/comp/_endpoint_/ep:7fffffffffffffff")
533        );
534
535        // Test parsing maximum value
536        let parsed = EtcdPath::parse(&format!(
537            "{ETCD_ROOT_PATH}ns/_component_/comp/_endpoint_/ep:7fffffffffffffff"
538        ))
539        .unwrap();
540        assert_eq!(parsed.lease_id, Some(i64::MAX));
541    }
542}
dynamo_runtime/transports/etcd/path.rs

dynamo_runtime/transports/etcd/
path.rs
