kumo_server_common/
start.rs

1use crate::diagnostic_logging::LoggingConfig;
2use crate::nodeid::NodeId;
3use anyhow::Context;
4use chrono::{DateTime, Utc};
5use config::RegisterFunc;
6use kumo_machine_info::MachineInfo;
7use kumo_prometheus::declare_metric;
8use kumo_server_lifecycle::LifeCycle;
9use kumo_server_runtime::rt_spawn;
10use parking_lot::Mutex;
11use std::future::Future;
12use std::path::Path;
13use std::sync::LazyLock;
14
15pub static ONLINE_SINCE: LazyLock<DateTime<Utc>> = LazyLock::new(Utc::now);
16pub static MACHINE_INFO: LazyLock<Mutex<Option<MachineInfo>>> = LazyLock::new(|| Mutex::new(None));
17
18pub struct StartConfig<'a> {
19    pub logging: LoggingConfig<'a>,
20    pub lua_funcs: &'a [RegisterFunc],
21    pub policy: &'a Path,
22}
23
24impl StartConfig<'_> {
25    pub async fn run<INIT, FINI>(
26        self,
27        init_future: INIT,
28        shutdown_future: FINI,
29    ) -> anyhow::Result<()>
30    where
31        INIT: Future<Output = anyhow::Result<()>> + Send + 'static,
32        FINI: Future<Output = ()> + Send + 'static,
33    {
34        LazyLock::force(&ONLINE_SINCE);
35        self.logging.init()?;
36
37        start_cpu_usage_monitor();
38        crate::log_backlog::start_monitor();
39
40        rustls::crypto::aws_lc_rs::default_provider()
41            .install_default()
42            .map_err(|_| anyhow::anyhow!("failed to install default crypto provider"))?;
43
44        kumo_server_memory::setup_memory_limit().context("setup_memory_limit")?;
45
46        prometheus::register(Box::new(
47            tokio_metrics_collector::default_runtime_collector(),
48        ))
49        .context("failed to configure tokio-metrics-collector")?;
50
51        for &func in self.lua_funcs {
52            config::register(func);
53        }
54
55        config::set_policy_path(self.policy.to_path_buf())
56            .await
57            .with_context(|| format!("Error evaluating policy file {:?}", self.policy))?;
58
59        tokio::spawn(async move {
60            let mut info = MachineInfo::new();
61            info.node_id.replace(NodeId::get().uuid.to_string());
62            info.query_cloud_provider().await;
63            *MACHINE_INFO.lock() = Some(info);
64        });
65
66        let mut life_cycle = LifeCycle::new();
67
68        let init_handle = rt_spawn("initialize", async move {
69            let mut error = None;
70            if let Err(err) = init_future.await {
71                let err = format!("{err:#}");
72                tracing::error!("problem initializing: {err}");
73                LifeCycle::request_shutdown(Ok(())).await;
74                error.replace(err);
75            }
76            // This log line is depended upon by the integration
77            // test harness. Do not change or remove it without
78            // making appropriate adjustments over there!
79            tracing::info!("initialization complete");
80            error
81        })?;
82
83        let final_result = life_cycle.wait_for_shutdown().await;
84
85        // after waiting for those to idle out, shut down logging
86        shutdown_future.await;
87        crate::acct::shutdown().await;
88
89        tracing::info!("Shutdown completed OK!");
90
91        if let Some(error) = init_handle.await? {
92            anyhow::bail!("Initialization raised an error: {error}");
93        }
94        if let Some(Err(error)) = final_result {
95            anyhow::bail!("Shutdown due to error: {error}");
96        }
97        Ok(())
98    }
99}
100
101declare_metric! {
102/// The sum of the system-wide CPU usage for each CPU in the system, can add up to more than 100%.
103///
104/// Each CPU has a value from 0-100% busy; a value of 100% in this metric
105/// indicates that the load is equivalent to one fully utilized CPU.
106///
107/// A multi-CPU system can report more than 100% in this metric; a dual-CPU
108/// system reporting 200% indicates that both CPUs are fully utilized.
109///
110/// See system_cpu_usage_normalized for a version of this metric that scales from
111/// 0% (totally idle) to 100% (totally saturated).
112///
113/// This metric is scoped to the system, reflecting the total load on the
114/// system, not just from the kumo related process(es).
115static SYS_CPU_USAGE: IntGauge("system_cpu_usage_sum");
116}
117
118declare_metric! {
119/// The sum of the system-wide CPU usage for each CPU in the system, divided by the number of CPUs.
120///
121/// 100% in this metric indicates that all CPU cores are 100% busy.
122///
123/// This metric is scoped to the system, reflecting the total load on the
124/// system, not just from the kumo related process(es).
125static SYS_CPU_USAGE_NORM: IntGauge("system_cpu_usage_normalized");
126}
127
128declare_metric! {
129/// The sum of the process CPU usage for each CPU in the system, can add up to more than 100%.
130///
131/// Each CPU has a value from 0-100% busy; a value of 100% in this metric
132/// indicates that the load is equivalent to one fully utilized CPU.
133///
134/// A multi-CPU system can report more than 100% in this metric; a dual-CPU
135/// system reporting 200% indicates that both CPUs are fully utilized.
136///
137/// See process_cpu_usage_normalized for a version of this metric that scales from
138/// 0% (totally idle) to 100% (totally saturated).
139///
140/// This metric is scoped to the service process, reflecting the CPU used only
141/// by the process and not the system as a whole.
142static PROC_CPU_USAGE: IntGauge("process_cpu_usage_sum");
143}
144
145declare_metric! {
146/// The sum of the process CPU usage for each CPU in the system, divided by the number of CPUs.
147///
148/// 100% in this metric indicates that all CPU cores are 100% busy.
149///
150/// This metric is scoped to the service process, reflecting the CPU used only
151/// by the process and not the system as a whole.
152static PROC_CPU_USAGE_NORM: IntGauge("process_cpu_usage_normalized");
153}
154
155fn start_cpu_usage_monitor() {
156    std::thread::spawn(|| {
157        use std::time::Duration;
158        use sysinfo::{
159            get_current_pid, CpuRefreshKind, ProcessRefreshKind, ProcessesToUpdate, RefreshKind,
160            System,
161        };
162
163        let mut sys = System::new_with_specifics(
164            RefreshKind::nothing().with_cpu(CpuRefreshKind::everything()),
165        );
166
167        let my_pid = get_current_pid().expect("failed to get own pid!?");
168
169        let update_interval = Duration::from_secs(3).max(sysinfo::MINIMUM_CPU_UPDATE_INTERVAL);
170        // Do two initial updates so that we have sufficient data for
171        // computing a delta; one here outside the loop, and another
172        // at the top of the loop.
173        sys.refresh_cpu_usage();
174        std::thread::sleep(sysinfo::MINIMUM_CPU_UPDATE_INTERVAL);
175
176        loop {
177            sys.refresh_cpu_usage();
178            sys.refresh_processes_specifics(
179                ProcessesToUpdate::Some(&[my_pid]),
180                true,
181                ProcessRefreshKind::everything().without_tasks(),
182            );
183
184            let num_cpus = sys.cpus().len() as i64;
185
186            let sys_usage = sys.cpus().iter().map(|cpu| cpu.cpu_usage()).sum::<f32>() as i64;
187            SYS_CPU_USAGE.set(sys_usage);
188            SYS_CPU_USAGE_NORM.set(sys_usage / num_cpus);
189
190            if let Some(p) = sys.process(my_pid) {
191                let proc_usage = p.cpu_usage() as i64;
192                PROC_CPU_USAGE.set(proc_usage);
193                PROC_CPU_USAGE_NORM.set(proc_usage / num_cpus);
194            }
195
196            std::thread::sleep(update_interval);
197        }
198    });
199}
kumo_server_common/start.rs

kumo_server_common/
start.rs
