1use std::path::{Path, PathBuf};
7use std::sync::Arc;
8use std::time::Duration;
9
10use async_trait::async_trait;
11use base64::Engine;
12use tempfile::TempDir;
13
14use super::backend::{BackendHandle, LambdaBackend, RuntimeError, WarmInstance};
15use super::env_rewrite::rewrite_localhost_envs;
16use crate::state::LambdaFunction;
17
18pub struct DockerBackend {
20 cli: String,
21 instance_id: String,
22 host_alias: String,
32 add_host_arg: Option<String>,
35 server_port: u16,
39 sibling_host: String,
49 registry_host: String,
53 docker_config: Option<Arc<TempDir>>,
57}
58
59impl DockerBackend {
60 pub fn auto_detect(server_port: u16) -> Option<Self> {
65 let cli = fakecloud_core::container_net::detect_container_cli()?;
73 let instance_id = format!("fakecloud-{}", std::process::id());
74 let net = fakecloud_core::container_net::HostNetworking::detect(&cli);
75
76 let docker_config = build_local_registry_docker_config(server_port).map(Arc::new);
77 Some(Self {
78 cli,
79 instance_id,
80 host_alias: net.host_alias,
81 add_host_arg: net.add_host_arg,
82 server_port,
83 sibling_host: net.sibling_host,
84 registry_host: std::env::var("FAKECLOUD_ECR_REGISTRY_HOST")
85 .unwrap_or_else(|_| "127.0.0.1".to_string()),
86 docker_config,
87 })
88 }
89
90 fn apply_host_alias(&self, cmd: &mut tokio::process::Command) {
94 if let Some(arg) = &self.add_host_arg {
95 cmd.arg("--add-host").arg(arg);
96 }
97 }
98
99 fn docker_config_path(&self) -> Option<PathBuf> {
100 self.docker_config.as_ref().map(|d| d.path().to_path_buf())
101 }
102
103 async fn start_image_container(
109 &self,
110 func: &LambdaFunction,
111 layers: &[Vec<u8>],
112 ) -> Result<WarmInstance, RuntimeError> {
113 let image = func.image_uri.as_deref().ok_or_else(|| {
114 RuntimeError::ContainerStartFailed("PackageType=Image function has no ImageUri".into())
115 })?;
116
117 let local_pull_uri = fakecloud_core::ecr_uri::translate_to_local_at(
121 image,
122 &self.registry_host,
123 self.server_port,
124 );
125 let pull_uri = local_pull_uri.as_deref().unwrap_or(image);
126
127 let mut pull_cmd = tokio::process::Command::new(&self.cli);
128 if let Some(p) = self.docker_config_path() {
129 pull_cmd.env("DOCKER_CONFIG", p);
130 }
131 let pull_out = pull_cmd
132 .args(["pull", pull_uri])
133 .output()
134 .await
135 .map_err(|e| RuntimeError::ContainerStartFailed(format!("docker pull: {e}")))?;
136 if !pull_out.status.success() {
137 return Err(RuntimeError::ContainerStartFailed(format!(
138 "docker pull failed: {}",
139 String::from_utf8_lossy(&pull_out.stderr)
140 )));
141 }
142 let run_image = if let Some(ref local_uri) = local_pull_uri {
147 if fakecloud_core::ecr_uri::is_digest_ref(image) {
148 local_uri.clone()
149 } else {
150 let _ = tokio::process::Command::new(&self.cli)
151 .args(["tag", local_uri, image])
152 .output()
153 .await;
154 image.to_string()
155 }
156 } else {
157 image.to_string()
158 };
159
160 let mut cmd = tokio::process::Command::new(&self.cli);
161 cmd.arg("create")
162 .arg("-p")
163 .arg(":8080")
164 .arg("--label")
165 .arg(format!("fakecloud-lambda={}", func.function_name))
166 .arg("--label")
167 .arg(format!("fakecloud-instance={}", self.instance_id));
168 self.apply_host_alias(&mut cmd);
169
170 for (key, value) in rewrite_localhost_envs(&func.environment, &self.host_alias) {
171 cmd.arg("-e").arg(format!("{key}={value}"));
172 }
173 cmd.arg("-e")
174 .arg(format!("AWS_LAMBDA_FUNCTION_TIMEOUT={}", func.timeout));
175
176 let tmpfs_arg = ephemeral_storage_tmpfs_arg(func.ephemeral_storage_size);
177 cmd.arg("--tmpfs").arg(tmpfs_arg);
178
179 cmd.arg(&run_image);
180
181 let output = cmd
182 .output()
183 .await
184 .map_err(|e| RuntimeError::ContainerStartFailed(e.to_string()))?;
185 if !output.status.success() {
186 return Err(RuntimeError::ContainerStartFailed(
187 String::from_utf8_lossy(&output.stderr).to_string(),
188 ));
189 }
190 let container_id = String::from_utf8_lossy(&output.stdout).trim().to_string();
191
192 if let Err(e) = self.copy_layers_into(&container_id, layers).await {
193 self.remove_container(&container_id).await;
194 return Err(e);
195 }
196
197 let start_result = tokio::process::Command::new(&self.cli)
198 .args(["start", &container_id])
199 .output()
200 .await
201 .map_err(|e| RuntimeError::ContainerStartFailed(e.to_string()))?;
202 if !start_result.status.success() {
203 self.remove_container(&container_id).await;
204 return Err(RuntimeError::ContainerStartFailed(format!(
205 "docker start failed: {}",
206 String::from_utf8_lossy(&start_result.stderr)
207 )));
208 }
209
210 let port = self.query_host_port(&container_id).await?;
211 self.wait_for_ready(&container_id, port).await?;
212
213 tracing::info!(
214 function = %func.function_name,
215 container_id = %container_id,
216 port = port,
217 image = %image,
218 "Lambda image container started"
219 );
220
221 Ok(WarmInstance {
222 endpoint: format!("{}:{port}", self.sibling_host),
223 handle: BackendHandle::Container { id: container_id },
224 })
225 }
226
227 async fn start_zip_container(
228 &self,
229 func: &LambdaFunction,
230 zip_bytes: &[u8],
231 layers: &[Vec<u8>],
232 ) -> Result<WarmInstance, RuntimeError> {
233 let image = runtime_to_image(&func.runtime)
234 .ok_or_else(|| RuntimeError::UnsupportedRuntime(func.runtime.clone()))?;
235
236 let code_dir =
239 TempDir::new().map_err(|e| RuntimeError::ZipExtractionFailed(e.to_string()))?;
240 let zip_bytes = zip_bytes.to_vec();
241 let code_path = code_dir.path().to_path_buf();
242 tokio::task::spawn_blocking(move || extract_zip(&zip_bytes, &code_path))
243 .await
244 .map_err(|e| RuntimeError::ZipExtractionFailed(e.to_string()))??;
245
246 let mut cmd = tokio::process::Command::new(&self.cli);
248 cmd.arg("create")
249 .arg("-p")
250 .arg(":8080")
251 .arg("--label")
252 .arg(format!("fakecloud-lambda={}", func.function_name))
253 .arg("--label")
254 .arg(format!("fakecloud-instance={}", self.instance_id));
255 self.apply_host_alias(&mut cmd);
256
257 for (key, value) in rewrite_localhost_envs(&func.environment, &self.host_alias) {
258 cmd.arg("-e").arg(format!("{key}={value}"));
259 }
260
261 cmd.arg("-e")
262 .arg(format!("AWS_LAMBDA_FUNCTION_TIMEOUT={}", func.timeout));
263
264 let tmpfs_arg = ephemeral_storage_tmpfs_arg(func.ephemeral_storage_size);
265 cmd.arg("--tmpfs").arg(tmpfs_arg);
266
267 cmd.arg(&image).arg(&func.handler);
268
269 let output = cmd
270 .output()
271 .await
272 .map_err(|e| RuntimeError::ContainerStartFailed(e.to_string()))?;
273
274 if !output.status.success() {
275 let stderr = String::from_utf8_lossy(&output.stderr);
276 return Err(RuntimeError::ContainerStartFailed(stderr.to_string()));
277 }
278
279 let container_id = String::from_utf8_lossy(&output.stdout).trim().to_string();
280
281 let cp_result = tokio::process::Command::new(&self.cli)
283 .arg("cp")
284 .arg(format!("{}/.", code_dir.path().display()))
285 .arg(format!("{}:/var/task", container_id))
286 .output()
287 .await
288 .map_err(|e| RuntimeError::ContainerStartFailed(e.to_string()))?;
289
290 if !cp_result.status.success() {
291 self.remove_container(&container_id).await;
292 let stderr = String::from_utf8_lossy(&cp_result.stderr);
293 return Err(RuntimeError::ContainerStartFailed(format!(
294 "docker cp failed: {stderr}"
295 )));
296 }
297
298 if func.runtime.starts_with("provided") {
300 let cp_runtime = tokio::process::Command::new(&self.cli)
301 .arg("cp")
302 .arg(format!("{}/.", code_dir.path().display()))
303 .arg(format!("{}:/var/runtime", container_id))
304 .output()
305 .await
306 .map_err(|e| RuntimeError::ContainerStartFailed(e.to_string()))?;
307
308 if !cp_runtime.status.success() {
309 self.remove_container(&container_id).await;
310 let stderr = String::from_utf8_lossy(&cp_runtime.stderr);
311 return Err(RuntimeError::ContainerStartFailed(format!(
312 "docker cp to /var/runtime failed: {stderr}"
313 )));
314 }
315 }
316
317 if let Err(e) = self.copy_layers_into(&container_id, layers).await {
318 self.remove_container(&container_id).await;
319 return Err(e);
320 }
321
322 let start_result = tokio::process::Command::new(&self.cli)
325 .args(["start", &container_id])
326 .output()
327 .await
328 .map_err(|e| RuntimeError::ContainerStartFailed(e.to_string()))?;
329
330 if !start_result.status.success() {
331 self.remove_container(&container_id).await;
332 let stderr = String::from_utf8_lossy(&start_result.stderr);
333 return Err(RuntimeError::ContainerStartFailed(format!(
334 "docker start failed: {stderr}"
335 )));
336 }
337
338 let port = self.query_host_port(&container_id).await?;
339 self.wait_for_ready(&container_id, port).await?;
340
341 tracing::info!(
342 function = %func.function_name,
343 container_id = %container_id,
344 port = port,
345 runtime = %func.runtime,
346 "Lambda container started"
347 );
348
349 Ok(WarmInstance {
350 endpoint: format!("{}:{port}", self.sibling_host),
351 handle: BackendHandle::Container { id: container_id },
352 })
353 }
354
355 async fn query_host_port(&self, container_id: &str) -> Result<u16, RuntimeError> {
356 let port_output = tokio::process::Command::new(&self.cli)
357 .args(["port", container_id, "8080"])
358 .output()
359 .await
360 .map_err(|e| RuntimeError::ContainerStartFailed(e.to_string()))?;
361 let port_str = String::from_utf8_lossy(&port_output.stdout);
362 port_str
363 .trim()
364 .rsplit(':')
365 .next()
366 .and_then(|p| p.parse().ok())
367 .ok_or_else(|| {
368 RuntimeError::ContainerStartFailed(format!(
369 "could not determine port from: {}",
370 port_str.trim()
371 ))
372 })
373 }
374
375 async fn wait_for_ready(&self, container_id: &str, port: u16) -> Result<(), RuntimeError> {
376 for _ in 0..20 {
377 tokio::time::sleep(Duration::from_millis(500)).await;
378 if tokio::net::TcpStream::connect(format!("{}:{port}", self.sibling_host))
379 .await
380 .is_ok()
381 {
382 return Ok(());
383 }
384 }
385 self.remove_container(container_id).await;
386 Err(RuntimeError::ContainerStartFailed(
387 "container did not become ready within 10 seconds".to_string(),
388 ))
389 }
390
391 async fn copy_layers_into(
398 &self,
399 container_id: &str,
400 layers: &[Vec<u8>],
401 ) -> Result<(), RuntimeError> {
402 if layers.is_empty() {
403 return Ok(());
404 }
405 let layers_dir =
406 TempDir::new().map_err(|e| RuntimeError::ZipExtractionFailed(e.to_string()))?;
407 let layers_path = layers_dir.path().to_path_buf();
408 let layers_owned: Vec<Vec<u8>> = layers.to_vec();
409 tokio::task::spawn_blocking(move || {
410 for bytes in &layers_owned {
411 extract_zip(bytes, &layers_path)?;
412 }
413 Ok::<_, RuntimeError>(())
414 })
415 .await
416 .map_err(|e| RuntimeError::ZipExtractionFailed(e.to_string()))??;
417
418 let cp_result = tokio::process::Command::new(&self.cli)
419 .arg("cp")
420 .arg(format!("{}/.", layers_dir.path().display()))
421 .arg(format!("{}:/opt", container_id))
422 .output()
423 .await
424 .map_err(|e| RuntimeError::ContainerStartFailed(e.to_string()))?;
425 if !cp_result.status.success() {
426 let stderr = String::from_utf8_lossy(&cp_result.stderr);
427 return Err(RuntimeError::ContainerStartFailed(format!(
428 "docker cp layers to /opt failed: {stderr}"
429 )));
430 }
431 Ok(())
432 }
433
434 async fn remove_container(&self, container_id: &str) {
436 let _ = tokio::process::Command::new(&self.cli)
437 .args(["rm", "-f", container_id])
438 .output()
439 .await;
440 }
441}
442
443#[async_trait]
444impl LambdaBackend for DockerBackend {
445 fn name(&self) -> &str {
446 &self.cli
447 }
448
449 async fn launch(
450 &self,
451 func: &LambdaFunction,
452 code_zip: Option<&[u8]>,
453 layers: &[Vec<u8>],
454 _deploy_id: &str,
455 ) -> Result<WarmInstance, RuntimeError> {
456 if func.package_type == "Image" {
457 self.start_image_container(func, layers).await
458 } else {
459 let bytes =
460 code_zip.ok_or_else(|| RuntimeError::NoCodeZip(func.function_name.clone()))?;
461 self.start_zip_container(func, bytes, layers).await
462 }
463 }
464
465 async fn terminate(&self, handle: &BackendHandle) {
466 match handle {
467 BackendHandle::Container { id } => self.remove_container(id).await,
468 BackendHandle::Pod { .. } => {}
471 }
472 }
473
474 async fn instance_logs(&self, handle: &BackendHandle) -> Option<String> {
475 let BackendHandle::Container { id } = handle else {
476 return None;
477 };
478 let output = tokio::process::Command::new(&self.cli)
482 .args(["logs", "--tail", "200", id])
483 .output()
484 .await
485 .ok()?;
486 let mut combined = String::from_utf8_lossy(&output.stdout).into_owned();
487 combined.push_str(&String::from_utf8_lossy(&output.stderr));
488 if combined.is_empty() {
489 None
490 } else {
491 Some(combined)
492 }
493 }
494
495 async fn prepull_image(&self, image: &str) -> Result<(), RuntimeError> {
496 let local_uri = fakecloud_core::ecr_uri::translate_to_local_at(
500 image,
501 &self.registry_host,
502 self.server_port,
503 );
504 let pull_uri = local_uri.as_deref().unwrap_or(image);
505
506 let mut cmd = tokio::process::Command::new(&self.cli);
507 if let Some(p) = self.docker_config_path() {
508 cmd.env("DOCKER_CONFIG", p);
509 }
510 let out = cmd
511 .args(["pull", pull_uri])
512 .output()
513 .await
514 .map_err(|e| RuntimeError::ContainerStartFailed(format!("docker pull: {e}")))?;
515 if !out.status.success() {
516 return Err(RuntimeError::ContainerStartFailed(format!(
517 "docker pull failed for {pull_uri}: {}",
518 String::from_utf8_lossy(&out.stderr)
519 )));
520 }
521 Ok(())
522 }
523}
524
525pub fn runtime_to_image(runtime: &str) -> Option<String> {
527 let (base, tag) = match runtime {
528 "python3.14" => ("python", "3.14"),
529 "python3.13" => ("python", "3.13"),
530 "python3.12" => ("python", "3.12"),
531 "python3.11" => ("python", "3.11"),
532 "python3.10" => ("python", "3.10"),
533 "python3.9" => ("python", "3.9"),
534 "python3.8" => ("python", "3.8"),
535 "nodejs24.x" => ("nodejs", "24"),
536 "nodejs22.x" => ("nodejs", "22"),
537 "nodejs20.x" => ("nodejs", "20"),
538 "nodejs18.x" => ("nodejs", "18"),
539 "nodejs16.x" => ("nodejs", "16"),
540 "ruby3.4" => ("ruby", "3.4"),
541 "ruby3.3" => ("ruby", "3.3"),
542 "java25" => ("java", "25"),
543 "java21" => ("java", "21"),
544 "java17" => ("java", "17"),
545 "java11" => ("java", "11"),
546 "dotnet10" => ("dotnet", "10"),
547 "dotnet8" => ("dotnet", "8"),
548 "go1.x" => ("go", "1"),
549 "provided.al2023" => ("provided", "al2023"),
550 "provided.al2" => ("provided", "al2"),
551 _ => return None,
552 };
553 Some(format!("public.ecr.aws/lambda/{base}:{tag}"))
554}
555
556pub(crate) fn ephemeral_storage_tmpfs_arg(size: Option<i64>) -> String {
568 let mib = size.unwrap_or(512).max(64);
569 format!("/tmp:size={mib}m,exec")
570}
571
572pub fn extract_zip(zip_bytes: &[u8], dest: &Path) -> Result<(), RuntimeError> {
574 let cursor = std::io::Cursor::new(zip_bytes);
575 let mut archive = zip::ZipArchive::new(cursor)
576 .map_err(|e| RuntimeError::ZipExtractionFailed(e.to_string()))?;
577
578 for i in 0..archive.len() {
579 let mut file = archive
580 .by_index(i)
581 .map_err(|e| RuntimeError::ZipExtractionFailed(e.to_string()))?;
582
583 let out_path = dest.join(file.enclosed_name().ok_or_else(|| {
584 RuntimeError::ZipExtractionFailed("invalid file name in ZIP".to_string())
585 })?);
586
587 if file.is_dir() {
588 std::fs::create_dir_all(&out_path)
589 .map_err(|e| RuntimeError::ZipExtractionFailed(e.to_string()))?;
590 } else {
591 if let Some(parent) = out_path.parent() {
592 std::fs::create_dir_all(parent)
593 .map_err(|e| RuntimeError::ZipExtractionFailed(e.to_string()))?;
594 }
595 let mut out_file = std::fs::File::create(&out_path)
596 .map_err(|e| RuntimeError::ZipExtractionFailed(e.to_string()))?;
597 std::io::copy(&mut file, &mut out_file)
598 .map_err(|e| RuntimeError::ZipExtractionFailed(e.to_string()))?;
599
600 #[cfg(unix)]
601 {
602 use std::os::unix::fs::PermissionsExt;
603 if let Some(mode) = file.unix_mode() {
604 std::fs::set_permissions(&out_path, std::fs::Permissions::from_mode(mode))
605 .map_err(|e| RuntimeError::ZipExtractionFailed(e.to_string()))?;
606 }
607 }
608 }
609 }
610 Ok(())
611}
612
613fn build_local_registry_docker_config(server_port: u16) -> Option<TempDir> {
614 let dir = TempDir::new().ok()?;
615 let auth = base64::engine::general_purpose::STANDARD.encode("AWS:fakecloud-lambda-runtime");
616 let auths: serde_json::Map<String, serde_json::Value> =
622 fakecloud_core::container_net::registry_auth_hosts(server_port)
623 .into_iter()
624 .map(|host| (host, serde_json::json!({ "auth": auth })))
625 .collect();
626 let config = serde_json::json!({ "auths": auths });
627 std::fs::write(dir.path().join("config.json"), config.to_string()).ok()?;
628 Some(dir)
629}
630
631#[cfg(test)]
632mod tests {
633 use std::io::{Read, Write};
634
635 use super::*;
636
637 #[test]
638 fn test_runtime_to_image() {
639 assert_eq!(
640 runtime_to_image("python3.12"),
641 Some("public.ecr.aws/lambda/python:3.12".to_string())
642 );
643 assert_eq!(
644 runtime_to_image("nodejs20.x"),
645 Some("public.ecr.aws/lambda/nodejs:20".to_string())
646 );
647 assert_eq!(
648 runtime_to_image("provided.al2023"),
649 Some("public.ecr.aws/lambda/provided:al2023".to_string())
650 );
651 assert_eq!(
652 runtime_to_image("ruby3.4"),
653 Some("public.ecr.aws/lambda/ruby:3.4".to_string())
654 );
655 assert_eq!(
656 runtime_to_image("java21"),
657 Some("public.ecr.aws/lambda/java:21".to_string())
658 );
659 assert_eq!(
660 runtime_to_image("dotnet8"),
661 Some("public.ecr.aws/lambda/dotnet:8".to_string())
662 );
663 assert_eq!(
664 runtime_to_image("nodejs16.x"),
665 Some("public.ecr.aws/lambda/nodejs:16".to_string())
666 );
667 assert_eq!(
668 runtime_to_image("python3.10"),
669 Some("public.ecr.aws/lambda/python:3.10".to_string())
670 );
671 assert_eq!(
672 runtime_to_image("python3.9"),
673 Some("public.ecr.aws/lambda/python:3.9".to_string())
674 );
675 assert_eq!(
676 runtime_to_image("python3.8"),
677 Some("public.ecr.aws/lambda/python:3.8".to_string())
678 );
679 assert_eq!(
680 runtime_to_image("java11"),
681 Some("public.ecr.aws/lambda/java:11".to_string())
682 );
683 assert_eq!(
684 runtime_to_image("go1.x"),
685 Some("public.ecr.aws/lambda/go:1".to_string())
686 );
687 assert_eq!(
688 runtime_to_image("nodejs24.x"),
689 Some("public.ecr.aws/lambda/nodejs:24".to_string())
690 );
691 assert_eq!(
692 runtime_to_image("python3.14"),
693 Some("public.ecr.aws/lambda/python:3.14".to_string())
694 );
695 assert_eq!(
696 runtime_to_image("java25"),
697 Some("public.ecr.aws/lambda/java:25".to_string())
698 );
699 assert_eq!(
700 runtime_to_image("dotnet10"),
701 Some("public.ecr.aws/lambda/dotnet:10".to_string())
702 );
703 assert_eq!(runtime_to_image("unknown"), None);
704 }
705
706 #[test]
707 fn test_extract_zip() {
708 let buf = Vec::new();
709 let cursor = std::io::Cursor::new(buf);
710 let mut writer = zip::ZipWriter::new(cursor);
711 let options = zip::write::SimpleFileOptions::default();
712 writer.start_file("handler.py", options).unwrap();
713 writer
714 .write_all(b"def handler(event, context):\n return {'statusCode': 200}\n")
715 .unwrap();
716 let cursor = writer.finish().unwrap();
717 let zip_bytes = cursor.into_inner();
718
719 let dir = TempDir::new().unwrap();
720 extract_zip(&zip_bytes, dir.path()).unwrap();
721
722 let handler_path = dir.path().join("handler.py");
723 assert!(handler_path.exists());
724
725 let mut content = String::new();
726 std::fs::File::open(&handler_path)
727 .unwrap()
728 .read_to_string(&mut content)
729 .unwrap();
730 assert!(content.contains("def handler"));
731 }
732
733 #[test]
734 fn ephemeral_storage_tmpfs_arg_defaults_to_512_when_none() {
735 assert_eq!(ephemeral_storage_tmpfs_arg(None), "/tmp:size=512m,exec");
739 }
740
741 #[test]
742 fn ephemeral_storage_tmpfs_arg_uses_supplied_size() {
743 assert_eq!(
744 ephemeral_storage_tmpfs_arg(Some(2048)),
745 "/tmp:size=2048m,exec"
746 );
747 assert_eq!(
748 ephemeral_storage_tmpfs_arg(Some(10240)),
749 "/tmp:size=10240m,exec"
750 );
751 }
752
753 #[test]
754 fn ephemeral_storage_tmpfs_arg_clamps_to_64_floor() {
755 assert_eq!(ephemeral_storage_tmpfs_arg(Some(0)), "/tmp:size=64m,exec");
759 assert_eq!(ephemeral_storage_tmpfs_arg(Some(32)), "/tmp:size=64m,exec");
760 }
761}