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
241/// CPU a daemon reserves when it does not say.
242pub fn default_daemon_cpu() -> ResourceSpec {
243    ResourceSpec {
244        min: "0.1".to_string(),
245        desired: "0.1".to_string(),
246    }
247}
248
249/// Memory a daemon reserves when it does not say.
250pub fn default_daemon_memory() -> ResourceSpec {
251    ResourceSpec {
252        min: "128Mi".to_string(),
253        desired: "128Mi".to_string(),
254    }
255}
256
257impl<S: daemon_builder::State> DaemonBuilder<S> {
258    pub fn link<R: ?Sized>(mut self, resource: &R) -> Self
259    where
260        for<'a> &'a R: Into<ResourceRef>,
261    {
262        let resource_ref: ResourceRef = resource.into();
263        self.links.push(resource_ref);
264        self
265    }
266
267    pub fn public_endpoint(mut self, endpoint: PublicEndpoint) -> Self {
268        self.public_endpoints.push(endpoint);
269        self
270    }
271}
272
273impl ResourceDefinition for Daemon {
274    fn get_resource_type(&self) -> ResourceType {
275        Self::RESOURCE_TYPE
276    }
277
278    fn id(&self) -> &str {
279        &self.id
280    }
281
282    fn replace_after_failed_create_is_safe(&self) -> bool {
283        // A create that timed out may already run replicas; deleting it stops them all, while
284        // creating it again updates the daemon in place.
285        false
286    }
287
288    fn get_dependencies(&self) -> Vec<ResourceRef> {
289        let mut dependencies = self.links.clone();
290        if let Some(cluster) = &self.cluster {
291            dependencies.push(ResourceRef::new(
292                ComputeCluster::RESOURCE_TYPE,
293                cluster.clone(),
294            ));
295        }
296        dependencies
297    }
298
299    fn get_permissions(&self) -> Option<&str> {
300        self.permissions.as_deref()
301    }
302
303    fn validate_update(&self, new_config: &dyn ResourceDefinition) -> Result<()> {
304        let new_daemon = new_config
305            .as_any()
306            .downcast_ref::<Daemon>()
307            .ok_or_else(|| {
308                AlienError::new(ErrorData::UnexpectedResourceType {
309                    resource_id: self.id.clone(),
310                    expected: Self::RESOURCE_TYPE,
311                    actual: new_config.get_resource_type(),
312                })
313            })?;
314
315        if self.id != new_daemon.id {
316            return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
317                resource_id: self.id.clone(),
318                reason: "the 'id' field is immutable".to_string(),
319            }));
320        }
321
322        self.validate_public_endpoints()?;
323        new_daemon.validate_public_endpoints()?;
324        self.validate_runtime()?;
325        new_daemon.validate_runtime()?;
326
327        if self.public_endpoints != new_daemon.public_endpoints {
328            return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
329                resource_id: self.id.clone(),
330                reason: "the 'publicEndpoints' field is immutable".to_string(),
331            }));
332        }
333
334        Ok(())
335    }
336
337    fn as_any(&self) -> &dyn Any {
338        self
339    }
340
341    fn as_any_mut(&mut self) -> &mut dyn Any {
342        self
343    }
344
345    fn box_clone(&self) -> Box<dyn ResourceDefinition> {
346        Box::new(self.clone())
347    }
348
349    fn resource_eq(&self, other: &dyn ResourceDefinition) -> bool {
350        other.as_any().downcast_ref::<Daemon>() == Some(self)
351    }
352
353    fn to_json_value(&self) -> serde_json::Result<serde_json::Value> {
354        serde_json::to_value(self)
355    }
356}
357
358#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
359#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
360#[serde(rename_all = "camelCase")]
361pub struct DaemonOutputs {
362    pub daemon_name: String,
363    pub running: bool,
364    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
365    pub public_endpoints: HashMap<String, PublicEndpointOutput>,
366}
367
368impl ResourceOutputsDefinition for DaemonOutputs {
369    fn get_resource_type(&self) -> ResourceType {
370        Daemon::RESOURCE_TYPE.clone()
371    }
372
373    fn as_any(&self) -> &dyn Any {
374        self
375    }
376
377    fn box_clone(&self) -> Box<dyn ResourceOutputsDefinition> {
378        Box::new(self.clone())
379    }
380
381    fn outputs_eq(&self, other: &dyn ResourceOutputsDefinition) -> bool {
382        other.as_any().downcast_ref::<DaemonOutputs>() == Some(self)
383    }
384
385    fn to_json_value(&self) -> serde_json::Result<serde_json::Value> {
386        serde_json::to_value(self)
387    }
388}
389
390#[cfg(test)]
391mod tests {
392    use super::*;
393
394    #[test]
395    fn daemon_without_permissions_preserves_links_without_an_identity() {
396        let storage = crate::Storage::new("objects".to_string()).build();
397        let daemon = Daemon::new("observer".to_string())
398            .code(DaemonCode::Image {
399                image: "observer:latest".to_string(),
400            })
401            .link(&storage)
402            .build();
403        assert_eq!(daemon.get_permissions(), None);
404        assert_eq!(ResourceDefinition::get_permissions(&daemon), None);
405        assert_eq!(daemon.get_dependencies(), daemon.links);
406        assert_eq!(daemon.links[0].id(), "objects");
407        let json = serde_json::to_value(&daemon).unwrap();
408        assert!(json.get("permissions").is_none());
409        let decoded: Daemon = serde_json::from_value(json).unwrap();
410        assert_eq!(decoded, daemon);
411
412        let mut explicit = serde_json::to_value(&daemon).unwrap();
413        explicit["permissions"] = serde_json::json!("reader");
414        let decoded: Daemon = serde_json::from_value(explicit.clone()).unwrap();
415        assert_eq!(decoded.get_permissions(), Some("reader"));
416        assert_eq!(
417            ResourceDefinition::get_permissions(&decoded),
418            Some("reader")
419        );
420        assert_eq!(serde_json::to_value(decoded).unwrap(), explicit);
421    }
422
423    #[test]
424    fn daemon_serializes_with_resource_type() {
425        let daemon = Daemon::new("endpoint-agent".to_string())
426            .code(DaemonCode::Source {
427                src: "./agent".to_string(),
428                toolchain: ToolchainConfig::Rust {
429                    binary_name: "agent".to_string(),
430                },
431            })
432            .permissions("execution".to_string())
433            .commands_enabled(true)
434            .build();
435
436        let resource = crate::Resource::new(daemon);
437        let json = serde_json::to_value(&resource).expect("daemon should serialize");
438        assert_eq!(json["type"], "daemon");
439
440        let roundtrip: crate::Resource =
441            serde_json::from_value(json).expect("daemon should deserialize");
442        assert_eq!(roundtrip.resource_type().as_ref(), "daemon");
443    }
444
445    #[test]
446    fn daemon_accepts_one_public_http_endpoint() {
447        let daemon = Daemon::new("gateway".to_string())
448            .code(DaemonCode::Image {
449                image: "gateway:latest".to_string(),
450            })
451            .public_endpoint(PublicEndpoint {
452                name: "public".to_string(),
453                port: 8080,
454                protocol: ExposeProtocol::Http,
455                host_label: Some("public".to_string()),
456                wildcard_subdomains: true,
457            })
458            .permissions("gateway".to_string())
459            .build();
460
461        assert!(daemon.validate_public_endpoints().is_ok());
462        assert_eq!(daemon.public_endpoints.len(), 1);
463        assert_eq!(
464            daemon.public_endpoints[0].host_label.as_deref(),
465            Some("public")
466        );
467        assert!(daemon.public_endpoints[0].wildcard_subdomains);
468    }
469
470    #[test]
471    fn daemon_serializes_stop_grace_period_when_set() {
472        let daemon = Daemon::new("gateway".to_string())
473            .code(DaemonCode::Image {
474                image: "gateway:latest".to_string(),
475            })
476            .permissions("gateway".to_string())
477            .stop_grace_period_seconds(21_600)
478            .build();
479
480        let json = serde_json::to_value(&daemon).expect("daemon should serialize");
481        assert_eq!(json["stopGracePeriodSeconds"], 21_600);
482    }
483
484    #[test]
485    fn daemon_omits_stop_grace_period_when_absent() {
486        let daemon = Daemon::new("gateway".to_string())
487            .code(DaemonCode::Image {
488                image: "gateway:latest".to_string(),
489            })
490            .permissions("gateway".to_string())
491            .build();
492
493        let json = serde_json::to_value(&daemon).expect("daemon should serialize");
494        assert!(json.get("stopGracePeriodSeconds").is_none());
495    }
496
497    #[test]
498    fn daemon_rejects_multiple_backend_ports_or_non_http_public_endpoints() {
499        let multiple = Daemon::new("gateway".to_string())
500            .code(DaemonCode::Image {
501                image: "gateway:latest".to_string(),
502            })
503            .public_endpoint(PublicEndpoint {
504                name: "api".to_string(),
505                port: 8080,
506                protocol: ExposeProtocol::Http,
507                host_label: None,
508                wildcard_subdomains: false,
509            })
510            .public_endpoint(PublicEndpoint {
511                name: "admin".to_string(),
512                port: 9090,
513                protocol: ExposeProtocol::Http,
514                host_label: None,
515                wildcard_subdomains: false,
516            })
517            .permissions("gateway".to_string())
518            .build();
519        assert!(multiple.validate_public_endpoints().is_err());
520
521        let tcp = Daemon::new("gateway".to_string())
522            .code(DaemonCode::Image {
523                image: "gateway:latest".to_string(),
524            })
525            .public_endpoint(PublicEndpoint {
526                name: "api".to_string(),
527                port: 8080,
528                protocol: ExposeProtocol::Tcp,
529                host_label: None,
530                wildcard_subdomains: false,
531            })
532            .permissions("gateway".to_string())
533            .build();
534        assert!(tcp.validate_public_endpoints().is_err());
535    }
536
537    #[test]
538    fn daemon_rejects_multiple_apex_public_endpoints() {
539        let daemon = Daemon::new("gateway".to_string())
540            .code(DaemonCode::Image {
541                image: "gateway:latest".to_string(),
542            })
543            .public_endpoint(PublicEndpoint {
544                name: "api".to_string(),
545                port: 8080,
546                protocol: ExposeProtocol::Http,
547                host_label: Some(APEX_HOST_LABEL.to_string()),
548                wildcard_subdomains: false,
549            })
550            .public_endpoint(PublicEndpoint {
551                name: "admin".to_string(),
552                port: 8080,
553                protocol: ExposeProtocol::Http,
554                host_label: Some(APEX_HOST_LABEL.to_string()),
555                wildcard_subdomains: false,
556            })
557            .permissions("gateway".to_string())
558            .build();
559
560        assert!(daemon.validate_public_endpoints().is_err());
561    }
562}
alien_core/resources/daemon.rs

alien_core/resources/
daemon.rs
