component.rs 25.1 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
36
    config::HealthStatus,
    discovery::Lease,
37
    metrics::{MetricsHierarchy, MetricsRegistry, prometheus_names},
38
    service::ServiceSet,
39
    transports::etcd::{ETCD_ROOT_PATH, EtcdPath},
40
};
Ryan Olson's avatar
Ryan Olson committed
41

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

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

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

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

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

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

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

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

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

105
impl Instance {
106
    pub fn id(&self) -> u64 {
107
        self.instance_id
108
    }
109
110
111
112
113
114
115
    pub fn endpoint_id(&self) -> EndpointId {
        EndpointId {
            namespace: self.namespace.clone(),
            component: self.component.clone(),
            name: self.endpoint.clone(),
        }
    }
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
141
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
142
/// A [Component] a discoverable entity in the distributed runtime.
Graham King's avatar
Graham King committed
143
144
/// 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
145
146
///
/// You can also issue a request to a [Component]'s [Endpoint] by creating a [Client].
147
#[derive(Educe, Builder, Clone, Validate)]
Ryan Olson's avatar
Ryan Olson committed
148
149
150
151
152
#[educe(Debug)]
#[builder(pattern = "owned")]
pub struct Component {
    #[builder(private)]
    #[educe(Debug(ignore))]
153
    drt: Arc<DistributedRuntime>,
Ryan Olson's avatar
Ryan Olson committed
154
155
156
157

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

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

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

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

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

179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
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 {}

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

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

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

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

220
221
222
223
224
225
226
227
228
229
230
231
232
233
    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
234
235
236
    }
}

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

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

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

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

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

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

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

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

281
    pub async fn list_instances(&self) -> anyhow::Result<Vec<Instance>> {
282
283
        let client = self.drt.store();
        let Some(bucket) = client.get_bucket(&self.instance_root()).await? else {
284
285
            return Ok(vec![]);
        };
286
287
288
289
        let entries = bucket.entries().await?;
        let mut instances = Vec::with_capacity(entries.len());
        for (name, bytes) in entries.into_iter() {
            let val = match serde_json::from_slice::<Instance>(&bytes) {
290
291
                Ok(val) => val,
                Err(err) => {
292
                    anyhow::bail!("Error converting storage response to Instance: {err}. {name}",);
293
294
                }
            };
295
            instances.push(val);
296
        }
297
298
        instances.sort();
        Ok(instances)
Ryan Olson's avatar
Ryan Olson committed
299
300
    }

301
302
    /// Scrape ServiceSet, which contains NATS stats as well as user defined stats
    /// embedded in data field of ServiceInfo.
303
    pub async fn scrape_stats(&self, timeout: Duration) -> Result<ServiceSet> {
304
        // Debug: scraping stats for component
Ryan Olson's avatar
Ryan Olson committed
305
        let service_name = self.service_name();
306
307
308
        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
309
        service_client
310
            .collect_services(&service_name, timeout)
Ryan Olson's avatar
Ryan Olson committed
311
312
313
            .await
    }

314
    /// Add Prometheus metrics for this component's NATS service stats.
315
    ///
316
    /// Starts a background task that periodically requests service statistics from NATS
317
318
319
320
    /// 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.
321
    pub fn start_scraping_nats_service_component_metrics(&self) -> Result<()> {
322
        const MAX_WAIT_MS: std::time::Duration = std::time::Duration::from_millis(9800); // Should be <= Prometheus scrape interval
323

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

        let component_clone = self.clone();

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

333
334
335
336
337
338
339
340
341
        // 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 {
342
343
            let timeout = std::time::Duration::from_millis(500);
            let mut interval = tokio::time::interval(MAX_WAIT_MS);
344
345
346
347
348
349
350
351
352
353
354
355
356
357
            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();
358
359
                    }
                }
360

361
362
                interval.tick().await;
            }
363
364
365
366
367
        });

        Ok(())
    }

Graham King's avatar
Graham King committed
368
369
    /// TODO
    ///
Ryan Olson's avatar
Ryan Olson committed
370
    /// This method will scrape the stats for all available services
Graham King's avatar
Graham King committed
371
    /// Returns a stream of `ServiceInfo` objects.
Ryan Olson's avatar
Ryan Olson committed
372
373
374
375
376
377
    /// 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.
    pub async fn stats_stream(&self) -> Result<()> {
        unimplemented!("collect_stats")
    }

378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
    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
            .component_registry
            .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?;

        let mut guard = self.drt.component_registry.inner.lock().await;
        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
420
421
422
423
    }
}

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

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

    // todo - restrict alphabet
    /// Endpoint name
    name: String,
436
437

    is_static: bool,
438
439
440

    /// Additional labels for metrics
    labels: Vec<(String, String)>,
441
442
443

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

446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
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 {}

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

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

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

481
482
483
484
485
486
487
488
489
490
491
492
493
494
    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
495
496
497
    }
}

Ryan Olson's avatar
Ryan Olson committed
498
impl Endpoint {
499
500
501
502
503
504
505
506
    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
507
508
509
510
    pub fn name(&self) -> &str {
        &self.name
    }

511
512
513
514
    pub fn component(&self) -> &Component {
        &self.component
    }

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

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

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

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

    /// Full path of this endpoint with forward slash separators, including lease id
548
    pub fn unique_path(&self, lease_id: u64) -> String {
549
550
551
552
        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
553
554
    }

555
556
557
558
559
560
561
    /// 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(),
562
                self.component.name(),
563
564
565
566
567
568
569
                &self.name,
                lease_id,
            )
            .expect("Endpoint name and component name should be valid")
        }
    }

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

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

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

591
    pub async fn client(&self) -> Result<client::Client> {
592
593
594
595
596
        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
597
598
599
600
601
602
603
    }

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

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

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

    is_static: bool,
614
615
616

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

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

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

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

633
634
635
636
637
638
639
640
641
642
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
        )
    }
}

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

649
650
651
652
653
654
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
655
impl Namespace {
656
    pub(crate) fn new(runtime: DistributedRuntime, name: String, is_static: bool) -> Result<Self> {
Ryan Olson's avatar
Ryan Olson committed
657
        Ok(NamespaceBuilder::default()
658
            .runtime(Arc::new(runtime))
Ryan Olson's avatar
Ryan Olson committed
659
            .name(name)
660
            .is_static(is_static)
Ryan Olson's avatar
Ryan Olson committed
661
662
663
            .build()?)
    }

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

673
674
675
676
677
678
679
680
681
682
683
    /// Create a [`Namespace`] in the parent namespace
    pub fn namespace(&self, name: impl Into<String>) -> Result<Namespace> {
        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 {
684
        format!("{ETCD_ROOT_PATH}{}", self.name())
685
686
687
688
689
690
691
    }

    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
692
    }
Ryan Olson's avatar
Ryan Olson committed
693
694
695
696
697
698
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
}

// 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);
//             }
//         }
//     }
// }