objectstore_server/
state.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use anyhow::Result;
5use objectstore_service::{StorageConfig, StorageService};
6use tokio::runtime::Handle;
7
8use crate::auth::PublicKeyDirectory;
9use crate::config::{Config, Storage};
10
11/// Shared reference to the objectstore [`Services`].
12pub type ServiceState = Arc<Services>;
13
14/// Reference to the objectstore business logic.
15///
16/// This structure is created during server startup and shared with all HTTP request handlers. It
17/// can be used to access the configured storage backends and other shared resources.
18///
19/// In request handlers, use `axum::extract::State<ServiceState>` to retrieve a shared reference to
20/// this structure.
21#[derive(Debug)]
22pub struct Services {
23    /// The server configuration.
24    pub config: Config,
25    /// Raw handle to the underlying storage service that does not enforce authorization checks.
26    ///
27    /// Consider using [`crate::auth::AuthAwareService`].
28    pub service: StorageService,
29    /// Directory for EdDSA public keys.
30    ///
31    /// The `kid` header field from incoming authorization tokens should correspond to a public key
32    /// in this directory that can be used to verify the token.
33    pub key_directory: PublicKeyDirectory,
34}
35
36impl Services {
37    /// Spawns all services and background tasks for objectstore.
38    ///
39    /// This returns a [`ServiceState`], which is a shared reference to the services suitable for
40    /// use in the web server.
41    pub async fn spawn(config: Config) -> Result<ServiceState> {
42        tokio::spawn(track_runtime_metrics(config.runtime.metrics_interval));
43
44        let high_volume = map_storage_config(&config.high_volume_storage);
45        let long_term = map_storage_config(&config.long_term_storage);
46        let service = StorageService::new(high_volume, long_term).await?;
47
48        let key_directory: PublicKeyDirectory = (&config.auth).try_into()?;
49
50        Ok(Arc::new(Self {
51            config,
52            service,
53            key_directory,
54        }))
55    }
56}
57
58fn map_storage_config(config: &'_ Storage) -> StorageConfig<'_> {
59    match config {
60        Storage::FileSystem { path } => StorageConfig::FileSystem { path },
61        Storage::S3Compatible { endpoint, bucket } => {
62            StorageConfig::S3Compatible { endpoint, bucket }
63        }
64        Storage::Gcs { endpoint, bucket } => StorageConfig::Gcs {
65            endpoint: endpoint.as_deref(),
66            bucket,
67        },
68        Storage::BigTable {
69            endpoint,
70            project_id,
71            instance_name,
72            table_name,
73            connections,
74        } => StorageConfig::BigTable {
75            endpoint: endpoint.as_deref(),
76            project_id,
77            instance_name,
78            table_name,
79            connections: *connections,
80        },
81    }
82}
83
84/// Periodically captures and reports internal Tokio runtime metrics.
85async fn track_runtime_metrics(interval: Duration) {
86    let mut ticker = tokio::time::interval(interval);
87    let metrics = Handle::current().metrics();
88
89    loop {
90        ticker.tick().await;
91        tracing::trace!("Capturing runtime metrics");
92
93        merni::gauge!("runtime.num_workers": metrics.num_workers());
94        merni::gauge!("runtime.num_alive_tasks": metrics.num_alive_tasks());
95        merni::gauge!("runtime.global_queue_depth": metrics.global_queue_depth());
96        merni::gauge!("runtime.num_blocking_threads": metrics.num_blocking_threads());
97        merni::gauge!("runtime.num_idle_blocking_threads": metrics.num_idle_blocking_threads());
98        merni::gauge!("runtime.blocking_queue_depth": metrics.blocking_queue_depth());
99
100        let registered_fds = metrics.io_driver_fd_registered_count();
101        let deregistered_fds = metrics.io_driver_fd_deregistered_count();
102        merni::gauge!("runtime.num_io_driver_fds": registered_fds - deregistered_fds);
103    }
104}
objectstore_server/state.rs

objectstore_server/
state.rs
