1use crate::cdi::{self, CdiContainerEdits, CdiRegistry};
9use crate::error::{AgentError, Result};
10use crate::runtime::ContainerId;
11use oci_spec::runtime::{
12 Capability, Hook, HookBuilder, Hooks, HooksBuilder, LinuxBuilder, LinuxCapabilitiesBuilder,
13 LinuxCpuBuilder, LinuxDeviceBuilder, LinuxDeviceCgroupBuilder, LinuxDeviceType,
14 LinuxMemoryBuilder, LinuxNamespaceBuilder, LinuxNamespaceType, LinuxResourcesBuilder, Mount,
15 MountBuilder, PosixRlimit, PosixRlimitBuilder, PosixRlimitType, ProcessBuilder, RootBuilder,
16 Spec, SpecBuilder, UserBuilder,
17};
18#[cfg(unix)]
21use oci_spec::runtime::LinuxIdMappingBuilder;
22use std::collections::{HashMap, HashSet};
23use std::path::{Path, PathBuf};
32use std::str::FromStr;
33use std::sync::Arc;
34use tokio::fs;
35use zlayer_secrets::{SecretScope, SecretsProvider};
36use zlayer_spec::{GpuSharingMode, ServiceSpec, ShardingSpec, StorageSpec, StorageTier, SwarmRole};
37
38const DEFAULT_MPS_PIPE_DIR: &str = "/tmp/nvidia-mps";
41
42const DEFAULT_MPS_LOG_DIR: &str = "/tmp/nvidia-log";
45
46const TIMESLICE_CONFIG_CONTAINER_PATH: &str = "/etc/nvidia/gpu-time-slicing.yaml";
51
52struct MpsDirs {
58 pipe_dir: PathBuf,
59 log_dir: PathBuf,
60}
61
62fn resolve_mps_dirs(gpu: &zlayer_spec::GpuSpec) -> Result<Option<MpsDirs>> {
72 if gpu.sharing != Some(GpuSharingMode::Mps) {
73 return Ok(None);
74 }
75
76 let pipe_dir = PathBuf::from(gpu.mps_pipe_dir.as_deref().unwrap_or(DEFAULT_MPS_PIPE_DIR));
77 let log_dir = PathBuf::from(gpu.mps_log_dir.as_deref().unwrap_or(DEFAULT_MPS_LOG_DIR));
78
79 if !pipe_dir.is_dir() {
80 return Err(AgentError::GpuSharingUnavailable {
81 mode: "mps".to_string(),
82 reason: format!(
83 "MPS pipe directory {} does not exist; ensure nvidia-cuda-mps-control is running",
84 pipe_dir.display()
85 ),
86 });
87 }
88 if !log_dir.is_dir() {
89 return Err(AgentError::GpuSharingUnavailable {
90 mode: "mps".to_string(),
91 reason: format!(
92 "MPS log directory {} does not exist; ensure nvidia-cuda-mps-control is running",
93 log_dir.display()
94 ),
95 });
96 }
97
98 Ok(Some(MpsDirs { pipe_dir, log_dir }))
99}
100
101#[must_use]
132pub(crate) fn swarm_ring_env(sharding: &ShardingSpec) -> Vec<(String, String)> {
133 struct Slot {
136 service: Option<String>,
137 layer_start: u32,
138 layer_end: u32,
139 }
140
141 let mut out: Vec<(String, String)> = Vec::new();
142 out.push(("ZLAYER_SWARM_ID".to_string(), sharding.swarm_id.clone()));
143
144 let role_str = match sharding.role {
145 SwarmRole::Stage => "stage",
146 SwarmRole::Coordinator => "coordinator",
147 };
148 out.push(("ZLAYER_SWARM_ROLE".to_string(), role_str.to_string()));
149
150 if matches!(sharding.role, SwarmRole::Stage) {
151 out.push((
152 "ZLAYER_SWARM_LAYER_START".to_string(),
153 sharding.layer_start.to_string(),
154 ));
155 out.push((
156 "ZLAYER_SWARM_LAYER_END".to_string(),
157 sharding.layer_end.to_string(),
158 ));
159 out.push((
160 "ZLAYER_SWARM_TOTAL_LAYERS".to_string(),
161 sharding.layer_count.to_string(),
162 ));
163 }
164
165 let coordinator = sharding.coordinator.clone();
166 if let Some(ref coord) = coordinator {
167 out.push(("ZLAYER_SWARM_COORDINATOR".to_string(), coord.clone()));
168 }
169
170 let mut peers: Vec<Slot> = sharding
172 .peers
173 .iter()
174 .map(|p| Slot {
175 service: Some(p.service.clone()),
176 layer_start: p.layer_start,
177 layer_end: p.layer_end,
178 })
179 .collect();
180 peers.sort_by_key(|s| s.layer_start);
181
182 let (next, prev): (Option<String>, Option<String>) = match sharding.role {
183 SwarmRole::Coordinator => {
184 let first = peers.iter().min_by_key(|s| s.layer_start);
186 let last = peers.iter().max_by_key(|s| s.layer_end);
187 (
188 first.and_then(|s| s.service.clone()),
189 last.and_then(|s| s.service.clone()),
190 )
191 }
192 SwarmRole::Stage => {
193 let mut ring: Vec<Slot> = peers;
195 ring.push(Slot {
196 service: None,
197 layer_start: sharding.layer_start,
198 layer_end: sharding.layer_end,
199 });
200 ring.sort_by_key(|s| s.layer_start);
201
202 let self_idx = ring.iter().position(|s| s.service.is_none());
204 match self_idx {
205 Some(i) => {
206 let prev = if i == 0 {
208 coordinator.clone()
209 } else {
210 ring[i - 1].service.clone()
211 };
212 let next = if i + 1 >= ring.len() {
214 coordinator.clone()
215 } else {
216 ring[i + 1].service.clone()
217 };
218 (next, prev)
219 }
220 None => (None, None),
223 }
224 }
225 };
226
227 if let Some(next) = next {
228 out.push(("ZLAYER_SWARM_NEXT_PEER".to_string(), next));
229 }
230 if let Some(prev) = prev {
231 out.push(("ZLAYER_SWARM_PREV_PEER".to_string(), prev));
232 }
233
234 out
235}
236
237fn cdi_node_to_oci_device(
246 node: &crate::cdi::CdiDeviceNode,
247) -> Result<oci_spec::runtime::LinuxDevice> {
248 let host_path = node.host_path.as_deref().unwrap_or(&node.path);
249
250 let dev_type = match node.device_type.as_deref() {
251 Some("c" | "u") => LinuxDeviceType::C,
252 Some("b") => LinuxDeviceType::B,
253 Some("p") => LinuxDeviceType::P,
254 _ => get_device_type(host_path).unwrap_or(LinuxDeviceType::C),
255 };
256
257 let (major, minor) = if let (Some(maj), Some(min)) = (node.major, node.minor) {
258 (maj, min)
259 } else {
260 get_device_major_minor(host_path).unwrap_or((0, 0))
261 };
262
263 let mut builder = LinuxDeviceBuilder::default()
264 .path(node.path.clone())
265 .typ(dev_type)
266 .major(major)
267 .minor(minor);
268 if let Some(mode) = node.file_mode {
269 builder = builder.file_mode(mode);
270 } else {
271 builder = builder.file_mode(0o666u32);
272 }
273 builder = builder.uid(node.uid.unwrap_or(0));
274 builder = builder.gid(node.gid.unwrap_or(0));
275
276 builder.build().map_err(|e| {
277 AgentError::InvalidSpec(format!(
278 "failed to build CDI device {path}: {e}",
279 path = node.path
280 ))
281 })
282}
283
284fn convert_cdi_hook(cdi_hook: &crate::cdi::CdiHook) -> Result<Hook> {
286 let mut builder = HookBuilder::default().path(PathBuf::from(&cdi_hook.path));
287 if !cdi_hook.args.is_empty() {
288 builder = builder.args(cdi_hook.args.clone());
289 }
290 if !cdi_hook.env.is_empty() {
291 builder = builder.env(cdi_hook.env.clone());
292 }
293 builder
294 .build()
295 .map_err(|e| AgentError::InvalidSpec(format!("failed to build CDI hook: {e}")))
296}
297
298const ALL_CAPABILITIES: &[Capability] = &[
300 Capability::AuditControl,
301 Capability::AuditRead,
302 Capability::AuditWrite,
303 Capability::BlockSuspend,
304 Capability::Bpf,
305 Capability::CheckpointRestore,
306 Capability::Chown,
307 Capability::DacOverride,
308 Capability::DacReadSearch,
309 Capability::Fowner,
310 Capability::Fsetid,
311 Capability::IpcLock,
312 Capability::IpcOwner,
313 Capability::Kill,
314 Capability::Lease,
315 Capability::LinuxImmutable,
316 Capability::MacAdmin,
317 Capability::MacOverride,
318 Capability::Mknod,
319 Capability::NetAdmin,
320 Capability::NetBindService,
321 Capability::NetBroadcast,
322 Capability::NetRaw,
323 Capability::Perfmon,
324 Capability::Setfcap,
325 Capability::Setgid,
326 Capability::Setpcap,
327 Capability::Setuid,
328 Capability::SysAdmin,
329 Capability::SysBoot,
330 Capability::SysChroot,
331 Capability::SysModule,
332 Capability::SysNice,
333 Capability::SysPacct,
334 Capability::SysPtrace,
335 Capability::SysRawio,
336 Capability::SysResource,
337 Capability::SysTime,
338 Capability::SysTtyConfig,
339 Capability::Syslog,
340 Capability::WakeAlarm,
341];
342
343#[must_use]
375pub fn generate_resolv_conf(nameservers: &[String], search_domains: &[String]) -> String {
376 let mut out = String::new();
377 for ns in nameservers {
378 out.push_str("nameserver ");
379 out.push_str(ns);
380 out.push('\n');
381 }
382 if !search_domains.is_empty() {
383 out.push_str("search ");
384 out.push_str(&search_domains.join(" "));
385 out.push('\n');
386 }
387 out.push_str("options edns0\n");
388 out
389}
390
391pub fn parse_memory_string(s: &str) -> std::result::Result<u64, String> {
394 let s = s.trim();
395 if s.is_empty() {
396 return Err("empty memory string".to_string());
397 }
398
399 let (num_str, multiplier) = if let Some(n) = s.strip_suffix("Ki") {
400 (n, 1024u64)
401 } else if let Some(n) = s.strip_suffix("Mi") {
402 (n, 1024u64 * 1024)
403 } else if let Some(n) = s.strip_suffix("Gi") {
404 (n, 1024u64 * 1024 * 1024)
405 } else if let Some(n) = s.strip_suffix("Ti") {
406 (n, 1024u64 * 1024 * 1024 * 1024)
407 } else if let Some(n) = s.strip_suffix('K').or_else(|| s.strip_suffix('k')) {
408 (n, 1000u64)
409 } else if let Some(n) = s.strip_suffix('M').or_else(|| s.strip_suffix('m')) {
410 (n, 1000u64 * 1000)
411 } else if let Some(n) = s.strip_suffix('G').or_else(|| s.strip_suffix('g')) {
412 (n, 1000u64 * 1000 * 1000)
413 } else if let Some(n) = s.strip_suffix('T').or_else(|| s.strip_suffix('t')) {
414 (n, 1000u64 * 1000 * 1000 * 1000)
415 } else {
416 (s, 1u64)
417 };
418
419 let num: u64 = num_str
420 .parse()
421 .map_err(|e| format!("invalid number: {e}"))?;
422
423 Ok(num * multiplier)
424}
425
426#[cfg(unix)]
435#[allow(clippy::cast_possible_wrap)]
436fn get_device_major_minor(path: &str) -> std::io::Result<(i64, i64)> {
437 use std::os::unix::fs::MetadataExt;
438 let metadata = std::fs::metadata(path)?;
439 let rdev = metadata.rdev();
440 let major = ((rdev >> 8) & 0xff) as i64;
442 let minor = (rdev & 0xff) as i64;
443 Ok((major, minor))
444}
445
446#[cfg(not(unix))]
448fn get_device_major_minor(_path: &str) -> std::io::Result<(i64, i64)> {
449 Err(std::io::Error::new(
450 std::io::ErrorKind::Unsupported,
451 "device-cgroup probes require Unix",
452 ))
453}
454
455fn ulimit_name_to_posix(name: &str) -> Option<PosixRlimitType> {
459 Some(match name.to_ascii_lowercase().as_str() {
460 "cpu" => PosixRlimitType::RlimitCpu,
461 "fsize" => PosixRlimitType::RlimitFsize,
462 "data" => PosixRlimitType::RlimitData,
463 "stack" => PosixRlimitType::RlimitStack,
464 "core" => PosixRlimitType::RlimitCore,
465 "rss" => PosixRlimitType::RlimitRss,
466 "nproc" => PosixRlimitType::RlimitNproc,
467 "nofile" => PosixRlimitType::RlimitNofile,
468 "memlock" => PosixRlimitType::RlimitMemlock,
469 "as" => PosixRlimitType::RlimitAs,
470 "locks" => PosixRlimitType::RlimitLocks,
471 "sigpending" => PosixRlimitType::RlimitSigpending,
472 "msgqueue" => PosixRlimitType::RlimitMsgqueue,
473 "nice" => PosixRlimitType::RlimitNice,
474 "rtprio" => PosixRlimitType::RlimitRtprio,
475 "rttime" => PosixRlimitType::RlimitRttime,
476 _ => return None,
477 })
478}
479
480#[cfg(test)]
481mod ulimit_translation_tests {
482 use super::{ulimit_name_to_posix, PosixRlimitType};
483
484 #[test]
485 fn known_names_map() {
486 assert_eq!(
487 ulimit_name_to_posix("nofile"),
488 Some(PosixRlimitType::RlimitNofile)
489 );
490 assert_eq!(
491 ulimit_name_to_posix("NOFILE"),
492 Some(PosixRlimitType::RlimitNofile)
493 );
494 assert_eq!(
495 ulimit_name_to_posix("nproc"),
496 Some(PosixRlimitType::RlimitNproc)
497 );
498 assert_eq!(ulimit_name_to_posix("as"), Some(PosixRlimitType::RlimitAs));
499 }
500
501 #[test]
502 fn unknown_names_return_none() {
503 assert!(ulimit_name_to_posix("not_a_real_ulimit").is_none());
504 assert!(ulimit_name_to_posix("").is_none());
505 }
506}
507
508#[cfg(unix)]
513fn get_device_type(path: &str) -> std::io::Result<LinuxDeviceType> {
514 use std::os::unix::fs::FileTypeExt;
515 let metadata = std::fs::metadata(path)?;
516 let file_type = metadata.file_type();
517 if file_type.is_char_device() {
518 Ok(LinuxDeviceType::C)
519 } else if file_type.is_block_device() {
520 Ok(LinuxDeviceType::B)
521 } else {
522 Ok(LinuxDeviceType::U) }
524}
525
526#[cfg(not(unix))]
528fn get_device_type(_path: &str) -> std::io::Result<LinuxDeviceType> {
529 Err(std::io::Error::new(
530 std::io::ErrorKind::Unsupported,
531 "device-cgroup probes require Unix",
532 ))
533}
534
535#[derive(Clone)]
549pub struct BundleBuilder {
550 bundle_dir: PathBuf,
552 rootfs_path: Option<PathBuf>,
554 hostname: Option<String>,
556 extra_env: Vec<(String, String)>,
558 cwd: Option<String>,
560 args: Option<Vec<String>>,
562 volume_paths: HashMap<String, PathBuf>,
564 image_config: Option<zlayer_registry::ImageConfig>,
566 host_network: bool,
568 netns_path: Option<PathBuf>,
576 secrets_provider: Option<Arc<dyn SecretsProvider>>,
578 deployment_scope: Option<SecretScope>,
580 socket_path: Option<String>,
582 docker_socket_path: Option<String>,
585 toolchain_cache: Option<PathBuf>,
591 cdi_registry: Option<Arc<CdiRegistry>>,
598}
599
600impl std::fmt::Debug for BundleBuilder {
601 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
602 f.debug_struct("BundleBuilder")
603 .field("bundle_dir", &self.bundle_dir)
604 .field("rootfs_path", &self.rootfs_path)
605 .field("hostname", &self.hostname)
606 .field("extra_env", &self.extra_env)
607 .field("cwd", &self.cwd)
608 .field("args", &self.args)
609 .field("volume_paths", &self.volume_paths)
610 .field("image_config", &self.image_config)
611 .field("host_network", &self.host_network)
612 .field("netns_path", &self.netns_path)
613 .field("secrets_provider", &self.secrets_provider.is_some())
614 .field("deployment_scope", &self.deployment_scope)
615 .field("socket_path", &self.socket_path)
616 .field("docker_socket_path", &self.docker_socket_path)
617 .field("toolchain_cache", &self.toolchain_cache)
618 .field("cdi_registry", &self.cdi_registry.is_some())
619 .finish()
620 }
621}
622
623#[cfg(unix)]
631fn build_rootless_id_mappings(
632 host_id: u32,
633 subid_path: &str,
634 username: &str,
635) -> Vec<oci_spec::runtime::LinuxIdMapping> {
636 let mut mappings = vec![LinuxIdMappingBuilder::default()
637 .container_id(0_u32)
638 .host_id(host_id)
639 .size(1_u32)
640 .build()
641 .unwrap()];
642 if !username.is_empty() {
643 if let Some((start, count)) = read_subid_range(subid_path, username) {
644 mappings.push(
645 LinuxIdMappingBuilder::default()
646 .container_id(1_u32)
647 .host_id(start)
648 .size(count)
649 .build()
650 .unwrap(),
651 );
652 }
653 }
654 mappings
655}
656
657#[cfg(unix)]
662fn build_single_id_mapping(host_id: u32) -> Vec<oci_spec::runtime::LinuxIdMapping> {
663 vec![LinuxIdMappingBuilder::default()
664 .container_id(0_u32)
665 .host_id(host_id)
666 .size(1_u32)
667 .build()
668 .unwrap()]
669}
670
671#[cfg(unix)]
679fn read_subid_range(path: &str, username: &str) -> Option<(u32, u32)> {
680 let contents = std::fs::read_to_string(path).ok()?;
681 for line in contents.lines() {
682 let mut parts = line.splitn(3, ':');
683 let user = parts.next()?;
684 if user != username {
685 continue;
686 }
687 let start: u32 = parts.next()?.parse().ok()?;
688 let count: u32 = parts.next()?.parse().ok()?;
689 return Some((start, count));
690 }
691 None
692}
693
694impl BundleBuilder {
695 #[must_use]
705 pub fn new(bundle_dir: PathBuf) -> Self {
706 Self {
707 bundle_dir,
708 rootfs_path: None,
709 hostname: None,
710 extra_env: Vec::new(),
711 cwd: None,
712 args: None,
713 volume_paths: HashMap::new(),
714 image_config: None,
715 host_network: false,
716 netns_path: None,
717 secrets_provider: None,
718 deployment_scope: None,
719 socket_path: None,
720 docker_socket_path: None,
721 toolchain_cache: None,
722 cdi_registry: None,
723 }
724 }
725
726 #[must_use]
733 pub fn with_cdi_registry(mut self, registry: Arc<CdiRegistry>) -> Self {
734 self.cdi_registry = Some(registry);
735 self
736 }
737
738 #[must_use]
740 pub fn for_container(container_id: &ContainerId) -> Self {
741 let bundle_dir = zlayer_paths::ZLayerDirs::system_default()
742 .bundles()
743 .join(container_id.to_string());
744 Self::new(bundle_dir)
745 }
746
747 #[must_use]
751 pub fn with_rootfs(mut self, rootfs_path: PathBuf) -> Self {
752 self.rootfs_path = Some(rootfs_path);
753 self
754 }
755
756 #[must_use]
758 pub fn with_hostname(mut self, hostname: String) -> Self {
759 self.hostname = Some(hostname);
760 self
761 }
762
763 #[must_use]
765 pub fn with_env(mut self, key: String, value: String) -> Self {
766 self.extra_env.push((key, value));
767 self
768 }
769
770 #[must_use]
772 pub fn with_cwd(mut self, cwd: String) -> Self {
773 self.cwd = Some(cwd);
774 self
775 }
776
777 #[must_use]
779 pub fn with_args(mut self, args: Vec<String>) -> Self {
780 self.args = Some(args);
781 self
782 }
783
784 #[must_use]
789 pub fn with_volume_paths(mut self, volume_paths: HashMap<String, PathBuf>) -> Self {
790 self.volume_paths = volume_paths;
791 self
792 }
793
794 #[must_use]
799 pub fn with_image_config(mut self, config: zlayer_registry::ImageConfig) -> Self {
800 self.image_config = Some(config);
801 self
802 }
803
804 #[must_use]
810 pub fn with_host_network(mut self, host_network: bool) -> Self {
811 self.host_network = host_network;
812 self
813 }
814
815 #[must_use]
825 pub fn with_netns_path(mut self, netns_path: Option<PathBuf>) -> Self {
826 self.netns_path = netns_path;
827 self
828 }
829
830 #[must_use]
835 pub fn with_secrets_provider(mut self, provider: Arc<dyn SecretsProvider>) -> Self {
836 self.secrets_provider = Some(provider);
837 self
838 }
839
840 #[must_use]
845 pub fn with_deployment_scope(mut self, scope: SecretScope) -> Self {
846 self.deployment_scope = Some(scope);
847 self
848 }
849
850 #[must_use]
853 pub fn with_socket_mount(mut self, path: impl Into<String>) -> Self {
854 self.socket_path = Some(path.into());
855 self
856 }
857
858 #[must_use]
862 pub fn with_docker_socket_mount(mut self, path: impl Into<String>) -> Self {
863 self.docker_socket_path = Some(path.into());
864 self
865 }
866
867 #[must_use]
869 pub fn with_toolchain_cache(mut self, dir: impl Into<PathBuf>) -> Self {
870 self.toolchain_cache = Some(dir.into());
871 self
872 }
873
874 #[must_use]
876 pub fn bundle_dir(&self) -> &Path {
877 &self.bundle_dir
878 }
879
880 #[cfg(unix)]
899 pub async fn build(&self, container_id: &ContainerId, spec: &ServiceSpec) -> Result<PathBuf> {
900 fs::create_dir_all(&self.bundle_dir)
902 .await
903 .map_err(|e| AgentError::CreateFailed {
904 id: container_id.to_string(),
905 reason: format!("failed to create bundle directory: {e}"),
906 })?;
907
908 let rootfs_in_bundle = self.bundle_dir.join("rootfs");
910 if let Some(ref rootfs_path) = self.rootfs_path {
911 let _ = fs::remove_file(&rootfs_in_bundle).await;
913 let _ = fs::remove_dir(&rootfs_in_bundle).await;
914
915 #[cfg(unix)]
920 tokio::fs::symlink(rootfs_path, &rootfs_in_bundle)
921 .await
922 .map_err(|e| AgentError::CreateFailed {
923 id: container_id.to_string(),
924 reason: format!(
925 "failed to symlink rootfs from {} to {}: {}",
926 rootfs_path.display(),
927 rootfs_in_bundle.display(),
928 e
929 ),
930 })?;
931
932 #[cfg(windows)]
933 tokio::fs::symlink_dir(rootfs_path, &rootfs_in_bundle)
934 .await
935 .map_err(|e| AgentError::CreateFailed {
936 id: container_id.to_string(),
937 reason: format!(
938 "failed to symlink rootfs from {} to {}: {}",
939 rootfs_path.display(),
940 rootfs_in_bundle.display(),
941 e
942 ),
943 })?;
944 } else {
945 fs::create_dir_all(&rootfs_in_bundle)
947 .await
948 .map_err(|e| AgentError::CreateFailed {
949 id: container_id.to_string(),
950 reason: format!("failed to create rootfs directory: {e}"),
951 })?;
952 }
953
954 let oci_spec = self
956 .build_spec_only(container_id, spec, &self.volume_paths)
957 .await?;
958
959 let config_path = self.bundle_dir.join("config.json");
961 let config_json =
962 serde_json::to_string_pretty(&oci_spec).map_err(|e| AgentError::CreateFailed {
963 id: container_id.to_string(),
964 reason: format!("failed to serialize OCI spec: {e}"),
965 })?;
966
967 fs::write(&config_path, config_json)
968 .await
969 .map_err(|e| AgentError::CreateFailed {
970 id: container_id.to_string(),
971 reason: format!("failed to write config.json: {e}"),
972 })?;
973
974 tracing::debug!(
975 "Created OCI bundle at {} for container {}",
976 self.bundle_dir.display(),
977 container_id
978 );
979
980 Ok(self.bundle_dir.clone())
981 }
982
983 pub async fn build_spec_only(
1003 &self,
1004 container_id: &ContainerId,
1005 spec: &ServiceSpec,
1006 volume_paths: &std::collections::HashMap<String, PathBuf>,
1007 ) -> Result<oci_spec::runtime::Spec> {
1008 self.build_oci_spec(container_id, spec, volume_paths).await
1009 }
1010
1011 fn resolve_cdi_edits(&self, spec: &ServiceSpec) -> Result<Option<Vec<CdiContainerEdits>>> {
1026 let Some(ref gpu) = spec.resources.gpu else {
1027 return Ok(None);
1028 };
1029
1030 let Some(kind) = cdi::vendor_to_cdi_kind(&gpu.vendor) else {
1033 return Ok(None);
1034 };
1035
1036 let (registry, strict) = if let Some(reg) = &self.cdi_registry {
1042 (reg.clone(), true)
1043 } else {
1044 let reg = Arc::new(CdiRegistry::discover());
1045 if reg.is_empty() {
1046 return Ok(None);
1047 }
1048 (reg, false)
1049 };
1050
1051 let device_names: Vec<String> = (0..gpu.count).map(|i| i.to_string()).collect();
1052
1053 match registry.resolve_for_kind(kind, &device_names) {
1054 Ok(edits) => Ok(Some(edits)),
1055 Err(err) => {
1056 if strict {
1057 Err(AgentError::InvalidSpec(format!(
1058 "CDI resolution failed for vendor '{}': {err}",
1059 gpu.vendor
1060 )))
1061 } else {
1062 tracing::warn!(
1063 vendor = %gpu.vendor,
1064 kind = %kind,
1065 error = %err,
1066 "CDI resolution failed; falling back to baked-in GPU device passthrough"
1067 );
1068 Ok(None)
1069 }
1070 }
1071 }
1072 }
1073
1074 #[allow(clippy::too_many_lines)]
1093 async fn build_oci_spec(
1094 &self,
1095 container_id: &ContainerId,
1096 spec: &ServiceSpec,
1097 volume_paths: &std::collections::HashMap<String, PathBuf>,
1098 ) -> Result<Spec> {
1099 let cdi_edits = self.resolve_cdi_edits(spec)?;
1103
1104 let user = {
1106 let (uid, gid) = if let Some(user_str) = self
1107 .image_config
1108 .as_ref()
1109 .and_then(|c| c.user.as_ref())
1110 .filter(|u| !u.is_empty())
1111 {
1112 let parts: Vec<&str> = user_str.splitn(2, ':').collect();
1114 let uid = parts[0].parse::<u32>().unwrap_or(0);
1115 let gid = if parts.len() > 1 {
1116 parts[1].parse::<u32>().unwrap_or(0)
1117 } else {
1118 uid
1119 };
1120 (uid, gid)
1121 } else {
1122 (0u32, 0u32)
1123 };
1124
1125 UserBuilder::default()
1126 .uid(uid)
1127 .gid(gid)
1128 .build()
1129 .map_err(|e| AgentError::InvalidSpec(format!("failed to build user: {e}")))?
1130 };
1131
1132 let mut env: Vec<String> = Vec::new();
1135 let mut env_keys: HashSet<String> = HashSet::new();
1136
1137 if let Some(img_env) = self.image_config.as_ref().and_then(|c| c.env.as_ref()) {
1139 for entry in img_env {
1140 if let Some(key) = entry.split('=').next() {
1141 env_keys.insert(key.to_string());
1142 }
1143 env.push(entry.clone());
1144 }
1145 }
1146
1147 if !env_keys.contains("PATH") {
1149 env.push(
1150 "PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin".to_string(),
1151 );
1152 env_keys.insert("PATH".to_string());
1153 }
1154
1155 if !env_keys.contains("TERM") {
1157 env.push("TERM=xterm".to_string());
1158 env_keys.insert("TERM".to_string());
1159 }
1160
1161 if self.toolchain_cache.is_some() {
1165 if !env_keys.contains("RUNNER_TOOL_CACHE") {
1166 env.push("RUNNER_TOOL_CACHE=/opt/zlayer/toolchains".to_string());
1167 env_keys.insert("RUNNER_TOOL_CACHE".to_string());
1168 }
1169 if !env_keys.contains("AGENT_TOOLSDIRECTORY") {
1170 env.push("AGENT_TOOLSDIRECTORY=/opt/zlayer/toolchains".to_string());
1171 env_keys.insert("AGENT_TOOLSDIRECTORY".to_string());
1172 }
1173 }
1174
1175 if let (Some(secrets_provider), Some(scope)) =
1182 (&self.secrets_provider, &self.deployment_scope)
1183 {
1184 let resolved_map = crate::env::resolve_env_with_secrets(
1185 &spec.env,
1186 secrets_provider.as_ref(),
1187 &scope.to_storage_scope(),
1188 )
1189 .await
1190 .map_err(|e| {
1191 AgentError::InvalidSpec(format!("environment variable resolution failed: {e}"))
1192 })?;
1193
1194 for (key, value) in &resolved_map {
1195 if env_keys.contains(key.as_str()) {
1196 env.retain(|e| e.split('=').next() != Some(key.as_str()));
1197 }
1198 env_keys.insert(key.clone());
1199 env.push(format!("{key}={value}"));
1200 }
1201 } else {
1202 let resolved = crate::env::resolve_env_vars_with_warnings(&spec.env).map_err(|e| {
1203 AgentError::InvalidSpec(format!("environment variable resolution failed: {e}"))
1204 })?;
1205
1206 for warning in &resolved.warnings {
1208 tracing::warn!(container = %container_id, "{}", warning);
1209 }
1210
1211 for var in &resolved.vars {
1213 if let Some(key) = var.split('=').next() {
1214 if env_keys.contains(key) {
1215 env.retain(|e| e.split('=').next() != Some(key));
1217 }
1218 env_keys.insert(key.to_string());
1219 }
1220 env.push(var.clone());
1221 }
1222 }
1223
1224 for (key, value) in &self.extra_env {
1226 if env_keys.contains(key.as_str()) {
1227 env.retain(|e| e.split('=').next() != Some(key.as_str()));
1228 }
1229 env_keys.insert(key.clone());
1230 env.push(format!("{key}={value}"));
1231 }
1232
1233 if let Some(ref edits_per_device) = cdi_edits {
1242 for edits in edits_per_device {
1243 for entry in &edits.env {
1244 if let Some(key) = entry.split('=').next() {
1245 if env_keys.contains(key) {
1246 env.retain(|e| e.split('=').next() != Some(key));
1247 }
1248 env_keys.insert(key.to_string());
1249 }
1250 env.push(entry.clone());
1251 }
1252 }
1253 } else if let Some(ref gpu) = spec.resources.gpu {
1254 let indices: Vec<String> = (0..gpu.count).map(|i| i.to_string()).collect();
1256 let device_list = indices.join(",");
1257 match gpu.vendor.as_str() {
1258 "nvidia" => {
1259 env.push(format!("NVIDIA_VISIBLE_DEVICES={device_list}"));
1260 env.push(format!("CUDA_VISIBLE_DEVICES={device_list}"));
1261 }
1262 "amd" => {
1263 env.push(format!("ROCR_VISIBLE_DEVICES={device_list}"));
1264 env.push(format!("HIP_VISIBLE_DEVICES={device_list}"));
1265 }
1266 "intel" => {
1267 env.push(format!("ZE_AFFINITY_MASK={device_list}"));
1268 }
1269 _ => {}
1270 }
1271 }
1272
1273 let mps_dirs = if let Some(ref gpu) = spec.resources.gpu {
1286 resolve_mps_dirs(gpu)?
1287 } else {
1288 None
1289 };
1290 if let Some(ref dirs) = mps_dirs {
1291 let pipe = format!("CUDA_MPS_PIPE_DIRECTORY={}", dirs.pipe_dir.display());
1292 let log = format!("CUDA_MPS_LOG_DIRECTORY={}", dirs.log_dir.display());
1293 if env_keys.contains("CUDA_MPS_PIPE_DIRECTORY") {
1294 env.retain(|e| e.split('=').next() != Some("CUDA_MPS_PIPE_DIRECTORY"));
1295 }
1296 if env_keys.contains("CUDA_MPS_LOG_DIRECTORY") {
1297 env.retain(|e| e.split('=').next() != Some("CUDA_MPS_LOG_DIRECTORY"));
1298 }
1299 env_keys.insert("CUDA_MPS_PIPE_DIRECTORY".to_string());
1300 env_keys.insert("CUDA_MPS_LOG_DIRECTORY".to_string());
1301 env.push(pipe);
1302 env.push(log);
1303 }
1304 if let Some(ref gpu) = spec.resources.gpu {
1305 if gpu.sharing == Some(GpuSharingMode::TimeSlice) {
1306 if let Some(idx) = gpu.time_slice_index {
1307 env.retain(|e| e.split('=').next() != Some("CUDA_VISIBLE_DEVICES"));
1312 env_keys.insert("CUDA_VISIBLE_DEVICES".to_string());
1313 env.push(format!("CUDA_VISIBLE_DEVICES={idx}"));
1314 }
1315 }
1316 }
1317
1318 if let Some(ref gpu) = spec.resources.gpu {
1322 if let Some(ref dist) = gpu.distributed {
1323 env.push(format!("MASTER_PORT={}", dist.master_port));
1324 env.push(format!("MASTER_ADDR={}", container_id.service));
1325 env.push("WORLD_SIZE=1".to_string());
1326 env.push("RANK=0".to_string());
1327 env.push("LOCAL_RANK=0".to_string());
1328 match dist.backend.as_str() {
1329 "nccl" => env.push("NCCL_SOCKET_IFNAME=eth0".to_string()),
1330 "gloo" => env.push("GLOO_SOCKET_IFNAME=eth0".to_string()),
1331 _ => {}
1332 }
1333 }
1334 }
1335
1336 if let Some(ref gpu) = spec.resources.gpu {
1340 if let Some(ref sharding) = gpu.sharding {
1341 for (k, v) in crate::bundle::swarm_ring_env(sharding) {
1342 env.push(format!("{k}={v}"));
1343 }
1344 }
1345 }
1346
1347 let capabilities = self.build_capabilities(spec)?;
1349
1350 let cwd = self
1352 .cwd
1353 .clone()
1354 .or_else(|| spec.command.workdir.clone())
1355 .or_else(|| {
1356 self.image_config
1357 .as_ref()
1358 .and_then(|c| c.working_dir.as_ref())
1359 .filter(|w| !w.is_empty())
1360 .cloned()
1361 })
1362 .unwrap_or_else(|| "/".to_string());
1363
1364 let process_args = if let Some(ref args) = self.args {
1366 args.clone()
1367 } else {
1368 Self::resolve_command_from_spec(spec, self.image_config.as_ref())
1369 };
1370
1371 let mut process_builder = ProcessBuilder::default()
1373 .terminal(false)
1374 .user(user)
1375 .env(env)
1376 .args(process_args)
1377 .cwd(cwd)
1378 .no_new_privileges(!spec.privileged && spec.capabilities.is_empty());
1379
1380 if let Some(caps) = capabilities {
1382 process_builder = process_builder.capabilities(caps);
1383 }
1384
1385 let mut rlimits: Vec<PosixRlimit> = Vec::with_capacity(spec.ulimits.len());
1391 for (name, limit) in &spec.ulimits {
1392 let typ = ulimit_name_to_posix(name).ok_or_else(|| {
1393 AgentError::InvalidSpec(format!(
1394 "unknown ulimit name `{name}` (expected one of: cpu, fsize, data, stack, \
1395 core, rss, nproc, nofile, memlock, as, locks, sigpending, msgqueue, nice, \
1396 rtprio, rttime)"
1397 ))
1398 })?;
1399 let entry = PosixRlimitBuilder::default()
1400 .typ(typ)
1401 .soft(u64::try_from(limit.soft.max(0)).unwrap_or(0))
1402 .hard(u64::try_from(limit.hard.max(0)).unwrap_or(0))
1403 .build()
1404 .map_err(|e| {
1405 AgentError::InvalidSpec(format!("failed to build rlimit `{name}`: {e}"))
1406 })?;
1407 rlimits.push(entry);
1408 }
1409 if !rlimits.is_empty() {
1410 process_builder = process_builder.rlimits(rlimits);
1411 }
1412
1413 let process = process_builder
1414 .build()
1415 .map_err(|e| AgentError::InvalidSpec(format!("failed to build process: {e}")))?;
1416
1417 let root = RootBuilder::default()
1420 .path("rootfs".to_string())
1421 .readonly(false)
1422 .build()
1423 .map_err(|e| AgentError::InvalidSpec(format!("failed to build root: {e}")))?;
1424
1425 let mut mounts = self.build_default_mounts(spec)?;
1427
1428 let storage_mounts = self.build_storage_mounts(spec, volume_paths)?;
1430 mounts.extend(storage_mounts);
1431
1432 if let Some(ref socket_path) = self.socket_path {
1436 mounts.push(
1437 MountBuilder::default()
1438 .destination(zlayer_paths::ZLayerDirs::default_socket_path())
1439 .typ("bind")
1440 .source(socket_path.clone())
1441 .options(vec!["rbind".into(), "ro".into()])
1442 .build()
1443 .expect("valid socket mount"),
1444 );
1445 }
1446
1447 if let Some(ref docker_socket_path) = self.docker_socket_path {
1451 mounts.push(
1452 MountBuilder::default()
1453 .destination("/var/run/docker.sock")
1454 .typ("bind")
1455 .source(docker_socket_path.clone())
1456 .options(vec!["rbind".into()])
1457 .build()
1458 .expect("valid docker socket mount"),
1459 );
1460 }
1461
1462 if let Some(ref toolchain_cache) = self.toolchain_cache {
1466 mounts.push(
1467 MountBuilder::default()
1468 .destination("/opt/zlayer/toolchains")
1469 .typ("bind")
1470 .source(toolchain_cache.clone())
1471 .options(vec!["rbind".into(), "rw".into()])
1472 .build()
1473 .expect("valid toolchain cache mount"),
1474 );
1475 }
1476
1477 let mut resolv_conf_source: Option<PathBuf> = None;
1501 if !spec.host_network && !spec.dns.is_empty() && self.bundle_dir.exists() {
1502 let resolv_path = self.bundle_dir.join("resolv.conf");
1503 let contents = generate_resolv_conf(&spec.dns, &spec.dns_search);
1504
1505 let mut wrote = true;
1509 if let Some(parent) = resolv_path.parent() {
1510 if let Err(e) = fs::create_dir_all(parent).await {
1511 tracing::warn!(
1512 bundle_dir = %parent.display(),
1513 error = %e,
1514 "failed to ensure bundle dir for resolv.conf; skipping DNS injection"
1515 );
1516 wrote = false;
1517 }
1518 }
1519
1520 if wrote {
1521 if let Err(e) = fs::write(&resolv_path, contents).await {
1522 tracing::warn!(
1523 path = %resolv_path.display(),
1524 error = %e,
1525 "failed to write resolv.conf to bundle; skipping DNS injection"
1526 );
1527 wrote = false;
1528 }
1529 }
1530
1531 if wrote && !resolv_path.exists() {
1535 tracing::warn!(
1536 path = %resolv_path.display(),
1537 "resolv.conf write reported success but file is absent; skipping DNS injection"
1538 );
1539 wrote = false;
1540 }
1541
1542 if wrote {
1543 resolv_conf_source = Some(resolv_path.clone());
1544 mounts.push(
1545 MountBuilder::default()
1546 .destination("/etc/resolv.conf".to_string())
1547 .typ("bind")
1548 .source(resolv_path.to_string_lossy().to_string())
1549 .options(vec!["rbind".to_string(), "ro".to_string()])
1550 .build()
1551 .map_err(|e| {
1552 AgentError::InvalidSpec(format!(
1553 "failed to build resolv.conf mount: {e}"
1554 ))
1555 })?,
1556 );
1557 }
1558 }
1559
1560 if let Some(ref edits_per_device) = cdi_edits {
1563 for edits in edits_per_device {
1564 for cdi_mount in &edits.mounts {
1565 let mut opts = cdi_mount.options.clone();
1566 if !opts.iter().any(|o| o == "bind" || o == "rbind") {
1567 opts.push("rbind".to_string());
1568 }
1569 mounts.push(
1570 MountBuilder::default()
1571 .destination(cdi_mount.container_path.clone())
1572 .typ("bind")
1573 .source(cdi_mount.host_path.clone())
1574 .options(opts)
1575 .build()
1576 .map_err(|e| {
1577 AgentError::InvalidSpec(format!("failed to build CDI mount: {e}"))
1578 })?,
1579 );
1580 }
1581 }
1582 }
1583
1584 if let Some(ref dirs) = mps_dirs {
1596 mounts.push(
1597 MountBuilder::default()
1598 .destination(dirs.pipe_dir.clone())
1599 .typ("bind")
1600 .source(dirs.pipe_dir.clone())
1601 .options(vec!["rbind".into(), "rw".into()])
1602 .build()
1603 .map_err(|e| {
1604 AgentError::InvalidSpec(format!("failed to build MPS pipe mount: {e}"))
1605 })?,
1606 );
1607 mounts.push(
1608 MountBuilder::default()
1609 .destination(dirs.log_dir.clone())
1610 .typ("bind")
1611 .source(dirs.log_dir.clone())
1612 .options(vec!["rbind".into(), "rw".into()])
1613 .build()
1614 .map_err(|e| {
1615 AgentError::InvalidSpec(format!("failed to build MPS log mount: {e}"))
1616 })?,
1617 );
1618 }
1619 if let Some(ref gpu) = spec.resources.gpu {
1620 if gpu.sharing == Some(GpuSharingMode::TimeSlice) {
1621 if let Some(ref cfg_path) = gpu.time_slicing_config_path {
1622 let host = PathBuf::from(cfg_path);
1623 if !host.is_file() {
1624 return Err(AgentError::GpuSharingUnavailable {
1625 mode: "time-slice".to_string(),
1626 reason: format!(
1627 "time-slicing config {} is not a regular file on the host",
1628 host.display()
1629 ),
1630 });
1631 }
1632 mounts.push(
1633 MountBuilder::default()
1634 .destination(PathBuf::from(TIMESLICE_CONFIG_CONTAINER_PATH))
1635 .typ("bind")
1636 .source(host)
1637 .options(vec!["rbind".into(), "ro".into()])
1638 .build()
1639 .map_err(|e| {
1640 AgentError::InvalidSpec(format!(
1641 "failed to build time-slicing config mount: {e}"
1642 ))
1643 })?,
1644 );
1645 }
1646 }
1647 }
1648
1649 Self::validate_host_bind_sources(&mounts, resolv_conf_source.as_deref())?;
1664
1665 let linux = self.build_linux_config(container_id, spec, cdi_edits.as_deref())?;
1667
1668 let hostname = self
1670 .hostname
1671 .clone()
1672 .unwrap_or_else(|| container_id.to_string());
1673
1674 let mut spec_builder = SpecBuilder::default()
1676 .version("1.0.2".to_string())
1677 .root(root)
1678 .process(process)
1679 .hostname(hostname)
1680 .mounts(mounts)
1681 .linux(linux);
1682
1683 if let Some(ref edits_per_device) = cdi_edits {
1684 if let Some(hooks) = Self::build_hooks_from_cdi(edits_per_device)? {
1685 spec_builder = spec_builder.hooks(hooks);
1686 }
1687 }
1688
1689 let oci_spec = spec_builder
1690 .build()
1691 .map_err(|e| AgentError::InvalidSpec(format!("failed to build OCI spec: {e}")))?;
1692
1693 Ok(oci_spec)
1694 }
1695
1696 fn validate_host_bind_sources(
1725 mounts: &[Mount],
1726 resolv_conf_source: Option<&Path>,
1727 ) -> Result<()> {
1728 const VIRTUAL_FS_TYPES: &[&str] = &[
1733 "proc",
1734 "tmpfs",
1735 "sysfs",
1736 "devpts",
1737 "mqueue",
1738 "cgroup",
1739 "cgroup2",
1740 "devtmpfs",
1741 "ramfs",
1742 "securityfs",
1743 "debugfs",
1744 "tracefs",
1745 "fusectl",
1746 "configfs",
1747 "pstore",
1748 "bpf",
1749 "binfmt_misc",
1750 "hugetlbfs",
1751 ];
1752
1753 for mount in mounts {
1754 let Some(source) = mount.source() else {
1755 continue;
1756 };
1757
1758 if !source.is_absolute() {
1761 continue;
1762 }
1763
1764 if let Some(typ) = mount.typ().as_deref() {
1766 if VIRTUAL_FS_TYPES.contains(&typ) {
1767 continue;
1768 }
1769 }
1770
1771 if let Some(resolv) = resolv_conf_source {
1773 if source == resolv {
1774 continue;
1775 }
1776 }
1777
1778 if !source.exists() {
1779 return Err(AgentError::MountSourceMissing {
1780 src_path: source.to_string_lossy().into_owned(),
1781 dest: mount.destination().to_string_lossy().into_owned(),
1782 });
1783 }
1784 }
1785 Ok(())
1786 }
1787
1788 fn build_hooks_from_cdi(edits_per_device: &[CdiContainerEdits]) -> Result<Option<Hooks>> {
1795 let mut prestart: Vec<Hook> = Vec::new();
1796 let mut create_runtime: Vec<Hook> = Vec::new();
1797 let mut create_container: Vec<Hook> = Vec::new();
1798 let mut start_container: Vec<Hook> = Vec::new();
1799 let mut poststart: Vec<Hook> = Vec::new();
1800 let mut poststop: Vec<Hook> = Vec::new();
1801
1802 for edits in edits_per_device {
1803 let Some(ref h) = edits.hooks else { continue };
1804 for hook in &h.prestart {
1805 prestart.push(convert_cdi_hook(hook)?);
1806 }
1807 for hook in &h.create_runtime {
1808 create_runtime.push(convert_cdi_hook(hook)?);
1809 }
1810 for hook in &h.create_container {
1811 create_container.push(convert_cdi_hook(hook)?);
1812 }
1813 for hook in &h.start_container {
1814 start_container.push(convert_cdi_hook(hook)?);
1815 }
1816 for hook in &h.poststart {
1817 poststart.push(convert_cdi_hook(hook)?);
1818 }
1819 for hook in &h.poststop {
1820 poststop.push(convert_cdi_hook(hook)?);
1821 }
1822 }
1823
1824 if prestart.is_empty()
1825 && create_runtime.is_empty()
1826 && create_container.is_empty()
1827 && start_container.is_empty()
1828 && poststart.is_empty()
1829 && poststop.is_empty()
1830 {
1831 return Ok(None);
1832 }
1833
1834 let mut builder = HooksBuilder::default();
1835 if !prestart.is_empty() {
1836 #[allow(deprecated)]
1837 {
1838 builder = builder.prestart(prestart);
1839 }
1840 }
1841 if !create_runtime.is_empty() {
1842 builder = builder.create_runtime(create_runtime);
1843 }
1844 if !create_container.is_empty() {
1845 builder = builder.create_container(create_container);
1846 }
1847 if !start_container.is_empty() {
1848 builder = builder.start_container(start_container);
1849 }
1850 if !poststart.is_empty() {
1851 builder = builder.poststart(poststart);
1852 }
1853 if !poststop.is_empty() {
1854 builder = builder.poststop(poststop);
1855 }
1856
1857 let hooks = builder
1858 .build()
1859 .map_err(|e| AgentError::InvalidSpec(format!("failed to build CDI hooks: {e}")))?;
1860 Ok(Some(hooks))
1861 }
1862
1863 #[allow(clippy::unused_self)]
1865 fn build_capabilities(
1866 &self,
1867 spec: &ServiceSpec,
1868 ) -> Result<Option<oci_spec::runtime::LinuxCapabilities>> {
1869 if spec.privileged {
1870 let all_caps: HashSet<Capability> = ALL_CAPABILITIES.iter().copied().collect();
1872 let empty_caps: HashSet<Capability> = HashSet::new();
1873
1874 let caps = LinuxCapabilitiesBuilder::default()
1875 .bounding(all_caps.clone())
1876 .effective(all_caps.clone())
1877 .permitted(all_caps)
1878 .inheritable(empty_caps.clone())
1879 .ambient(empty_caps)
1880 .build()
1881 .map_err(|e| {
1882 AgentError::InvalidSpec(format!("failed to build capabilities: {e}"))
1883 })?;
1884
1885 Ok(Some(caps))
1886 } else if !spec.capabilities.is_empty() {
1887 let caps: HashSet<Capability> = spec
1889 .capabilities
1890 .iter()
1891 .filter_map(|c| {
1892 let cap_name = if c.starts_with("CAP_") {
1894 c.to_uppercase()
1895 } else {
1896 format!("CAP_{}", c.to_uppercase())
1897 };
1898 Capability::from_str(&cap_name).ok()
1899 })
1900 .collect();
1901
1902 let empty_caps: HashSet<Capability> = HashSet::new();
1903
1904 let built_caps = LinuxCapabilitiesBuilder::default()
1905 .bounding(caps.clone())
1906 .effective(caps.clone())
1907 .permitted(caps)
1908 .inheritable(empty_caps.clone())
1909 .ambient(empty_caps)
1910 .build()
1911 .map_err(|e| {
1912 AgentError::InvalidSpec(format!("failed to build capabilities: {e}"))
1913 })?;
1914
1915 Ok(Some(built_caps))
1916 } else {
1917 let default_caps: HashSet<Capability> = [
1919 Capability::Chown,
1920 Capability::DacOverride,
1921 Capability::Fsetid,
1922 Capability::Fowner,
1923 Capability::Mknod,
1924 Capability::NetRaw,
1925 Capability::Setgid,
1926 Capability::Setuid,
1927 Capability::Setfcap,
1928 Capability::Setpcap,
1929 Capability::NetBindService,
1930 Capability::SysChroot,
1931 Capability::Kill,
1932 Capability::AuditWrite,
1933 ]
1934 .into_iter()
1935 .collect();
1936
1937 let empty_caps: HashSet<Capability> = HashSet::new();
1938
1939 let built_caps = LinuxCapabilitiesBuilder::default()
1940 .bounding(default_caps.clone())
1941 .effective(default_caps.clone())
1942 .permitted(default_caps)
1943 .inheritable(empty_caps.clone())
1944 .ambient(empty_caps)
1945 .build()
1946 .map_err(|e| {
1947 AgentError::InvalidSpec(format!("failed to build capabilities: {e}"))
1948 })?;
1949
1950 Ok(Some(built_caps))
1951 }
1952 }
1953
1954 #[allow(clippy::unused_self, clippy::too_many_lines)]
1956 fn build_default_mounts(&self, spec: &ServiceSpec) -> Result<Vec<Mount>> {
1957 let mut mounts = Vec::new();
1958
1959 mounts.push(
1961 MountBuilder::default()
1962 .destination("/proc".to_string())
1963 .typ("proc".to_string())
1964 .source("proc".to_string())
1965 .options(vec![
1966 "nosuid".to_string(),
1967 "noexec".to_string(),
1968 "nodev".to_string(),
1969 ])
1970 .build()
1971 .map_err(|e| {
1972 AgentError::InvalidSpec(format!("failed to build /proc mount: {e}"))
1973 })?,
1974 );
1975
1976 mounts.push(
1978 MountBuilder::default()
1979 .destination("/dev".to_string())
1980 .typ("tmpfs".to_string())
1981 .source("tmpfs".to_string())
1982 .options(vec![
1983 "nosuid".to_string(),
1984 "strictatime".to_string(),
1985 "mode=755".to_string(),
1986 "size=65536k".to_string(),
1987 ])
1988 .build()
1989 .map_err(|e| AgentError::InvalidSpec(format!("failed to build /dev mount: {e}")))?,
1990 );
1991
1992 mounts.push(
1994 MountBuilder::default()
1995 .destination("/dev/pts".to_string())
1996 .typ("devpts".to_string())
1997 .source("devpts".to_string())
1998 .options(vec![
1999 "nosuid".to_string(),
2000 "noexec".to_string(),
2001 "newinstance".to_string(),
2002 "ptmxmode=0666".to_string(),
2003 "mode=0620".to_string(),
2004 "gid=5".to_string(),
2005 ])
2006 .build()
2007 .map_err(|e| {
2008 AgentError::InvalidSpec(format!("failed to build /dev/pts mount: {e}"))
2009 })?,
2010 );
2011
2012 mounts.push(
2014 MountBuilder::default()
2015 .destination("/dev/shm".to_string())
2016 .typ("tmpfs".to_string())
2017 .source("shm".to_string())
2018 .options(vec![
2019 "nosuid".to_string(),
2020 "noexec".to_string(),
2021 "nodev".to_string(),
2022 "mode=1777".to_string(),
2023 "size=65536k".to_string(),
2024 ])
2025 .build()
2026 .map_err(|e| {
2027 AgentError::InvalidSpec(format!("failed to build /dev/shm mount: {e}"))
2028 })?,
2029 );
2030
2031 mounts.push(
2033 MountBuilder::default()
2034 .destination("/dev/mqueue".to_string())
2035 .typ("mqueue".to_string())
2036 .source("mqueue".to_string())
2037 .options(vec![
2038 "nosuid".to_string(),
2039 "noexec".to_string(),
2040 "nodev".to_string(),
2041 ])
2042 .build()
2043 .map_err(|e| {
2044 AgentError::InvalidSpec(format!("failed to build /dev/mqueue mount: {e}"))
2045 })?,
2046 );
2047
2048 let sys_options = if spec.privileged {
2050 vec![
2051 "nosuid".to_string(),
2052 "noexec".to_string(),
2053 "nodev".to_string(),
2054 ]
2055 } else {
2056 vec![
2057 "nosuid".to_string(),
2058 "noexec".to_string(),
2059 "nodev".to_string(),
2060 "ro".to_string(),
2061 ]
2062 };
2063
2064 mounts.push(
2065 MountBuilder::default()
2066 .destination("/sys".to_string())
2067 .typ("sysfs".to_string())
2068 .source("sysfs".to_string())
2069 .options(sys_options)
2070 .build()
2071 .map_err(|e| AgentError::InvalidSpec(format!("failed to build /sys mount: {e}")))?,
2072 );
2073
2074 mounts.push(
2076 MountBuilder::default()
2077 .destination("/sys/fs/cgroup".to_string())
2078 .typ("cgroup2".to_string())
2079 .source("cgroup".to_string())
2080 .options(vec![
2081 "nosuid".to_string(),
2082 "noexec".to_string(),
2083 "nodev".to_string(),
2084 "relatime".to_string(),
2085 ])
2086 .build()
2087 .map_err(|e| {
2088 AgentError::InvalidSpec(format!("failed to build cgroup mount: {e}"))
2089 })?,
2090 );
2091
2092 Ok(mounts)
2093 }
2094
2095 #[allow(clippy::unused_self, clippy::too_many_lines)]
2101 fn build_storage_mounts(
2102 &self,
2103 spec: &ServiceSpec,
2104 volume_paths: &std::collections::HashMap<String, PathBuf>,
2105 ) -> Result<Vec<Mount>> {
2106 let mut mounts = Vec::new();
2107
2108 for storage in &spec.storage {
2109 let mount = match storage {
2110 StorageSpec::Bind {
2111 source,
2112 target,
2113 readonly,
2114 } => {
2115 let mut options = vec!["rbind".to_string()];
2116 if *readonly {
2117 options.push("ro".to_string());
2118 } else {
2119 options.push("rw".to_string());
2120 }
2121
2122 MountBuilder::default()
2123 .destination(target.clone())
2124 .typ("none".to_string())
2125 .source(source.clone())
2126 .options(options)
2127 .build()
2128 .map_err(|e| {
2129 AgentError::InvalidSpec(format!(
2130 "failed to build bind mount for {target}: {e}"
2131 ))
2132 })?
2133 }
2134
2135 StorageSpec::Named {
2136 name,
2137 target,
2138 readonly,
2139 tier,
2140 ..
2141 } => {
2142 let source = volume_paths.get(name).ok_or_else(|| {
2144 AgentError::InvalidSpec(format!(
2145 "volume '{name}' not prepared - ensure StorageManager.ensure_volume() was called"
2146 ))
2147 })?;
2148
2149 if matches!(tier, StorageTier::Network) {
2151 tracing::warn!(
2152 volume = %name,
2153 tier = ?tier,
2154 "Network storage tier is NOT SQLite-safe. Avoid using SQLite databases on this volume."
2155 );
2156 }
2157
2158 let mut options = vec!["rbind".to_string()];
2159 if *readonly {
2160 options.push("ro".to_string());
2161 } else {
2162 options.push("rw".to_string());
2163 }
2164
2165 MountBuilder::default()
2166 .destination(target.clone())
2167 .typ("none".to_string())
2168 .source(source.to_string_lossy().to_string())
2169 .options(options)
2170 .build()
2171 .map_err(|e| {
2172 AgentError::InvalidSpec(format!(
2173 "failed to build named volume mount for {target}: {e}"
2174 ))
2175 })?
2176 }
2177
2178 StorageSpec::Anonymous { target, tier } => {
2179 let key = format!("_anon_{}", target.trim_start_matches('/').replace('/', "_"));
2182 let source = volume_paths.get(&key).ok_or_else(|| {
2183 AgentError::InvalidSpec(format!(
2184 "anonymous volume for '{target}' not prepared"
2185 ))
2186 })?;
2187
2188 if matches!(tier, StorageTier::Network) {
2189 tracing::warn!(
2190 target = %target,
2191 tier = ?tier,
2192 "Network storage tier is NOT SQLite-safe."
2193 );
2194 }
2195
2196 let options = vec!["rbind".to_string(), "rw".to_string()];
2197
2198 MountBuilder::default()
2199 .destination(target.clone())
2200 .typ("none".to_string())
2201 .source(source.to_string_lossy().to_string())
2202 .options(options)
2203 .build()
2204 .map_err(|e| {
2205 AgentError::InvalidSpec(format!(
2206 "failed to build anonymous volume mount for {target}: {e}"
2207 ))
2208 })?
2209 }
2210
2211 StorageSpec::Tmpfs { target, size, mode } => {
2212 let mut options = vec!["nosuid".to_string(), "nodev".to_string()];
2213
2214 if let Some(size_str) = size {
2215 options.push(format!("size={size_str}"));
2216 }
2217
2218 if let Some(mode_val) = mode {
2219 options.push(format!("mode={mode_val:o}"));
2220 }
2221
2222 MountBuilder::default()
2223 .destination(target.clone())
2224 .typ("tmpfs".to_string())
2225 .source("tmpfs".to_string())
2226 .options(options)
2227 .build()
2228 .map_err(|e| {
2229 AgentError::InvalidSpec(format!(
2230 "failed to build tmpfs mount for {target}: {e}"
2231 ))
2232 })?
2233 }
2234
2235 StorageSpec::S3 {
2236 bucket,
2237 prefix,
2238 target,
2239 readonly,
2240 endpoint: _,
2241 credentials: _,
2242 } => {
2243 let key = format!("_s3_{}_{}", bucket, prefix.as_deref().unwrap_or(""));
2246 let source = volume_paths.get(&key).ok_or_else(|| {
2247 AgentError::InvalidSpec(format!(
2248 "S3 volume for bucket '{bucket}' not mounted - ensure StorageManager.mount_s3() was called"
2249 ))
2250 })?;
2251
2252 tracing::warn!(
2253 bucket = %bucket,
2254 target = %target,
2255 "S3 storage is NOT SQLite-safe. Use for read-heavy workloads only."
2256 );
2257
2258 let mut options = vec!["rbind".to_string()];
2259 if *readonly {
2260 options.push("ro".to_string());
2261 } else {
2262 options.push("rw".to_string());
2263 }
2264
2265 MountBuilder::default()
2266 .destination(target.clone())
2267 .typ("none".to_string())
2268 .source(source.to_string_lossy().to_string())
2269 .options(options)
2270 .build()
2271 .map_err(|e| {
2272 AgentError::InvalidSpec(format!(
2273 "failed to build S3 mount for {target}: {e}"
2274 ))
2275 })?
2276 }
2277 };
2278
2279 mounts.push(mount);
2280 }
2281
2282 Ok(mounts)
2283 }
2284
2285 #[allow(clippy::similar_names)] #[allow(clippy::too_many_lines)]
2288 fn build_linux_config(
2289 &self,
2290 container_id: &ContainerId,
2291 spec: &ServiceSpec,
2292 cdi_edits: Option<&[CdiContainerEdits]>,
2293 ) -> Result<oci_spec::runtime::Linux> {
2294 let mut namespaces = vec![
2296 LinuxNamespaceBuilder::default()
2297 .typ(LinuxNamespaceType::Pid)
2298 .build()
2299 .unwrap(),
2300 LinuxNamespaceBuilder::default()
2301 .typ(LinuxNamespaceType::Ipc)
2302 .build()
2303 .unwrap(),
2304 LinuxNamespaceBuilder::default()
2305 .typ(LinuxNamespaceType::Uts)
2306 .build()
2307 .unwrap(),
2308 LinuxNamespaceBuilder::default()
2309 .typ(LinuxNamespaceType::Mount)
2310 .build()
2311 .unwrap(),
2312 ];
2313
2314 if self.host_network {
2329 } else if let Some(netns_path) = self.netns_path.as_ref() {
2331 namespaces.push(
2332 LinuxNamespaceBuilder::default()
2333 .typ(LinuxNamespaceType::Network)
2334 .path(netns_path.clone())
2335 .build()
2336 .unwrap(),
2337 );
2338 } else {
2339 namespaces.push(
2340 LinuxNamespaceBuilder::default()
2341 .typ(LinuxNamespaceType::Network)
2342 .build()
2343 .unwrap(),
2344 );
2345 }
2346
2347 #[cfg(unix)]
2359 let daemon_rootless = std::env::var_os("ZLAYER_ROOTLESS").is_some();
2360 #[cfg(unix)]
2361 let rootless = daemon_rootless || !nix::unistd::geteuid().is_root();
2362 #[cfg(not(unix))]
2363 let rootless = false;
2364
2365 if rootless {
2366 namespaces.push(
2367 LinuxNamespaceBuilder::default()
2368 .typ(LinuxNamespaceType::User)
2369 .build()
2370 .unwrap(),
2371 );
2372 namespaces.push(
2373 LinuxNamespaceBuilder::default()
2374 .typ(LinuxNamespaceType::Cgroup)
2375 .build()
2376 .unwrap(),
2377 );
2378 }
2379
2380 let mut linux_builder = LinuxBuilder::default().namespaces(namespaces);
2381
2382 #[cfg(unix)]
2383 if rootless {
2384 if daemon_rootless {
2385 linux_builder = linux_builder
2391 .uid_mappings(build_single_id_mapping(0))
2392 .gid_mappings(build_single_id_mapping(0));
2393 } else {
2394 let euid = nix::unistd::geteuid();
2395 let egid = nix::unistd::getegid();
2396 let username = nix::unistd::User::from_uid(euid)
2397 .ok()
2398 .flatten()
2399 .map(|u| u.name)
2400 .unwrap_or_default();
2401 linux_builder = linux_builder
2402 .uid_mappings(build_rootless_id_mappings(
2403 euid.as_raw(),
2404 "/etc/subuid",
2405 &username,
2406 ))
2407 .gid_mappings(build_rootless_id_mappings(
2408 egid.as_raw(),
2409 "/etc/subgid",
2410 &username,
2411 ));
2412 }
2413 }
2414
2415 let resources = self.build_resources(spec)?;
2417 if let Some(resources) = resources {
2418 linux_builder = linux_builder.resources(resources);
2419 }
2420
2421 let mut devices = self.build_devices(spec, None, cdi_edits.is_some())?;
2428 if let Some(edits_per_device) = cdi_edits {
2429 for edits in edits_per_device {
2430 for node in &edits.device_nodes {
2431 devices.push(cdi_node_to_oci_device(node)?);
2432 }
2433 }
2434 }
2435 if !devices.is_empty() {
2436 linux_builder = linux_builder.devices(devices);
2437 }
2438
2439 linux_builder = linux_builder.rootfs_propagation("private".to_string());
2441
2442 if spec.privileged {
2444 linux_builder = linux_builder.masked_paths(vec![]).readonly_paths(vec![]);
2446 } else {
2447 let masked_paths = vec![
2449 "/proc/acpi".to_string(),
2450 "/proc/asound".to_string(),
2451 "/proc/kcore".to_string(),
2452 "/proc/keys".to_string(),
2453 "/proc/latency_stats".to_string(),
2454 "/proc/timer_list".to_string(),
2455 "/proc/timer_stats".to_string(),
2456 "/proc/sched_debug".to_string(),
2457 "/proc/scsi".to_string(),
2458 "/sys/firmware".to_string(),
2459 ];
2460
2461 let readonly_paths = vec![
2463 "/proc/bus".to_string(),
2464 "/proc/fs".to_string(),
2465 "/proc/irq".to_string(),
2466 "/proc/sys".to_string(),
2467 "/proc/sysrq-trigger".to_string(),
2468 ];
2469
2470 linux_builder = linux_builder
2471 .masked_paths(masked_paths)
2472 .readonly_paths(readonly_paths);
2473 }
2474
2475 let cid = container_id.to_string();
2487
2488 let explicit_parent: Option<(String, &'static str)> =
2492 if let Some(p) = spec.cgroup_parent.as_deref().filter(|s| !s.is_empty()) {
2493 Some((p.to_string(), "spec"))
2494 } else if let Some(p) = std::env::var("ZLAYER_CGROUP_PARENT")
2495 .ok()
2496 .filter(|s| !s.is_empty())
2497 {
2498 Some((p, "env"))
2499 } else {
2500 None
2501 };
2502
2503 #[cfg(target_os = "linux")]
2509 let auto_parent: Option<(String, &'static str)> = {
2510 let host_mode = crate::capability::DaemonCapabilities::get().can_write_cgroup_root;
2522 if host_mode {
2523 if let Some(p) = crate::capability::ensure_host_container_parent() {
2524 Some((p, "auto-host"))
2525 } else if let Some(p) = crate::capability::ensure_daemon_leaf_and_container_parent()
2526 {
2527 Some((p, "auto-init"))
2530 } else if let Some(p) = crate::capability::current_cgroup_v2_path() {
2531 Some((p, "auto"))
2532 } else {
2533 None
2534 }
2535 } else if let Some(p) = crate::capability::ensure_daemon_leaf_and_container_parent() {
2536 Some((p, "auto-init"))
2541 } else if let Some(p) = crate::capability::current_cgroup_v2_path() {
2542 Some((p, "auto"))
2545 } else {
2546 None
2547 }
2548 };
2549 #[cfg(not(target_os = "linux"))]
2550 let auto_parent: Option<(String, &'static str)> = None;
2551
2552 let (cgroup_parent_value, cgroup_parent_source): (Option<String>, &'static str) =
2553 explicit_parent
2554 .or(auto_parent)
2555 .map_or((None, "none"), |(p, s)| (Some(p), s));
2556
2557 #[cfg(target_os = "linux")]
2564 if cgroup_parent_value.is_none() && crate::capability::DaemonCapabilities::get().is_nested {
2565 tracing::warn!(
2566 container_id = %cid,
2567 "capability survey reports nested daemon but cgroup_parent could not be resolved — proceeding with v2 root"
2568 );
2569 }
2570
2571 if let Some(parent) = cgroup_parent_value {
2572 let parent = parent.trim_end_matches('/');
2573 let full = format!("{parent}/{cid}");
2574 match cgroup_parent_source {
2575 "spec" => tracing::info!(
2576 container_id = %cid,
2577 source = "spec",
2578 path = %full,
2579 "cgroup_parent selected"
2580 ),
2581 "env" => tracing::info!(
2582 container_id = %cid,
2583 source = "env",
2584 path = %full,
2585 "cgroup_parent selected"
2586 ),
2587 "auto" => tracing::info!(
2588 container_id = %cid,
2589 source = "auto",
2590 path = %full,
2591 "cgroup_parent selected (from /proc/self/cgroup)"
2592 ),
2593 "auto-init" => tracing::info!(
2594 container_id = %cid,
2595 source = "auto-init",
2596 path = %full,
2597 "cgroup_parent selected (migrated daemon to <scope>/init; containers go under <scope>/containers)"
2598 ),
2599 "auto-host" => tracing::info!(
2600 container_id = %cid,
2601 source = "auto-host",
2602 path = %full,
2603 "cgroup_parent selected (host daemon; containers rooted at top-level /zlayer/containers, outside the unit cgroup)"
2604 ),
2605 _ => unreachable!(),
2606 }
2607 linux_builder = linux_builder.cgroups_path(std::path::PathBuf::from(full));
2608 } else {
2609 #[cfg(target_os = "linux")]
2618 {
2619 let caps = crate::capability::DaemonCapabilities::get();
2620 if !caps.can_write_cgroup_root {
2621 return Err(AgentError::InvalidSpec(format!(
2622 "cannot create container {cid}: no writable cgroup parent. \
2623 /proc/self/cgroup reports the cgroup-v2 root, and \
2624 /sys/fs/cgroup is read-only to this process. Fix one of: \
2625 (a) run the daemon's outer container with --cgroupns=host \
2626 so /proc/self/cgroup reports a real parent; \
2627 (b) set ZLAYER_CGROUP_PARENT=/path/to/writable/cgroup; \
2628 (c) grant the daemon write access to /sys/fs/cgroup."
2629 )));
2630 }
2631 tracing::info!(
2632 container_id = %cid,
2633 "cgroup_parent unset — libcontainer will use v2 root (cgroup root is writable here)"
2634 );
2635 }
2636 #[cfg(not(target_os = "linux"))]
2637 tracing::debug!(
2638 container_id = %cid,
2639 "non-Linux host — cgroup_parent unset; libcontainer inside the WSL distro will resolve a parent from its cgroup-v2 root"
2640 );
2641 }
2642
2643 linux_builder
2644 .build()
2645 .map_err(|e| AgentError::InvalidSpec(format!("failed to build linux config: {e}")))
2646 }
2647
2648 #[allow(clippy::cast_possible_truncation, clippy::cast_possible_wrap)]
2650 fn build_resources(
2651 &self,
2652 spec: &ServiceSpec,
2653 ) -> Result<Option<oci_spec::runtime::LinuxResources>> {
2654 let mut resources_builder = LinuxResourcesBuilder::default();
2655 let mut has_resources = false;
2656
2657 if let Some(cpu_limit) = spec.resources.cpu {
2659 let quota = (cpu_limit * 100_000.0) as i64;
2662 let cpu = LinuxCpuBuilder::default()
2663 .quota(quota)
2664 .period(100_000u64)
2665 .build()
2666 .map_err(|e| AgentError::InvalidSpec(format!("failed to build CPU limits: {e}")))?;
2667
2668 resources_builder = resources_builder.cpu(cpu);
2669 has_resources = true;
2670 }
2671
2672 if let Some(ref memory_str) = spec.resources.memory {
2674 let bytes = parse_memory_string(memory_str)
2675 .map_err(|e| AgentError::InvalidSpec(format!("invalid memory limit: {e}")))?;
2676
2677 let memory = LinuxMemoryBuilder::default()
2678 .limit(bytes as i64)
2679 .build()
2680 .map_err(|e| {
2681 AgentError::InvalidSpec(format!("failed to build memory limits: {e}"))
2682 })?;
2683
2684 resources_builder = resources_builder.memory(memory);
2685 has_resources = true;
2686 }
2687
2688 let device_rules = self.build_device_cgroup_rules(spec, None)?;
2690 if !device_rules.is_empty() {
2691 resources_builder = resources_builder.devices(device_rules);
2692 has_resources = true;
2693 }
2694
2695 if has_resources {
2696 let resources = resources_builder
2697 .build()
2698 .map_err(|e| AgentError::InvalidSpec(format!("failed to build resources: {e}")))?;
2699 Ok(Some(resources))
2700 } else {
2701 Ok(None)
2702 }
2703 }
2704
2705 #[allow(clippy::unused_self, clippy::too_many_lines)]
2707 fn build_device_cgroup_rules(
2708 &self,
2709 spec: &ServiceSpec,
2710 _gpu_indices: Option<&[u32]>,
2711 ) -> Result<Vec<oci_spec::runtime::LinuxDeviceCgroup>> {
2712 let mut rules = Vec::new();
2713
2714 if spec.privileged {
2715 let rule = LinuxDeviceCgroupBuilder::default()
2717 .allow(true)
2718 .access("rwm".to_string())
2719 .build()
2720 .map_err(|e| {
2721 AgentError::InvalidSpec(format!("failed to build device cgroup rule: {e}"))
2722 })?;
2723 rules.push(rule);
2724 } else {
2725 let deny_all = LinuxDeviceCgroupBuilder::default()
2727 .allow(false)
2728 .access("rwm".to_string())
2729 .build()
2730 .map_err(|e| AgentError::InvalidSpec(format!("failed to build deny rule: {e}")))?;
2731 rules.push(deny_all);
2732
2733 let standard_char_devices = [
2736 (1, 3, "rwm"), (1, 5, "rwm"), (1, 7, "rwm"), (1, 8, "rwm"), (1, 9, "rwm"), (5, 0, "rwm"), (5, 1, "rwm"), (5, 2, "rwm"), (136, -1, "rwm"), ];
2746
2747 for (major, minor, access) in standard_char_devices {
2748 let mut builder = LinuxDeviceCgroupBuilder::default()
2749 .allow(true)
2750 .typ(LinuxDeviceType::C)
2751 .major(i64::from(major))
2752 .access(access.to_string());
2753
2754 if minor >= 0 {
2755 builder = builder.minor(i64::from(minor));
2756 }
2757
2758 let rule = builder.build().map_err(|e| {
2759 AgentError::InvalidSpec(format!("failed to build char device rule: {e}"))
2760 })?;
2761 rules.push(rule);
2762 }
2763
2764 #[cfg(unix)]
2768 for device in &spec.devices {
2769 if let Ok((major, minor)) = get_device_major_minor(&device.path) {
2770 let dev_type = get_device_type(&device.path).unwrap_or(LinuxDeviceType::C);
2771
2772 let mut access = String::new();
2774 if device.read {
2775 access.push('r');
2776 }
2777 if device.write {
2778 access.push('w');
2779 }
2780 if device.mknod {
2781 access.push('m');
2782 }
2783 if access.is_empty() {
2784 access = "rw".to_string();
2785 }
2786
2787 let rule = LinuxDeviceCgroupBuilder::default()
2788 .allow(true)
2789 .typ(dev_type)
2790 .major(major)
2791 .minor(minor)
2792 .access(access)
2793 .build()
2794 .map_err(|e| {
2795 AgentError::InvalidSpec(format!(
2796 "failed to build device rule for {}: {}",
2797 device.path, e
2798 ))
2799 })?;
2800 rules.push(rule);
2801 } else {
2802 tracing::warn!("Failed to get device info for {}, skipping", device.path);
2803 }
2804 }
2805
2806 if let Some(ref gpu) = spec.resources.gpu {
2808 match gpu.vendor.as_str() {
2809 "nvidia" => {
2810 let rule = LinuxDeviceCgroupBuilder::default()
2812 .allow(true)
2813 .typ(LinuxDeviceType::C)
2814 .major(195i64)
2815 .access("rwm".to_string())
2816 .build()
2817 .map_err(|e| {
2818 AgentError::InvalidSpec(format!(
2819 "failed to build GPU cgroup rule: {e}"
2820 ))
2821 })?;
2822 rules.push(rule);
2823
2824 let uvm_rule = LinuxDeviceCgroupBuilder::default()
2826 .allow(true)
2827 .typ(LinuxDeviceType::C)
2828 .major(510i64)
2829 .access("rwm".to_string())
2830 .build()
2831 .map_err(|e| {
2832 AgentError::InvalidSpec(format!(
2833 "failed to build GPU UVM cgroup rule: {e}"
2834 ))
2835 })?;
2836 rules.push(uvm_rule);
2837 }
2838 "amd" => {
2839 let dri_rule = LinuxDeviceCgroupBuilder::default()
2841 .allow(true)
2842 .typ(LinuxDeviceType::C)
2843 .major(226i64)
2844 .access("rwm".to_string())
2845 .build()
2846 .map_err(|e| {
2847 AgentError::InvalidSpec(format!(
2848 "failed to build AMD DRI cgroup rule: {e}"
2849 ))
2850 })?;
2851 rules.push(dri_rule);
2852
2853 let kfd_rule = LinuxDeviceCgroupBuilder::default()
2855 .allow(true)
2856 .typ(LinuxDeviceType::C)
2857 .major(234i64)
2858 .access("rwm".to_string())
2859 .build()
2860 .map_err(|e| {
2861 AgentError::InvalidSpec(format!(
2862 "failed to build AMD KFD cgroup rule: {e}"
2863 ))
2864 })?;
2865 rules.push(kfd_rule);
2866 }
2867 "intel" => {
2868 let dri_rule = LinuxDeviceCgroupBuilder::default()
2870 .allow(true)
2871 .typ(LinuxDeviceType::C)
2872 .major(226i64)
2873 .access("rwm".to_string())
2874 .build()
2875 .map_err(|e| {
2876 AgentError::InvalidSpec(format!(
2877 "failed to build Intel DRI cgroup rule: {e}"
2878 ))
2879 })?;
2880 rules.push(dri_rule);
2881 }
2882 other => {
2883 tracing::warn!(
2885 vendor = %other,
2886 "Unknown GPU vendor, allowing DRI devices (major 226)"
2887 );
2888 let dri_rule = LinuxDeviceCgroupBuilder::default()
2889 .allow(true)
2890 .typ(LinuxDeviceType::C)
2891 .major(226i64)
2892 .access("rwm".to_string())
2893 .build()
2894 .map_err(|e| {
2895 AgentError::InvalidSpec(format!(
2896 "failed to build GPU DRI cgroup rule: {e}"
2897 ))
2898 })?;
2899 rules.push(dri_rule);
2900 }
2901 }
2902 }
2903 }
2904
2905 Ok(rules)
2906 }
2907
2908 #[allow(clippy::unused_self, clippy::too_many_lines)]
2917 #[cfg_attr(not(unix), allow(clippy::unnecessary_wraps, clippy::needless_return))]
2918 fn build_devices(
2919 &self,
2920 spec: &ServiceSpec,
2921 gpu_indices: Option<&[u32]>,
2922 skip_gpu_defaults: bool,
2923 ) -> Result<Vec<oci_spec::runtime::LinuxDevice>> {
2924 #[cfg(not(unix))]
2925 {
2926 let _ = (spec, gpu_indices, skip_gpu_defaults);
2927 return Ok(Vec::new());
2928 }
2929
2930 #[cfg(unix)]
2931 {
2932 let mut devices = Vec::new();
2933
2934 for device in &spec.devices {
2935 if let Ok((major, minor)) = get_device_major_minor(&device.path) {
2936 let dev_type = get_device_type(&device.path).unwrap_or(LinuxDeviceType::C);
2937
2938 let linux_device = LinuxDeviceBuilder::default()
2939 .path(device.path.clone())
2940 .typ(dev_type)
2941 .major(major)
2942 .minor(minor)
2943 .file_mode(0o666u32)
2944 .uid(0u32)
2945 .gid(0u32)
2946 .build()
2947 .map_err(|e| {
2948 AgentError::InvalidSpec(format!(
2949 "failed to build device {}: {}",
2950 device.path, e
2951 ))
2952 })?;
2953
2954 devices.push(linux_device);
2955 }
2956 }
2957
2958 if skip_gpu_defaults {
2963 return Ok(devices);
2964 }
2965
2966 if let Some(ref gpu) = spec.resources.gpu {
2968 let indices: Vec<u32> =
2969 gpu_indices.map_or_else(|| (0..gpu.count).collect(), <[u32]>::to_vec);
2970
2971 match gpu.vendor.as_str() {
2972 "nvidia" => {
2973 let always_devices =
2975 ["/dev/nvidiactl", "/dev/nvidia-uvm", "/dev/nvidia-uvm-tools"];
2976 for dev_path in &always_devices {
2977 if let Ok((major, minor)) = get_device_major_minor(dev_path) {
2978 let dev_type =
2979 get_device_type(dev_path).unwrap_or(LinuxDeviceType::C);
2980 let linux_device = LinuxDeviceBuilder::default()
2981 .path((*dev_path).to_string())
2982 .typ(dev_type)
2983 .major(major)
2984 .minor(minor)
2985 .file_mode(0o666u32)
2986 .uid(0u32)
2987 .gid(0u32)
2988 .build()
2989 .map_err(|e| {
2990 AgentError::InvalidSpec(format!(
2991 "failed to build GPU device {dev_path}: {e}"
2992 ))
2993 })?;
2994 devices.push(linux_device);
2995 } else {
2996 tracing::warn!(
2997 "GPU device {} not found on host, skipping",
2998 dev_path
2999 );
3000 }
3001 }
3002
3003 for i in &indices {
3005 let dev_path = format!("/dev/nvidia{i}");
3006 if let Ok((major, minor)) = get_device_major_minor(&dev_path) {
3007 let dev_type =
3008 get_device_type(&dev_path).unwrap_or(LinuxDeviceType::C);
3009 let linux_device = LinuxDeviceBuilder::default()
3010 .path(dev_path.clone())
3011 .typ(dev_type)
3012 .major(major)
3013 .minor(minor)
3014 .file_mode(0o666u32)
3015 .uid(0u32)
3016 .gid(0u32)
3017 .build()
3018 .map_err(|e| {
3019 AgentError::InvalidSpec(format!(
3020 "failed to build GPU device {dev_path}: {e}"
3021 ))
3022 })?;
3023 devices.push(linux_device);
3024 } else {
3025 tracing::warn!(
3026 "GPU device {} not found on host, skipping",
3027 dev_path
3028 );
3029 }
3030 }
3031 }
3032 "amd" => {
3033 let amd_always_devices = ["/dev/kfd"];
3035 for dev_path in &amd_always_devices {
3036 if let Ok((major, minor)) = get_device_major_minor(dev_path) {
3037 let dev_type =
3038 get_device_type(dev_path).unwrap_or(LinuxDeviceType::C);
3039 let linux_device = LinuxDeviceBuilder::default()
3040 .path((*dev_path).to_string())
3041 .typ(dev_type)
3042 .major(major)
3043 .minor(minor)
3044 .file_mode(0o666u32)
3045 .uid(0u32)
3046 .gid(0u32)
3047 .build()
3048 .map_err(|e| {
3049 AgentError::InvalidSpec(format!(
3050 "failed to build GPU device {dev_path}: {e}"
3051 ))
3052 })?;
3053 devices.push(linux_device);
3054 } else {
3055 tracing::warn!(
3056 "GPU device {} not found on host, skipping",
3057 dev_path
3058 );
3059 }
3060 }
3061
3062 for i in &indices {
3064 let dev_path = format!("/dev/dri/renderD{}", 128 + i);
3065 if let Ok((major, minor)) = get_device_major_minor(&dev_path) {
3066 let dev_type =
3067 get_device_type(&dev_path).unwrap_or(LinuxDeviceType::C);
3068 let linux_device = LinuxDeviceBuilder::default()
3069 .path(dev_path.clone())
3070 .typ(dev_type)
3071 .major(major)
3072 .minor(minor)
3073 .file_mode(0o666u32)
3074 .uid(0u32)
3075 .gid(0u32)
3076 .build()
3077 .map_err(|e| {
3078 AgentError::InvalidSpec(format!(
3079 "failed to build GPU device {dev_path}: {e}"
3080 ))
3081 })?;
3082 devices.push(linux_device);
3083 } else {
3084 tracing::warn!(
3085 "GPU device {} not found on host, skipping",
3086 dev_path
3087 );
3088 }
3089 }
3090
3091 for i in &indices {
3093 let dev_path = format!("/dev/dri/card{i}");
3094 if let Ok((major, minor)) = get_device_major_minor(&dev_path) {
3095 let dev_type =
3096 get_device_type(&dev_path).unwrap_or(LinuxDeviceType::C);
3097 let linux_device = LinuxDeviceBuilder::default()
3098 .path(dev_path.clone())
3099 .typ(dev_type)
3100 .major(major)
3101 .minor(minor)
3102 .file_mode(0o666u32)
3103 .uid(0u32)
3104 .gid(0u32)
3105 .build()
3106 .map_err(|e| {
3107 AgentError::InvalidSpec(format!(
3108 "failed to build GPU device {dev_path}: {e}"
3109 ))
3110 })?;
3111 devices.push(linux_device);
3112 } else {
3113 tracing::warn!(
3114 "GPU device {} not found on host, skipping",
3115 dev_path
3116 );
3117 }
3118 }
3119 }
3120 "intel" => {
3121 for i in &indices {
3123 let dev_path = format!("/dev/dri/renderD{}", 128 + i);
3124 if let Ok((major, minor)) = get_device_major_minor(&dev_path) {
3125 let dev_type =
3126 get_device_type(&dev_path).unwrap_or(LinuxDeviceType::C);
3127 let linux_device = LinuxDeviceBuilder::default()
3128 .path(dev_path.clone())
3129 .typ(dev_type)
3130 .major(major)
3131 .minor(minor)
3132 .file_mode(0o666u32)
3133 .uid(0u32)
3134 .gid(0u32)
3135 .build()
3136 .map_err(|e| {
3137 AgentError::InvalidSpec(format!(
3138 "failed to build GPU device {dev_path}: {e}"
3139 ))
3140 })?;
3141 devices.push(linux_device);
3142 } else {
3143 tracing::warn!(
3144 "GPU device {} not found on host, skipping",
3145 dev_path
3146 );
3147 }
3148 }
3149
3150 for i in &indices {
3152 let dev_path = format!("/dev/dri/card{i}");
3153 if let Ok((major, minor)) = get_device_major_minor(&dev_path) {
3154 let dev_type =
3155 get_device_type(&dev_path).unwrap_or(LinuxDeviceType::C);
3156 let linux_device = LinuxDeviceBuilder::default()
3157 .path(dev_path.clone())
3158 .typ(dev_type)
3159 .major(major)
3160 .minor(minor)
3161 .file_mode(0o666u32)
3162 .uid(0u32)
3163 .gid(0u32)
3164 .build()
3165 .map_err(|e| {
3166 AgentError::InvalidSpec(format!(
3167 "failed to build GPU device {dev_path}: {e}"
3168 ))
3169 })?;
3170 devices.push(linux_device);
3171 } else {
3172 tracing::warn!(
3173 "GPU device {} not found on host, skipping",
3174 dev_path
3175 );
3176 }
3177 }
3178 }
3179 other => {
3180 tracing::warn!(
3182 vendor = %other,
3183 "Unknown GPU vendor, attempting DRI device passthrough"
3184 );
3185 for i in &indices {
3186 let dev_path = format!("/dev/dri/renderD{}", 128 + i);
3187 if let Ok((major, minor)) = get_device_major_minor(&dev_path) {
3188 let dev_type =
3189 get_device_type(&dev_path).unwrap_or(LinuxDeviceType::C);
3190 let linux_device = LinuxDeviceBuilder::default()
3191 .path(dev_path.clone())
3192 .typ(dev_type)
3193 .major(major)
3194 .minor(minor)
3195 .file_mode(0o666u32)
3196 .uid(0u32)
3197 .gid(0u32)
3198 .build()
3199 .map_err(|e| {
3200 AgentError::InvalidSpec(format!(
3201 "failed to build GPU device {dev_path}: {e}"
3202 ))
3203 })?;
3204 devices.push(linux_device);
3205 } else {
3206 tracing::warn!(
3207 "GPU device {} not found on host, skipping",
3208 dev_path
3209 );
3210 }
3211 }
3212 }
3213 }
3214 }
3215
3216 Ok(devices)
3217 } }
3219
3220 pub async fn write_config(
3232 &self,
3233 container_id: &ContainerId,
3234 spec: &ServiceSpec,
3235 ) -> Result<PathBuf> {
3236 let oci_spec = self
3238 .build_spec_only(container_id, spec, &self.volume_paths)
3239 .await?;
3240
3241 let config_path = self.bundle_dir.join("config.json");
3243 let config_json =
3244 serde_json::to_string_pretty(&oci_spec).map_err(|e| AgentError::CreateFailed {
3245 id: container_id.to_string(),
3246 reason: format!("failed to serialize OCI spec: {e}"),
3247 })?;
3248
3249 fs::write(&config_path, config_json)
3250 .await
3251 .map_err(|e| AgentError::CreateFailed {
3252 id: container_id.to_string(),
3253 reason: format!("failed to write config.json: {e}"),
3254 })?;
3255
3256 tracing::debug!(
3257 "Wrote OCI config.json at {} for container {}",
3258 config_path.display(),
3259 container_id
3260 );
3261
3262 Ok(self.bundle_dir.clone())
3263 }
3264
3265 fn resolve_command_from_spec(
3274 spec: &ServiceSpec,
3275 image_config: Option<&zlayer_registry::ImageConfig>,
3276 ) -> Vec<String> {
3277 let mut args = Vec::new();
3278
3279 match (&spec.command.entrypoint, &spec.command.args) {
3280 (Some(entrypoint), Some(cmd_args)) => {
3281 args.extend_from_slice(entrypoint);
3282 args.extend_from_slice(cmd_args);
3283 }
3284 (Some(entrypoint), None) => {
3285 args.extend_from_slice(entrypoint);
3286 }
3287 (None, Some(cmd_args)) if !cmd_args.is_empty() => {
3288 args.extend_from_slice(cmd_args);
3289 }
3290 _ => {
3291 if let Some(img_cmd) =
3293 image_config.and_then(zlayer_registry::ImageConfig::full_command)
3294 {
3295 if img_cmd.is_empty() {
3296 args.push("/bin/sh".to_string());
3297 } else {
3298 args.extend(img_cmd);
3299 }
3300 } else {
3301 args.push("/bin/sh".to_string());
3302 }
3303 }
3304 }
3305
3306 args
3307 }
3308
3309 pub async fn cleanup(&self) -> Result<()> {
3316 if self.bundle_dir.exists() {
3317 fs::remove_dir_all(&self.bundle_dir)
3318 .await
3319 .map_err(|e| AgentError::CreateFailed {
3320 id: "cleanup".to_string(),
3321 reason: format!(
3322 "failed to remove bundle directory {}: {}",
3323 self.bundle_dir.display(),
3324 e
3325 ),
3326 })?;
3327 }
3328 Ok(())
3329 }
3330}
3331
3332#[cfg(unix)]
3345pub async fn create_bundle(
3346 container_id: &ContainerId,
3347 spec: &ServiceSpec,
3348 rootfs_path: Option<PathBuf>,
3349) -> Result<PathBuf> {
3350 let mut builder =
3351 BundleBuilder::for_container(container_id).with_host_network(spec.host_network);
3352
3353 if let Some(rootfs) = rootfs_path {
3354 builder = builder.with_rootfs(rootfs);
3355 }
3356
3357 builder.build(container_id, spec).await
3358}
3359
3360pub async fn cleanup_bundle(container_id: &ContainerId) -> Result<()> {
3367 let builder = BundleBuilder::for_container(container_id);
3368 builder.cleanup().await
3369}
3370
3371#[cfg(test)]
3372mod tests {
3373 use super::*;
3374 use zlayer_spec::*;
3375
3376 fn mock_spec() -> ServiceSpec {
3377 serde_yaml::from_str::<DeploymentSpec>(
3378 r"
3379version: v1
3380deployment: test
3381services:
3382 test:
3383 rtype: service
3384 image:
3385 name: test:latest
3386 endpoints:
3387 - name: http
3388 protocol: http
3389 port: 8080
3390",
3391 )
3392 .unwrap()
3393 .services
3394 .remove("test")
3395 .unwrap()
3396 }
3397
3398 #[cfg(target_os = "linux")]
3399 fn mock_spec_with_resources() -> ServiceSpec {
3400 serde_yaml::from_str::<DeploymentSpec>(
3401 r"
3402version: v1
3403deployment: test
3404services:
3405 test:
3406 rtype: service
3407 image:
3408 name: test:latest
3409 resources:
3410 cpu: 0.5
3411 memory: 512Mi
3412 env:
3413 MY_VAR: my_value
3414 ANOTHER: value2
3415 endpoints:
3416 - name: http
3417 protocol: http
3418 port: 8080
3419",
3420 )
3421 .unwrap()
3422 .services
3423 .remove("test")
3424 .unwrap()
3425 }
3426
3427 #[cfg(target_os = "linux")]
3428 fn mock_privileged_spec() -> ServiceSpec {
3429 serde_yaml::from_str::<DeploymentSpec>(
3430 r"
3431version: v1
3432deployment: test
3433services:
3434 test:
3435 rtype: service
3436 image:
3437 name: test:latest
3438 privileged: true
3439 endpoints:
3440 - name: http
3441 protocol: http
3442 port: 8080
3443",
3444 )
3445 .unwrap()
3446 .services
3447 .remove("test")
3448 .unwrap()
3449 }
3450
3451 #[test]
3452 fn test_parse_memory_string() {
3453 assert_eq!(parse_memory_string("512Mi").unwrap(), 512 * 1024 * 1024);
3454 assert_eq!(parse_memory_string("1Gi").unwrap(), 1024 * 1024 * 1024);
3455 assert_eq!(parse_memory_string("2G").unwrap(), 2 * 1000 * 1000 * 1000);
3456 assert_eq!(parse_memory_string("1024").unwrap(), 1024);
3457 assert_eq!(parse_memory_string("512Ki").unwrap(), 512 * 1024);
3458 }
3459
3460 #[test]
3461 fn test_parse_memory_string_errors() {
3462 assert!(parse_memory_string("").is_err());
3463 assert!(parse_memory_string("abc").is_err());
3464 assert!(parse_memory_string("12.5Mi").is_err());
3465 }
3466
3467 #[test]
3468 fn test_generate_resolv_conf_single_nameserver() {
3469 let out = generate_resolv_conf(&["10.42.0.1".to_string()], &[]);
3470 assert_eq!(out, "nameserver 10.42.0.1\noptions edns0\n");
3471 }
3472
3473 #[test]
3474 fn test_generate_resolv_conf_two_nameservers() {
3475 let out = generate_resolv_conf(&["10.42.0.1".to_string(), "fd00::1".to_string()], &[]);
3476 assert_eq!(
3477 out,
3478 "nameserver 10.42.0.1\nnameserver fd00::1\noptions edns0\n"
3479 );
3480 }
3481
3482 #[test]
3483 fn test_generate_resolv_conf_emits_search_domains() {
3484 let out = generate_resolv_conf(
3485 &["10.200.0.5".to_string()],
3486 &[
3487 "forgejo-stack.zlayer.local".to_string(),
3488 "zlayer.local".to_string(),
3489 ],
3490 );
3491 assert_eq!(
3492 out,
3493 "nameserver 10.200.0.5\nsearch forgejo-stack.zlayer.local zlayer.local\noptions edns0\n"
3494 );
3495 }
3496
3497 #[cfg(target_os = "linux")]
3498 #[tokio::test]
3499 async fn test_build_oci_spec_injects_resolv_conf_mount() {
3500 let dir = tempfile::tempdir().unwrap();
3501 let id = ContainerId::new("test".to_string(), 1);
3502 let mut spec = mock_spec();
3503 spec.dns = vec!["10.42.0.1".to_string()];
3504 let builder = BundleBuilder::new(dir.path().to_path_buf());
3505
3506 let oci_spec = builder
3507 .build_spec_only(&id, &spec, &std::collections::HashMap::new())
3508 .await
3509 .unwrap();
3510
3511 let mounts = oci_spec.mounts().as_ref().expect("mounts present");
3512 let resolv_mount = mounts
3513 .iter()
3514 .find(|m| m.destination() == Path::new("/etc/resolv.conf"))
3515 .expect("resolv.conf mount injected");
3516 let source = resolv_mount.source().as_ref().unwrap();
3517 let written = std::fs::read_to_string(source).unwrap();
3518 assert_eq!(written, "nameserver 10.42.0.1\noptions edns0\n");
3519 }
3520
3521 #[cfg(target_os = "linux")]
3522 #[tokio::test]
3523 async fn test_build_oci_spec_no_resolv_conf_when_dns_empty() {
3524 let dir = tempfile::tempdir().unwrap();
3525 let id = ContainerId::new("test".to_string(), 1);
3526 let spec = mock_spec(); let builder = BundleBuilder::new(dir.path().to_path_buf());
3528
3529 let oci_spec = builder
3530 .build_spec_only(&id, &spec, &std::collections::HashMap::new())
3531 .await
3532 .unwrap();
3533
3534 let mounts = oci_spec.mounts().as_ref().expect("mounts present");
3535 assert!(
3536 !mounts
3537 .iter()
3538 .any(|m| m.destination() == Path::new("/etc/resolv.conf")),
3539 "no resolv.conf mount should be injected for empty spec.dns"
3540 );
3541 }
3542
3543 #[cfg(target_os = "linux")]
3544 #[tokio::test]
3545 async fn test_build_oci_spec_no_resolv_conf_when_host_network() {
3546 let dir = tempfile::tempdir().unwrap();
3547 let id = ContainerId::new("test".to_string(), 1);
3548 let mut spec = mock_spec();
3549 spec.dns = vec!["10.42.0.1".to_string()];
3550 spec.host_network = true;
3551 let builder = BundleBuilder::new(dir.path().to_path_buf());
3552
3553 let oci_spec = builder
3554 .build_spec_only(&id, &spec, &std::collections::HashMap::new())
3555 .await
3556 .unwrap();
3557
3558 let mounts = oci_spec.mounts().as_ref().expect("mounts present");
3559 assert!(
3560 !mounts
3561 .iter()
3562 .any(|m| m.destination() == Path::new("/etc/resolv.conf")),
3563 "host_network containers must inherit the host resolv.conf"
3564 );
3565 }
3566
3567 #[cfg(target_os = "linux")]
3568 #[tokio::test]
3569 async fn test_build_oci_spec_errors_on_missing_bind_source() {
3570 let dir = tempfile::tempdir().unwrap();
3574 let id = ContainerId::new("test".to_string(), 1);
3575 let mut spec = mock_spec();
3576 let missing = dir.path().join("definitely-does-not-exist");
3577 spec.storage.push(StorageSpec::Bind {
3578 source: missing.to_string_lossy().into_owned(),
3579 target: "/data".to_string(),
3580 readonly: false,
3581 });
3582 let builder = BundleBuilder::new(dir.path().to_path_buf());
3583
3584 let err = builder
3585 .build_spec_only(&id, &spec, &std::collections::HashMap::new())
3586 .await
3587 .expect_err("missing bind source should error");
3588
3589 match err {
3590 AgentError::MountSourceMissing { src_path, dest } => {
3591 assert_eq!(src_path, missing.to_string_lossy());
3592 assert_eq!(dest, "/data");
3593 }
3594 other => panic!("expected MountSourceMissing, got {other:?}"),
3595 }
3596 }
3597
3598 #[cfg(target_os = "linux")]
3599 #[tokio::test]
3600 async fn test_build_oci_spec_accepts_existing_bind_source() {
3601 let dir = tempfile::tempdir().unwrap();
3603 let src = dir.path().join("present");
3604 std::fs::create_dir_all(&src).unwrap();
3605 let id = ContainerId::new("test".to_string(), 1);
3606 let mut spec = mock_spec();
3607 spec.storage.push(StorageSpec::Bind {
3608 source: src.to_string_lossy().into_owned(),
3609 target: "/data".to_string(),
3610 readonly: false,
3611 });
3612 let builder = BundleBuilder::new(dir.path().to_path_buf());
3613
3614 let oci_spec = builder
3615 .build_spec_only(&id, &spec, &std::collections::HashMap::new())
3616 .await
3617 .expect("existing bind source should build");
3618
3619 let mounts = oci_spec.mounts().as_ref().expect("mounts present");
3620 assert!(
3621 mounts.iter().any(|m| m.destination() == Path::new("/data")),
3622 "bind mount with existing source should be present"
3623 );
3624 }
3625
3626 #[cfg(target_os = "linux")]
3627 #[tokio::test]
3628 async fn test_resolv_conf_present_when_source_writable() {
3629 let dir = tempfile::tempdir().unwrap();
3632 let id = ContainerId::new("test".to_string(), 1);
3633 let mut spec = mock_spec();
3634 spec.dns = vec!["10.42.0.1".to_string()];
3635 let builder = BundleBuilder::new(dir.path().to_path_buf());
3636
3637 let oci_spec = builder
3638 .build_spec_only(&id, &spec, &std::collections::HashMap::new())
3639 .await
3640 .expect("writable bundle dir should build with resolv.conf");
3641
3642 let mounts = oci_spec.mounts().as_ref().expect("mounts present");
3643 let resolv = mounts
3644 .iter()
3645 .find(|m| m.destination() == Path::new("/etc/resolv.conf"))
3646 .expect("resolv.conf mount injected when source is writable");
3647 let source = resolv.source().as_ref().unwrap();
3648 assert!(source.exists(), "resolv.conf source must exist on disk");
3649 }
3650
3651 #[cfg(unix)]
3660 #[test]
3661 fn test_validate_host_bind_sources_skips_virtual_and_resolv() {
3662 let resolv = std::path::PathBuf::from("/nonexistent/bundle/resolv.conf");
3666
3667 let proc = MountBuilder::default()
3669 .destination("/proc".to_string())
3670 .typ("proc".to_string())
3671 .source("proc".to_string())
3672 .build()
3673 .unwrap();
3674 let resolv_mount = MountBuilder::default()
3675 .destination("/etc/resolv.conf".to_string())
3676 .typ("bind".to_string())
3677 .source(resolv.to_string_lossy().to_string())
3678 .options(vec!["rbind".to_string(), "ro".to_string()])
3679 .build()
3680 .unwrap();
3681 BundleBuilder::validate_host_bind_sources(&[proc, resolv_mount], Some(resolv.as_path()))
3682 .expect("virtual + exempt-resolv mounts must validate");
3683
3684 let storage = MountBuilder::default()
3686 .destination("/data".to_string())
3687 .typ("none".to_string())
3688 .source("/definitely/not/here".to_string())
3689 .options(vec!["rbind".to_string(), "rw".to_string()])
3690 .build()
3691 .unwrap();
3692 let err = BundleBuilder::validate_host_bind_sources(&[storage], None)
3693 .expect_err("missing storage bind source must error");
3694 assert!(matches!(err, AgentError::MountSourceMissing { .. }));
3695 }
3696
3697 #[test]
3698 fn test_bundle_builder_new() {
3699 let builder = BundleBuilder::new("/tmp/test-bundle".into());
3700 assert_eq!(builder.bundle_dir(), Path::new("/tmp/test-bundle"));
3701 assert!(builder.rootfs_path.is_none());
3702 }
3703
3704 #[test]
3705 fn test_bundle_builder_for_container() {
3706 let dirs = zlayer_paths::ZLayerDirs::system_default();
3707 let id = ContainerId::new("myservice".to_string(), 1);
3708 let builder = BundleBuilder::for_container(&id);
3709 assert_eq!(builder.bundle_dir(), dirs.bundles().join("myservice-rep-1"));
3710 }
3711
3712 #[test]
3713 fn test_bundle_builder_with_rootfs() {
3714 let dirs = zlayer_paths::ZLayerDirs::system_default();
3715 let builder = BundleBuilder::new("/tmp/test-bundle".into())
3716 .with_rootfs(dirs.rootfs().join("myimage"));
3717 assert_eq!(builder.rootfs_path, Some(dirs.rootfs().join("myimage")));
3718 }
3719
3720 #[cfg(target_os = "linux")]
3721 #[tokio::test]
3722 async fn test_build_oci_spec_basic() {
3723 let id = ContainerId::new("test".to_string(), 1);
3724 let spec = mock_spec();
3725 let builder = BundleBuilder::new("/tmp/test-bundle".into());
3726
3727 let oci_spec = builder
3728 .build_spec_only(&id, &spec, &std::collections::HashMap::new())
3729 .await
3730 .unwrap();
3731
3732 assert_eq!(oci_spec.version(), "1.0.2");
3733 assert!(oci_spec.root().is_some());
3734 assert_eq!(
3735 oci_spec.root().as_ref().unwrap().path(),
3736 std::path::Path::new("rootfs")
3737 );
3738 assert!(oci_spec.process().is_some());
3739 assert!(oci_spec.linux().is_some());
3740 }
3741
3742 #[cfg(target_os = "linux")]
3743 #[tokio::test]
3744 async fn test_build_oci_spec_with_resources() {
3745 let id = ContainerId::new("test".to_string(), 1);
3746 let spec = mock_spec_with_resources();
3747 let builder = BundleBuilder::new("/tmp/test-bundle".into());
3748
3749 let oci_spec = builder
3750 .build_spec_only(&id, &spec, &std::collections::HashMap::new())
3751 .await
3752 .unwrap();
3753
3754 let linux = oci_spec.linux().as_ref().unwrap();
3756 let resources = linux.resources().as_ref().unwrap();
3757
3758 let cpu = resources.cpu().as_ref().unwrap();
3760 assert_eq!(cpu.quota(), Some(50_000)); assert_eq!(cpu.period(), Some(100_000));
3762
3763 let memory = resources.memory().as_ref().unwrap();
3765 assert_eq!(memory.limit(), Some(512 * 1024 * 1024)); }
3767
3768 #[cfg(target_os = "linux")]
3769 #[tokio::test]
3770 async fn test_build_oci_spec_translates_ulimits() {
3771 let id = ContainerId::new("test".to_string(), 1);
3772 let mut spec = mock_spec();
3773 spec.ulimits.insert(
3774 "nofile".to_string(),
3775 UlimitSpec {
3776 soft: 100_000,
3777 hard: 200_000,
3778 },
3779 );
3780 spec.ulimits
3782 .insert("nproc".to_string(), UlimitSpec { soft: -1, hard: -5 });
3783 let builder = BundleBuilder::new("/tmp/test-bundle".into());
3784
3785 let oci_spec = builder
3786 .build_spec_only(&id, &spec, &std::collections::HashMap::new())
3787 .await
3788 .unwrap();
3789
3790 let process = oci_spec.process().as_ref().expect("process present");
3791 let rlimits = process.rlimits().as_ref().expect("rlimits present");
3792
3793 let nofile: Vec<_> = rlimits
3797 .iter()
3798 .filter(|r| r.typ() == PosixRlimitType::RlimitNofile)
3799 .collect();
3800 assert_eq!(nofile.len(), 1, "nofile must not be duplicated");
3801 assert_eq!(nofile[0].soft(), 100_000);
3802 assert_eq!(nofile[0].hard(), 200_000);
3803
3804 let nproc = rlimits
3805 .iter()
3806 .find(|r| r.typ() == PosixRlimitType::RlimitNproc)
3807 .expect("nproc rlimit present");
3808 assert_eq!(nproc.soft(), 0, "negative soft clamps to 0");
3809 assert_eq!(nproc.hard(), 0, "negative hard clamps to 0");
3810 }
3811
3812 #[cfg(target_os = "linux")]
3813 #[tokio::test]
3814 async fn test_build_oci_spec_rejects_unknown_ulimit() {
3815 let id = ContainerId::new("test".to_string(), 1);
3816 let mut spec = mock_spec();
3817 spec.ulimits.insert(
3818 "not_a_real_ulimit".to_string(),
3819 UlimitSpec { soft: 1, hard: 1 },
3820 );
3821 let builder = BundleBuilder::new("/tmp/test-bundle".into());
3822
3823 let err = builder
3824 .build_spec_only(&id, &spec, &std::collections::HashMap::new())
3825 .await
3826 .expect_err("unknown ulimit name must be rejected");
3827 assert!(
3828 err.to_string().contains("not_a_real_ulimit"),
3829 "error should name the unknown ulimit: {err}"
3830 );
3831 }
3832
3833 #[cfg(target_os = "linux")]
3834 #[tokio::test]
3835 async fn test_build_oci_spec_keeps_oci_default_rlimits_when_ulimits_empty() {
3836 let id = ContainerId::new("test".to_string(), 1);
3842 let spec = mock_spec();
3843 let builder = BundleBuilder::new("/tmp/test-bundle".into());
3844
3845 let oci_spec = builder
3846 .build_spec_only(&id, &spec, &std::collections::HashMap::new())
3847 .await
3848 .unwrap();
3849
3850 let process = oci_spec.process().as_ref().expect("process present");
3851 let rlimits = process
3852 .rlimits()
3853 .as_ref()
3854 .expect("oci default rlimits present");
3855 let nofile = rlimits
3856 .iter()
3857 .find(|r| r.typ() == PosixRlimitType::RlimitNofile)
3858 .expect("default nofile rlimit present");
3859 assert_eq!(nofile.soft(), 1024);
3862 assert_eq!(nofile.hard(), 1024);
3863 }
3864
3865 #[cfg(target_os = "linux")]
3866 #[tokio::test]
3867 async fn test_build_oci_spec_privileged() {
3868 let id = ContainerId::new("test".to_string(), 1);
3869 let spec = mock_privileged_spec();
3870 let builder = BundleBuilder::new("/tmp/test-bundle".into());
3871
3872 let oci_spec = builder
3873 .build_spec_only(&id, &spec, &std::collections::HashMap::new())
3874 .await
3875 .unwrap();
3876
3877 let process = oci_spec.process().as_ref().unwrap();
3879 let caps = process.capabilities().as_ref().unwrap();
3880 let bounding = caps.bounding().as_ref().unwrap();
3881
3882 assert!(bounding.contains(&Capability::SysAdmin));
3884 assert!(bounding.contains(&Capability::NetAdmin));
3885
3886 let linux = oci_spec.linux().as_ref().unwrap();
3888 assert!(
3889 linux.masked_paths().is_none() || linux.masked_paths().as_ref().unwrap().is_empty()
3890 );
3891 }
3892
3893 #[cfg(target_os = "linux")]
3894 #[tokio::test]
3895 async fn test_build_oci_spec_environment() {
3896 let id = ContainerId::new("test".to_string(), 1);
3897 let spec = mock_spec_with_resources();
3898 let builder = BundleBuilder::new("/tmp/test-bundle".into())
3899 .with_env("EXTRA_VAR".to_string(), "extra_value".to_string());
3900
3901 let oci_spec = builder
3902 .build_spec_only(&id, &spec, &std::collections::HashMap::new())
3903 .await
3904 .unwrap();
3905
3906 let process = oci_spec.process().as_ref().unwrap();
3907 let env = process.env().as_ref().unwrap();
3908
3909 assert!(env.iter().any(|e| e == "MY_VAR=my_value"));
3911 assert!(env.iter().any(|e| e == "ANOTHER=value2"));
3912 assert!(env.iter().any(|e| e == "EXTRA_VAR=extra_value"));
3914 assert!(env.iter().any(|e| e.starts_with("PATH=")));
3916 }
3917
3918 mod secret_scope_regression {
3937 use super::*;
3938 use async_trait::async_trait;
3939 use zlayer_secrets::{Secret, SecretMetadata, SecretsError, SecretsProvider};
3940
3941 struct ScopePinnedProvider {
3946 expect_scope: String,
3947 expect_name: String,
3948 value: String,
3949 }
3950
3951 #[async_trait]
3952 impl SecretsProvider for ScopePinnedProvider {
3953 async fn get_secret(&self, scope: &str, name: &str) -> zlayer_secrets::Result<Secret> {
3954 if scope == self.expect_scope && name == self.expect_name {
3955 Ok(Secret::new(&self.value))
3956 } else {
3957 Err(SecretsError::NotFound {
3958 name: name.to_string(),
3959 })
3960 }
3961 }
3962
3963 async fn get_secrets(
3964 &self,
3965 scope: &str,
3966 names: &[&str],
3967 ) -> zlayer_secrets::Result<HashMap<String, Secret>> {
3968 let mut out = HashMap::new();
3969 for name in names {
3970 if let Ok(secret) = self.get_secret(scope, name).await {
3971 out.insert((*name).to_string(), secret);
3972 }
3973 }
3974 Ok(out)
3975 }
3976
3977 async fn list_secrets(
3978 &self,
3979 _scope: &str,
3980 ) -> zlayer_secrets::Result<Vec<SecretMetadata>> {
3981 Ok(Vec::new())
3982 }
3983
3984 async fn exists(&self, scope: &str, name: &str) -> zlayer_secrets::Result<bool> {
3985 Ok(self.get_secret(scope, name).await.is_ok())
3986 }
3987 }
3988
3989 fn spec_with_secret_env() -> ServiceSpec {
3991 serde_yaml::from_str::<DeploymentSpec>(
3992 r"
3993version: v1
3994deployment: zatabase
3995services:
3996 test:
3997 rtype: service
3998 image:
3999 name: test:latest
4000 env:
4001 ZATA_SECRETS_KEY_HEX: $S:ZATA_SECRETS_KEY_HEX
4002 endpoints:
4003 - name: http
4004 protocol: http
4005 port: 8080
4006",
4007 )
4008 .unwrap()
4009 .services
4010 .remove("test")
4011 .unwrap()
4012 }
4013
4014 #[test]
4018 fn for_env_storage_scope_grammar_is_stable() {
4019 assert_eq!(
4020 SecretScope::for_env(Some("zatabase"), "env-uuid").to_storage_scope(),
4021 "project:zatabase:env:env-uuid"
4022 );
4023 assert_eq!(
4026 SecretScope::for_env(None, "env-uuid").to_storage_scope(),
4027 "env:env-uuid"
4028 );
4029 }
4030
4031 #[tokio::test]
4034 async fn bundle_gate_resolves_secret_with_correct_scope() {
4035 let provider = Arc::new(ScopePinnedProvider {
4036 expect_scope: "project:zatabase:env:env-uuid".to_string(),
4037 expect_name: "ZATA_SECRETS_KEY_HEX".to_string(),
4038 value: "deadbeef".to_string(),
4039 });
4040
4041 let id = ContainerId::new("test".to_string(), 1);
4042 let spec = spec_with_secret_env();
4043 let builder = BundleBuilder::new("/tmp/test-bundle-secret-ok".into())
4044 .with_secrets_provider(provider)
4045 .with_deployment_scope(SecretScope::for_env(Some("zatabase"), "env-uuid"));
4046
4047 let oci_spec = builder
4048 .build_spec_only(&id, &spec, &std::collections::HashMap::new())
4049 .await
4050 .expect("build should succeed when the secret resolves");
4051
4052 let env = oci_spec
4053 .process()
4054 .as_ref()
4055 .unwrap()
4056 .env()
4057 .as_ref()
4058 .unwrap()
4059 .clone();
4060
4061 assert!(
4063 env.iter().any(|e| e == "ZATA_SECRETS_KEY_HEX=deadbeef"),
4064 "expected resolved secret in OCI env, got: {env:?}"
4065 );
4066 assert!(
4068 !env.iter()
4069 .any(|e| e == "ZATA_SECRETS_KEY_HEX=$S:ZATA_SECRETS_KEY_HEX"),
4070 "literal $S: ref leaked into container env (the original bug): {env:?}"
4071 );
4072 }
4073
4074 #[tokio::test]
4080 async fn bundle_gate_wrong_scope_fails_to_resolve() {
4081 let provider = Arc::new(ScopePinnedProvider {
4082 expect_scope: "project:zatabase:env:env-uuid".to_string(),
4083 expect_name: "ZATA_SECRETS_KEY_HEX".to_string(),
4084 value: "deadbeef".to_string(),
4085 });
4086
4087 let id = ContainerId::new("test".to_string(), 1);
4088 let spec = spec_with_secret_env();
4089 let builder = BundleBuilder::new("/tmp/test-bundle-secret-wrong".into())
4091 .with_secrets_provider(provider)
4092 .with_deployment_scope(SecretScope::for_env(None, "env-uuid"));
4093
4094 let result = builder
4095 .build_spec_only(&id, &spec, &std::collections::HashMap::new())
4096 .await;
4097
4098 assert!(
4099 result.is_err(),
4100 "wrong scope must fail secret resolution, not silently pass the literal ref"
4101 );
4102 }
4103 }
4104
4105 #[cfg(target_os = "linux")]
4106 #[tokio::test]
4107 async fn test_build_namespaces() {
4108 let id = ContainerId::new("test".to_string(), 1);
4109 let spec = mock_spec();
4110 let builder = BundleBuilder::new("/tmp/test-bundle".into());
4111
4112 let oci_spec = builder
4113 .build_spec_only(&id, &spec, &std::collections::HashMap::new())
4114 .await
4115 .unwrap();
4116 let linux = oci_spec.linux().as_ref().unwrap();
4117 let namespaces = linux.namespaces().as_ref().unwrap();
4118
4119 let namespace_types: Vec<_> = namespaces
4121 .iter()
4122 .map(oci_spec::runtime::LinuxNamespace::typ)
4123 .collect();
4124 assert!(namespace_types.contains(&LinuxNamespaceType::Pid));
4125 assert!(namespace_types.contains(&LinuxNamespaceType::Ipc));
4126 assert!(namespace_types.contains(&LinuxNamespaceType::Uts));
4127 assert!(namespace_types.contains(&LinuxNamespaceType::Mount));
4128 assert!(namespace_types.contains(&LinuxNamespaceType::Network));
4129 }
4130
4131 #[cfg(target_os = "linux")]
4132 #[tokio::test]
4133 async fn test_build_namespaces_host_network() {
4134 let id = ContainerId::new("test".to_string(), 1);
4135 let spec = mock_spec();
4136 let builder = BundleBuilder::new("/tmp/test-bundle".into()).with_host_network(true);
4137
4138 let oci_spec = builder
4139 .build_spec_only(&id, &spec, &std::collections::HashMap::new())
4140 .await
4141 .unwrap();
4142 let linux = oci_spec.linux().as_ref().unwrap();
4143 let namespaces = linux.namespaces().as_ref().unwrap();
4144
4145 let namespace_types: Vec<_> = namespaces
4147 .iter()
4148 .map(oci_spec::runtime::LinuxNamespace::typ)
4149 .collect();
4150 assert!(namespace_types.contains(&LinuxNamespaceType::Pid));
4151 assert!(namespace_types.contains(&LinuxNamespaceType::Ipc));
4152 assert!(namespace_types.contains(&LinuxNamespaceType::Uts));
4153 assert!(namespace_types.contains(&LinuxNamespaceType::Mount));
4154 assert!(
4155 !namespace_types.contains(&LinuxNamespaceType::Network),
4156 "Network namespace should NOT be present in host_network mode"
4157 );
4158 }
4159
4160 #[cfg(target_os = "linux")]
4165 #[tokio::test]
4166 async fn test_build_namespaces_netns_path_cases() {
4167 use std::path::PathBuf;
4168
4169 let id = ContainerId::new("test".to_string(), 1);
4170 let spec = mock_spec();
4171 let empty = std::collections::HashMap::new();
4172
4173 let net_ns = |oci: &oci_spec::runtime::Spec| -> Option<Option<PathBuf>> {
4174 oci.linux()
4175 .as_ref()
4176 .and_then(|l| l.namespaces().as_ref())
4177 .and_then(|ns| {
4178 ns.iter()
4179 .find(|n| n.typ() == LinuxNamespaceType::Network)
4180 .map(|n| n.path().clone())
4181 })
4182 };
4183
4184 let host = BundleBuilder::new("/tmp/test-bundle".into())
4186 .with_host_network(true)
4187 .with_netns_path(Some(PathBuf::from("/proc/1234/ns/net")))
4188 .build_spec_only(&id, &spec, &empty)
4189 .await
4190 .unwrap();
4191 assert!(
4192 net_ns(&host).is_none(),
4193 "host_network must emit no Network namespace even when netns_path is set"
4194 );
4195
4196 let join_path = PathBuf::from("/proc/4242/ns/net");
4198 let joined = BundleBuilder::new("/tmp/test-bundle".into())
4199 .with_netns_path(Some(join_path.clone()))
4200 .build_spec_only(&id, &spec, &empty)
4201 .await
4202 .unwrap();
4203 assert_eq!(
4204 net_ns(&joined),
4205 Some(Some(join_path)),
4206 "netns_path must produce a Network namespace carrying that path"
4207 );
4208
4209 let fresh = BundleBuilder::new("/tmp/test-bundle".into())
4211 .build_spec_only(&id, &spec, &empty)
4212 .await
4213 .unwrap();
4214 assert_eq!(
4215 net_ns(&fresh),
4216 Some(None),
4217 "default mode must produce a Network namespace with no path"
4218 );
4219 }
4220
4221 #[test]
4222 fn test_build_default_mounts() {
4223 let spec = mock_spec();
4224 let builder = BundleBuilder::new("/tmp/test-bundle".into());
4225
4226 let mounts = builder.build_default_mounts(&spec).unwrap();
4227
4228 let mount_destinations: Vec<_> = mounts
4230 .iter()
4231 .map(|m| m.destination().to_string_lossy().to_string())
4232 .collect();
4233 assert!(mount_destinations.contains(&"/proc".to_string()));
4234 assert!(mount_destinations.contains(&"/dev".to_string()));
4235 assert!(mount_destinations.contains(&"/dev/pts".to_string()));
4236 assert!(mount_destinations.contains(&"/dev/shm".to_string()));
4237 assert!(mount_destinations.contains(&"/sys".to_string()));
4238 }
4239
4240 #[test]
4241 fn test_build_storage_mounts_bind() {
4242 let spec = serde_yaml::from_str::<zlayer_spec::DeploymentSpec>(
4243 r"
4244version: v1
4245deployment: test
4246services:
4247 test:
4248 image:
4249 name: test:latest
4250 storage:
4251 - type: bind
4252 source: /host/data
4253 target: /app/data
4254 readonly: true
4255",
4256 )
4257 .unwrap()
4258 .services
4259 .remove("test")
4260 .unwrap();
4261
4262 let builder = BundleBuilder::new("/tmp/test-bundle".into());
4263 let volume_paths = std::collections::HashMap::new();
4264
4265 let mounts = builder.build_storage_mounts(&spec, &volume_paths).unwrap();
4266
4267 assert_eq!(mounts.len(), 1);
4268 assert_eq!(mounts[0].destination().to_string_lossy(), "/app/data");
4269 assert_eq!(
4270 mounts[0]
4271 .source()
4272 .as_ref()
4273 .map(|s| s.to_string_lossy().to_string()),
4274 Some("/host/data".to_string())
4275 );
4276 let options = mounts[0].options().as_ref().unwrap();
4277 assert!(options.contains(&"rbind".to_string()));
4278 assert!(options.contains(&"ro".to_string()));
4279 }
4280
4281 #[test]
4282 fn test_build_storage_mounts_named() {
4283 let spec = serde_yaml::from_str::<zlayer_spec::DeploymentSpec>(
4284 r"
4285version: v1
4286deployment: test
4287services:
4288 test:
4289 image:
4290 name: test:latest
4291 storage:
4292 - type: named
4293 name: my-volume
4294 target: /app/data
4295",
4296 )
4297 .unwrap()
4298 .services
4299 .remove("test")
4300 .unwrap();
4301
4302 let dirs = zlayer_paths::ZLayerDirs::system_default();
4303 let builder = BundleBuilder::new("/tmp/test-bundle".into());
4304 let mut volume_paths = std::collections::HashMap::new();
4305 volume_paths.insert("my-volume".to_string(), dirs.volumes().join("my-volume"));
4306
4307 let mounts = builder.build_storage_mounts(&spec, &volume_paths).unwrap();
4308
4309 assert_eq!(mounts.len(), 1);
4310 assert_eq!(mounts[0].destination().to_string_lossy(), "/app/data");
4311 assert_eq!(
4312 mounts[0]
4313 .source()
4314 .as_ref()
4315 .map(|s| s.to_string_lossy().to_string()),
4316 Some(
4317 dirs.volumes()
4318 .join("my-volume")
4319 .to_string_lossy()
4320 .into_owned()
4321 )
4322 );
4323 }
4324
4325 #[test]
4326 fn test_build_storage_mounts_tmpfs() {
4327 let spec = serde_yaml::from_str::<zlayer_spec::DeploymentSpec>(
4328 r"
4329version: v1
4330deployment: test
4331services:
4332 test:
4333 image:
4334 name: test:latest
4335 storage:
4336 - type: tmpfs
4337 target: /app/tmp
4338 size: 256Mi
4339 mode: 1777
4340",
4341 )
4342 .unwrap()
4343 .services
4344 .remove("test")
4345 .unwrap();
4346
4347 let builder = BundleBuilder::new("/tmp/test-bundle".into());
4348 let volume_paths = std::collections::HashMap::new();
4349
4350 let mounts = builder.build_storage_mounts(&spec, &volume_paths).unwrap();
4351
4352 assert_eq!(mounts.len(), 1);
4353 assert_eq!(mounts[0].destination().to_string_lossy(), "/app/tmp");
4354 assert_eq!(mounts[0].typ().as_ref().map(String::as_str), Some("tmpfs"));
4355 let options = mounts[0].options().as_ref().unwrap();
4356 assert!(options.iter().any(|o| o.starts_with("size=")));
4357 assert!(options.iter().any(|o| o.starts_with("mode=")));
4358 }
4359
4360 #[test]
4361 fn test_build_storage_mounts_multiple() {
4362 let spec = serde_yaml::from_str::<zlayer_spec::DeploymentSpec>(
4363 r"
4364version: v1
4365deployment: test
4366services:
4367 test:
4368 image:
4369 name: test:latest
4370 storage:
4371 - type: bind
4372 source: /etc/config
4373 target: /app/config
4374 readonly: true
4375 - type: named
4376 name: app-data
4377 target: /app/data
4378 - type: tmpfs
4379 target: /app/tmp
4380",
4381 )
4382 .unwrap()
4383 .services
4384 .remove("test")
4385 .unwrap();
4386
4387 let dirs = zlayer_paths::ZLayerDirs::system_default();
4388 let builder = BundleBuilder::new("/tmp/test-bundle".into());
4389 let mut volume_paths = std::collections::HashMap::new();
4390 volume_paths.insert("app-data".to_string(), dirs.volumes().join("app-data"));
4391
4392 let mounts = builder.build_storage_mounts(&spec, &volume_paths).unwrap();
4393
4394 assert_eq!(mounts.len(), 3);
4395
4396 let destinations: Vec<String> = mounts
4398 .iter()
4399 .map(|m| m.destination().to_string_lossy().to_string())
4400 .collect();
4401 assert!(destinations.contains(&"/app/config".to_string()));
4402 assert!(destinations.contains(&"/app/data".to_string()));
4403 assert!(destinations.contains(&"/app/tmp".to_string()));
4404 }
4405
4406 #[test]
4407 fn test_build_storage_mounts_anonymous_missing_path() {
4408 let spec = serde_yaml::from_str::<zlayer_spec::DeploymentSpec>(
4409 r"
4410version: v1
4411deployment: test
4412services:
4413 test:
4414 image:
4415 name: test:latest
4416 storage:
4417 - type: anonymous
4418 target: /app/cache
4419",
4420 )
4421 .unwrap()
4422 .services
4423 .remove("test")
4424 .unwrap();
4425
4426 let builder = BundleBuilder::new("/tmp/test-bundle".into());
4427 let volume_paths = std::collections::HashMap::new(); let result = builder.build_storage_mounts(&spec, &volume_paths);
4430
4431 assert!(result.is_err());
4433 }
4434
4435 #[cfg(target_os = "linux")]
4436 #[tokio::test]
4437 async fn test_oci_spec_includes_storage_mounts() {
4438 let id = ContainerId::new("test".to_string(), 1);
4439 let src_dir = tempfile::tempdir().unwrap();
4443 let src_path = src_dir.path().to_string_lossy().into_owned();
4444 let spec = serde_yaml::from_str::<zlayer_spec::DeploymentSpec>(&format!(
4445 r"
4446version: v1
4447deployment: test
4448services:
4449 test:
4450 image:
4451 name: test:latest
4452 storage:
4453 - type: bind
4454 source: {src_path}
4455 target: /app/data
4456 - type: tmpfs
4457 target: /app/tmp
4458"
4459 ))
4460 .unwrap()
4461 .services
4462 .remove("test")
4463 .unwrap();
4464
4465 let builder = BundleBuilder::new("/tmp/test-bundle".into());
4466 let volume_paths = std::collections::HashMap::new();
4467
4468 let oci_spec = builder
4469 .build_spec_only(&id, &spec, &volume_paths)
4470 .await
4471 .unwrap();
4472
4473 let mounts = oci_spec.mounts().as_ref().unwrap();
4475 let destinations: Vec<String> = mounts
4476 .iter()
4477 .map(|m| m.destination().to_string_lossy().to_string())
4478 .collect();
4479
4480 assert!(destinations.contains(&"/proc".to_string())); assert!(destinations.contains(&"/dev".to_string())); assert!(destinations.contains(&"/app/data".to_string())); assert!(destinations.contains(&"/app/tmp".to_string())); }
4486
4487 fn mock_gpu_spec(vendor: &str, count: u32) -> ServiceSpec {
4488 let yaml = format!(
4489 "
4490version: v1
4491deployment: test
4492services:
4493 test:
4494 rtype: service
4495 image:
4496 name: test:latest
4497 resources:
4498 gpu:
4499 count: {count}
4500 vendor: {vendor}
4501 endpoints:
4502 - name: http
4503 protocol: http
4504 port: 8080
4505"
4506 );
4507 serde_yaml::from_str::<DeploymentSpec>(&yaml)
4508 .unwrap()
4509 .services
4510 .remove("test")
4511 .unwrap()
4512 }
4513
4514 fn write_nvidia_cdi_fixture(dir: &std::path::Path, json: &str) {
4515 std::fs::write(dir.join("nvidia.json"), json).unwrap();
4516 }
4517
4518 fn nvidia_cdi_fixture() -> &'static str {
4519 r#"{
4520 "cdiVersion": "0.6.0",
4521 "kind": "nvidia.com/gpu",
4522 "devices": [{
4523 "name": "0",
4524 "containerEdits": {
4525 "deviceNodes": [
4526 {"path": "/dev/nvidia0", "type": "c", "major": 195, "minor": 0}
4527 ],
4528 "env": ["NVIDIA_VISIBLE_DEVICES=0"],
4529 "hooks": {
4530 "createContainer": [{
4531 "path": "/usr/bin/nvidia-container-runtime-hook",
4532 "args": ["nvidia-container-runtime-hook", "prestart"]
4533 }]
4534 }
4535 }
4536 }]
4537 }"#
4538 }
4539
4540 #[cfg(target_os = "linux")]
4541 #[tokio::test]
4542 async fn gpu_spec_translates_to_cdi_device_nodes() {
4543 let dir = tempfile::tempdir().unwrap();
4544 write_nvidia_cdi_fixture(dir.path(), nvidia_cdi_fixture());
4545 let registry = std::sync::Arc::new(crate::cdi::CdiRegistry::discover_from(&[dir.path()]));
4546
4547 let id = ContainerId::new("test".to_string(), 1);
4548 let spec = mock_gpu_spec("nvidia", 1);
4549 let builder = BundleBuilder::new("/tmp/test-bundle-cdi".into()).with_cdi_registry(registry);
4550
4551 let oci_spec = builder
4552 .build_oci_spec(&id, &spec, &std::collections::HashMap::new())
4553 .await
4554 .expect("build with CDI fixture");
4555
4556 let linux = oci_spec.linux().as_ref().expect("linux config present");
4558 let devices = linux.devices().as_ref().expect("devices present");
4559 assert!(
4560 devices
4561 .iter()
4562 .any(|d| d.path() == std::path::Path::new("/dev/nvidia0")),
4563 "expected /dev/nvidia0 from CDI fixture; got {:?}",
4564 devices
4565 .iter()
4566 .map(oci_spec::runtime::LinuxDevice::path)
4567 .collect::<Vec<_>>()
4568 );
4569
4570 let process = oci_spec.process().as_ref().expect("process present");
4572 let env = process.env().as_ref().expect("env present");
4573 assert!(
4574 env.iter().any(|e| e == "NVIDIA_VISIBLE_DEVICES=0"),
4575 "expected NVIDIA_VISIBLE_DEVICES=0 in env; got {env:?}"
4576 );
4577
4578 let hooks = oci_spec.hooks().as_ref().expect("hooks present");
4580 let create_container = hooks
4581 .create_container()
4582 .as_ref()
4583 .expect("createContainer hooks present");
4584 assert_eq!(create_container.len(), 1);
4585 assert_eq!(
4586 create_container[0].path(),
4587 &std::path::PathBuf::from("/usr/bin/nvidia-container-runtime-hook")
4588 );
4589 }
4590
4591 #[tokio::test]
4592 async fn gpu_spec_with_missing_cdi_returns_error() {
4593 let dir = tempfile::tempdir().unwrap();
4595 let registry = std::sync::Arc::new(crate::cdi::CdiRegistry::discover_from(&[dir.path()]));
4596
4597 let id = ContainerId::new("test".to_string(), 1);
4598 let spec = mock_gpu_spec("nvidia", 1);
4599 let builder =
4600 BundleBuilder::new("/tmp/test-bundle-cdi-missing".into()).with_cdi_registry(registry);
4601
4602 let err = builder
4603 .build_oci_spec(&id, &spec, &std::collections::HashMap::new())
4604 .await
4605 .expect_err("should fail when CDI registry is empty");
4606
4607 match err {
4608 AgentError::InvalidSpec(msg) => {
4609 assert!(
4610 msg.contains("nvidia") || msg.contains("CDI"),
4611 "error should mention CDI / vendor; got: {msg}"
4612 );
4613 }
4614 other => panic!("expected InvalidSpec, got {other:?}"),
4615 }
4616 }
4617
4618 #[tokio::test]
4619 async fn gpu_spec_with_unknown_device_returns_error() {
4620 let dir = tempfile::tempdir().unwrap();
4623 write_nvidia_cdi_fixture(dir.path(), nvidia_cdi_fixture());
4624 let registry = std::sync::Arc::new(crate::cdi::CdiRegistry::discover_from(&[dir.path()]));
4625
4626 let id = ContainerId::new("test".to_string(), 1);
4627 let spec = mock_gpu_spec("nvidia", 2);
4628 let builder =
4629 BundleBuilder::new("/tmp/test-bundle-cdi-unknown".into()).with_cdi_registry(registry);
4630
4631 let err = builder
4632 .build_oci_spec(&id, &spec, &std::collections::HashMap::new())
4633 .await
4634 .expect_err("should fail when device '1' is not declared");
4635 match err {
4636 AgentError::InvalidSpec(msg) => {
4637 assert!(
4638 msg.contains("'1'") || msg.contains("device"),
4639 "error should mention the missing device; got: {msg}"
4640 );
4641 }
4642 other => panic!("expected InvalidSpec, got {other:?}"),
4643 }
4644 }
4645
4646 #[cfg(target_os = "linux")]
4647 #[tokio::test]
4648 async fn gpu_spec_with_all_devices_expands_to_all_in_spec() {
4649 let dir = tempfile::tempdir().unwrap();
4651 let fixture = r#"{
4652 "cdiVersion": "0.6.0",
4653 "kind": "nvidia.com/gpu",
4654 "devices": [
4655 {
4656 "name": "0",
4657 "containerEdits": {
4658 "env": ["NVIDIA_VISIBLE_DEVICES=0"],
4659 "deviceNodes": [
4660 {"path": "/dev/nvidia0", "type": "c", "major": 195, "minor": 0}
4661 ]
4662 }
4663 },
4664 {
4665 "name": "1",
4666 "containerEdits": {
4667 "env": ["NVIDIA_VISIBLE_DEVICES=1"],
4668 "deviceNodes": [
4669 {"path": "/dev/nvidia1", "type": "c", "major": 195, "minor": 1}
4670 ]
4671 }
4672 }
4673 ]
4674 }"#;
4675 write_nvidia_cdi_fixture(dir.path(), fixture);
4676 let registry = std::sync::Arc::new(crate::cdi::CdiRegistry::discover_from(&[dir.path()]));
4677
4678 let edits = registry
4681 .resolve_for_kind("nvidia.com/gpu", &["all".to_string()])
4682 .expect("resolve all");
4683 assert_eq!(edits.len(), 2);
4684
4685 let id = ContainerId::new("test".to_string(), 1);
4688 let spec = mock_gpu_spec("nvidia", 2);
4689 let builder =
4690 BundleBuilder::new("/tmp/test-bundle-cdi-all".into()).with_cdi_registry(registry);
4691
4692 let oci_spec = builder
4693 .build_oci_spec(&id, &spec, &std::collections::HashMap::new())
4694 .await
4695 .expect("build with 2-device fixture");
4696
4697 let devices = oci_spec
4698 .linux()
4699 .as_ref()
4700 .unwrap()
4701 .devices()
4702 .as_ref()
4703 .expect("devices present");
4704 let paths: Vec<_> = devices.iter().map(|d| d.path().clone()).collect();
4705 assert!(paths.contains(&std::path::PathBuf::from("/dev/nvidia0")));
4706 assert!(paths.contains(&std::path::PathBuf::from("/dev/nvidia1")));
4707 }
4708
4709 fn build_nvidia_cdi_registry(dir: &std::path::Path) -> std::sync::Arc<crate::cdi::CdiRegistry> {
4714 write_nvidia_cdi_fixture(dir, nvidia_cdi_fixture());
4715 std::sync::Arc::new(crate::cdi::CdiRegistry::discover_from(&[dir]))
4716 }
4717
4718 #[cfg(target_os = "linux")]
4719 #[tokio::test]
4720 async fn gpu_spec_with_mps_sharing_injects_env_and_mounts() {
4721 let cdi_dir = tempfile::tempdir().unwrap();
4725 let mps_root = tempfile::tempdir().unwrap();
4726 let pipe_dir = mps_root.path().join("nvidia-mps");
4727 let log_dir = mps_root.path().join("nvidia-log");
4728 std::fs::create_dir(&pipe_dir).unwrap();
4729 std::fs::create_dir(&log_dir).unwrap();
4730 let registry = build_nvidia_cdi_registry(cdi_dir.path());
4731
4732 let id = ContainerId::new("test".to_string(), 1);
4733 let mut spec = mock_gpu_spec("nvidia", 1);
4734 let gpu = spec.resources.gpu.as_mut().expect("gpu spec set");
4735 gpu.sharing = Some(zlayer_spec::GpuSharingMode::Mps);
4736 gpu.mps_pipe_dir = Some(pipe_dir.to_string_lossy().into_owned());
4737 gpu.mps_log_dir = Some(log_dir.to_string_lossy().into_owned());
4738
4739 let builder =
4740 BundleBuilder::new("/tmp/test-bundle-mps-env".into()).with_cdi_registry(registry);
4741 let oci_spec = builder
4742 .build_oci_spec(&id, &spec, &std::collections::HashMap::new())
4743 .await
4744 .expect("build with MPS sharing");
4745
4746 let env = oci_spec
4747 .process()
4748 .as_ref()
4749 .and_then(|p| p.env().as_ref())
4750 .expect("env present");
4751 let pipe_expect = format!("CUDA_MPS_PIPE_DIRECTORY={}", pipe_dir.display());
4752 let log_expect = format!("CUDA_MPS_LOG_DIRECTORY={}", log_dir.display());
4753 assert!(
4754 env.iter().any(|e| e == &pipe_expect),
4755 "expected {pipe_expect} in env; got {env:?}"
4756 );
4757 assert!(
4758 env.iter().any(|e| e == &log_expect),
4759 "expected {log_expect} in env; got {env:?}"
4760 );
4761
4762 let mounts = oci_spec.mounts().as_ref().expect("mounts present");
4763 assert!(
4764 mounts
4765 .iter()
4766 .any(|m| m.destination() == &pipe_dir && m.source().as_ref() == Some(&pipe_dir)),
4767 "expected bind mount of MPS pipe dir {}; got destinations {:?}",
4768 pipe_dir.display(),
4769 mounts.iter().map(Mount::destination).collect::<Vec<_>>()
4770 );
4771 assert!(
4772 mounts
4773 .iter()
4774 .any(|m| m.destination() == &log_dir && m.source().as_ref() == Some(&log_dir)),
4775 "expected bind mount of MPS log dir {}",
4776 log_dir.display()
4777 );
4778 }
4779
4780 #[tokio::test]
4781 async fn gpu_spec_with_mps_sharing_fails_when_pipe_dir_missing() {
4782 let cdi_dir = tempfile::tempdir().unwrap();
4783 let registry = build_nvidia_cdi_registry(cdi_dir.path());
4784
4785 let id = ContainerId::new("test".to_string(), 1);
4786 let mut spec = mock_gpu_spec("nvidia", 1);
4787 let gpu = spec.resources.gpu.as_mut().expect("gpu spec set");
4788 gpu.sharing = Some(zlayer_spec::GpuSharingMode::Mps);
4789 let missing = tempfile::tempdir().unwrap();
4792 let missing_path = missing.path().join("definitely-not-here");
4793 gpu.mps_pipe_dir = Some(missing_path.to_string_lossy().into_owned());
4794
4795 let builder =
4796 BundleBuilder::new("/tmp/test-bundle-mps-missing".into()).with_cdi_registry(registry);
4797 let err = builder
4798 .build_oci_spec(&id, &spec, &std::collections::HashMap::new())
4799 .await
4800 .expect_err("should fail when MPS pipe dir is missing");
4801 match err {
4802 AgentError::GpuSharingUnavailable { mode, reason } => {
4803 assert_eq!(mode, "mps");
4804 assert!(
4805 reason.contains("pipe") || reason.contains(&missing_path.display().to_string()),
4806 "reason should mention the missing path; got: {reason}"
4807 );
4808 }
4809 other => panic!("expected GpuSharingUnavailable, got {other:?}"),
4810 }
4811 }
4812
4813 #[cfg(target_os = "linux")]
4814 #[tokio::test]
4815 async fn gpu_spec_with_timeslicing_injects_visible_devices() {
4816 let cdi_dir = tempfile::tempdir().unwrap();
4817 let registry = build_nvidia_cdi_registry(cdi_dir.path());
4818
4819 let id = ContainerId::new("test".to_string(), 1);
4820 let mut spec = mock_gpu_spec("nvidia", 1);
4821 let gpu = spec.resources.gpu.as_mut().expect("gpu spec set");
4822 gpu.sharing = Some(zlayer_spec::GpuSharingMode::TimeSlice);
4823 gpu.time_slice_index = Some(2);
4824
4825 let builder =
4826 BundleBuilder::new("/tmp/test-bundle-timeslice".into()).with_cdi_registry(registry);
4827 let oci_spec = builder
4828 .build_oci_spec(&id, &spec, &std::collections::HashMap::new())
4829 .await
4830 .expect("build with time-slicing");
4831
4832 let env = oci_spec
4833 .process()
4834 .as_ref()
4835 .and_then(|p| p.env().as_ref())
4836 .expect("env present");
4837 let cuda_entries: Vec<&String> = env
4840 .iter()
4841 .filter(|e| e.starts_with("CUDA_VISIBLE_DEVICES="))
4842 .collect();
4843 assert_eq!(
4844 cuda_entries.len(),
4845 1,
4846 "exactly one CUDA_VISIBLE_DEVICES expected; got {cuda_entries:?}"
4847 );
4848 assert_eq!(cuda_entries[0], "CUDA_VISIBLE_DEVICES=2");
4849 }
4850
4851 #[cfg(target_os = "linux")]
4852 #[tokio::test]
4853 async fn gpu_spec_without_sharing_omits_mps_env() {
4854 let cdi_dir = tempfile::tempdir().unwrap();
4855 let registry = build_nvidia_cdi_registry(cdi_dir.path());
4856
4857 let id = ContainerId::new("test".to_string(), 1);
4858 let spec = mock_gpu_spec("nvidia", 1);
4859 assert!(spec.resources.gpu.as_ref().unwrap().sharing.is_none());
4860
4861 let builder =
4862 BundleBuilder::new("/tmp/test-bundle-no-sharing".into()).with_cdi_registry(registry);
4863 let oci_spec = builder
4864 .build_oci_spec(&id, &spec, &std::collections::HashMap::new())
4865 .await
4866 .expect("build without sharing");
4867
4868 let env = oci_spec
4869 .process()
4870 .as_ref()
4871 .and_then(|p| p.env().as_ref())
4872 .expect("env present");
4873 assert!(
4874 !env.iter().any(|e| e.starts_with("CUDA_MPS_")),
4875 "no CUDA_MPS_* env should be present without sharing; got {env:?}"
4876 );
4877
4878 let mounts = oci_spec.mounts().as_ref().expect("mounts present");
4882 assert!(
4883 !mounts
4884 .iter()
4885 .any(|m| { m.destination().to_string_lossy().contains("nvidia-mps") }),
4886 "no MPS pipe mount should be present without sharing"
4887 );
4888 }
4889
4890 #[cfg(unix)]
4891 mod subid_tests {
4892 use super::super::{build_single_id_mapping, read_subid_range};
4893 use std::io::Write;
4894
4895 #[test]
4896 fn build_single_id_mapping_yields_one_identity_mapping() {
4897 let mappings = build_single_id_mapping(0);
4898 assert_eq!(mappings.len(), 1);
4899 assert_eq!(mappings[0].container_id(), 0);
4900 assert_eq!(mappings[0].host_id(), 0);
4901 assert_eq!(mappings[0].size(), 1);
4902 }
4903
4904 #[test]
4905 fn read_subid_range_returns_range_for_user() {
4906 let mut tmp = tempfile::NamedTempFile::new().unwrap();
4907 writeln!(tmp, "alice:100000:65536").unwrap();
4908 writeln!(tmp, "bob:165536:65536").unwrap();
4909 tmp.flush().unwrap();
4910 let path = tmp.path().to_str().unwrap();
4911 assert_eq!(read_subid_range(path, "bob"), Some((165_536, 65_536)));
4912 assert_eq!(read_subid_range(path, "alice"), Some((100_000, 65_536)));
4913 }
4914
4915 #[test]
4916 fn read_subid_range_returns_none_for_unknown_user() {
4917 let mut tmp = tempfile::NamedTempFile::new().unwrap();
4918 writeln!(tmp, "alice:100000:65536").unwrap();
4919 tmp.flush().unwrap();
4920 assert_eq!(
4921 read_subid_range(tmp.path().to_str().unwrap(), "carol"),
4922 None
4923 );
4924 }
4925
4926 #[test]
4927 fn read_subid_range_returns_none_on_missing_file() {
4928 assert_eq!(
4929 read_subid_range("/this/path/does/not/exist/subuid", "anyone"),
4930 None
4931 );
4932 }
4933 }
4934
4935 fn swarm_env_map(sharding: &ShardingSpec) -> std::collections::HashMap<String, String> {
4937 swarm_ring_env(sharding).into_iter().collect()
4938 }
4939
4940 #[test]
4941 fn swarm_ring_env_middle_stage() {
4942 let sharding = ShardingSpec {
4945 swarm_id: "swarm-x".to_string(),
4946 layer_start: 12,
4947 layer_end: 24,
4948 layer_count: 36,
4949 role: SwarmRole::Stage,
4950 manifest_ref: None,
4951 peers: vec![
4952 SwarmPeer {
4953 service: "svc-a".to_string(),
4954 layer_start: 0,
4955 layer_end: 12,
4956 },
4957 SwarmPeer {
4958 service: "svc-c".to_string(),
4959 layer_start: 24,
4960 layer_end: 36,
4961 },
4962 ],
4963 coordinator: Some("svc-coord".to_string()),
4964 };
4965 let env = swarm_env_map(&sharding);
4966 assert_eq!(
4967 env.get("ZLAYER_SWARM_ID").map(String::as_str),
4968 Some("swarm-x")
4969 );
4970 assert_eq!(
4971 env.get("ZLAYER_SWARM_ROLE").map(String::as_str),
4972 Some("stage")
4973 );
4974 assert_eq!(
4975 env.get("ZLAYER_SWARM_LAYER_START").map(String::as_str),
4976 Some("12")
4977 );
4978 assert_eq!(
4979 env.get("ZLAYER_SWARM_LAYER_END").map(String::as_str),
4980 Some("24")
4981 );
4982 assert_eq!(
4983 env.get("ZLAYER_SWARM_TOTAL_LAYERS").map(String::as_str),
4984 Some("36")
4985 );
4986 assert_eq!(
4987 env.get("ZLAYER_SWARM_COORDINATOR").map(String::as_str),
4988 Some("svc-coord")
4989 );
4990 assert_eq!(
4991 env.get("ZLAYER_SWARM_NEXT_PEER").map(String::as_str),
4992 Some("svc-c")
4993 );
4994 assert_eq!(
4995 env.get("ZLAYER_SWARM_PREV_PEER").map(String::as_str),
4996 Some("svc-a")
4997 );
4998 }
4999
5000 #[test]
5001 fn swarm_ring_env_first_stage_prev_is_coordinator() {
5002 let sharding = ShardingSpec {
5005 swarm_id: "swarm-y".to_string(),
5006 layer_start: 0,
5007 layer_end: 12,
5008 layer_count: 36,
5009 role: SwarmRole::Stage,
5010 manifest_ref: None,
5011 peers: vec![
5012 SwarmPeer {
5013 service: "svc-b".to_string(),
5014 layer_start: 12,
5015 layer_end: 24,
5016 },
5017 SwarmPeer {
5018 service: "svc-c".to_string(),
5019 layer_start: 24,
5020 layer_end: 36,
5021 },
5022 ],
5023 coordinator: Some("svc-coord".to_string()),
5024 };
5025 let env = swarm_env_map(&sharding);
5026 assert_eq!(
5027 env.get("ZLAYER_SWARM_ROLE").map(String::as_str),
5028 Some("stage")
5029 );
5030 assert_eq!(
5031 env.get("ZLAYER_SWARM_PREV_PEER").map(String::as_str),
5032 Some("svc-coord")
5033 );
5034 assert_eq!(
5035 env.get("ZLAYER_SWARM_NEXT_PEER").map(String::as_str),
5036 Some("svc-b")
5037 );
5038 }
5039
5040 #[test]
5041 fn swarm_ring_env_last_stage_next_is_coordinator() {
5042 let sharding = ShardingSpec {
5045 swarm_id: "swarm-z".to_string(),
5046 layer_start: 24,
5047 layer_end: 36,
5048 layer_count: 36,
5049 role: SwarmRole::Stage,
5050 manifest_ref: None,
5051 peers: vec![
5052 SwarmPeer {
5053 service: "svc-a".to_string(),
5054 layer_start: 0,
5055 layer_end: 12,
5056 },
5057 SwarmPeer {
5058 service: "svc-b".to_string(),
5059 layer_start: 12,
5060 layer_end: 24,
5061 },
5062 ],
5063 coordinator: Some("svc-coord".to_string()),
5064 };
5065 let env = swarm_env_map(&sharding);
5066 assert_eq!(
5067 env.get("ZLAYER_SWARM_NEXT_PEER").map(String::as_str),
5068 Some("svc-coord")
5069 );
5070 assert_eq!(
5071 env.get("ZLAYER_SWARM_PREV_PEER").map(String::as_str),
5072 Some("svc-b")
5073 );
5074 }
5075
5076 #[test]
5077 fn swarm_ring_env_coordinator_role() {
5078 let sharding = ShardingSpec {
5081 swarm_id: "swarm-c".to_string(),
5082 layer_start: 0,
5083 layer_end: 0,
5084 layer_count: 36,
5085 role: SwarmRole::Coordinator,
5086 manifest_ref: None,
5087 peers: vec![
5088 SwarmPeer {
5089 service: "svc-a".to_string(),
5090 layer_start: 0,
5091 layer_end: 12,
5092 },
5093 SwarmPeer {
5094 service: "svc-b".to_string(),
5095 layer_start: 12,
5096 layer_end: 24,
5097 },
5098 SwarmPeer {
5099 service: "svc-c".to_string(),
5100 layer_start: 24,
5101 layer_end: 36,
5102 },
5103 ],
5104 coordinator: None,
5105 };
5106 let env = swarm_env_map(&sharding);
5107 assert_eq!(
5108 env.get("ZLAYER_SWARM_ROLE").map(String::as_str),
5109 Some("coordinator")
5110 );
5111 assert_eq!(
5112 env.get("ZLAYER_SWARM_NEXT_PEER").map(String::as_str),
5113 Some("svc-a")
5114 );
5115 assert_eq!(
5116 env.get("ZLAYER_SWARM_PREV_PEER").map(String::as_str),
5117 Some("svc-c")
5118 );
5119 assert!(!env.contains_key("ZLAYER_SWARM_LAYER_START"));
5121 assert!(!env.contains_key("ZLAYER_SWARM_TOTAL_LAYERS"));
5122 }
5123
5124 #[test]
5125 fn swarm_ring_env_single_stage_no_peers_uses_coordinator() {
5126 let sharding = ShardingSpec {
5128 swarm_id: "solo".to_string(),
5129 layer_start: 0,
5130 layer_end: 36,
5131 layer_count: 36,
5132 role: SwarmRole::Stage,
5133 manifest_ref: None,
5134 peers: vec![],
5135 coordinator: Some("svc-coord".to_string()),
5136 };
5137 let env = swarm_env_map(&sharding);
5138 assert_eq!(
5139 env.get("ZLAYER_SWARM_NEXT_PEER").map(String::as_str),
5140 Some("svc-coord")
5141 );
5142 assert_eq!(
5143 env.get("ZLAYER_SWARM_PREV_PEER").map(String::as_str),
5144 Some("svc-coord")
5145 );
5146 }
5147
5148 #[test]
5149 fn swarm_ring_env_single_stage_no_peers_no_coordinator_emits_no_neighbors() {
5150 let sharding = ShardingSpec {
5152 swarm_id: "solo2".to_string(),
5153 layer_start: 0,
5154 layer_end: 36,
5155 layer_count: 36,
5156 role: SwarmRole::Stage,
5157 manifest_ref: None,
5158 peers: vec![],
5159 coordinator: None,
5160 };
5161 let env = swarm_env_map(&sharding);
5162 assert!(!env.contains_key("ZLAYER_SWARM_NEXT_PEER"));
5163 assert!(!env.contains_key("ZLAYER_SWARM_PREV_PEER"));
5164 assert!(!env.contains_key("ZLAYER_SWARM_COORDINATOR"));
5165 }
5166}