component.rs 25.2 KB
Newer Older
1
2
// SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0
Ryan Olson's avatar
Ryan Olson committed
3
4
5

//! The [Component] module defines the top-level API for building distributed applications.
//!
Graham King's avatar
Graham King committed
6
7
8
//! A distributed application consists of a set of [Component] that can host one
//! or more [Endpoint]. Each [Endpoint] is a network-accessible service
//! that can be accessed by other [Component] in the distributed application.
Ryan Olson's avatar
Ryan Olson committed
9
10
11
12
//!
//! A [Component] is made discoverable by registering it with the distributed runtime under
//! a [`Namespace`].
//!
Graham King's avatar
Graham King committed
13
//! A [`Namespace`] is a logical grouping of [Component] that are grouped together.
Ryan Olson's avatar
Ryan Olson committed
14
15
16
17
18
19
20
21
22
23
24
25
//!
//! We might extend namespace to include grouping behavior, which would define groups of
//! components that are tightly coupled.
//!
//! A [Component] is the core building block of a distributed application. It is a logical
//! unit of work such as a `Preprocessor` or `SmartRouter` that has a well-defined role in the
//! distributed application.
//!
//! A [Component] can present to the distributed application one or more configuration files
//! which define how that component was constructed/configured and what capabilities it can
//! provide.
//!
Graham King's avatar
Graham King committed
26
//! Other [Component] can write to watching locations within a [Component] etcd
Ryan Olson's avatar
Ryan Olson committed
27
28
29
30
31
//! path. This allows the [Component] to take dynamic actions depending on the watch
//! triggers.
//!
//! TODO: Top-level Overview of Endpoints/Functions

32
33
use std::fmt;

34
use crate::{
35
    config::HealthStatus,
36
    metrics::{MetricsHierarchy, MetricsRegistry, prometheus_names},
37
    service::ServiceSet,
38
    transports::etcd::{ETCD_ROOT_PATH, EtcdPath},
39
};
Ryan Olson's avatar
Ryan Olson committed
40

Ryan Olson's avatar
Ryan Olson committed
41
use super::{
42
    DistributedRuntime, Runtime,
43
44
45
46
    traits::*,
    transports::etcd::{COMPONENT_KEYWORD, ENDPOINT_KEYWORD},
    transports::nats::Slug,
    utils::Duration,
Ryan Olson's avatar
Ryan Olson committed
47
};
Ryan Olson's avatar
Ryan Olson committed
48

49
use crate::pipeline::network::{PushWorkHandler, ingress::push_endpoint::PushEndpoint};
50
use crate::protocols::EndpointId;
51
use crate::service::ComponentNatsServerPrometheusMetrics;
Ryan Olson's avatar
Ryan Olson committed
52
53
54
55
56
57
58
59
use async_nats::{
    rustls::quic,
    service::{Service, ServiceExt},
};
use derive_builder::Builder;
use derive_getters::Getters;
use educe::Educe;
use serde::{Deserialize, Serialize};
60
use service::EndpointStatsHandler;
61
use std::{collections::HashMap, hash::Hash, sync::Arc};
Ryan Olson's avatar
Ryan Olson committed
62
63
64
use validator::{Validate, ValidationError};

mod client;
65
66
#[allow(clippy::module_inception)]
mod component;
Ryan Olson's avatar
Ryan Olson committed
67
mod endpoint;
Ryan Olson's avatar
Ryan Olson committed
68
mod namespace;
Ryan Olson's avatar
Ryan Olson committed
69
mod registry;
70
pub mod service;
Ryan Olson's avatar
Ryan Olson committed
71

72
73
pub use client::{Client, InstanceSource};

74
/// The root key-value path where each instance registers itself in.
75
/// An instance is namespace+component+endpoint+lease_id and must be unique.
76
pub const INSTANCE_ROOT_PATH: &str = "v1/instances";
77

78
#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq, Hash)]
Ryan Olson's avatar
Ryan Olson committed
79
80
81
82
83
#[serde(rename_all = "snake_case")]
pub enum TransportType {
    NatsTcp(String),
}

84
85
86
#[derive(Default)]
pub struct RegistryInner {
    services: HashMap<String, Service>,
87
    stats_handlers: HashMap<String, Arc<parking_lot::Mutex<HashMap<String, EndpointStatsHandler>>>>,
88
89
}

Ryan Olson's avatar
Ryan Olson committed
90
91
#[derive(Clone)]
pub struct Registry {
92
    inner: Arc<tokio::sync::Mutex<RegistryInner>>,
Ryan Olson's avatar
Ryan Olson committed
93
94
}

95
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
96
pub struct Instance {
Ryan Olson's avatar
Ryan Olson committed
97
98
99
    pub component: String,
    pub endpoint: String,
    pub namespace: String,
100
    pub instance_id: u64,
Ryan Olson's avatar
Ryan Olson committed
101
102
103
    pub transport: TransportType,
}

104
impl Instance {
105
    pub fn id(&self) -> u64 {
106
        self.instance_id
107
    }
108
109
110
111
112
113
114
    pub fn endpoint_id(&self) -> EndpointId {
        EndpointId {
            namespace: self.namespace.clone(),
            component: self.component.clone(),
            name: self.endpoint.clone(),
        }
    }
115
116
}

117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
impl fmt::Display for Instance {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "{}/{}/{}/{}",
            self.namespace, self.component, self.endpoint, self.instance_id
        )
    }
}

/// Sort by string name
impl std::cmp::Ord for Instance {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        self.to_string().cmp(&other.to_string())
    }
}

impl PartialOrd for Instance {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        // Since Ord is fully implemented, the comparison is always total.
        Some(self.cmp(other))
    }
}

Ryan Olson's avatar
Ryan Olson committed
141
/// A [Component] a discoverable entity in the distributed runtime.
Graham King's avatar
Graham King committed
142
143
/// You can host [Endpoint] on a [Component] by first creating
/// a [Service] then adding one or more [Endpoint] to the [Service].
Ryan Olson's avatar
Ryan Olson committed
144
145
///
/// You can also issue a request to a [Component]'s [Endpoint] by creating a [Client].
146
#[derive(Educe, Builder, Clone, Validate)]
Ryan Olson's avatar
Ryan Olson committed
147
148
149
150
151
#[educe(Debug)]
#[builder(pattern = "owned")]
pub struct Component {
    #[builder(private)]
    #[educe(Debug(ignore))]
152
    drt: Arc<DistributedRuntime>,
Ryan Olson's avatar
Ryan Olson committed
153
154
155
156

    // todo - restrict the namespace to a-z0-9-_A-Z
    /// Name of the component
    #[builder(setter(into))]
157
    #[validate(custom(function = "validate_allowed_chars"))]
Ryan Olson's avatar
Ryan Olson committed
158
159
    name: String,

160
161
162
163
    /// Additional labels for metrics
    #[builder(default = "Vec::new()")]
    labels: Vec<(String, String)>,

Ryan Olson's avatar
Ryan Olson committed
164
165
166
    // todo - restrict the namespace to a-z0-9-_A-Z
    /// Namespace
    #[builder(setter(into))]
167
    namespace: Namespace,
168
169
170
171

    // A static component's endpoints cannot be discovered via etcd, they are
    // fixed at startup time.
    is_static: bool,
172
173
174
175

    /// This hierarchy's own metrics registry
    #[builder(default = "crate::MetricsRegistry::new()")]
    metrics_registry: crate::MetricsRegistry,
Ryan Olson's avatar
Ryan Olson committed
176
177
}

178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
impl Hash for Component {
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        self.namespace.name().hash(state);
        self.name.hash(state);
        self.is_static.hash(state);
    }
}

impl PartialEq for Component {
    fn eq(&self, other: &Self) -> bool {
        self.namespace.name() == other.namespace.name()
            && self.name == other.name
            && self.is_static == other.is_static
    }
}

impl Eq for Component {}

196
197
impl std::fmt::Display for Component {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
198
        write!(f, "{}.{}", self.namespace.name(), self.name)
199
200
201
    }
}

202
203
204
205
206
207
208
209
210
211
212
213
impl DistributedRuntimeProvider for Component {
    fn drt(&self) -> &DistributedRuntime {
        &self.drt
    }
}

impl RuntimeProvider for Component {
    fn rt(&self) -> &Runtime {
        self.drt.rt()
    }
}

214
impl MetricsHierarchy for Component {
215
216
217
218
    fn basename(&self) -> String {
        self.name.clone()
    }

219
220
221
222
223
224
225
226
227
228
229
230
231
232
    fn parent_hierarchies(&self) -> Vec<&dyn MetricsHierarchy> {
        let mut parents = vec![];

        // Get all ancestors of namespace (DRT, parent namespaces, etc.)
        parents.extend(self.namespace.parent_hierarchies());

        // Add namespace itself
        parents.push(&self.namespace as &dyn MetricsHierarchy);

        parents
    }

    fn get_metrics_registry(&self) -> &MetricsRegistry {
        &self.metrics_registry
233
234
235
    }
}

Ryan Olson's avatar
Ryan Olson committed
236
impl Component {
237
238
    /// The component part of an instance path in key-value store.
    pub fn instance_root(&self) -> String {
239
240
241
        let ns = self.namespace.name();
        let cp = &self.name;
        format!("{INSTANCE_ROOT_PATH}/{ns}/{cp}")
Ryan Olson's avatar
Ryan Olson committed
242
243
    }

244
    pub fn service_name(&self) -> String {
245
        let service_name = format!("{}_{}", self.namespace.name(), self.name);
246
        Slug::slugify(&service_name).to_string()
Ryan Olson's avatar
Ryan Olson committed
247
248
    }

249
    pub fn path(&self) -> String {
250
        format!("{}/{}", self.namespace.name(), self.name)
251
252
    }

253
254
255
256
257
    pub fn etcd_path(&self) -> EtcdPath {
        EtcdPath::new_component(&self.namespace.name(), &self.name)
            .expect("Component name and namespace should be valid")
    }

258
    pub fn namespace(&self) -> &Namespace {
259
260
261
        &self.namespace
    }

262
263
    pub fn name(&self) -> &str {
        &self.name
264
265
    }

266
267
268
269
    pub fn labels(&self) -> &[(String, String)] {
        &self.labels
    }

Ryan Olson's avatar
Ryan Olson committed
270
271
272
273
    pub fn endpoint(&self, endpoint: impl Into<String>) -> Endpoint {
        Endpoint {
            component: self.clone(),
            name: endpoint.into(),
274
            is_static: self.is_static,
275
            labels: Vec::new(),
276
            metrics_registry: crate::MetricsRegistry::new(),
Ryan Olson's avatar
Ryan Olson committed
277
278
279
        }
    }

280
    pub async fn list_instances(&self) -> anyhow::Result<Vec<Instance>> {
281
282
283
284
285
        let discovery = self.drt.discovery();

        let discovery_query = crate::discovery::DiscoveryQuery::ComponentEndpoints {
            namespace: self.namespace.name(),
            component: self.name.clone(),
286
        };
287
288
289
290
291
292
293
294
295
296
297
298

        let discovery_instances = discovery.list(discovery_query).await?;

        // Extract Instance from DiscoveryInstance::Endpoint wrapper
        let mut instances: Vec<Instance> = discovery_instances
            .into_iter()
            .filter_map(|di| match di {
                crate::discovery::DiscoveryInstance::Endpoint(instance) => Some(instance),
                _ => None, // Ignore all other variants (ModelCard, etc.)
            })
            .collect();

299
300
        instances.sort();
        Ok(instances)
Ryan Olson's avatar
Ryan Olson committed
301
302
    }

303
304
    /// Scrape ServiceSet, which contains NATS stats as well as user defined stats
    /// embedded in data field of ServiceInfo.
305
    pub async fn scrape_stats(&self, timeout: Duration) -> anyhow::Result<ServiceSet> {
306
        // Debug: scraping stats for component
Ryan Olson's avatar
Ryan Olson committed
307
        let service_name = self.service_name();
308
309
310
        let Some(service_client) = self.drt().service_client() else {
            anyhow::bail!("ServiceSet is gathered via NATS, do not call this in non-NATS setups.");
        };
Ryan Olson's avatar
Ryan Olson committed
311
        service_client
312
            .collect_services(&service_name, timeout)
Ryan Olson's avatar
Ryan Olson committed
313
314
315
            .await
    }

316
    /// Add Prometheus metrics for this component's NATS service stats.
317
    ///
318
    /// Starts a background task that periodically requests service statistics from NATS
319
320
321
322
    /// and updates the corresponding Prometheus metrics. The first scrape happens immediately,
    /// then subsequent scrapes occur at a fixed interval of 9.8 seconds (MAX_WAIT_MS),
    /// which should be near or smaller than typical Prometheus scraping intervals to ensure
    /// metrics are fresh when Prometheus collects them.
323
    pub fn start_scraping_nats_service_component_metrics(&self) -> anyhow::Result<()> {
324
        const MAX_WAIT_MS: std::time::Duration = std::time::Duration::from_millis(9800); // Should be <= Prometheus scrape interval
325

326
327
        // If there is another component with the same service name, this will fail.
        let component_metrics = ComponentNatsServerPrometheusMetrics::new(self)?;
328
329
330

        let component_clone = self.clone();

331
332
333
334
        // Start a background task that scrapes stats every 5 seconds
        let m = component_metrics.clone();
        let c = component_clone.clone();

335
336
337
338
339
340
341
342
343
        // Use the DRT's runtime handle to spawn the background task.
        // We cannot use regular `tokio::spawn` here because:
        // 1. This method may be called from contexts without an active Tokio runtime
        //    (e.g., tests that create a DRT in a blocking context)
        // 2. Tests often create a temporary runtime just to build the DRT, then drop it
        // 3. `tokio::spawn` requires being called from within a runtime context
        // By using the DRT's own runtime handle, we ensure the task runs in the
        // correct runtime that will persist for the lifetime of the component.
        c.drt().runtime().secondary().spawn(async move {
344
345
            let timeout = std::time::Duration::from_millis(500);
            let mut interval = tokio::time::interval(MAX_WAIT_MS);
346
347
348
349
350
351
352
353
354
355
356
357
358
359
            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

            loop {
                match c.scrape_stats(timeout).await {
                    Ok(service_set) => {
                        m.update_from_service_set(&service_set);
                    }
                    Err(err) => {
                        tracing::error!(
                            "Background scrape failed for {}: {}",
                            c.service_name(),
                            err
                        );
                        m.reset_to_zeros();
360
361
                    }
                }
362

363
364
                interval.tick().await;
            }
365
366
367
368
369
        });

        Ok(())
    }

Graham King's avatar
Graham King committed
370
371
    /// TODO
    ///
Ryan Olson's avatar
Ryan Olson committed
372
    /// This method will scrape the stats for all available services
Graham King's avatar
Graham King committed
373
    /// Returns a stream of `ServiceInfo` objects.
Ryan Olson's avatar
Ryan Olson committed
374
375
    /// This should be consumed by a `[tokio::time::timeout_at`] because each services
    /// will only respond once, but there is no way to know when all services have responded.
376
    pub async fn stats_stream(&self) -> anyhow::Result<()> {
Ryan Olson's avatar
Ryan Olson committed
377
378
379
        unimplemented!("collect_stats")
    }

380
381
382
383
384
385
    pub async fn add_stats_service(&mut self) -> anyhow::Result<()> {
        let service_name = self.service_name();

        // Pre-check to save cost of creating the service, but don't hold the lock
        if self
            .drt
386
            .component_registry()
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
            .inner
            .lock()
            .await
            .services
            .contains_key(&service_name)
        {
            anyhow::bail!("Service {service_name} already exists");
        }

        let Some(nats_client) = self.drt.nats_client() else {
            anyhow::bail!("Cannot create NATS service without NATS.");
        };
        let description = None;
        let (nats_service, stats_reg) =
            service::build_nats_service(nats_client, self, description).await?;

403
        let mut guard = self.drt.component_registry().inner.lock().await;
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
        if !guard.services.contains_key(&service_name) {
            // Normal case
            guard.services.insert(service_name.clone(), nats_service);
            guard.stats_handlers.insert(service_name.clone(), stats_reg);
            drop(guard);
        } else {
            drop(guard);
            let _ = nats_service.stop().await;
            return Err(anyhow::anyhow!(
                "Service create race for {service_name}, now already exists"
            ));
        }

        // Register metrics callback. CRITICAL: Never fail service creation for metrics issues.
        if let Err(err) = self.start_scraping_nats_service_component_metrics() {
            tracing::debug!(service_name, error = %err, "Metrics registration failed");
        }
        Ok(())
Ryan Olson's avatar
Ryan Olson committed
422
423
424
425
    }
}

impl ComponentBuilder {
426
    pub fn from_runtime(drt: Arc<DistributedRuntime>) -> Self {
Ryan Olson's avatar
Ryan Olson committed
427
428
429
430
431
432
433
434
435
436
437
        Self::default().drt(drt)
    }
}

#[derive(Debug, Clone)]
pub struct Endpoint {
    component: Component,

    // todo - restrict alphabet
    /// Endpoint name
    name: String,
438
439

    is_static: bool,
440
441
442

    /// Additional labels for metrics
    labels: Vec<(String, String)>,
443
444
445

    /// This hierarchy's own metrics registry
    metrics_registry: crate::MetricsRegistry,
Ryan Olson's avatar
Ryan Olson committed
446
447
}

448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
impl Hash for Endpoint {
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        self.component.hash(state);
        self.name.hash(state);
        self.is_static.hash(state);
    }
}

impl PartialEq for Endpoint {
    fn eq(&self, other: &Self) -> bool {
        self.component == other.component
            && self.name == other.name
            && self.is_static == other.is_static
    }
}

impl Eq for Endpoint {}

466
467
468
469
470
471
472
473
474
475
476
477
impl DistributedRuntimeProvider for Endpoint {
    fn drt(&self) -> &DistributedRuntime {
        self.component.drt()
    }
}

impl RuntimeProvider for Endpoint {
    fn rt(&self) -> &Runtime {
        self.component.rt()
    }
}

478
impl MetricsHierarchy for Endpoint {
479
480
481
482
    fn basename(&self) -> String {
        self.name.clone()
    }

483
484
485
486
487
488
489
490
491
492
493
494
495
496
    fn parent_hierarchies(&self) -> Vec<&dyn MetricsHierarchy> {
        let mut parents = vec![];

        // Get all ancestors of component (DRT, Namespace, etc.)
        parents.extend(self.component.parent_hierarchies());

        // Add component itself
        parents.push(&self.component as &dyn MetricsHierarchy);

        parents
    }

    fn get_metrics_registry(&self) -> &MetricsRegistry {
        &self.metrics_registry
497
498
499
    }
}

Ryan Olson's avatar
Ryan Olson committed
500
impl Endpoint {
501
502
503
504
505
506
507
508
    pub fn id(&self) -> EndpointId {
        EndpointId {
            namespace: self.component.namespace().name().to_string(),
            component: self.component.name().to_string(),
            name: self.name().to_string(),
        }
    }

Ryan Olson's avatar
Ryan Olson committed
509
510
511
512
    pub fn name(&self) -> &str {
        &self.name
    }

513
514
515
516
    pub fn component(&self) -> &Component {
        &self.component
    }

517
    // todo(ryan): deprecate this as we move to Discovery traits and Component Identifiers
518
    pub fn path(&self) -> String {
519
520
521
522
523
524
        format!(
            "{}/{}/{}",
            self.component.path(),
            ENDPOINT_KEYWORD,
            self.name
        )
525
526
    }

527
528
    /// The endpoint part of an instance path in etcd
    pub fn etcd_root(&self) -> String {
529
        let component_path = self.component.instance_root();
530
531
        let endpoint_name = &self.name;
        format!("{component_path}/{endpoint_name}")
Ryan Olson's avatar
Ryan Olson committed
532
533
    }

534
535
536
537
    /// The endpoint as an EtcdPath object
    pub fn etcd_path(&self) -> EtcdPath {
        EtcdPath::new_endpoint(
            &self.component.namespace().name(),
538
            self.component.name(),
539
540
541
542
543
            &self.name,
        )
        .expect("Endpoint name and component name should be valid")
    }

544
    /// The fully path of an instance in etcd
545
    pub fn etcd_path_with_lease_id(&self, lease_id: u64) -> String {
546
547
548
549
        format!("{INSTANCE_ROOT_PATH}/{}", self.unique_path(lease_id))
    }

    /// Full path of this endpoint with forward slash separators, including lease id
550
    pub fn unique_path(&self, lease_id: u64) -> String {
551
552
553
554
        let ns = self.component.namespace().name();
        let cp = self.component.name();
        let ep = self.name();
        format!("{ns}/{cp}/{ep}/{lease_id:x}")
Ryan Olson's avatar
Ryan Olson committed
555
556
    }

557
558
559
560
561
562
563
    /// The endpoint as an EtcdPath object with lease ID
    pub fn etcd_path_object_with_lease_id(&self, lease_id: i64) -> EtcdPath {
        if self.is_static {
            self.etcd_path()
        } else {
            EtcdPath::new_endpoint_with_lease(
                &self.component.namespace().name(),
564
                self.component.name(),
565
566
567
568
569
570
571
                &self.name,
                lease_id,
            )
            .expect("Endpoint name and component name should be valid")
        }
    }

572
    pub fn name_with_id(&self, lease_id: u64) -> String {
573
574
575
576
577
        if self.is_static {
            self.name.clone()
        } else {
            format!("{}-{:x}", self.name, lease_id)
        }
Ryan Olson's avatar
Ryan Olson committed
578
579
    }

Ryan Olson's avatar
Ryan Olson committed
580
581
582
583
584
    pub fn subject(&self) -> String {
        format!("{}.{}", self.component.service_name(), self.name)
    }

    /// Subject to an instance of the [Endpoint] with a specific lease id
585
    pub fn subject_to(&self, lease_id: u64) -> String {
Ryan Olson's avatar
Ryan Olson committed
586
587
588
589
590
        format!(
            "{}.{}",
            self.component.service_name(),
            self.name_with_id(lease_id)
        )
Ryan Olson's avatar
Ryan Olson committed
591
592
    }

593
    pub async fn client(&self) -> anyhow::Result<client::Client> {
594
595
596
597
598
        if self.is_static {
            client::Client::new_static(self.clone()).await
        } else {
            client::Client::new_dynamic(self.clone()).await
        }
Ryan Olson's avatar
Ryan Olson committed
599
600
601
602
603
604
605
    }

    pub fn endpoint_builder(&self) -> endpoint::EndpointConfigBuilder {
        endpoint::EndpointConfigBuilder::from_endpoint(self.clone())
    }
}

606
#[derive(Builder, Clone, Validate)]
Ryan Olson's avatar
Ryan Olson committed
607
608
609
#[builder(pattern = "owned")]
pub struct Namespace {
    #[builder(private)]
610
    runtime: Arc<DistributedRuntime>,
Ryan Olson's avatar
Ryan Olson committed
611

612
    #[validate(custom(function = "validate_allowed_chars"))]
Ryan Olson's avatar
Ryan Olson committed
613
    name: String,
614
615

    is_static: bool,
616
617
618

    #[builder(default = "None")]
    parent: Option<Arc<Namespace>>,
619
620
621
622

    /// Additional labels for metrics
    #[builder(default = "Vec::new()")]
    labels: Vec<(String, String)>,
623
624
625
626

    /// This hierarchy's own metrics registry
    #[builder(default = "crate::MetricsRegistry::new()")]
    metrics_registry: crate::MetricsRegistry,
Ryan Olson's avatar
Ryan Olson committed
627
628
}

629
630
631
632
633
634
impl DistributedRuntimeProvider for Namespace {
    fn drt(&self) -> &DistributedRuntime {
        &self.runtime
    }
}

635
636
637
638
639
640
641
642
643
644
impl std::fmt::Debug for Namespace {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "Namespace {{ name: {}; is_static: {}; parent: {:?} }}",
            self.name, self.is_static, self.parent
        )
    }
}

645
646
647
648
649
650
impl RuntimeProvider for Namespace {
    fn rt(&self) -> &Runtime {
        self.runtime.rt()
    }
}

651
652
653
654
655
656
impl std::fmt::Display for Namespace {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.name)
    }
}

Ryan Olson's avatar
Ryan Olson committed
657
impl Namespace {
658
659
660
661
662
    pub(crate) fn new(
        runtime: DistributedRuntime,
        name: String,
        is_static: bool,
    ) -> anyhow::Result<Self> {
Ryan Olson's avatar
Ryan Olson committed
663
        Ok(NamespaceBuilder::default()
664
            .runtime(Arc::new(runtime))
Ryan Olson's avatar
Ryan Olson committed
665
            .name(name)
666
            .is_static(is_static)
Ryan Olson's avatar
Ryan Olson committed
667
668
669
            .build()?)
    }

670
    /// Create a [`Component`] in the namespace who's endpoints can be discovered with etcd
671
    pub fn component(&self, name: impl Into<String>) -> anyhow::Result<Component> {
672
        Ok(ComponentBuilder::from_runtime(self.runtime.clone())
Ryan Olson's avatar
Ryan Olson committed
673
            .name(name)
674
            .namespace(self.clone())
675
            .is_static(self.is_static)
676
            .build()?)
Ryan Olson's avatar
Ryan Olson committed
677
    }
Ryan Olson's avatar
Ryan Olson committed
678

679
    /// Create a [`Namespace`] in the parent namespace
680
    pub fn namespace(&self, name: impl Into<String>) -> anyhow::Result<Namespace> {
681
682
683
684
685
686
687
688
689
        Ok(NamespaceBuilder::default()
            .runtime(self.runtime.clone())
            .name(name.into())
            .is_static(self.is_static)
            .parent(Some(Arc::new(self.clone())))
            .build()?)
    }

    pub fn etcd_path(&self) -> String {
690
        format!("{ETCD_ROOT_PATH}{}", self.name())
691
692
693
694
695
696
697
    }

    pub fn name(&self) -> String {
        match &self.parent {
            Some(parent) => format!("{}.{}", parent.name(), self.name),
            None => self.name.clone(),
        }
Ryan Olson's avatar
Ryan Olson committed
698
    }
Ryan Olson's avatar
Ryan Olson committed
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
}

// Custom validator function
fn validate_allowed_chars(input: &str) -> Result<(), ValidationError> {
    // Define the allowed character set using a regex
    let regex = regex::Regex::new(r"^[a-z0-9-_]+$").unwrap();

    if regex.is_match(input) {
        Ok(())
    } else {
        Err(ValidationError::new("invalid_characters"))
    }
}

// TODO - enable restrictions to the character sets allowed for namespaces,
// components, and endpoints.
//
// Put Validate traits on the struct and use the `validate_allowed_chars` method
// to validate the fields.

// #[cfg(test)]
// mod tests {
//     use super::*;
//     use validator::Validate;

//     #[test]
//     fn test_valid_names() {
//         // Valid strings
//         let valid_inputs = vec![
//             "abc",        // Lowercase letters
//             "abc123",     // Letters and numbers
//             "a-b-c",      // Letters with hyphens
//             "a_b_c",      // Letters with underscores
//             "a-b_c-123",  // Mixed valid characters
//             "a",          // Single character
//             "a_b",        // Short valid pattern
//             "123456",     // Only numbers
//             "a---b_c123", // Repeated hyphens/underscores
//         ];

//         for input in valid_inputs {
//             let result = validate_allowed_chars(input);
//             assert!(result.is_ok(), "Expected '{}' to be valid", input);
//         }
//     }

//     #[test]
//     fn test_invalid_names() {
//         // Invalid strings
//         let invalid_inputs = vec![
//             "abc!",     // Invalid character `!`
//             "abc@",     // Invalid character `@`
//             "123$",     // Invalid character `$`
//             "foo.bar",  // Invalid character `.`
//             "foo/bar",  // Invalid character `/`
//             "foo\\bar", // Invalid character `\`
//             "abc#",     // Invalid character `#`
//             "abc def",  // Spaces are not allowed
//             "foo,",     // Invalid character `,`
//             "",         // Empty string
//         ];

//         for input in invalid_inputs {
//             let result = validate_allowed_chars(input);
//             assert!(result.is_err(), "Expected '{}' to be invalid", input);
//         }
//     }

//     // #[test]
//     // fn test_struct_validation_valid() {
//     //     // Struct with valid data
//     //     let valid_data = InputData {
//     //         name: "valid-name_123".to_string(),
//     //     };
//     //     assert!(valid_data.validate().is_ok());
//     // }

//     // #[test]
//     // fn test_struct_validation_invalid() {
//     //     // Struct with invalid data
//     //     let invalid_data = InputData {
//     //         name: "invalid!name".to_string(),
//     //     };
//     //     let result = invalid_data.validate();
//     //     assert!(result.is_err());

//     //     if let Err(errors) = result {
//     //         let error_map = errors.field_errors();
//     //         assert!(error_map.contains_key("name"));
//     //         let name_errors = &error_map["name"];
//     //         assert_eq!(name_errors[0].code, "invalid_characters");
//     //     }
//     // }

//     #[test]
//     fn test_edge_cases() {
//         // Edge cases
//         let edge_inputs = vec![
//             ("-", true),   // Single hyphen
//             ("_", true),   // Single underscore
//             ("a-", true),  // Letter with hyphen
//             ("-", false),  // Repeated hyphens
//             ("-a", false), // Hyphen at the beginning
//             ("a-", false), // Hyphen at the end
//         ];

//         for (input, expected_validity) in edge_inputs {
//             let result = validate_allowed_chars(input);
//             if expected_validity {
//                 assert!(result.is_ok(), "Expected '{}' to be valid", input);
//             } else {
//                 assert!(result.is_err(), "Expected '{}' to be invalid", input);
//             }
//         }
//     }
// }