alien_core/resources/
daemon.rs

1use crate::error::{ErrorData, Result};
2use crate::resource::{ResourceDefinition, ResourceOutputsDefinition, ResourceRef, ResourceType};
3use crate::resources::{
4    ComputeCluster, ExposeProtocol, HealthCheck, PublicEndpoint, PublicEndpointOutput,
5    ResourceSpec, ToolchainConfig, APEX_HOST_LABEL,
6};
7use alien_error::AlienError;
8use bon::Builder;
9use serde::{Deserialize, Serialize};
10use std::any::Any;
11use std::collections::HashMap;
12use std::fmt::Debug;
13
14#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
15#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
16#[serde(rename_all = "camelCase", tag = "type")]
17pub enum DaemonCode {
18    #[serde(rename_all = "camelCase")]
19    Image { image: String },
20    #[serde(rename_all = "camelCase")]
21    Source {
22        src: String,
23        toolchain: ToolchainConfig,
24    },
25}
26
27#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
28#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
29#[serde(rename_all = "camelCase", deny_unknown_fields)]
30pub struct DaemonRuntimeMount {
31    /// Absolute host path to mount into the daemon container.
32    pub source: String,
33    /// Absolute container path where the source is mounted.
34    pub target: String,
35    /// Optional mount options understood by the backend runtime.
36    #[serde(skip_serializing_if = "Option::is_none")]
37    pub options: Option<String>,
38}
39
40#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
41#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
42#[serde(rename_all = "camelCase", deny_unknown_fields)]
43pub struct DaemonRuntime {
44    /// Run the daemon container with elevated host capabilities.
45    #[serde(skip_serializing_if = "Option::is_none")]
46    pub privileged: Option<bool>,
47    /// Process namespace mode. Supported values are `host` and `private`.
48    #[serde(skip_serializing_if = "Option::is_none")]
49    pub pid_namespace: Option<String>,
50    /// Network mode. Supported values are `host` and `appnet`.
51    #[serde(skip_serializing_if = "Option::is_none")]
52    pub network_mode: Option<String>,
53    /// Host mounts exposed to the daemon container.
54    #[serde(default, skip_serializing_if = "Vec::is_empty")]
55    pub mounts: Vec<DaemonRuntimeMount>,
56    /// Runtime user, as a numeric uid or uid:gid string.
57    #[serde(skip_serializing_if = "Option::is_none")]
58    pub user: Option<String>,
59}
60
61#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Builder)]
62#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
63#[serde(rename_all = "camelCase", deny_unknown_fields)]
64#[builder(start_fn = new)]
65pub struct Daemon {
66    #[builder(start_fn)]
67    pub id: String,
68    #[builder(field)]
69    pub links: Vec<ResourceRef>,
70    /// Public endpoints exposed by the daemon.
71    #[builder(field)]
72    #[serde(default, skip_serializing_if = "Vec::is_empty")]
73    pub public_endpoints: Vec<PublicEndpoint>,
74    /// HTTP health check for public daemon endpoint load balancers.
75    #[serde(skip_serializing_if = "Option::is_none")]
76    pub health_check: Option<HealthCheck>,
77    /// ComputeCluster resource ID that this daemon runs on for managed cloud
78    /// compute backends. Kubernetes and Local runtimes ignore this field.
79    #[serde(skip_serializing_if = "Option::is_none")]
80    pub cluster: Option<String>,
81    /// Named workload permission profile. Absent means no workload cloud identity.
82    #[serde(default, skip_serializing_if = "Option::is_none")]
83    pub permissions: Option<String>,
84    pub code: DaemonCode,
85    /// CPU resource requirements for each daemon instance.
86    #[builder(default = default_daemon_cpu())]
87    #[serde(default = "default_daemon_cpu")]
88    pub cpu: ResourceSpec,
89    /// Memory resource requirements for each daemon instance.
90    #[builder(default = default_daemon_memory())]
91    #[serde(default = "default_daemon_memory")]
92    pub memory: ResourceSpec,
93    /// Capacity group/pool to run on for backends that expose machine pools.
94    #[serde(skip_serializing_if = "Option::is_none")]
95    pub pool: Option<String>,
96    /// Command to override the image default.
97    #[serde(skip_serializing_if = "Option::is_none")]
98    pub command: Option<Vec<String>>,
99    /// Grace period in seconds for stopping daemon instances during updates, drains, and deletes.
100    ///
101    /// When omitted, the runtime backend applies its default. Valid values are
102    /// 1 second through 24 hours.
103    #[serde(skip_serializing_if = "Option::is_none")]
104    #[cfg_attr(feature = "openapi", schema(minimum = 1, maximum = 86400))]
105    pub stop_grace_period_seconds: Option<u32>,
106    /// Optional backend runtime settings for trusted daemons.
107    ///
108    /// These settings are intended for daemon-style infrastructure that must
109    /// operate on the host. Backends that do not support a setting may reject
110    /// it during provisioning.
111    #[serde(skip_serializing_if = "Option::is_none")]
112    pub runtime: Option<DaemonRuntime>,
113    #[builder(default)]
114    #[serde(default)]
115    pub environment: HashMap<String, String>,
116    /// Whether an app-owned command receiver can lease pending commands for
117    /// this Daemon and execute registered handlers.
118    #[builder(default = default_commands_enabled())]
119    #[serde(default = "default_commands_enabled")]
120    #[cfg_attr(feature = "openapi", schema(default = default_commands_enabled))]
121    pub commands_enabled: bool,
122}
123
124impl Daemon {
125    pub const RESOURCE_TYPE: ResourceType = ResourceType::from_static("daemon");
126
127    pub fn get_permissions(&self) -> Option<&str> {
128        self.permissions.as_deref()
129    }
130
131    fn validate_public_endpoints(&self) -> Result<()> {
132        let mut endpoint_names = std::collections::HashSet::new();
133        let mut backend_ports = std::collections::HashSet::new();
134        let mut apex_endpoint_name: Option<&str> = None;
135
136        for endpoint in &self.public_endpoints {
137            endpoint.validate_for_resource(&self.id)?;
138            if !endpoint_names.insert(endpoint.name.as_str()) {
139                return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
140                    resource_id: self.id.clone(),
141                    reason: format!("duplicate public endpoint name '{}'", endpoint.name),
142                }));
143            }
144            if endpoint.host_label.as_deref() == Some(APEX_HOST_LABEL) {
145                if let Some(existing_name) = apex_endpoint_name {
146                    return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
147                        resource_id: self.id.clone(),
148                        reason: format!(
149                            "only one apex public endpoint is allowed per resource; '{}' already uses hostLabel '@'",
150                            existing_name
151                        ),
152                    }));
153                }
154                apex_endpoint_name = Some(endpoint.name.as_str());
155            }
156            backend_ports.insert(endpoint.port);
157            if endpoint.protocol != ExposeProtocol::Http {
158                return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
159                    resource_id: self.id.clone(),
160                    reason: "daemon public endpoints currently support only HTTP".to_string(),
161                }));
162            }
163        }
164
165        if backend_ports.len() > 1 {
166            return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
167                resource_id: self.id.clone(),
168                reason:
169                    "public endpoints on one daemon must currently route to the same backend port"
170                        .to_string(),
171            }));
172        }
173
174        Ok(())
175    }
176
177    fn validate_runtime(&self) -> Result<()> {
178        let Some(runtime) = &self.runtime else {
179            return Ok(());
180        };
181
182        if let Some(pid_namespace) = &runtime.pid_namespace {
183            if pid_namespace != "host" && pid_namespace != "private" {
184                return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
185                    resource_id: self.id.clone(),
186                    reason: "runtime.pidNamespace must be 'host' or 'private'".to_string(),
187                }));
188            }
189        }
190
191        if let Some(network_mode) = &runtime.network_mode {
192            if network_mode != "host" && network_mode != "appnet" {
193                return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
194                    resource_id: self.id.clone(),
195                    reason: "runtime.networkMode must be 'host' or 'appnet'".to_string(),
196                }));
197            }
198        }
199
200        if let Some(user) = &runtime.user {
201            let valid = match user.split_once(':') {
202                Some((uid, gid)) => {
203                    !uid.is_empty()
204                        && !gid.is_empty()
205                        && uid.chars().all(|c| c.is_ascii_digit())
206                        && gid.chars().all(|c| c.is_ascii_digit())
207                }
208                None => !user.is_empty() && user.chars().all(|c| c.is_ascii_digit()),
209            };
210            if !valid {
211                return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
212                    resource_id: self.id.clone(),
213                    reason: "runtime.user must be a numeric uid or uid:gid".to_string(),
214                }));
215            }
216        }
217
218        for mount in &runtime.mounts {
219            if mount.source.is_empty() || mount.target.is_empty() {
220                return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
221                    resource_id: self.id.clone(),
222                    reason: "runtime.mounts source and target must be non-empty".to_string(),
223                }));
224            }
225            if !mount.source.starts_with('/') || !mount.target.starts_with('/') {
226                return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
227                    resource_id: self.id.clone(),
228                    reason: "runtime.mounts source and target must be absolute paths".to_string(),
229                }));
230            }
231        }
232
233        Ok(())
234    }
235}
236
237fn default_commands_enabled() -> bool {
238    false
239}
240
241fn default_daemon_cpu() -> ResourceSpec {
242    ResourceSpec {
243        min: "0.1".to_string(),
244        desired: "0.1".to_string(),
245    }
246}
247
248fn default_daemon_memory() -> ResourceSpec {
249    ResourceSpec {
250        min: "128Mi".to_string(),
251        desired: "128Mi".to_string(),
252    }
253}
254
255impl<S: daemon_builder::State> DaemonBuilder<S> {
256    pub fn link<R: ?Sized>(mut self, resource: &R) -> Self
257    where
258        for<'a> &'a R: Into<ResourceRef>,
259    {
260        let resource_ref: ResourceRef = resource.into();
261        self.links.push(resource_ref);
262        self
263    }
264
265    pub fn public_endpoint(mut self, endpoint: PublicEndpoint) -> Self {
266        self.public_endpoints.push(endpoint);
267        self
268    }
269}
270
271impl ResourceDefinition for Daemon {
272    fn get_resource_type(&self) -> ResourceType {
273        Self::RESOURCE_TYPE
274    }
275
276    fn id(&self) -> &str {
277        &self.id
278    }
279
280    fn replace_after_failed_create_is_safe(&self) -> bool {
281        // A create that timed out may already run replicas; deleting it stops them all, while
282        // creating it again updates the daemon in place.
283        false
284    }
285
286    fn get_dependencies(&self) -> Vec<ResourceRef> {
287        let mut dependencies = self.links.clone();
288        if let Some(cluster) = &self.cluster {
289            dependencies.push(ResourceRef::new(
290                ComputeCluster::RESOURCE_TYPE,
291                cluster.clone(),
292            ));
293        }
294        dependencies
295    }
296
297    fn get_permissions(&self) -> Option<&str> {
298        self.permissions.as_deref()
299    }
300
301    fn validate_update(&self, new_config: &dyn ResourceDefinition) -> Result<()> {
302        let new_daemon = new_config
303            .as_any()
304            .downcast_ref::<Daemon>()
305            .ok_or_else(|| {
306                AlienError::new(ErrorData::UnexpectedResourceType {
307                    resource_id: self.id.clone(),
308                    expected: Self::RESOURCE_TYPE,
309                    actual: new_config.get_resource_type(),
310                })
311            })?;
312
313        if self.id != new_daemon.id {
314            return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
315                resource_id: self.id.clone(),
316                reason: "the 'id' field is immutable".to_string(),
317            }));
318        }
319
320        self.validate_public_endpoints()?;
321        new_daemon.validate_public_endpoints()?;
322        self.validate_runtime()?;
323        new_daemon.validate_runtime()?;
324
325        if self.public_endpoints != new_daemon.public_endpoints {
326            return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
327                resource_id: self.id.clone(),
328                reason: "the 'publicEndpoints' field is immutable".to_string(),
329            }));
330        }
331
332        Ok(())
333    }
334
335    fn as_any(&self) -> &dyn Any {
336        self
337    }
338
339    fn as_any_mut(&mut self) -> &mut dyn Any {
340        self
341    }
342
343    fn box_clone(&self) -> Box<dyn ResourceDefinition> {
344        Box::new(self.clone())
345    }
346
347    fn resource_eq(&self, other: &dyn ResourceDefinition) -> bool {
348        other.as_any().downcast_ref::<Daemon>() == Some(self)
349    }
350
351    fn to_json_value(&self) -> serde_json::Result<serde_json::Value> {
352        serde_json::to_value(self)
353    }
354}
355
356#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
357#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
358#[serde(rename_all = "camelCase")]
359pub struct DaemonOutputs {
360    pub daemon_name: String,
361    pub running: bool,
362    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
363    pub public_endpoints: HashMap<String, PublicEndpointOutput>,
364}
365
366impl ResourceOutputsDefinition for DaemonOutputs {
367    fn get_resource_type(&self) -> ResourceType {
368        Daemon::RESOURCE_TYPE.clone()
369    }
370
371    fn as_any(&self) -> &dyn Any {
372        self
373    }
374
375    fn box_clone(&self) -> Box<dyn ResourceOutputsDefinition> {
376        Box::new(self.clone())
377    }
378
379    fn outputs_eq(&self, other: &dyn ResourceOutputsDefinition) -> bool {
380        other.as_any().downcast_ref::<DaemonOutputs>() == Some(self)
381    }
382
383    fn to_json_value(&self) -> serde_json::Result<serde_json::Value> {
384        serde_json::to_value(self)
385    }
386}
387
388#[cfg(test)]
389mod tests {
390    use super::*;
391
392    #[test]
393    fn daemon_without_permissions_preserves_links_without_an_identity() {
394        let storage = crate::Storage::new("objects".to_string()).build();
395        let daemon = Daemon::new("observer".to_string())
396            .code(DaemonCode::Image {
397                image: "observer:latest".to_string(),
398            })
399            .link(&storage)
400            .build();
401        assert_eq!(daemon.get_permissions(), None);
402        assert_eq!(ResourceDefinition::get_permissions(&daemon), None);
403        assert_eq!(daemon.get_dependencies(), daemon.links);
404        assert_eq!(daemon.links[0].id(), "objects");
405        let json = serde_json::to_value(&daemon).unwrap();
406        assert!(json.get("permissions").is_none());
407        let decoded: Daemon = serde_json::from_value(json).unwrap();
408        assert_eq!(decoded, daemon);
409
410        let mut explicit = serde_json::to_value(&daemon).unwrap();
411        explicit["permissions"] = serde_json::json!("reader");
412        let decoded: Daemon = serde_json::from_value(explicit.clone()).unwrap();
413        assert_eq!(decoded.get_permissions(), Some("reader"));
414        assert_eq!(
415            ResourceDefinition::get_permissions(&decoded),
416            Some("reader")
417        );
418        assert_eq!(serde_json::to_value(decoded).unwrap(), explicit);
419    }
420
421    #[test]
422    fn daemon_serializes_with_resource_type() {
423        let daemon = Daemon::new("endpoint-agent".to_string())
424            .code(DaemonCode::Source {
425                src: "./agent".to_string(),
426                toolchain: ToolchainConfig::Rust {
427                    binary_name: "agent".to_string(),
428                },
429            })
430            .permissions("execution".to_string())
431            .commands_enabled(true)
432            .build();
433
434        let resource = crate::Resource::new(daemon);
435        let json = serde_json::to_value(&resource).expect("daemon should serialize");
436        assert_eq!(json["type"], "daemon");
437
438        let roundtrip: crate::Resource =
439            serde_json::from_value(json).expect("daemon should deserialize");
440        assert_eq!(roundtrip.resource_type().as_ref(), "daemon");
441    }
442
443    #[test]
444    fn daemon_accepts_one_public_http_endpoint() {
445        let daemon = Daemon::new("gateway".to_string())
446            .code(DaemonCode::Image {
447                image: "gateway:latest".to_string(),
448            })
449            .public_endpoint(PublicEndpoint {
450                name: "public".to_string(),
451                port: 8080,
452                protocol: ExposeProtocol::Http,
453                host_label: Some("public".to_string()),
454                wildcard_subdomains: true,
455            })
456            .permissions("gateway".to_string())
457            .build();
458
459        assert!(daemon.validate_public_endpoints().is_ok());
460        assert_eq!(daemon.public_endpoints.len(), 1);
461        assert_eq!(
462            daemon.public_endpoints[0].host_label.as_deref(),
463            Some("public")
464        );
465        assert!(daemon.public_endpoints[0].wildcard_subdomains);
466    }
467
468    #[test]
469    fn daemon_serializes_stop_grace_period_when_set() {
470        let daemon = Daemon::new("gateway".to_string())
471            .code(DaemonCode::Image {
472                image: "gateway:latest".to_string(),
473            })
474            .permissions("gateway".to_string())
475            .stop_grace_period_seconds(21_600)
476            .build();
477
478        let json = serde_json::to_value(&daemon).expect("daemon should serialize");
479        assert_eq!(json["stopGracePeriodSeconds"], 21_600);
480    }
481
482    #[test]
483    fn daemon_omits_stop_grace_period_when_absent() {
484        let daemon = Daemon::new("gateway".to_string())
485            .code(DaemonCode::Image {
486                image: "gateway:latest".to_string(),
487            })
488            .permissions("gateway".to_string())
489            .build();
490
491        let json = serde_json::to_value(&daemon).expect("daemon should serialize");
492        assert!(json.get("stopGracePeriodSeconds").is_none());
493    }
494
495    #[test]
496    fn daemon_rejects_multiple_backend_ports_or_non_http_public_endpoints() {
497        let multiple = Daemon::new("gateway".to_string())
498            .code(DaemonCode::Image {
499                image: "gateway:latest".to_string(),
500            })
501            .public_endpoint(PublicEndpoint {
502                name: "api".to_string(),
503                port: 8080,
504                protocol: ExposeProtocol::Http,
505                host_label: None,
506                wildcard_subdomains: false,
507            })
508            .public_endpoint(PublicEndpoint {
509                name: "admin".to_string(),
510                port: 9090,
511                protocol: ExposeProtocol::Http,
512                host_label: None,
513                wildcard_subdomains: false,
514            })
515            .permissions("gateway".to_string())
516            .build();
517        assert!(multiple.validate_public_endpoints().is_err());
518
519        let tcp = Daemon::new("gateway".to_string())
520            .code(DaemonCode::Image {
521                image: "gateway:latest".to_string(),
522            })
523            .public_endpoint(PublicEndpoint {
524                name: "api".to_string(),
525                port: 8080,
526                protocol: ExposeProtocol::Tcp,
527                host_label: None,
528                wildcard_subdomains: false,
529            })
530            .permissions("gateway".to_string())
531            .build();
532        assert!(tcp.validate_public_endpoints().is_err());
533    }
534
535    #[test]
536    fn daemon_rejects_multiple_apex_public_endpoints() {
537        let daemon = Daemon::new("gateway".to_string())
538            .code(DaemonCode::Image {
539                image: "gateway:latest".to_string(),
540            })
541            .public_endpoint(PublicEndpoint {
542                name: "api".to_string(),
543                port: 8080,
544                protocol: ExposeProtocol::Http,
545                host_label: Some(APEX_HOST_LABEL.to_string()),
546                wildcard_subdomains: false,
547            })
548            .public_endpoint(PublicEndpoint {
549                name: "admin".to_string(),
550                port: 8080,
551                protocol: ExposeProtocol::Http,
552                host_label: Some(APEX_HOST_LABEL.to_string()),
553                wildcard_subdomains: false,
554            })
555            .permissions("gateway".to_string())
556            .build();
557
558        assert!(daemon.validate_public_endpoints().is_err());
559    }
560}
alien_core/resources/daemon.rs

alien_core/resources/
daemon.rs
