solti_api/grpc/convert/
task.rs

1//! # Task Conversion
2//!
3//! Converts create and apply manifests into domain desired state.
4//! Converts stored tasks, list pages, and watch events into protobuf responses.
5
6use solti_model::{Annotations, TASK_API_VERSION, TASK_KIND, Task, TaskManifest};
7
8use super::spec::{convert_labels, convert_task_spec, spec_to_proto};
9use crate::error::ApiError;
10use crate::proto_api;
11
12impl TryFrom<Task> for proto_api::Task {
13    type Error = ApiError;
14
15    fn try_from(task: Task) -> Result<Self, Self::Error> {
16        let (type_meta, metadata, spec, status) = task.into_parts();
17        let (observed_generation, phase, attempt, exit_code, error, conditions) =
18            status.into_parts();
19
20        Ok(proto_api::Task {
21            api_version: type_meta.api_version().to_owned(),
22            kind: type_meta.kind().to_owned(),
23            metadata: Some(proto_api::ObjectMeta::try_from(&metadata)?),
24            spec: Some(spec_to_proto(&spec)?),
25            status: Some(proto_api::TaskStatus {
26                observed_generation,
27                phase: proto_api::TaskPhase::try_from(phase)? as i32,
28                exit_code,
29                attempt,
30                error,
31                conditions: conditions
32                    .into_iter()
33                    .map(proto_api::TaskCondition::try_from)
34                    .collect::<Result<_, _>>()?,
35            }),
36        })
37    }
38}
39
40/// Converts a create or apply manifest into domain desired state.
41pub(crate) fn task_manifest_from_proto(
42    manifest: proto_api::TaskManifest,
43) -> Result<TaskManifest, ApiError> {
44    if manifest.api_version != TASK_API_VERSION {
45        return Err(ApiError::InvalidRequest(format!(
46            "Task apiVersion must be `{TASK_API_VERSION}`",
47        )));
48    }
49    if manifest.kind != TASK_KIND {
50        return Err(ApiError::InvalidRequest(format!(
51            "Task kind must be `{TASK_KIND}`",
52        )));
53    }
54    let metadata = manifest
55        .metadata
56        .ok_or_else(|| ApiError::InvalidRequest("missing metadata".into()))?;
57    let spec = manifest
58        .spec
59        .ok_or_else(|| ApiError::InvalidRequest("missing spec".into()))?;
60
61    let mut annotations = Annotations::new();
62    for (key, value) in metadata.annotations {
63        annotations.insert(key, value);
64    }
65    let manifest = TaskManifest::new(metadata.name, convert_task_spec(spec)?)
66        .and_then(|manifest| manifest.with_labels(convert_labels(metadata.labels)))
67        .and_then(|manifest| manifest.with_annotations(annotations))
68        .map_err(|error| ApiError::InvalidRequest(error.to_string()))?;
69    manifest
70        .validate()
71        .map_err(|error| ApiError::InvalidRequest(error.to_string()))?;
72    Ok(manifest)
73}
74
75/// Converts one domain task page into a protobuf list response.
76pub(crate) fn tasks_page_to_proto(
77    page: solti_model::TaskPage<solti_model::Task>,
78) -> Result<proto_api::ListTasksResponse, ApiError> {
79    let remaining_item_count = u64::try_from(page.remaining_item_count).map_err(|_| {
80        ApiError::Internal("remaining task count is outside the protobuf range".into())
81    })?;
82    let continuation = page
83        .continuation
84        .map(crate::continuation::encode)
85        .transpose()?
86        .unwrap_or_default();
87    let tasks: Vec<proto_api::Task> = page
88        .items
89        .into_iter()
90        .map(proto_api::Task::try_from)
91        .collect::<Result<_, _>>()?;
92
93    Ok(proto_api::ListTasksResponse {
94        tasks,
95        resource_version: page.resource_version,
96        r#continue: continuation,
97        remaining_item_count: (remaining_item_count > 0).then_some(remaining_item_count),
98    })
99}
100
101/// Converts one domain watch event into protobuf.
102pub(crate) fn task_watch_event_to_proto(
103    event: solti_model::TaskWatchEvent,
104) -> Result<proto_api::WatchTasksResponse, ApiError> {
105    let (event_type, task) = match event {
106        solti_model::TaskWatchEvent::Added(task) => (proto_api::TaskWatchEventType::Added, task),
107        solti_model::TaskWatchEvent::Modified(task) => {
108            (proto_api::TaskWatchEventType::Modified, task)
109        }
110        solti_model::TaskWatchEvent::Deleted(task) => {
111            (proto_api::TaskWatchEventType::Deleted, task)
112        }
113    };
114
115    Ok(proto_api::WatchTasksResponse {
116        r#type: event_type as i32,
117        object: Some(proto_api::Task::try_from(task)?),
118    })
119}
120
121#[cfg(test)]
122mod tests {
123    use super::*;
124    use solti_model::{
125        EmbeddedSpec, Flag, SubprocessMode, SubprocessSpec, TaskContinuation, TaskEnv, TaskFilter,
126        TaskId, TaskPhase, TaskSpec, TaskWorkload,
127    };
128    use std::time::UNIX_EPOCH;
129
130    fn subprocess_workload() -> TaskWorkload {
131        TaskWorkload::Subprocess(SubprocessSpec::new(
132            SubprocessMode::Command {
133                command: "ls".into(),
134                args: vec![],
135            },
136            TaskEnv::new(),
137            None,
138            Flag::from(true),
139        ))
140    }
141
142    #[test]
143    fn task_converts_correctly() {
144        let spec = TaskSpec::builder("my-slot", subprocess_workload(), 5_000_u64)
145            .build()
146            .unwrap();
147        let mut task = Task::new("task-42", spec).unwrap();
148        task.set_resource_version("1").unwrap();
149
150        task.transition_starting(1, 1, "2").unwrap();
151        task.transition_finished(1, 1, TaskPhase::Failed, Some("first".into()), None, "3")
152            .unwrap();
153        task.transition_starting(1, 2, "4").unwrap();
154        task.transition_finished(1, 2, TaskPhase::Failed, Some("boom".into()), None, "5")
155            .unwrap();
156        task.transition_starting(1, 3, "6").unwrap();
157        task.transition_finished(1, 3, TaskPhase::Failed, Some("boom".into()), None, "7")
158            .unwrap();
159
160        let created_ms = task
161            .metadata()
162            .creation_timestamp()
163            .duration_since(UNIX_EPOCH)
164            .unwrap()
165            .as_millis() as i64;
166        let bumped_version = task.metadata().resource_version().to_owned();
167
168        let proto = proto_api::Task::try_from(task).expect("conversion must succeed");
169        assert_eq!(proto.api_version, TASK_API_VERSION);
170        assert_eq!(proto.kind, TASK_KIND);
171
172        let meta = proto.metadata.unwrap();
173        assert_eq!(meta.name, "task-42");
174        assert!(!meta.uid.is_empty());
175        assert_eq!(meta.creation_timestamp, created_ms);
176        assert_eq!(meta.generation, 1);
177        assert_eq!(meta.resource_version, bumped_version);
178
179        let spec = proto.spec.unwrap();
180        assert_eq!(spec.slot, "my-slot");
181
182        let status = proto.status.unwrap();
183        assert_eq!(status.phase, proto_api::TaskPhase::Failed as i32);
184        assert_eq!(status.attempt, 3);
185        assert_eq!(status.error, Some("boom".to_string()));
186        assert_eq!(status.conditions.len(), 1);
187        let condition = &status.conditions[0];
188        assert_eq!(condition.r#type, "Reconciled");
189        assert_eq!(condition.status, proto_api::ConditionStatus::True as i32);
190        assert_eq!(condition.observed_generation, 1);
191        assert_eq!(condition.reason, "RuntimeAccepted");
192        assert!(!condition.message.is_empty());
193        assert!(condition.last_transition_time > 0);
194    }
195
196    #[test]
197    fn task_no_error() {
198        let spec = TaskSpec::builder("slot", subprocess_workload(), 5_000_u64)
199            .build()
200            .unwrap();
201        let mut task = Task::new("task-1", spec).unwrap();
202        task.transition_starting(1, 1, "1").unwrap();
203        task.transition_finished(1, 1, TaskPhase::Succeeded, None, Some(0), "2")
204            .unwrap();
205
206        let proto = proto_api::Task::try_from(task).expect("conversion must succeed");
207        let status = proto.status.unwrap();
208        assert_eq!(status.error, None);
209        assert_eq!(status.exit_code, Some(0));
210    }
211
212    #[test]
213    fn list_response_carries_snapshot_continuation_and_remaining_count() {
214        let mk = |id: &str| {
215            let spec = TaskSpec::builder("slot", subprocess_workload(), 5_000_u64)
216                .build()
217                .unwrap();
218            Task::new(id, spec).unwrap()
219        };
220        let continuation =
221            TaskContinuation::new("store:9", TaskFilter::new(), TaskId::new("task-2").unwrap())
222                .unwrap();
223        let page = solti_model::TaskPage {
224            items: vec![mk("task-1"), mk("task-2")],
225            resource_version: "store:9".into(),
226            continuation: Some(continuation.clone()),
227            remaining_item_count: 3,
228        };
229
230        let resp = tasks_page_to_proto(page).expect("conversion must succeed");
231
232        assert_eq!(resp.tasks.len(), 2);
233        assert_eq!(resp.resource_version, "store:9");
234        assert_eq!(resp.remaining_item_count, Some(3));
235        assert_eq!(
236            crate::continuation::decode(&resp.r#continue).unwrap(),
237            continuation
238        );
239    }
240
241    #[test]
242    fn handler_output_with_embedded_workload_is_an_internal_error() {
243        let spec = TaskSpec::builder(
244            "slot",
245            TaskWorkload::Embedded(EmbeddedSpec::new("test-v1").unwrap()),
246            5_000_u64,
247        )
248        .build()
249        .unwrap();
250        let task = Task::new("task-1", spec).unwrap();
251        let err = proto_api::Task::try_from(task).unwrap_err();
252        assert!(matches!(&err, ApiError::Internal(msg) if msg.contains("Embedded")));
253    }
254
255    #[test]
256    fn request_rejects_wrong_task_gvk() {
257        let proto = proto_api::TaskManifest {
258            api_version: "other.io/v1".into(),
259            kind: TASK_KIND.into(),
260            metadata: Some(proto_api::TaskManifestMeta {
261                name: "task-1".into(),
262                ..Default::default()
263            }),
264            spec: None,
265        };
266
267        let err = task_manifest_from_proto(proto).unwrap_err();
268        assert!(matches!(err, ApiError::InvalidRequest(msg) if msg.contains("apiVersion")));
269    }
270
271    #[test]
272    fn request_rejects_invalid_user_metadata() {
273        let spec = TaskSpec::builder("slot", subprocess_workload(), 5_000_u64)
274            .build()
275            .unwrap();
276        let mut labels = std::collections::HashMap::new();
277        labels.insert("bad key".to_owned(), "value".to_owned());
278        let proto = proto_api::TaskManifest {
279            api_version: TASK_API_VERSION.into(),
280            kind: TASK_KIND.into(),
281            metadata: Some(proto_api::TaskManifestMeta {
282                name: "task-1".into(),
283                labels,
284                ..Default::default()
285            }),
286            spec: Some(spec_to_proto(&spec).unwrap()),
287        };
288
289        let err = task_manifest_from_proto(proto).unwrap_err();
290        assert!(matches!(err, ApiError::InvalidRequest(msg) if msg.contains("label key")));
291    }
292
293    #[test]
294    fn request_manifest_converts_only_desired_state() {
295        let spec = TaskSpec::builder("slot", subprocess_workload(), 5_000_u64)
296            .build()
297            .unwrap();
298        let proto = proto_api::TaskManifest {
299            api_version: TASK_API_VERSION.into(),
300            kind: TASK_KIND.into(),
301            metadata: Some(proto_api::TaskManifestMeta {
302                name: "task-1".into(),
303                labels: [("app.kubernetes.io/name".into(), "worker".into())]
304                    .into_iter()
305                    .collect(),
306                annotations: [("example.io/note".into(), "desired".into())]
307                    .into_iter()
308                    .collect(),
309            }),
310            spec: Some(spec_to_proto(&spec).unwrap()),
311        };
312
313        let manifest = task_manifest_from_proto(proto).unwrap();
314
315        assert_eq!(manifest.name(), "task-1");
316        assert_eq!(
317            manifest.metadata().labels().get("app.kubernetes.io/name"),
318            Some("worker")
319        );
320        assert_eq!(
321            manifest.metadata().annotations().get("example.io/note"),
322            Some("desired")
323        );
324    }
325}
solti_api/grpc/convert/task.rs

solti_api/grpc/convert/
task.rs
