dynamo_runtime/component/
service.rs

1// SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4use derive_getters::Dissolve;
5use std::collections::HashMap;
6use std::sync::Mutex;
7
8use super::*;
9
10pub use super::endpoint::EndpointStats;
11pub type StatsHandler =
12    Box<dyn FnMut(String, EndpointStats) -> serde_json::Value + Send + Sync + 'static>;
13pub type EndpointStatsHandler =
14    Box<dyn FnMut(EndpointStats) -> serde_json::Value + Send + Sync + 'static>;
15
16pub const PROJECT_NAME: &str = "Dynamo";
17
18#[derive(Educe, Builder, Dissolve)]
19#[educe(Debug)]
20#[builder(pattern = "owned", build_fn(private, name = "build_internal"))]
21pub struct ServiceConfig {
22    #[builder(private)]
23    component: Component,
24
25    /// Description
26    #[builder(default)]
27    description: Option<String>,
28}
29
30impl ServiceConfigBuilder {
31    /// Create the [`Component`]'s service and store it in the registry.
32    pub async fn create(self) -> Result<Component> {
33        let (component, description) = self.build_internal()?.dissolve();
34
35        let version = "0.0.1".to_string();
36
37        let service_name = component.service_name();
38        log::debug!("component: {component}; creating, service_name: {service_name}");
39
40        let description = description.unwrap_or(format!(
41            "{PROJECT_NAME} component {} in namespace {}",
42            component.name, component.namespace
43        ));
44
45        let stats_handler_registry: Arc<Mutex<HashMap<String, EndpointStatsHandler>>> =
46            Arc::new(Mutex::new(HashMap::new()));
47
48        let stats_handler_registry_clone = stats_handler_registry.clone();
49
50        let mut guard = component.drt.component_registry.inner.lock().await;
51
52        if guard.services.contains_key(&service_name) {
53            return Err(anyhow::anyhow!("Service already exists"));
54        }
55
56        // create service on the secondary runtime
57        let builder = component.drt.nats_client.client().service_builder();
58
59        tracing::debug!("Starting service: {}", service_name);
60        let service_builder = builder
61            .description(description)
62            .stats_handler(move |name, stats| {
63                log::trace!("stats_handler: {name}, {stats:?}");
64                let mut guard = stats_handler_registry.lock().unwrap();
65                match guard.get_mut(&name) {
66                    Some(handler) => handler(stats),
67                    None => serde_json::Value::Null,
68                }
69            });
70        tracing::debug!("Got builder");
71        let service = service_builder
72            .start(service_name.clone(), version)
73            .await
74            .map_err(|e| anyhow::anyhow!("Failed to start service: {e}"))?;
75
76        // new copy of service_name as the previous one is moved into the task above
77        let service_name = component.service_name();
78
79        // insert the service into the registry
80        guard.services.insert(service_name.clone(), service);
81
82        // insert the stats handler into the registry
83        guard
84            .stats_handlers
85            .insert(service_name, stats_handler_registry_clone);
86
87        // drop the guard to unlock the mutex
88        drop(guard);
89
90        // Register metrics callback. CRITICAL: Never fail service creation for metrics issues.
91        if let Err(err) = component.start_scraping_nats_service_component_metrics() {
92            tracing::debug!(
93                "Metrics registration failed for '{}': {}",
94                component.service_name(),
95                err
96            );
97        }
98
99        Ok(component)
100    }
101}
102
103impl ServiceConfigBuilder {
104    pub(crate) fn from_component(component: Component) -> Self {
105        Self::default().component(component)
106    }
107}
dynamo_runtime/component/service.rs

dynamo_runtime/component/
service.rs
