rskit_workload/
manager.rs

1//! Workload lifecycle manager and optional capability traits.
2//!
3//! Every backend implements [`Manager`].
4//! Backends opt into extra capabilities (exec, stats, log streaming, event watching) by implementing the focused capability traits,
5//! mirroring gokit's optional provider interfaces.
6
7use async_trait::async_trait;
8use futures::stream::BoxStream;
9use rskit_errors::AppResult;
10
11use crate::report::{
12    DeployResult, DiskUsage, ExecResult, ImageDetail, ImageEvent, SystemInfo, WaitResult,
13    WorkloadEvent, WorkloadInfo, WorkloadStats, WorkloadStatus,
14};
15use crate::spec::{DeployRequest, ImageEventFilter, ListFilter, LogOptions};
16
17/// Core workload lifecycle operations. All backends implement this trait.
18///
19/// Every method takes an explicit workload identifier
20/// and is cancellation-aware through the caller's async runtime;
21/// backends must apply their own timeouts to remote calls.
22#[async_trait]
23pub trait Manager: Send + Sync {
24    /// Create and start a workload.
25    async fn deploy(&self, request: DeployRequest) -> AppResult<DeployResult>;
26
27    /// Gracefully stop a running workload.
28    async fn stop(&self, id: &str) -> AppResult<()>;
29
30    /// Remove a stopped workload and clean up its resources.
31    async fn remove(&self, id: &str) -> AppResult<()>;
32
33    /// Stop and restart a workload.
34    async fn restart(&self, id: &str) -> AppResult<()>;
35
36    /// Return the current status of a workload.
37    async fn status(&self, id: &str) -> AppResult<WorkloadStatus>;
38
39    /// Block until the workload exits, returning its exit status.
40    async fn wait(&self, id: &str) -> AppResult<WaitResult>;
41
42    /// Return buffered log lines for a workload.
43    async fn logs(&self, id: &str, options: LogOptions) -> AppResult<Vec<String>>;
44
45    /// List workloads matching `filter`.
46    async fn list(&self, filter: ListFilter) -> AppResult<Vec<WorkloadInfo>>;
47
48    /// Verify the backend runtime is reachable and healthy.
49    async fn health_check(&self) -> AppResult<()>;
50}
51
52/// Implemented by backends that can execute a command inside a running workload.
53#[async_trait]
54pub trait ExecCapable: Send + Sync {
55    /// Execute `command` inside the workload identified by `id`.
56    async fn exec(&self, id: &str, command: &[String]) -> AppResult<ExecResult>;
57}
58
59/// Implemented by backends that expose real-time resource usage statistics.
60#[async_trait]
61pub trait StatsCapable: Send + Sync {
62    /// Return a resource usage snapshot for the workload identified by `id`.
63    async fn stats(&self, id: &str) -> AppResult<WorkloadStats>;
64}
65
66/// Implemented by backends that can stream log lines in real time.
67#[async_trait]
68pub trait LogStreamer: Send + Sync {
69    /// Stream log lines for the workload identified by `id`.
70    async fn stream_logs(
71        &self,
72        id: &str,
73        options: LogOptions,
74    ) -> AppResult<BoxStream<'static, AppResult<String>>>;
75}
76
77/// Implemented by backends that can watch workload lifecycle events.
78#[async_trait]
79pub trait EventWatcher: Send + Sync {
80    /// Stream lifecycle events for workloads matching `filter`.
81    async fn watch_events(
82        &self,
83        filter: ListFilter,
84    ) -> AppResult<BoxStream<'static, AppResult<WorkloadEvent>>>;
85}
86
87/// Implemented by backends that report host-level system information.
88#[async_trait]
89pub trait SystemInfoCapable: Send + Sync {
90    /// Return host information for the backend runtime.
91    async fn system_info(&self) -> AppResult<SystemInfo>;
92}
93
94/// Implemented by backends that report runtime disk usage.
95#[async_trait]
96pub trait DiskUsageCapable: Send + Sync {
97    /// Return disk usage for the backend runtime, broken down by category.
98    async fn disk_usage(&self) -> AppResult<DiskUsage>;
99}
100
101/// Implemented by backends that can inspect image metadata.
102#[async_trait]
103pub trait ImageInspector: Send + Sync {
104    /// Return detailed metadata for the image identified by `reference`.
105    async fn image_inspect(&self, reference: &str) -> AppResult<ImageDetail>;
106}
107
108/// Implemented by backends that can watch image lifecycle events.
109#[async_trait]
110pub trait ImageEventWatcher: Send + Sync {
111    /// Stream image lifecycle events matching `filter`.
112    async fn watch_image_events(
113        &self,
114        filter: ImageEventFilter,
115    ) -> AppResult<BoxStream<'static, AppResult<ImageEvent>>>;
116}
117
118#[cfg(test)]
119mod tests {
120    use std::sync::Arc;
121
122    use futures::StreamExt;
123
124    use super::*;
125    use crate::test_support::FakeManager;
126
127    #[tokio::test]
128    async fn manager_is_object_safe_and_drives_every_method() {
129        let manager: Arc<dyn Manager> = Arc::new(FakeManager);
130        let result = manager
131            .deploy(DeployRequest::new("api", "nginx"))
132            .await
133            .unwrap();
134        assert_eq!(result.id, "api-1");
135        assert!(manager.status(&result.id).await.unwrap().running);
136        assert!(manager.health_check().await.is_ok());
137        assert_eq!(
138            manager
139                .logs(&result.id, LogOptions::default())
140                .await
141                .unwrap()
142                .len(),
143            1
144        );
145        assert!(manager.wait(&result.id).await.is_ok());
146        assert!(
147            manager
148                .list(ListFilter::default())
149                .await
150                .unwrap()
151                .is_empty()
152        );
153        manager.restart(&result.id).await.unwrap();
154        manager.stop(&result.id).await.unwrap();
155        manager.remove(&result.id).await.unwrap();
156    }
157
158    #[tokio::test]
159    async fn capability_traits_are_usable_as_trait_objects() {
160        let exec: Arc<dyn ExecCapable> = Arc::new(FakeManager);
161        assert_eq!(
162            exec.exec("id", &["ls".to_string()])
163                .await
164                .unwrap()
165                .exit_code,
166            0
167        );
168
169        let stats: Arc<dyn StatsCapable> = Arc::new(FakeManager);
170        assert_eq!(stats.stats("id").await.unwrap().pids, 0);
171
172        let streamer: Arc<dyn LogStreamer> = Arc::new(FakeManager);
173        let lines: Vec<_> = streamer
174            .stream_logs("id", LogOptions::default())
175            .await
176            .unwrap()
177            .filter_map(|line| async move { line.ok() })
178            .collect()
179            .await;
180        assert_eq!(lines, vec!["a".to_string(), "b".to_string()]);
181
182        let watcher: Arc<dyn EventWatcher> = Arc::new(FakeManager);
183        let events: Vec<_> = watcher
184            .watch_events(ListFilter::default())
185            .await
186            .unwrap()
187            .collect()
188            .await;
189        assert_eq!(events.len(), 1);
190    }
191
192    #[tokio::test]
193    async fn optional_runtime_capabilities_are_usable_as_trait_objects() {
194        let sysinfo: Arc<dyn SystemInfoCapable> = Arc::new(FakeManager);
195        assert_eq!(sysinfo.system_info().await.unwrap().cpus, 8);
196
197        let disk: Arc<dyn DiskUsageCapable> = Arc::new(FakeManager);
198        assert_eq!(disk.disk_usage().await.unwrap().images_size, 1024);
199
200        let inspector: Arc<dyn ImageInspector> = Arc::new(FakeManager);
201        assert_eq!(
202            inspector.image_inspect("nginx:latest").await.unwrap().id,
203            "nginx:latest"
204        );
205
206        let img_watcher: Arc<dyn ImageEventWatcher> = Arc::new(FakeManager);
207        let events: Vec<_> = img_watcher
208            .watch_image_events(ImageEventFilter::default())
209            .await
210            .unwrap()
211            .collect()
212            .await;
213        assert_eq!(events.len(), 1);
214    }
215}
rskit_workload/manager.rs

rskit_workload/
manager.rs
