camel_function/provider/
container.rs

1use crate::pool::{RunnerHandle, RunnerPoolKey};
2use crate::protocol::ProtocolClient;
3use camel_api::Exchange;
4use camel_api::function::*;
5use dashmap::DashMap;
6use std::sync::Arc;
7use tokio_util::sync::CancellationToken;
8
9use super::{FunctionHealthStatus, FunctionProvider, ProviderError};
10
11struct ContainerEntry {
12    container_id: String,
13    endpoint: String,
14}
15
16#[derive(Debug, Clone)]
17pub enum PullPolicy {
18    Always,
19    Never,
20    IfMissing,
21}
22
23impl std::fmt::Debug for ContainerProvider {
24    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
25        f.debug_struct("ContainerProvider")
26            .field("image", &self.image)
27            .field("active_containers", &self.containers_by_handle.len())
28            .finish()
29    }
30}
31
32#[derive(Debug, Clone)]
33pub struct ContainerProviderBuilder {
34    image: String,
35    boot_timeout: std::time::Duration,
36    pull_policy: PullPolicy,
37    instance_id: Option<String>,
38}
39
40impl Default for ContainerProviderBuilder {
41    fn default() -> Self {
42        Self {
43            image: "kennycallado/deno-runner:latest".to_string(),
44            boot_timeout: std::time::Duration::from_secs(10),
45            pull_policy: PullPolicy::IfMissing,
46            instance_id: None,
47        }
48    }
49}
50
51impl ContainerProviderBuilder {
52    pub fn new() -> Self {
53        Self::default()
54    }
55
56    pub fn image(mut self, image: impl Into<String>) -> Self {
57        self.image = image.into();
58        self
59    }
60
61    pub fn boot_timeout(mut self, timeout: std::time::Duration) -> Self {
62        self.boot_timeout = timeout;
63        self
64    }
65
66    pub fn pull_policy(mut self, policy: PullPolicy) -> Self {
67        self.pull_policy = policy;
68        self
69    }
70
71    pub fn instance_id(mut self, id: impl Into<String>) -> Self {
72        self.instance_id = Some(id.into());
73        self
74    }
75
76    pub fn build(self) -> Result<ContainerProvider, ProviderError> {
77        let docker = bollard::Docker::connect_with_local_defaults()
78            .map_err(|e| ProviderError::SpawnFailed(format!("docker connect: {e}")))?;
79        let instance_id = self.instance_id.unwrap_or_else(|| {
80            let hash = blake3::hash(
81                format!(
82                    "{}-{}",
83                    self.image,
84                    std::time::SystemTime::now()
85                        .duration_since(std::time::UNIX_EPOCH)
86                        .unwrap_or_default()
87                        .as_nanos()
88                )
89                .as_bytes(),
90            );
91            format!("inst-{}", &hash.to_hex()[..12])
92        });
93        Ok(ContainerProvider {
94            docker,
95            image: self.image,
96            boot_timeout: self.boot_timeout,
97            pull_policy: self.pull_policy,
98            client: ProtocolClient::new(),
99            containers_by_handle: DashMap::new(),
100            instance_id,
101        })
102    }
103}
104
105pub struct ContainerProvider {
106    docker: bollard::Docker,
107    image: String,
108    instance_id: String,
109    boot_timeout: std::time::Duration,
110    pull_policy: PullPolicy,
111    client: ProtocolClient,
112    containers_by_handle: DashMap<String, ContainerEntry>,
113}
114
115impl ContainerProvider {
116    pub fn builder() -> ContainerProviderBuilder {
117        ContainerProviderBuilder::new()
118    }
119
120    pub async fn cleanup_all(&self) {
121        let entries: Vec<(String, String)> = self
122            .containers_by_handle
123            .iter()
124            .map(|e| (e.key().clone(), e.container_id.clone()))
125            .collect();
126        for (handle_id, container_id) in entries {
127            self.stop_and_remove_container(&container_id).await;
128            self.containers_by_handle.remove(&handle_id);
129        }
130    }
131
132    pub fn instance_id(&self) -> &str {
133        &self.instance_id
134    }
135
136    pub async fn is_clean(&self) -> bool {
137        self.list_instance_containers().await.is_empty()
138    }
139
140    pub async fn list_instance_containers(&self) -> Vec<String> {
141        let options = bollard::query_parameters::ListContainersOptions {
142            filters: Some(std::collections::HashMap::from([(
143                "label".to_string(),
144                vec![format!("camel.function.instance={}", self.instance_id)],
145            )])),
146            ..Default::default()
147        };
148        match self.docker.list_containers(Some(options)).await {
149            Ok(containers) => containers.into_iter().filter_map(|c| c.id).collect(),
150            Err(_) => vec![],
151        }
152    }
153
154    async fn stop_and_remove_container(&self, container_id: &str) {
155        let _ = self.docker.stop_container(container_id, None).await;
156        match self
157            .docker
158            .remove_container(
159                container_id,
160                Some(bollard::query_parameters::RemoveContainerOptions {
161                    force: true,
162                    ..Default::default()
163                }),
164            )
165            .await
166        {
167            Ok(()) => {}
168            Err(bollard::errors::Error::DockerResponseServerError {
169                status_code: 404, ..
170            }) => {}
171            Err(e) => {
172                tracing::warn!(target: "camel_function::container", %container_id, "remove error: {e}");
173            }
174        }
175    }
176
177    async fn allocate_host_port(&self) -> Result<u16, ProviderError> {
178        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
179            .await
180            .map_err(|e| ProviderError::SpawnFailed(format!("allocate port: {e}")))?;
181        let port = listener
182            .local_addr()
183            .map_err(|e| ProviderError::SpawnFailed(format!("get port: {e}")))?
184            .port();
185        drop(listener);
186        Ok(port)
187    }
188
189    pub async fn spawn_runner(&self, runtime: &str) -> Result<RunnerHandle, ProviderError> {
190        let key = RunnerPoolKey {
191            runtime: runtime.to_string(),
192        };
193        FunctionProvider::spawn(self, &key).await
194    }
195
196    pub async fn shutdown_runner(&self, handle: RunnerHandle) -> Result<(), ProviderError> {
197        FunctionProvider::shutdown(self, handle).await
198    }
199
200    pub async fn health_runner(
201        &self,
202        handle: &RunnerHandle,
203    ) -> Result<FunctionHealthStatus, ProviderError> {
204        FunctionProvider::health(self, handle).await
205    }
206
207    pub async fn register_function(
208        &self,
209        handle: &RunnerHandle,
210        def: &FunctionDefinition,
211    ) -> Result<(), ProviderError> {
212        FunctionProvider::register(self, handle, def).await
213    }
214
215    pub async fn unregister_function(
216        &self,
217        handle: &RunnerHandle,
218        id: &FunctionId,
219    ) -> Result<(), ProviderError> {
220        FunctionProvider::unregister(self, handle, id).await
221    }
222
223    pub async fn invoke_function(
224        &self,
225        handle: &RunnerHandle,
226        function_id: &FunctionId,
227        exchange: &Exchange,
228    ) -> Result<ExchangePatch, ProviderError> {
229        FunctionProvider::invoke(
230            self,
231            handle,
232            function_id,
233            exchange,
234            std::time::Duration::from_millis(5000),
235        )
236        .await
237    }
238
239    /// Pull the configured image according to the current [PullPolicy].
240    async fn pull_image_if_needed(&self) -> Result<(), ProviderError> {
241        match self.pull_policy {
242            PullPolicy::Never => {}
243            PullPolicy::Always => {
244                self.pull_image().await?;
245            }
246            PullPolicy::IfMissing => match self.docker.inspect_image(&self.image).await {
247                Ok(_) => {}
248                Err(bollard::errors::Error::DockerResponseServerError {
249                    status_code: 404, ..
250                }) => {
251                    self.pull_image().await?;
252                }
253                Err(e) => {
254                    return Err(inspect_error_to_spawn_failed(&self.image, e));
255                }
256            },
257        }
258        Ok(())
259    }
260}
261
262/// Converts a non-404 Docker inspect error into a [`ProviderError::SpawnFailed`].
263///
264/// Extracted as a free function so the classification logic can be unit-tested
265/// without a live Docker daemon.
266fn inspect_error_to_spawn_failed(image: &str, e: bollard::errors::Error) -> ProviderError {
267    ProviderError::SpawnFailed(format!("failed to inspect image '{image}': {e}"))
268}
269
270/// Build the container creation request with security hardening.
271fn build_container_config(
272    image: &str,
273    host_port: u16,
274    instance_id: &str,
275    handle_id: &str,
276) -> bollard::models::ContainerCreateBody {
277    let labels = std::collections::HashMap::from([
278        ("camel.function.runner".to_string(), "true".to_string()),
279        ("camel.function.context".to_string(), handle_id.to_string()),
280        (
281            "camel.function.instance".to_string(),
282            instance_id.to_string(),
283        ),
284    ]);
285
286    bollard::models::ContainerCreateBody {
287        image: Some(image.to_string()),
288        env: Some(vec![
289            "PORT=8080".to_string(),
290            "DENO_NO_PROMPT=1".to_string(),
291            "DENO_DIR=/tmp/deno".to_string(),
292        ]),
293        labels: Some(labels),
294        exposed_ports: Some(vec!["8080/tcp".to_string()]),
295        host_config: Some(bollard::models::HostConfig {
296            port_bindings: Some(std::collections::HashMap::from([(
297                "8080/tcp".to_string(),
298                Some(vec![bollard::models::PortBinding {
299                    host_ip: Some("127.0.0.1".to_string()),
300                    host_port: Some(host_port.to_string()),
301                }]),
302            )])),
303            init: Some(true),
304            auto_remove: Some(false),
305            cap_drop: Some(vec!["ALL".to_string()]),
306            security_opt: Some(vec!["no-new-privileges".to_string()]),
307            readonly_rootfs: Some(true),
308            memory: Some(256 * 1024 * 1024),
309            nano_cpus: Some(1_000_000_000),
310            pids_limit: Some(100),
311            tmpfs: Some(std::collections::HashMap::from([(
312                "/tmp".to_string(),
313                String::new(),
314            )])),
315            dns_search: Some(vec![".".to_string()]),
316            ..Default::default()
317        }),
318        ..Default::default()
319    }
320}
321
322impl ContainerProvider {
323    /// Execute the image pull via bollard's `create_image` API.
324    async fn pull_image(&self) -> Result<(), ProviderError> {
325        use futures::StreamExt;
326
327        let options = bollard::query_parameters::CreateImageOptionsBuilder::default()
328            .from_image(&self.image)
329            .build();
330
331        let mut stream = self.docker.create_image(Some(options), None, None);
332        while let Some(item) = stream.next().await {
333            match item {
334                Ok(_) => {}
335                Err(e) => {
336                    return Err(ProviderError::SpawnFailed(format!(
337                        "image pull failed for '{}': {e}",
338                        self.image
339                    )));
340                }
341            }
342        }
343        Ok(())
344    }
345
346    fn spawn_log_forwarder(&self, container_id: String) {
347        use futures::StreamExt;
348
349        let docker = self.docker.clone();
350        tokio::spawn(async move {
351            let options = bollard::query_parameters::LogsOptions {
352                follow: true,
353                stdout: true,
354                stderr: true,
355                ..Default::default()
356            };
357            let mut stream = docker.logs(&container_id, Some(options));
358
359            while let Some(msg) = stream.next().await {
360                match msg {
361                    Ok(log_output) => {
362                        let text = match &log_output {
363                            bollard::container::LogOutput::StdOut { message } => {
364                                String::from_utf8_lossy(message).into_owned()
365                            }
366                            bollard::container::LogOutput::StdErr { message } => {
367                                String::from_utf8_lossy(message).into_owned()
368                            }
369                            _ => continue,
370                        };
371                        let trimmed = text.trim_end();
372                        if trimmed.is_empty() {
373                            continue;
374                        }
375                        match &log_output {
376                            bollard::container::LogOutput::StdOut { .. } => {
377                                tracing::info!(target: "camel_function::runner", "{trimmed}");
378                            }
379                            bollard::container::LogOutput::StdErr { .. } => {
380                                tracing::warn!(target: "camel_function::runner", "{trimmed}");
381                            }
382                            _ => {}
383                        }
384                    }
385                    Err(e) => {
386                        tracing::debug!(target: "camel_function::container", "log stream error: {e}");
387                        break;
388                    }
389                }
390            }
391        });
392    }
393}
394
395impl super::sealed::Sealed for ContainerProvider {}
396
397#[async_trait::async_trait]
398impl FunctionProvider for ContainerProvider {
399    async fn spawn(&self, _key: &RunnerPoolKey) -> Result<RunnerHandle, ProviderError> {
400        let hash = blake3::hash(
401            format!(
402                "{}",
403                std::time::SystemTime::now()
404                    .duration_since(std::time::UNIX_EPOCH)
405                    .unwrap_or_default()
406                    .as_nanos()
407            )
408            .as_bytes(),
409        )
410        .to_hex();
411        let handle_id = format!("deno-{}", &hash[..16]);
412        let host_port = self.allocate_host_port().await?;
413
414        tracing::debug!(
415            target: "camel_function::container",
416            %handle_id,
417            image = %self.image,
418            "spawning container"
419        );
420
421        self.pull_image_if_needed().await?;
422
423        let config = build_container_config(&self.image, host_port, &self.instance_id, &handle_id);
424
425        let create_opts = bollard::query_parameters::CreateContainerOptions {
426            name: Some(handle_id.clone()),
427            ..Default::default()
428        };
429
430        let create_result = self
431            .docker
432            .create_container(Some(create_opts), config)
433            .await
434            .map_err(|e| ProviderError::SpawnFailed(format!("create container: {e}")))?;
435
436        let container_id = create_result.id;
437
438        if let Err(e) = self.docker.start_container(&container_id, None).await {
439            let _ = self.stop_and_remove_container(&container_id).await;
440            return Err(ProviderError::SpawnFailed(format!("start container: {e}")));
441        }
442
443        let endpoint = format!("http://127.0.0.1:{host_port}");
444
445        // Wait for the container to become healthy within boot_timeout.
446        let boot_timeout = self.boot_timeout;
447        let client = &self.client;
448        let endpoint_clone = endpoint.clone();
449        let ready_result = tokio::time::timeout(boot_timeout, async {
450            loop {
451                tokio::time::sleep(std::time::Duration::from_millis(100)).await;
452                if client.health(&endpoint_clone).await.is_ok() {
453                    return;
454                }
455            }
456        })
457        .await;
458
459        if ready_result.is_err() {
460            let _ = self.stop_and_remove_container(&container_id).await;
461            return Err(ProviderError::SpawnFailed("container boot timeout".into()));
462        }
463
464        self.spawn_log_forwarder(container_id.clone());
465
466        self.containers_by_handle.insert(
467            handle_id.clone(),
468            ContainerEntry {
469                container_id,
470                endpoint,
471            },
472        );
473
474        Ok(RunnerHandle {
475            id: handle_id,
476            state: Arc::new(std::sync::Mutex::new(crate::pool::RunnerState::Booting)),
477            cancel: CancellationToken::new(),
478        })
479    }
480
481    async fn shutdown(&self, handle: RunnerHandle) -> Result<(), ProviderError> {
482        handle.cancel.cancel();
483        let entry = match self.containers_by_handle.remove(&handle.id) {
484            Some((_, entry)) => entry,
485            None => return Ok(()),
486        };
487        let _ = self.client.shutdown(&entry.endpoint).await;
488        self.stop_and_remove_container(&entry.container_id).await;
489        Ok(())
490    }
491
492    async fn health(&self, handle: &RunnerHandle) -> Result<FunctionHealthStatus, ProviderError> {
493        let endpoint = self
494            .containers_by_handle
495            .get(&handle.id)
496            .ok_or_else(|| ProviderError::HealthFailed(format!("unknown handle {}", handle.id)))?
497            .endpoint
498            .clone();
499        self.client.health(&endpoint).await
500    }
501
502    async fn register(
503        &self,
504        handle: &RunnerHandle,
505        def: &FunctionDefinition,
506    ) -> Result<(), ProviderError> {
507        let endpoint = self
508            .containers_by_handle
509            .get(&handle.id)
510            .ok_or_else(|| ProviderError::RegisterFailed(format!("unknown handle {}", handle.id)))?
511            .endpoint
512            .clone();
513        self.client.register(&endpoint, def).await
514    }
515
516    async fn unregister(
517        &self,
518        handle: &RunnerHandle,
519        id: &FunctionId,
520    ) -> Result<(), ProviderError> {
521        let endpoint = self
522            .containers_by_handle
523            .get(&handle.id)
524            .ok_or_else(|| {
525                ProviderError::UnregisterFailed(format!("unknown handle {}", handle.id))
526            })?
527            .endpoint
528            .clone();
529        self.client.unregister(&endpoint, id).await
530    }
531
532    async fn invoke(
533        &self,
534        handle: &RunnerHandle,
535        id: &FunctionId,
536        ex: &Exchange,
537        timeout: std::time::Duration,
538    ) -> Result<ExchangePatch, ProviderError> {
539        let endpoint = self
540            .containers_by_handle
541            .get(&handle.id)
542            .ok_or_else(|| ProviderError::InvokeFailed(format!("unknown handle {}", handle.id)))?
543            .endpoint
544            .clone();
545        let resp = self.client.invoke(&endpoint, id, ex, timeout).await?;
546        if resp.ok {
547            let patch = resp.patch.unwrap_or_default();
548            Ok(patch
549                .to_exchange_patch()
550                .map_err(|e| ProviderError::InvokeFailed(e.to_string()))?)
551        } else {
552            let err = resp.error.unwrap_or_else(|| crate::protocol::ErrorWire {
553                kind: "unknown".into(),
554                message: "no error body".into(),
555                stack: None,
556            });
557            Err(ProviderError::InvokeFailed(format!(
558                "{}: {}",
559                err.kind, err.message
560            )))
561        }
562    }
563}
564
565impl Drop for ContainerProvider {
566    fn drop(&mut self) {
567        if self.containers_by_handle.is_empty() {
568            return;
569        }
570        let docker = self.docker.clone();
571        let container_ids: Vec<String> = self
572            .containers_by_handle
573            .iter()
574            .map(|e| e.container_id.clone())
575            .collect();
576        match tokio::runtime::Handle::try_current() {
577            Ok(handle) => {
578                drop(handle.spawn(async move {
579                    for id in container_ids {
580                        let _ = docker.stop_container(&id, None).await;
581                        let _ = docker
582                            .remove_container(
583                                &id,
584                                Some(bollard::query_parameters::RemoveContainerOptions {
585                                    force: true,
586                                    ..Default::default()
587                                }),
588                            )
589                            .await;
590                    }
591                }));
592            }
593            Err(_) => {
594                tracing::warn!(target: "camel_function::container", "container cleanup skipped: no tokio runtime");
595            }
596        }
597    }
598}
599
600#[cfg(test)]
601mod tests {
602    use super::*;
603
604    #[test]
605    fn inspect_non_404_becomes_spawn_failed() {
606        let err = bollard::errors::Error::DockerResponseServerError {
607            status_code: 500,
608            message: "internal server error".into(),
609        };
610        let result = inspect_error_to_spawn_failed("my-image:latest", err);
611        assert!(
612            matches!(result, ProviderError::SpawnFailed(ref msg) if msg.contains("my-image:latest")),
613            "expected SpawnFailed with image name, got: {result:?}"
614        );
615    }
616
617    #[test]
618    fn inspect_permission_denied_becomes_spawn_failed() {
619        let err = bollard::errors::Error::DockerResponseServerError {
620            status_code: 403,
621            message: "permission denied".into(),
622        };
623        let result = inspect_error_to_spawn_failed("private/image:1.0", err);
624        assert!(matches!(
625            result,
626            ProviderError::SpawnFailed(ref msg) if msg.contains("private/image:1.0")
627        ));
628    }
629
630    #[test]
631    fn container_config_has_security_hardening() {
632        let config =
633            build_container_config("denoland/deno:2.1.4", 8080, "test-instance", "test-handle");
634
635        let hc = config.host_config.expect("host_config must be set");
636
637        let cap_drop = hc.cap_drop.expect("cap_drop must be set");
638        assert!(
639            cap_drop.contains(&"ALL".to_string()),
640            "cap_drop must include ALL"
641        );
642
643        let security_opt = hc.security_opt.expect("security_opt must be set");
644        assert!(
645            security_opt.contains(&"no-new-privileges".to_string()),
646            "security_opt must include no-new-privileges"
647        );
648
649        assert_eq!(
650            hc.readonly_rootfs,
651            Some(true),
652            "readonly_rootfs must be true"
653        );
654
655        let mem = hc.memory.expect("memory limit must be set");
656        assert!(
657            mem > 0 && mem <= 512 * 1024 * 1024,
658            "memory must be a sane limit (got {mem})"
659        );
660
661        let cpus = hc.nano_cpus.expect("nano_cpus must be set");
662        assert!(cpus > 0, "nano_cpus must be positive");
663
664        let pids = hc.pids_limit.expect("pids_limit must be set");
665        assert!(
666            pids > 0 && pids <= 500,
667            "pids_limit must be a sane value (got {pids})"
668        );
669
670        let tmpfs = hc.tmpfs.expect("tmpfs must be set");
671        assert!(tmpfs.contains_key("/tmp"), "tmpfs must mount /tmp");
672
673        let dns_search = hc.dns_search.expect("dns_search must be set");
674        assert!(
675            dns_search.contains(&".".to_string()),
676            "dns_search must suppress defaults"
677        );
678
679        let env = config.env.expect("env must be set");
680        assert!(
681            env.iter().any(|e| e.contains("DENO_DIR=/tmp")),
682            "env must set DENO_DIR=/tmp for readonly rootfs"
683        );
684    }
685}
camel_function/provider/container.rs

camel_function/provider/
container.rs
