component.rs 13.7 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
// SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
Ryan Olson's avatar
Ryan Olson committed
15
16
17

//! The [Component] module defines the top-level API for building distributed applications.
//!
Graham King's avatar
Graham King committed
18
19
20
//! 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
21
22
23
24
//!
//! A [Component] is made discoverable by registering it with the distributed runtime under
//! a [`Namespace`].
//!
Graham King's avatar
Graham King committed
25
//! A [`Namespace`] is a logical grouping of [Component] that are grouped together.
Ryan Olson's avatar
Ryan Olson committed
26
27
28
29
30
31
32
33
34
35
36
37
//!
//! 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
38
//! Other [Component] can write to watching locations within a [Component] etcd
Ryan Olson's avatar
Ryan Olson committed
39
40
41
42
43
//! path. This allows the [Component] to take dynamic actions depending on the watch
//! triggers.
//!
//! TODO: Top-level Overview of Endpoints/Functions

Ryan Olson's avatar
Ryan Olson committed
44
use crate::{discovery::Lease, service::ServiceSet};
Ryan Olson's avatar
Ryan Olson committed
45

Ryan Olson's avatar
Ryan Olson committed
46
47
48
use super::{
    error, traits::*, transports::nats::Slug, utils::Duration, DistributedRuntime, Result, Runtime,
};
Ryan Olson's avatar
Ryan Olson committed
49
50
51
52
53
54
55
56
57
58

use crate::pipeline::network::{ingress::push_endpoint::PushEndpoint, PushWorkHandler};
use async_nats::{
    rustls::quic,
    service::{Service, ServiceExt},
};
use derive_builder::Builder;
use derive_getters::Getters;
use educe::Educe;
use serde::{Deserialize, Serialize};
59
use service::EndpointStatsHandler;
Ryan Olson's avatar
Ryan Olson committed
60
61
62
63
use std::{collections::HashMap, sync::Arc};
use validator::{Validate, ValidationError};

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

71
pub use client::{Client, EndpointSource};
Ryan Olson's avatar
Ryan Olson committed
72
73
74
75
76
77
78

#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum TransportType {
    NatsTcp(String),
}

79
80
81
82
83
84
#[derive(Default)]
pub struct RegistryInner {
    services: HashMap<String, Service>,
    stats_handlers: HashMap<String, Arc<std::sync::Mutex<HashMap<String, EndpointStatsHandler>>>>,
}

Ryan Olson's avatar
Ryan Olson committed
85
86
#[derive(Clone)]
pub struct Registry {
87
    inner: Arc<tokio::sync::Mutex<RegistryInner>>,
Ryan Olson's avatar
Ryan Olson committed
88
89
90
91
92
93
94
95
96
97
98
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ComponentEndpointInfo {
    pub component: String,
    pub endpoint: String,
    pub namespace: String,
    pub lease_id: i64,
    pub transport: TransportType,
}

99
100
101
102
103
104
impl ComponentEndpointInfo {
    pub fn id(&self) -> i64 {
        self.lease_id
    }
}

Ryan Olson's avatar
Ryan Olson committed
105
/// A [Component] a discoverable entity in the distributed runtime.
Graham King's avatar
Graham King committed
106
107
/// 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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
///
/// You can also issue a request to a [Component]'s [Endpoint] by creating a [Client].
#[derive(Educe, Builder, Clone)]
#[educe(Debug)]
#[builder(pattern = "owned")]
pub struct Component {
    #[builder(private)]
    #[educe(Debug(ignore))]
    drt: DistributedRuntime,

    // todo - restrict the namespace to a-z0-9-_A-Z
    /// Name of the component
    #[builder(setter(into))]
    name: String,

    // todo - restrict the namespace to a-z0-9-_A-Z
    /// Namespace
    #[builder(setter(into))]
126
    namespace: Namespace,
127
128
129
130

    // A static component's endpoints cannot be discovered via etcd, they are
    // fixed at startup time.
    is_static: bool,
Ryan Olson's avatar
Ryan Olson committed
131
132
}

133
134
impl std::fmt::Display for Component {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
135
        write!(f, "{}.{}", self.namespace.name(), self.name)
136
137
138
    }
}

139
140
141
142
143
144
145
146
147
148
149
150
impl DistributedRuntimeProvider for Component {
    fn drt(&self) -> &DistributedRuntime {
        &self.drt
    }
}

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

Ryan Olson's avatar
Ryan Olson committed
151
152
impl Component {
    pub fn etcd_path(&self) -> String {
153
        format!("{}/components/{}", self.namespace.name(), self.name)
Ryan Olson's avatar
Ryan Olson committed
154
155
    }

156
    pub fn service_name(&self) -> String {
157
        let service_name = format!("{}_{}", self.namespace.name(), self.name);
158
        Slug::slugify(&service_name).to_string()
Ryan Olson's avatar
Ryan Olson committed
159
160
    }

161
    pub fn path(&self) -> String {
162
        format!("{}/{}", self.namespace.name(), self.name)
163
164
    }

165
    pub fn namespace(&self) -> &Namespace {
166
167
168
        &self.namespace
    }

169
170
171
172
    pub fn name(&self) -> String {
        self.name.clone()
    }

Ryan Olson's avatar
Ryan Olson committed
173
174
175
176
    pub fn endpoint(&self, endpoint: impl Into<String>) -> Endpoint {
        Endpoint {
            component: self.clone(),
            name: endpoint.into(),
177
            is_static: self.is_static,
Ryan Olson's avatar
Ryan Olson committed
178
179
180
181
182
183
184
185
186
        }
    }

    /// Get keys from etcd on the slug, splitting the endpoints and only returning the
    /// set of unique endpoints.
    pub async fn list_endpoints(&self) -> Vec<Endpoint> {
        unimplemented!("endpoints")
    }

187
    pub async fn scrape_stats(&self, timeout: Duration) -> Result<ServiceSet> {
Ryan Olson's avatar
Ryan Olson committed
188
189
190
        let service_name = self.service_name();
        let service_client = self.drt().service_client();
        service_client
191
            .collect_services(&service_name, timeout)
Ryan Olson's avatar
Ryan Olson committed
192
193
194
            .await
    }

Graham King's avatar
Graham King committed
195
196
    /// TODO
    ///
Ryan Olson's avatar
Ryan Olson committed
197
    /// This method will scrape the stats for all available services
Graham King's avatar
Graham King committed
198
    /// Returns a stream of `ServiceInfo` objects.
Ryan Olson's avatar
Ryan Olson committed
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
    /// 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")
    }

    pub fn service_builder(&self) -> service::ServiceConfigBuilder {
        service::ServiceConfigBuilder::from_component(self.clone())
    }
}

impl ComponentBuilder {
    pub fn from_runtime(drt: DistributedRuntime) -> Self {
        Self::default().drt(drt)
    }
}

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

    // todo - restrict alphabet
    /// Endpoint name
    name: String,
223
224

    is_static: bool,
Ryan Olson's avatar
Ryan Olson committed
225
226
}

227
228
229
230
231
232
233
234
235
236
237
238
impl DistributedRuntimeProvider for Endpoint {
    fn drt(&self) -> &DistributedRuntime {
        self.component.drt()
    }
}

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

Ryan Olson's avatar
Ryan Olson committed
239
240
241
242
243
impl Endpoint {
    pub fn name(&self) -> &str {
        &self.name
    }

244
245
246
247
    pub fn component(&self) -> &Component {
        &self.component
    }

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

Ryan Olson's avatar
Ryan Olson committed
252
253
254
255
256
    pub fn etcd_path(&self) -> String {
        format!("{}/{}", self.component.etcd_path(), self.name)
    }

    pub fn etcd_path_with_id(&self, lease_id: i64) -> String {
257
258
259
260
261
        if self.is_static {
            self.etcd_path()
        } else {
            format!("{}:{:x}", self.etcd_path(), lease_id)
        }
Ryan Olson's avatar
Ryan Olson committed
262
263
264
    }

    pub fn name_with_id(&self, lease_id: i64) -> String {
265
266
267
268
269
        if self.is_static {
            self.name.clone()
        } else {
            format!("{}-{:x}", self.name, lease_id)
        }
Ryan Olson's avatar
Ryan Olson committed
270
271
    }

Ryan Olson's avatar
Ryan Olson committed
272
273
274
275
276
277
278
279
280
281
282
    pub fn subject(&self) -> String {
        format!("{}.{}", self.component.service_name(), self.name)
    }

    /// Subject to an instance of the [Endpoint] with a specific lease id
    pub fn subject_to(&self, lease_id: i64) -> String {
        format!(
            "{}.{}",
            self.component.service_name(),
            self.name_with_id(lease_id)
        )
Ryan Olson's avatar
Ryan Olson committed
283
284
    }

285
    pub async fn client(&self) -> Result<client::Client> {
286
287
288
289
290
        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
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
    }

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

#[derive(Educe, Builder, Clone, Validate)]
#[educe(Debug)]
#[builder(pattern = "owned")]
pub struct Namespace {
    #[builder(private)]
    #[educe(Debug(ignore))]
    runtime: DistributedRuntime,

    #[validate()]
    name: String,
308
309

    is_static: bool,
Ryan Olson's avatar
Ryan Olson committed
310
311
}

312
313
314
315
316
317
318
319
320
321
322
323
impl DistributedRuntimeProvider for Namespace {
    fn drt(&self) -> &DistributedRuntime {
        &self.runtime
    }
}

impl RuntimeProvider for Namespace {
    fn rt(&self) -> &Runtime {
        self.runtime.rt()
    }
}

324
325
326
327
328
329
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
330
impl Namespace {
331
    pub(crate) fn new(runtime: DistributedRuntime, name: String, is_static: bool) -> Result<Self> {
Ryan Olson's avatar
Ryan Olson committed
332
333
334
        Ok(NamespaceBuilder::default()
            .runtime(runtime)
            .name(name)
335
            .is_static(is_static)
Ryan Olson's avatar
Ryan Olson committed
336
337
338
            .build()?)
    }

339
    /// Create a [`Component`] in the namespace who's endpoints can be discovered with etcd
Ryan Olson's avatar
Ryan Olson committed
340
341
342
    pub fn component(&self, name: impl Into<String>) -> Result<Component> {
        Ok(ComponentBuilder::from_runtime(self.runtime.clone())
            .name(name)
343
            .namespace(self.clone())
344
            .is_static(self.is_static)
Ryan Olson's avatar
Ryan Olson committed
345
346
            .build()?)
    }
Ryan Olson's avatar
Ryan Olson committed
347
348
349
350

    pub fn name(&self) -> &str {
        &self.name
    }
Ryan Olson's avatar
Ryan Olson committed
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
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
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
}

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