hydro_lang/deploy/
deploy_graph_containerized.rs

1//! Deployment backend for Hydro that uses Docker to provision and launch services.
2
3use std::cell::RefCell;
4use std::collections::HashMap;
5use std::pin::Pin;
6use std::rc::Rc;
7
8use bollard::Docker;
9use bollard::models::{ContainerCreateBody, EndpointSettings, HostConfig, NetworkCreateRequest};
10use bollard::query_parameters::{
11    BuildImageOptions, CreateContainerOptions, InspectContainerOptions, KillContainerOptions,
12    RemoveContainerOptions, StartContainerOptions,
13};
14use bollard::secret::NetworkingConfig;
15use bytes::Bytes;
16use dfir_lang::graph::{AsCodeOptions, DfirGraph};
17use futures::{Sink, SinkExt, Stream, StreamExt};
18use http_body_util::Full;
19// Re-export LinuxCompileType so users can configure compile type without depending on hydro_deploy directly.
20pub use hydro_deploy::LinuxCompileType;
21use hydro_deploy::RustCrate;
22use hydro_deploy::rust_crate::build::{BuildError, build_crate_memoized};
23use nanoid::nanoid;
24use proc_macro2::Span;
25use sinktools::lazy::LazySink;
26use stageleft::QuotedWithContext;
27use syn::parse_quote;
28use tar::{Builder, Header};
29use tokio::net::TcpStream;
30use tokio_util::codec::{FramedRead, FramedWrite, LengthDelimitedCodec};
31use tracing::{Instrument, instrument, trace, warn};
32
33use super::deploy_runtime_containerized::*;
34use crate::compile::builder::ExternalPortId;
35use crate::compile::deploy::DeployResult;
36use crate::compile::deploy_provider::{
37    ClusterSpec, Deploy, ExternalSpec, Node, ProcessSpec, RegisterPort,
38};
39use crate::compile::trybuild::generate::{LinkingMode, create_graph_trybuild};
40use crate::location::dynamic::LocationId;
41use crate::location::member_id::TaglessMemberId;
42use crate::location::{LocationKey, MembershipEvent, NetworkHint};
43
44/// represents a docker network
45#[derive(Clone, Debug)]
46pub struct DockerNetwork {
47    name: String,
48}
49
50impl DockerNetwork {
51    /// creates a new docker network (will actually be created when deployment.start() is called).
52    pub fn new(name: String) -> Self {
53        Self {
54            name: format!("{name}-{}", nanoid::nanoid!(6, &CONTAINER_ALPHABET)),
55        }
56    }
57}
58
59/// Represents a process running in a docker container
60#[derive(Clone)]
61pub struct DockerDeployProcess {
62    key: LocationKey,
63    name: String,
64    next_port: Rc<RefCell<u16>>,
65    rust_crate: Rc<RefCell<Option<RustCrate>>>,
66
67    exposed_ports: Rc<RefCell<Vec<u16>>>,
68
69    docker_container_name: Rc<RefCell<Option<String>>>,
70
71    rustflags: Option<String>,
72
73    config: Vec<String>,
74
75    network: DockerNetwork,
76
77    base_image: Option<String>,
78
79    linux_compile_type: LinuxCompileType,
80
81    features: Vec<String>,
82}
83
84impl Node for DockerDeployProcess {
85    type Port = u16;
86    type Meta = ();
87    type InstantiateEnv = DockerDeploy;
88
89    #[instrument(level = "trace", skip_all, ret, fields(key = %self.key, name = self.name))]
90    fn next_port(&self) -> Self::Port {
91        let port = {
92            let mut borrow = self.next_port.borrow_mut();
93            let port = *borrow;
94            *borrow += 1;
95            port
96        };
97
98        port
99    }
100
101    #[instrument(level = "trace", skip_all, fields(key = %self.key, name = self.name))]
102    fn update_meta(&self, _meta: &Self::Meta) {}
103
104    #[instrument(level = "trace", skip_all, fields(key = %self.key, name = self.name, ?meta, extra_stmts = extra_stmts.len(), sidecars = sidecars.len()))]
105    fn instantiate(
106        &self,
107        _env: &mut Self::InstantiateEnv,
108        meta: &mut Self::Meta,
109        graph: DfirGraph,
110        extra_stmts: &[syn::Stmt],
111        sidecars: &[syn::Expr],
112        as_code_options: &AsCodeOptions,
113    ) {
114        let (bin_name, config) = create_graph_trybuild(
115            graph,
116            extra_stmts,
117            sidecars,
118            as_code_options,
119            Some(&self.name),
120            crate::compile::trybuild::generate::DeployMode::Containerized,
121            LinkingMode::Static,
122        );
123
124        let mut ret = RustCrate::new(&config.project_dir, &config.project_dir)
125            .target_dir(config.target_dir)
126            .example(bin_name)
127            .no_default_features();
128
129        ret = ret.display_name("test_display_name");
130
131        ret = ret.features(vec!["hydro___feature_docker_runtime".to_owned()]);
132
133        if let Some(features) = config.features {
134            ret = ret.features(features);
135        }
136
137        if !self.features.is_empty() {
138            ret = ret.features(self.features.clone());
139        }
140
141        ret = ret.build_env("STAGELEFT_TRYBUILD_BUILD_STAGED", "1");
142        ret = ret.config("build.incremental = false");
143
144        *self.rust_crate.borrow_mut() = Some(ret);
145    }
146}
147
148/// Represents a logical cluster, which can be a variable amount of individual containers.
149#[derive(Clone)]
150pub struct DockerDeployCluster {
151    key: LocationKey,
152    name: String,
153    next_port: Rc<RefCell<u16>>,
154    rust_crate: Rc<RefCell<Option<RustCrate>>>,
155
156    exposed_ports: Rc<RefCell<Vec<u16>>>,
157
158    docker_container_name: Rc<RefCell<Vec<String>>>,
159
160    rustflags: Option<String>,
161
162    config: Vec<String>,
163
164    count: usize,
165
166    base_image: Option<String>,
167
168    linux_compile_type: LinuxCompileType,
169
170    features: Vec<String>,
171}
172
173impl Node for DockerDeployCluster {
174    type Port = u16;
175    type Meta = ();
176    type InstantiateEnv = DockerDeploy;
177
178    #[instrument(level = "trace", skip_all, ret, fields(key = %self.key, name = self.name))]
179    fn next_port(&self) -> Self::Port {
180        let port = {
181            let mut borrow = self.next_port.borrow_mut();
182            let port = *borrow;
183            *borrow += 1;
184            port
185        };
186
187        port
188    }
189
190    #[instrument(level = "trace", skip_all, fields(key = %self.key, name = self.name))]
191    fn update_meta(&self, _meta: &Self::Meta) {}
192
193    #[instrument(level = "trace", skip_all, fields(key = %self.key, name = self.name, extra_stmts = extra_stmts.len()))]
194    fn instantiate(
195        &self,
196        _env: &mut Self::InstantiateEnv,
197        _meta: &mut Self::Meta,
198        graph: DfirGraph,
199        extra_stmts: &[syn::Stmt],
200        sidecars: &[syn::Expr],
201        as_code_options: &AsCodeOptions,
202    ) {
203        let (bin_name, config) = create_graph_trybuild(
204            graph,
205            extra_stmts,
206            sidecars,
207            as_code_options,
208            Some(&self.name),
209            crate::compile::trybuild::generate::DeployMode::Containerized,
210            LinkingMode::Static,
211        );
212
213        let mut ret = RustCrate::new(&config.project_dir, &config.project_dir)
214            .target_dir(config.target_dir)
215            .example(bin_name)
216            .no_default_features();
217
218        ret = ret.display_name("test_display_name");
219
220        ret = ret.features(vec!["hydro___feature_docker_runtime".to_owned()]);
221
222        if let Some(features) = config.features {
223            ret = ret.features(features);
224        }
225
226        if !self.features.is_empty() {
227            ret = ret.features(self.features.clone());
228        }
229
230        ret = ret.build_env("STAGELEFT_TRYBUILD_BUILD_STAGED", "1");
231        ret = ret.config("build.incremental = false");
232
233        *self.rust_crate.borrow_mut() = Some(ret);
234    }
235}
236
237/// Represents an external process, outside the control of this deployment but still with some communication into this deployment.
238#[derive(Clone, Debug)]
239#[expect(
240    dead_code,
241    reason = "fields used via Rc<RefCell> in RegisterPort impl and ExternalBytesPort construction"
242)]
243pub struct DockerDeployExternal {
244    /// The location key for this external, used for port handle construction.
245    pub(crate) key: LocationKey,
246    name: String,
247    next_port: Rc<RefCell<u16>>,
248
249    /// Counter for generating ExternalPortId values at deploy time.
250    next_external_port_id: Rc<RefCell<crate::Counter<ExternalPortId>>>,
251
252    ports: Rc<RefCell<HashMap<ExternalPortId, u16>>>,
253
254    connection_info: Rc<RefCell<HashMap<u16, (Rc<RefCell<Option<String>>>, u16, DockerNetwork)>>>,
255}
256
257impl Node for DockerDeployExternal {
258    type Port = u16;
259    type Meta = ();
260    type InstantiateEnv = DockerDeploy;
261
262    #[instrument(level = "trace", skip_all, ret, fields(name = self.name))]
263    fn next_port(&self) -> Self::Port {
264        let port = {
265            let mut borrow = self.next_port.borrow_mut();
266            let port = *borrow;
267            *borrow += 1;
268            port
269        };
270
271        port
272    }
273
274    #[instrument(level = "trace", skip_all, fields(name = self.name))]
275    fn update_meta(&self, _meta: &Self::Meta) {}
276
277    #[instrument(level = "trace", skip_all, fields(name = self.name, ?meta, extra_stmts = extra_stmts.len(), sidecars = sidecars.len()))]
278    fn instantiate(
279        &self,
280        _env: &mut Self::InstantiateEnv,
281        meta: &mut Self::Meta,
282        graph: DfirGraph,
283        extra_stmts: &[syn::Stmt],
284        sidecars: &[syn::Expr],
285        _as_code_options: &AsCodeOptions,
286    ) {
287        trace!(name: "surface", surface = graph.surface_syntax_string());
288    }
289}
290
291impl DockerDeployProcess {
292    /// Expose a TCP port on this process for external access.
293    ///
294    /// The binary running on this process must bind a `TcpListener` on this port.
295    /// This method ensures the port appears in the Docker image's `EXPOSE` directives
296    /// and is available for endpoint discovery via [`Self::get_tcp_endpoint`].
297    pub fn expose_port(&self, port: u16) {
298        self.exposed_ports.borrow_mut().push(port);
299    }
300
301    /// Returns the TCP endpoint `(host, port)` for this process exposing
302    /// the given container port. Queries Docker for the dynamically allocated
303    /// host port mapping.
304    pub async fn get_tcp_endpoint(&self, container_port: u16) -> (String, u16) {
305        let name = self
306            .docker_container_name
307            .borrow()
308            .as_ref()
309            .expect("container not yet started")
310            .clone();
311        let host_port = find_dynamically_allocated_docker_port(&name, container_port).await;
312        ("localhost".to_owned(), host_port)
313    }
314}
315
316impl DockerDeployCluster {
317    /// Expose a TCP port on every member of this cluster for external access.
318    ///
319    /// The binary running on this cluster must bind a `TcpListener` on this port.
320    /// This method ensures the port appears in the Docker image's `EXPOSE` directives
321    /// and is available for endpoint discovery via [`Self::get_all_tcp_endpoints`].
322    pub fn expose_port(&self, port: u16) {
323        self.exposed_ports.borrow_mut().push(port);
324    }
325
326    /// Returns TCP endpoints `(host, port)` for all cluster members exposing
327    /// the given container port. Queries Docker for the dynamically allocated
328    /// host port mapping.
329    pub async fn get_all_tcp_endpoints(&self, container_port: u16) -> Vec<(String, u16)> {
330        let names = self.docker_container_name.borrow().clone();
331        let mut endpoints = Vec::with_capacity(names.len());
332        for name in names {
333            let host_port = find_dynamically_allocated_docker_port(&name, container_port).await;
334            endpoints.push(("localhost".to_owned(), host_port));
335        }
336        endpoints
337    }
338}
339
340type DynSourceSink<Out, In, InErr> = (
341    Pin<Box<dyn Stream<Item = Out>>>,
342    Pin<Box<dyn Sink<In, Error = InErr>>>,
343);
344
345impl<'a> RegisterPort<'a, DockerDeploy> for DockerDeployExternal {
346    #[instrument(level = "trace", skip_all, fields(name = self.name, %external_port_id, %port))]
347    fn register(&self, external_port_id: ExternalPortId, port: Self::Port) {
348        self.ports.borrow_mut().insert(external_port_id, port);
349    }
350
351    fn as_bytes_bidi(
352        &self,
353        external_port_id: ExternalPortId,
354    ) -> impl Future<
355        Output = DynSourceSink<Result<bytes::BytesMut, std::io::Error>, Bytes, std::io::Error>,
356    > + 'a {
357        let guard =
358            tracing::trace_span!("as_bytes_bidi", name = %self.name, %external_port_id).entered();
359
360        let local_port = *self.ports.borrow().get(&external_port_id).unwrap();
361        let (docker_container_name, remote_port, _) = self
362            .connection_info
363            .borrow()
364            .get(&local_port)
365            .unwrap()
366            .clone();
367
368        let docker_container_name = docker_container_name.borrow().as_ref().unwrap().clone();
369
370        async move {
371            let local_port =
372                find_dynamically_allocated_docker_port(&docker_container_name, remote_port).await;
373            let remote_ip_address = "localhost";
374
375            trace!(name: "as_bytes_bidi_connecting", to = %remote_ip_address, to_port = %local_port);
376
377            let stream = TcpStream::connect(format!("{remote_ip_address}:{local_port}"))
378                .await
379                .unwrap();
380
381            trace!(name: "as_bytes_bidi_connected", to = %remote_ip_address, to_port = %local_port);
382
383            let (rx, tx) = stream.into_split();
384
385            let source = Box::pin(
386                FramedRead::new(rx, LengthDelimitedCodec::new()),
387            ) as Pin<Box<dyn Stream<Item = Result<bytes::BytesMut, std::io::Error>>>>;
388
389            let sink = Box::pin(FramedWrite::new(tx, LengthDelimitedCodec::new()))
390                as Pin<Box<dyn Sink<Bytes, Error = std::io::Error>>>;
391
392            (source, sink)
393        }
394        .instrument(guard.exit())
395    }
396
397    fn as_bincode_bidi<InT, OutT>(
398        &self,
399        external_port_id: ExternalPortId,
400    ) -> impl Future<Output = DynSourceSink<OutT, InT, std::io::Error>> + 'a
401    where
402        InT: serde::Serialize + 'static,
403        OutT: serde::de::DeserializeOwned + 'static,
404    {
405        let guard =
406            tracing::trace_span!("as_bincode_bidi", name = %self.name, %external_port_id).entered();
407
408        let local_port = *self.ports.borrow().get(&external_port_id).unwrap();
409        let (docker_container_name, remote_port, _) = self
410            .connection_info
411            .borrow()
412            .get(&local_port)
413            .unwrap()
414            .clone();
415
416        let docker_container_name = docker_container_name.borrow().as_ref().unwrap().clone();
417
418        async move {
419            let local_port =
420                find_dynamically_allocated_docker_port(&docker_container_name, remote_port).await;
421            let remote_ip_address = "localhost";
422
423            trace!(name: "as_bincode_bidi_connecting", to = %remote_ip_address, to_port = %local_port);
424
425            let stream = TcpStream::connect(format!("{remote_ip_address}:{local_port}"))
426                .await
427                .unwrap();
428
429            trace!(name: "as_bincode_bidi_connected", to = %remote_ip_address, to_port = %local_port);
430
431            let (rx, tx) = stream.into_split();
432
433            let source = Box::pin(
434                FramedRead::new(rx, LengthDelimitedCodec::new())
435                    .map(|v| bincode::deserialize(&v.unwrap()).unwrap()),
436            ) as Pin<Box<dyn Stream<Item = OutT>>>;
437
438            let sink = Box::pin(
439                FramedWrite::new(tx, LengthDelimitedCodec::new()).with(move |v: InT| async move {
440                    Ok::<_, std::io::Error>(Bytes::from(bincode::serialize(&v).unwrap()))
441                }),
442            ) as Pin<Box<dyn Sink<InT, Error = std::io::Error>>>;
443
444            (source, sink)
445        }
446        .instrument(guard.exit())
447    }
448
449    fn as_bincode_sink<T>(
450        &self,
451        external_port_id: ExternalPortId,
452    ) -> impl Future<Output = Pin<Box<dyn Sink<T, Error = std::io::Error>>>> + 'a
453    where
454        T: serde::Serialize + 'static,
455    {
456        let guard =
457            tracing::trace_span!("as_bincode_sink", name = %self.name, %external_port_id).entered();
458
459        let local_port = *self.ports.borrow().get(&external_port_id).unwrap();
460        let (docker_container_name, remote_port, _) = self
461            .connection_info
462            .borrow()
463            .get(&local_port)
464            .unwrap()
465            .clone();
466
467        let docker_container_name = docker_container_name.borrow().as_ref().unwrap().clone();
468
469        async move {
470            let local_port = find_dynamically_allocated_docker_port(&docker_container_name, remote_port).await;
471            let remote_ip_address = "localhost";
472
473            Box::pin(
474                LazySink::new(move || {
475                    Box::pin(async move {
476                        trace!(name: "as_bincode_sink_connecting", to = %remote_ip_address, to_port = %local_port);
477
478                        let stream =
479                            TcpStream::connect(format!("{remote_ip_address}:{local_port}"))
480                                .await?;
481
482                        trace!(name: "as_bincode_sink_connected", to = %remote_ip_address, to_port = %local_port);
483
484                        Result::<_, std::io::Error>::Ok(FramedWrite::new(
485                            stream,
486                            LengthDelimitedCodec::new(),
487                        ))
488                    })
489                })
490                .with(move |v| async move {
491                    Ok(Bytes::from(bincode::serialize(&v).unwrap()))
492                }),
493            ) as Pin<Box<dyn Sink<T, Error = std::io::Error>>>
494        }
495        .instrument(guard.exit())
496    }
497
498    fn as_bincode_source<T>(
499        &self,
500        external_port_id: ExternalPortId,
501    ) -> impl Future<Output = Pin<Box<dyn Stream<Item = T>>>> + 'a
502    where
503        T: serde::de::DeserializeOwned + 'static,
504    {
505        let guard =
506            tracing::trace_span!("as_bincode_sink", name = %self.name, %external_port_id).entered();
507
508        let local_port = *self.ports.borrow().get(&external_port_id).unwrap();
509        let (docker_container_name, remote_port, _) = self
510            .connection_info
511            .borrow()
512            .get(&local_port)
513            .unwrap()
514            .clone();
515
516        let docker_container_name = docker_container_name.borrow().as_ref().unwrap().clone();
517
518        async move {
519
520            let local_port = find_dynamically_allocated_docker_port(&docker_container_name, remote_port).await;
521            let remote_ip_address = "localhost";
522
523            trace!(name: "as_bincode_source_connecting", to = %remote_ip_address, to_port = %local_port);
524
525            let stream = TcpStream::connect(format!("{remote_ip_address}:{local_port}"))
526                .await
527                .unwrap();
528
529            trace!(name: "as_bincode_source_connected", to = %remote_ip_address, to_port = %local_port);
530
531            Box::pin(
532                FramedRead::new(stream, LengthDelimitedCodec::new())
533                    .map(|v| bincode::deserialize(&v.unwrap()).unwrap()),
534            ) as Pin<Box<dyn Stream<Item = T>>>
535        }
536        .instrument(guard.exit())
537    }
538}
539
540#[instrument(level = "trace", skip_all, fields(%docker_container_name, %destination_port))]
541async fn find_dynamically_allocated_docker_port(
542    docker_container_name: &str,
543    destination_port: u16,
544) -> u16 {
545    let docker = Docker::connect_with_local_defaults().unwrap();
546
547    let container_info = docker
548        .inspect_container(docker_container_name, None::<InspectContainerOptions>)
549        .await
550        .unwrap();
551
552    trace!(name: "port struct", container_info = ?container_info.network_settings.as_ref().unwrap().ports.as_ref().unwrap());
553
554    // container_info={"1001/tcp": Some([PortBinding { host_ip: Some("0.0.0.0"), host_port: Some("32771") }, PortBinding { host_ip: Some("::"), host_port: Some("32771") }])} destination_port=1001
555    let remote_port = container_info
556        .network_settings
557        .as_ref()
558        .unwrap()
559        .ports
560        .as_ref()
561        .unwrap()
562        .get(&format!("{destination_port}/tcp"))
563        .unwrap()
564        .as_ref()
565        .unwrap()
566        .iter()
567        .find(|v| v.host_ip == Some("0.0.0.0".to_owned()))
568        .unwrap()
569        .host_port
570        .as_ref()
571        .unwrap()
572        .parse()
573        .unwrap();
574
575    remote_port
576}
577
578/// For deploying to a local docker instance
579pub struct DockerDeploy {
580    docker_processes: Vec<DockerDeployProcessSpec>,
581    docker_clusters: Vec<DockerDeployClusterSpec>,
582    network: DockerNetwork,
583    deployment_instance: String,
584}
585
586#[instrument(level = "trace", skip_all, fields(%image_name, %container_name, %network_name, %deployment_instance))]
587async fn create_and_start_container(
588    docker: &Docker,
589    container_name: &str,
590    image_name: &str,
591    network_name: &str,
592    deployment_instance: &str,
593) -> Result<(), anyhow::Error> {
594    let config = ContainerCreateBody {
595        image: Some(image_name.to_owned()),
596        hostname: Some(container_name.to_owned()),
597        host_config: Some(HostConfig {
598            binds: Some(vec!["/var/run/docker.sock:/var/run/docker.sock".to_owned()]),
599            publish_all_ports: Some(true),
600            port_bindings: Some(HashMap::new()), /* Due to a bug in docker, if you don't send empty port bindings with publish_all_ports set to true and with a docker image that has EXPOSE directives in it, docker will crash because it will try to write to a map in memory that it has not initialized yet. Setting port_bindings explicitly to an empty map will initialize it first so that it does not break. */
601            ..Default::default()
602        }),
603        env: Some(vec![
604            format!("CONTAINER_NAME={container_name}"),
605            format!("DEPLOYMENT_INSTANCE={deployment_instance}"),
606            "RUST_LOG=trace".to_owned(),
607        ]),
608        networking_config: Some(NetworkingConfig {
609            endpoints_config: Some(HashMap::from([(
610                network_name.to_owned(),
611                EndpointSettings {
612                    ..Default::default()
613                },
614            )])),
615        }),
616        tty: Some(true),
617        ..Default::default()
618    };
619
620    let options = CreateContainerOptions {
621        name: Some(container_name.to_owned()),
622        ..Default::default()
623    };
624
625    tracing::error!("Config: {}", serde_json::to_string_pretty(&config).unwrap());
626    docker.create_container(Some(options), config).await?;
627    docker
628        .start_container(container_name, None::<StartContainerOptions>)
629        .await?;
630
631    Ok(())
632}
633
634#[instrument(level = "trace", skip_all, fields(%image_name))]
635async fn build_and_create_image(
636    rust_crate: &Rc<RefCell<Option<RustCrate>>>,
637    rustflags: Option<&str>,
638    config: &[String],
639    exposed_ports: &[u16],
640    image_name: &str,
641    base_image: Option<&str>,
642    linux_compile_type: LinuxCompileType,
643) -> Result<(), anyhow::Error> {
644    let mut rust_crate = rust_crate.borrow_mut().take().unwrap();
645
646    if let Some(rustflags) = rustflags {
647        rust_crate = rust_crate.rustflags(rustflags);
648    }
649
650    for cfg in config {
651        rust_crate = rust_crate.config(cfg);
652    }
653
654    let build_output = match build_crate_memoized(
655        rust_crate.get_build_params(hydro_deploy::HostTargetType::Linux(linux_compile_type)),
656    )
657    .await
658    {
659        Ok(build_output) => build_output,
660        Err(BuildError::FailedToBuildCrate {
661            exit_status,
662            diagnostics,
663            text_lines,
664            stderr_lines,
665        }) => {
666            let diagnostics = diagnostics
667                .into_iter()
668                .map(|d| d.rendered.unwrap())
669                .collect::<Vec<_>>()
670                .join("\n");
671            let text_lines = text_lines.join("\n");
672            let stderr_lines = stderr_lines.join("\n");
673
674            anyhow::bail!(
675                r#"
676Failed to build crate {exit_status:?}
677--- diagnostics
678---
679{diagnostics}
680---
681---
682---
683
684--- text_lines
685---
686---
687{text_lines}
688---
689---
690---
691
692--- stderr_lines
693---
694---
695{stderr_lines}
696---
697---
698---"#
699            );
700        }
701        Err(err) => {
702            anyhow::bail!("Failed to build crate {err:?}");
703        }
704    };
705
706    let docker = Docker::connect_with_local_defaults()?;
707
708    let mut tar_data = Vec::new();
709    {
710        let mut tar = Builder::new(&mut tar_data);
711
712        let exposed_ports = exposed_ports
713            .iter()
714            .map(|port| format!("EXPOSE {port}/tcp"))
715            .collect::<Vec<_>>()
716            .join("\n");
717
718        let from_image = base_image.unwrap_or("scratch");
719        let dockerfile_content = format!(
720            r#"
721                FROM {from_image}
722                {exposed_ports}
723                COPY app /app
724                CMD ["/app"]
725            "#,
726        );
727
728        trace!(name: "dockerfile", %dockerfile_content);
729
730        let mut header = Header::new_gnu();
731        header.set_path("Dockerfile")?;
732        header.set_size(dockerfile_content.len() as u64);
733        header.set_cksum();
734        tar.append(&header, dockerfile_content.as_bytes())?;
735
736        let mut header = Header::new_gnu();
737        header.set_path("app")?;
738        header.set_size(build_output.bin_data.len() as u64);
739        header.set_mode(0o755);
740        header.set_cksum();
741        tar.append(&header, &build_output.bin_data[..])?;
742
743        tar.finish()?;
744    }
745
746    let build_options = BuildImageOptions {
747        dockerfile: "Dockerfile".to_owned(),
748        t: Some(image_name.to_owned()),
749        rm: true,
750        ..Default::default()
751    };
752
753    use bollard::errors::Error;
754
755    let body = http_body_util::Either::Left(Full::new(Bytes::from(tar_data)));
756    let mut build_stream = docker.build_image(build_options, None, Some(body));
757    while let Some(msg) = build_stream.next().await {
758        match msg {
759            Ok(_) => {}
760            Err(e) => match e {
761                Error::DockerStreamError { error } => {
762                    return Err(anyhow::anyhow!(
763                        "Docker build failed: DockerStreamError: {{ error: {error} }}"
764                    ));
765                }
766                _ => return Err(anyhow::anyhow!("Docker build failed: {}", e)),
767            },
768        }
769    }
770
771    Ok(())
772}
773
774impl DockerDeploy {
775    /// Create a new deployment
776    pub fn new(network: DockerNetwork) -> Self {
777        Self {
778            docker_processes: Vec::new(),
779            docker_clusters: Vec::new(),
780            network,
781            deployment_instance: nanoid!(6, &CONTAINER_ALPHABET),
782        }
783    }
784
785    /// Add an internal docker service to the deployment.
786    pub fn add_localhost_docker(
787        &mut self,
788        rustflags: Option<String>,
789        config: Vec<String>,
790    ) -> DockerDeployProcessSpec {
791        let process = DockerDeployProcessSpec {
792            rustflags,
793            config,
794            network: self.network.clone(),
795            deployment_instance: self.deployment_instance.clone(),
796            base_image: None,
797            linux_compile_type: LinuxCompileType::Musl,
798            features: vec![],
799        };
800
801        self.docker_processes.push(process.clone());
802
803        process
804    }
805
806    /// Add an internal docker cluster to the deployment.
807    pub fn add_localhost_docker_cluster(
808        &mut self,
809        rustflags: Option<String>,
810        config: Vec<String>,
811        count: usize,
812    ) -> DockerDeployClusterSpec {
813        let cluster = DockerDeployClusterSpec {
814            rustflags,
815            config,
816            count,
817            deployment_instance: self.deployment_instance.clone(),
818            base_image: None,
819            linux_compile_type: LinuxCompileType::Musl,
820            features: vec![],
821        };
822
823        self.docker_clusters.push(cluster.clone());
824
825        cluster
826    }
827
828    /// Add an external process to the deployment.
829    pub fn add_external(&self, name: String) -> DockerDeployExternalSpec {
830        DockerDeployExternalSpec { name }
831    }
832
833    /// Get the deployment instance from this deployment.
834    pub fn get_deployment_instance(&self) -> String {
835        self.deployment_instance.clone()
836    }
837
838    /// Create docker images.
839    #[instrument(level = "trace", skip_all)]
840    pub async fn provision(&self, nodes: &DeployResult<'_, Self>) -> Result<(), anyhow::Error> {
841        for (_, _, process) in nodes.get_all_processes() {
842            let exposed_ports = process.exposed_ports.borrow().clone();
843
844            build_and_create_image(
845                &process.rust_crate,
846                process.rustflags.as_deref(),
847                &process.config,
848                &exposed_ports,
849                &process.name,
850                process.base_image.as_deref(),
851                process.linux_compile_type,
852            )
853            .await?;
854        }
855
856        for (_, _, cluster) in nodes.get_all_clusters() {
857            let exposed_ports = cluster.exposed_ports.borrow().clone();
858            build_and_create_image(
859                &cluster.rust_crate,
860                cluster.rustflags.as_deref(),
861                &cluster.config,
862                &exposed_ports,
863                &cluster.name,
864                cluster.base_image.as_deref(),
865                cluster.linux_compile_type,
866            )
867            .await?;
868        }
869
870        Ok(())
871    }
872
873    /// Start the deployment, tell docker to create containers from the existing provisioned images.
874    #[instrument(level = "trace", skip_all)]
875    pub async fn start(&self, nodes: &DeployResult<'_, Self>) -> Result<(), anyhow::Error> {
876        let docker = Docker::connect_with_local_defaults()?;
877
878        match docker
879            .create_network(NetworkCreateRequest {
880                name: self.network.name.clone(),
881                driver: Some("bridge".to_owned()),
882                ..Default::default()
883            })
884            .await
885        {
886            Ok(v) => v.id,
887            Err(e) => {
888                panic!("Failed to create docker network: {e:?}");
889            }
890        };
891
892        for (_, _, process) in nodes.get_all_processes() {
893            let docker_container_name: String = get_docker_container_name(&process.name, None);
894            *process.docker_container_name.borrow_mut() = Some(docker_container_name.clone());
895
896            create_and_start_container(
897                &docker,
898                &docker_container_name,
899                &process.name,
900                &self.network.name,
901                &self.deployment_instance,
902            )
903            .await?;
904        }
905
906        for (_, _, cluster) in nodes.get_all_clusters() {
907            for num in 0..cluster.count {
908                let docker_container_name = get_docker_container_name(&cluster.name, Some(num));
909                cluster
910                    .docker_container_name
911                    .borrow_mut()
912                    .push(docker_container_name.clone());
913
914                create_and_start_container(
915                    &docker,
916                    &docker_container_name,
917                    &cluster.name,
918                    &self.network.name,
919                    &self.deployment_instance,
920                )
921                .await?;
922            }
923        }
924
925        Ok(())
926    }
927
928    /// Stop the deployment, destroy all containers
929    #[instrument(level = "trace", skip_all)]
930    pub async fn stop(&mut self, nodes: &DeployResult<'_, Self>) -> Result<(), anyhow::Error> {
931        let docker = Docker::connect_with_local_defaults()?;
932
933        for (_, _, process) in nodes.get_all_processes() {
934            let docker_container_name: String = get_docker_container_name(&process.name, None);
935
936            docker
937                .kill_container(&docker_container_name, None::<KillContainerOptions>)
938                .await?;
939        }
940
941        for (_, _, cluster) in nodes.get_all_clusters() {
942            for num in 0..cluster.count {
943                let docker_container_name = get_docker_container_name(&cluster.name, Some(num));
944
945                docker
946                    .kill_container(&docker_container_name, None::<KillContainerOptions>)
947                    .await?;
948            }
949        }
950
951        Ok(())
952    }
953
954    /// remove containers, images, and networks.
955    #[instrument(level = "trace", skip_all)]
956    pub async fn cleanup(&mut self, nodes: &DeployResult<'_, Self>) -> Result<(), anyhow::Error> {
957        let docker = Docker::connect_with_local_defaults()?;
958
959        for (_, _, process) in nodes.get_all_processes() {
960            let docker_container_name: String = get_docker_container_name(&process.name, None);
961
962            docker
963                .remove_container(&docker_container_name, None::<RemoveContainerOptions>)
964                .await?;
965        }
966
967        for (_, _, cluster) in nodes.get_all_clusters() {
968            for num in 0..cluster.count {
969                let docker_container_name = get_docker_container_name(&cluster.name, Some(num));
970
971                docker
972                    .remove_container(&docker_container_name, None::<RemoveContainerOptions>)
973                    .await?;
974            }
975        }
976
977        docker
978            .remove_network(&self.network.name)
979            .await
980            .map_err(|e| anyhow::anyhow!("Failed to remove docker network: {e:?}"))?;
981
982        use bollard::query_parameters::RemoveImageOptions;
983
984        for (_, _, process) in nodes.get_all_processes() {
985            docker
986                .remove_image(&process.name, None::<RemoveImageOptions>, None)
987                .await?;
988        }
989
990        for (_, _, cluster) in nodes.get_all_clusters() {
991            docker
992                .remove_image(&cluster.name, None::<RemoveImageOptions>, None)
993                .await?;
994        }
995
996        Ok(())
997    }
998}
999
1000impl<'a> Deploy<'a> for DockerDeploy {
1001    type Meta = ();
1002    type InstantiateEnv = Self;
1003
1004    type Process = DockerDeployProcess;
1005    type Cluster = DockerDeployCluster;
1006    type External = DockerDeployExternal;
1007
1008    #[instrument(level = "trace", skip_all, fields(p1 = p1.name, %p1_port, p2 = p2.name, %p2_port))]
1009    fn o2o_sink_source(
1010        _env: &mut Self::InstantiateEnv,
1011        p1: &Self::Process,
1012        p1_port: &<Self::Process as Node>::Port,
1013        p2: &Self::Process,
1014        p2_port: &<Self::Process as Node>::Port,
1015        name: Option<&str>,
1016        networking_info: &crate::networking::NetworkingInfo,
1017        _external_types: Option<(&syn::Type, &syn::Type)>,
1018    ) -> (syn::Expr, syn::Expr) {
1019        match networking_info {
1020            crate::networking::NetworkingInfo::Tcp {
1021                fault: crate::networking::TcpFault::FailStop,
1022            } => {}
1023            _ => panic!("Unsupported networking info: {:?}", networking_info),
1024        }
1025
1026        deploy_containerized_o2o(
1027            &p2.name,
1028            name.expect("channel name is required for containerized deployment"),
1029        )
1030    }
1031
1032    #[instrument(level = "trace", skip_all, fields(p1 = p1.name, %p1_port, p2 = p2.name, %p2_port))]
1033    fn o2o_connect(
1034        p1: &Self::Process,
1035        p1_port: &<Self::Process as Node>::Port,
1036        p2: &Self::Process,
1037        p2_port: &<Self::Process as Node>::Port,
1038    ) -> Box<dyn FnOnce()> {
1039        let serialized = format!("o2o_connect {}:{p1_port} -> {}:{p2_port}", p1.name, p2.name);
1040
1041        Box::new(move || {
1042            trace!(name: "o2o_connect thunk", %serialized);
1043        })
1044    }
1045
1046    #[instrument(level = "trace", skip_all, fields(p1 = p1.name, %p1_port, c2 = c2.name, %c2_port))]
1047    fn o2m_sink_source(
1048        _env: &mut Self::InstantiateEnv,
1049        p1: &Self::Process,
1050        p1_port: &<Self::Process as Node>::Port,
1051        c2: &Self::Cluster,
1052        c2_port: &<Self::Cluster as Node>::Port,
1053        name: Option<&str>,
1054        networking_info: &crate::networking::NetworkingInfo,
1055        _external_types: Option<(&syn::Type, &syn::Type)>,
1056    ) -> (syn::Expr, syn::Expr) {
1057        match networking_info {
1058            crate::networking::NetworkingInfo::Tcp {
1059                fault: crate::networking::TcpFault::FailStop,
1060            } => {}
1061            _ => panic!("Unsupported networking info: {:?}", networking_info),
1062        }
1063
1064        deploy_containerized_o2m(
1065            name.expect("channel name is required for containerized deployment"),
1066        )
1067    }
1068
1069    #[instrument(level = "trace", skip_all, fields(p1 = p1.name, %p1_port, c2 = c2.name, %c2_port))]
1070    fn o2m_connect(
1071        p1: &Self::Process,
1072        p1_port: &<Self::Process as Node>::Port,
1073        c2: &Self::Cluster,
1074        c2_port: &<Self::Cluster as Node>::Port,
1075    ) -> Box<dyn FnOnce()> {
1076        let serialized = format!("o2m_connect {}:{p1_port} -> {}:{c2_port}", p1.name, c2.name);
1077
1078        Box::new(move || {
1079            trace!(name: "o2m_connect thunk", %serialized);
1080        })
1081    }
1082
1083    #[instrument(level = "trace", skip_all, fields(c1 = c1.name, %c1_port, p2 = p2.name, %p2_port))]
1084    fn m2o_sink_source(
1085        _env: &mut Self::InstantiateEnv,
1086        c1: &Self::Cluster,
1087        c1_port: &<Self::Cluster as Node>::Port,
1088        p2: &Self::Process,
1089        p2_port: &<Self::Process as Node>::Port,
1090        name: Option<&str>,
1091        networking_info: &crate::networking::NetworkingInfo,
1092        _external_types: Option<(&syn::Type, &syn::Type)>,
1093    ) -> (syn::Expr, syn::Expr) {
1094        match networking_info {
1095            crate::networking::NetworkingInfo::Tcp {
1096                fault: crate::networking::TcpFault::FailStop,
1097            } => {}
1098            _ => panic!("Unsupported networking info: {:?}", networking_info),
1099        }
1100
1101        deploy_containerized_m2o(
1102            &p2.name,
1103            name.expect("channel name is required for containerized deployment"),
1104        )
1105    }
1106
1107    #[instrument(level = "trace", skip_all, fields(c1 = c1.name, %c1_port, p2 = p2.name, %p2_port))]
1108    fn m2o_connect(
1109        c1: &Self::Cluster,
1110        c1_port: &<Self::Cluster as Node>::Port,
1111        p2: &Self::Process,
1112        p2_port: &<Self::Process as Node>::Port,
1113    ) -> Box<dyn FnOnce()> {
1114        let serialized = format!("o2m_connect {}:{c1_port} -> {}:{p2_port}", c1.name, p2.name);
1115
1116        Box::new(move || {
1117            trace!(name: "m2o_connect thunk", %serialized);
1118        })
1119    }
1120
1121    #[instrument(level = "trace", skip_all, fields(c1 = c1.name, %c1_port, c2 = c2.name, %c2_port))]
1122    fn m2m_sink_source(
1123        _env: &mut Self::InstantiateEnv,
1124        c1: &Self::Cluster,
1125        c1_port: &<Self::Cluster as Node>::Port,
1126        c2: &Self::Cluster,
1127        c2_port: &<Self::Cluster as Node>::Port,
1128        name: Option<&str>,
1129        networking_info: &crate::networking::NetworkingInfo,
1130        _external_types: Option<(&syn::Type, &syn::Type)>,
1131    ) -> (syn::Expr, syn::Expr) {
1132        match networking_info {
1133            crate::networking::NetworkingInfo::Tcp {
1134                fault: crate::networking::TcpFault::FailStop,
1135            } => {}
1136            _ => panic!("Unsupported networking info: {:?}", networking_info),
1137        }
1138
1139        deploy_containerized_m2m(
1140            name.expect("channel name is required for containerized deployment"),
1141        )
1142    }
1143
1144    #[instrument(level = "trace", skip_all, fields(c1 = c1.name, %c1_port, c2 = c2.name, %c2_port))]
1145    fn m2m_connect(
1146        c1: &Self::Cluster,
1147        c1_port: &<Self::Cluster as Node>::Port,
1148        c2: &Self::Cluster,
1149        c2_port: &<Self::Cluster as Node>::Port,
1150    ) -> Box<dyn FnOnce()> {
1151        let serialized = format!("m2m_connect {}:{c1_port} -> {}:{c2_port}", c1.name, c2.name);
1152
1153        Box::new(move || {
1154            trace!(name: "m2m_connect thunk", %serialized);
1155        })
1156    }
1157
1158    #[instrument(level = "trace", skip_all, fields(p2 = p2.name, %p2_port, %shared_handle, extra_stmts = extra_stmts.len()))]
1159    fn e2o_many_source(
1160        extra_stmts: &mut Vec<syn::Stmt>,
1161        p2: &Self::Process,
1162        p2_port: &<Self::Process as Node>::Port,
1163        codec_type: &syn::Type,
1164        shared_handle: String,
1165    ) -> syn::Expr {
1166        p2.exposed_ports.borrow_mut().push(*p2_port);
1167
1168        let socket_ident = syn::Ident::new(
1169            &format!("__hydro_deploy_many_{}_socket", shared_handle),
1170            Span::call_site(),
1171        );
1172
1173        let source_ident = syn::Ident::new(
1174            &format!("__hydro_deploy_many_{}_source", shared_handle),
1175            Span::call_site(),
1176        );
1177
1178        let sink_ident = syn::Ident::new(
1179            &format!("__hydro_deploy_many_{}_sink", shared_handle),
1180            Span::call_site(),
1181        );
1182
1183        let membership_ident = syn::Ident::new(
1184            &format!("__hydro_deploy_many_{}_membership", shared_handle),
1185            Span::call_site(),
1186        );
1187
1188        let bind_addr = format!("0.0.0.0:{}", p2_port);
1189
1190        extra_stmts.push(syn::parse_quote! {
1191            let #socket_ident = tokio::net::TcpListener::bind(#bind_addr).await.unwrap();
1192        });
1193
1194        let root = crate::staging_util::get_this_crate();
1195
1196        extra_stmts.push(syn::parse_quote! {
1197            let (#source_ident, #sink_ident, #membership_ident) = #root::runtime_support::hydro_deploy_integration::multi_connection::tcp_multi_connection::<_, #codec_type>(#socket_ident);
1198        });
1199
1200        parse_quote!(#source_ident)
1201    }
1202
1203    #[instrument(level = "trace", skip_all, fields(%shared_handle))]
1204    fn e2o_many_sink(shared_handle: String) -> syn::Expr {
1205        let sink_ident = syn::Ident::new(
1206            &format!("__hydro_deploy_many_{}_sink", shared_handle),
1207            Span::call_site(),
1208        );
1209        parse_quote!(#sink_ident)
1210    }
1211
1212    #[instrument(level = "trace", skip_all, fields(p1 = p1.name, %p1_port, p2 = p2.name, %p2_port, %shared_handle))]
1213    fn e2o_source(
1214        extra_stmts: &mut Vec<syn::Stmt>,
1215        p1: &Self::External,
1216        p1_port: &<Self::External as Node>::Port,
1217        p2: &Self::Process,
1218        p2_port: &<Self::Process as Node>::Port,
1219        _codec_type: &syn::Type,
1220        shared_handle: String,
1221    ) -> syn::Expr {
1222        p1.connection_info.borrow_mut().insert(
1223            *p1_port,
1224            (
1225                p2.docker_container_name.clone(),
1226                *p2_port,
1227                p2.network.clone(),
1228            ),
1229        );
1230
1231        p2.exposed_ports.borrow_mut().push(*p2_port);
1232
1233        let socket_ident = syn::Ident::new(
1234            &format!("__hydro_deploy_{}_socket", shared_handle),
1235            Span::call_site(),
1236        );
1237
1238        let source_ident = syn::Ident::new(
1239            &format!("__hydro_deploy_{}_source", shared_handle),
1240            Span::call_site(),
1241        );
1242
1243        let sink_ident = syn::Ident::new(
1244            &format!("__hydro_deploy_{}_sink", shared_handle),
1245            Span::call_site(),
1246        );
1247
1248        let bind_addr = format!("0.0.0.0:{}", p2_port);
1249
1250        extra_stmts.push(syn::parse_quote! {
1251            let #socket_ident = tokio::net::TcpListener::bind(#bind_addr).await.unwrap();
1252        });
1253
1254        let create_expr = deploy_containerized_external_sink_source_ident(socket_ident);
1255
1256        extra_stmts.push(syn::parse_quote! {
1257            let (#sink_ident, #source_ident) = (#create_expr).split();
1258        });
1259
1260        parse_quote!(#source_ident)
1261    }
1262
1263    #[instrument(level = "trace", skip_all, fields(p1 = p1.name, %p1_port, p2 = p2.name, %p2_port, ?many, ?server_hint))]
1264    fn e2o_connect(
1265        p1: &Self::External,
1266        p1_port: &<Self::External as Node>::Port,
1267        p2: &Self::Process,
1268        p2_port: &<Self::Process as Node>::Port,
1269        many: bool,
1270        server_hint: NetworkHint,
1271    ) -> Box<dyn FnOnce()> {
1272        if server_hint != NetworkHint::Auto {
1273            panic!(
1274                "Docker deployment only supports NetworkHint::Auto, got {:?}",
1275                server_hint
1276            );
1277        }
1278
1279        // For many connections, we need to populate connection_info so as_bincode_bidi can find it
1280        if many {
1281            p1.connection_info.borrow_mut().insert(
1282                *p1_port,
1283                (
1284                    p2.docker_container_name.clone(),
1285                    *p2_port,
1286                    p2.network.clone(),
1287                ),
1288            );
1289        }
1290
1291        let serialized = format!("e2o_connect {}:{p1_port} -> {}:{p2_port}", p1.name, p2.name);
1292
1293        Box::new(move || {
1294            trace!(name: "e2o_connect thunk", %serialized);
1295        })
1296    }
1297
1298    #[instrument(level = "trace", skip_all, fields(p1 = p1.name, %p1_port, p2 = p2.name, %p2_port, %shared_handle))]
1299    fn o2e_sink(
1300        p1: &Self::Process,
1301        p1_port: &<Self::Process as Node>::Port,
1302        p2: &Self::External,
1303        p2_port: &<Self::External as Node>::Port,
1304        shared_handle: String,
1305    ) -> syn::Expr {
1306        let sink_ident = syn::Ident::new(
1307            &format!("__hydro_deploy_{}_sink", shared_handle),
1308            Span::call_site(),
1309        );
1310        parse_quote!(#sink_ident)
1311    }
1312
1313    #[instrument(level = "trace", skip_all, fields(%of_cluster))]
1314    fn cluster_ids(
1315        of_cluster: LocationKey,
1316    ) -> impl QuotedWithContext<'a, &'a [TaglessMemberId], ()> + Clone + 'a {
1317        cluster_ids()
1318    }
1319
1320    #[instrument(level = "trace", skip_all)]
1321    fn cluster_self_id() -> impl QuotedWithContext<'a, TaglessMemberId, ()> + Clone + 'a {
1322        cluster_self_id()
1323    }
1324
1325    #[instrument(level = "trace", skip_all, fields(?location_id))]
1326    fn cluster_membership_stream(
1327        _env: &mut Self::InstantiateEnv,
1328        _at_location: &LocationId,
1329        location_id: &LocationId,
1330    ) -> impl QuotedWithContext<'a, Box<dyn Stream<Item = (TaglessMemberId, MembershipEvent)> + Unpin>, ()>
1331    {
1332        cluster_membership_stream(location_id)
1333    }
1334}
1335
1336const CONTAINER_ALPHABET: [char; 36] = [
1337    '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i',
1338    'j', 'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z',
1339];
1340
1341fn is_valid_docker_image_name(name: &str) -> bool {
1342    regex::Regex::new(r"^[a-z0-9]+([._-][a-z0-9]+)*$")
1343        .unwrap()
1344        .is_match(name)
1345}
1346
1347#[instrument(level = "trace", skip_all, ret, fields(%name_hint, %location_key, %deployment_instance))]
1348fn get_docker_image_name(
1349    name_hint: &str,
1350    location_key: LocationKey,
1351    deployment_instance: &str,
1352) -> String {
1353    let name_hint: String = name_hint
1354        .split("::")
1355        .last()
1356        .unwrap()
1357        .to_ascii_lowercase()
1358        .split(['.', '_', '-'])
1359        .filter(|s| !s.is_empty())
1360        .collect::<Vec<_>>()
1361        .join("-");
1362
1363    let image_name = format!("hy-{name_hint}-{deployment_instance}-{location_key}");
1364
1365    if !is_valid_docker_image_name(&image_name) {
1366        panic!(
1367            "Generated Docker image name '{image_name}' is not a valid Docker image name. \
1368             Docker image names may only contain lowercase alphanumeric characters \
1369             separated by single '.', '_', or '-' characters, and must start and end \
1370             with an alphanumeric character. The most likely cause is your location \
1371             struct name '{name_hint}'"
1372        );
1373    }
1374
1375    image_name
1376}
1377
1378#[instrument(level = "trace", skip_all, ret, fields(%image_name, ?instance))]
1379fn get_docker_container_name(image_name: &str, instance: Option<usize>) -> String {
1380    if let Some(instance) = instance {
1381        format!("{image_name}-{instance}")
1382    } else {
1383        image_name.to_owned()
1384    }
1385}
1386/// Represents a Process running in a docker container
1387#[derive(Clone)]
1388pub struct DockerDeployProcessSpec {
1389    rustflags: Option<String>,
1390    config: Vec<String>,
1391    network: DockerNetwork,
1392    deployment_instance: String,
1393    base_image: Option<String>,
1394    linux_compile_type: LinuxCompileType,
1395    features: Vec<String>,
1396}
1397
1398impl<'a> ProcessSpec<'a, DockerDeploy> for DockerDeployProcessSpec {
1399    #[instrument(level = "trace", skip_all, fields(%key, %name_hint))]
1400    fn build(self, key: LocationKey, name_hint: &'_ str) -> <DockerDeploy as Deploy<'a>>::Process {
1401        DockerDeployProcess {
1402            key,
1403            name: get_docker_image_name(name_hint, key, &self.deployment_instance),
1404
1405            next_port: Rc::new(RefCell::new(1000)),
1406            rust_crate: Rc::new(RefCell::new(None)),
1407
1408            exposed_ports: Rc::new(RefCell::new(Vec::new())),
1409
1410            docker_container_name: Rc::new(RefCell::new(None)),
1411
1412            rustflags: self.rustflags,
1413            config: self.config,
1414
1415            network: self.network.clone(),
1416
1417            base_image: self.base_image,
1418            linux_compile_type: self.linux_compile_type,
1419            features: self.features,
1420        }
1421    }
1422}
1423
1424/// Represents a Cluster running across `count` docker containers.
1425#[derive(Clone)]
1426pub struct DockerDeployClusterSpec {
1427    rustflags: Option<String>,
1428    config: Vec<String>,
1429    count: usize,
1430    deployment_instance: String,
1431    base_image: Option<String>,
1432    linux_compile_type: LinuxCompileType,
1433    features: Vec<String>,
1434}
1435
1436impl<'a> ClusterSpec<'a, DockerDeploy> for DockerDeployClusterSpec {
1437    #[instrument(level = "trace", skip_all, fields(%key, %name_hint))]
1438    fn build(self, key: LocationKey, name_hint: &str) -> <DockerDeploy as Deploy<'a>>::Cluster {
1439        DockerDeployCluster {
1440            key,
1441            name: get_docker_image_name(name_hint, key, &self.deployment_instance),
1442
1443            next_port: Rc::new(RefCell::new(1000)),
1444            rust_crate: Rc::new(RefCell::new(None)),
1445
1446            exposed_ports: Rc::new(RefCell::new(Vec::new())),
1447
1448            docker_container_name: Rc::new(RefCell::new(Vec::new())),
1449
1450            rustflags: self.rustflags,
1451            config: self.config,
1452
1453            count: self.count,
1454
1455            base_image: self.base_image,
1456            linux_compile_type: self.linux_compile_type,
1457            features: self.features,
1458        }
1459    }
1460}
1461
1462impl DockerDeployProcessSpec {
1463    /// Set the base Docker image for this process.
1464    /// Defaults to `scratch` if not specified.
1465    pub fn base_image(mut self, image: impl Into<String>) -> Self {
1466        self.base_image = Some(image.into());
1467        self
1468    }
1469
1470    /// Set the Linux compile type (glibc or musl) for this process.
1471    /// Defaults to `Musl` if not specified.
1472    pub fn linux_compile_type(mut self, compile_type: LinuxCompileType) -> Self {
1473        self.linux_compile_type = compile_type;
1474        self
1475    }
1476
1477    /// Add features to enable when compiling the final binary.
1478    pub fn features(mut self, features: impl IntoIterator<Item = impl Into<String>>) -> Self {
1479        self.features.extend(features.into_iter().map(Into::into));
1480        self
1481    }
1482
1483    /// Add a single feature to enable when compiling the final binary.
1484    pub fn feature(mut self, feature: impl Into<String>) -> Self {
1485        self.features.push(feature.into());
1486        self
1487    }
1488}
1489
1490impl DockerDeployClusterSpec {
1491    /// Set the base Docker image for this cluster.
1492    /// Defaults to `scratch` if not specified.
1493    pub fn base_image(mut self, image: impl Into<String>) -> Self {
1494        self.base_image = Some(image.into());
1495        self
1496    }
1497
1498    /// Set the Linux compile type (glibc or musl) for this cluster.
1499    /// Defaults to `Musl` if not specified.
1500    pub fn linux_compile_type(mut self, compile_type: LinuxCompileType) -> Self {
1501        self.linux_compile_type = compile_type;
1502        self
1503    }
1504
1505    /// Add features to enable when compiling the final binary.
1506    pub fn features(mut self, features: impl IntoIterator<Item = impl Into<String>>) -> Self {
1507        self.features.extend(features.into_iter().map(Into::into));
1508        self
1509    }
1510
1511    /// Add a single feature to enable when compiling the final binary.
1512    pub fn feature(mut self, feature: impl Into<String>) -> Self {
1513        self.features.push(feature.into());
1514        self
1515    }
1516}
1517
1518/// Represents an external process outside of the management of hydro deploy.
1519pub struct DockerDeployExternalSpec {
1520    name: String,
1521}
1522
1523impl<'a> ExternalSpec<'a, DockerDeploy> for DockerDeployExternalSpec {
1524    #[instrument(level = "trace", skip_all, fields(%key, %name_hint))]
1525    fn build(self, key: LocationKey, name_hint: &str) -> <DockerDeploy as Deploy<'a>>::External {
1526        DockerDeployExternal {
1527            key,
1528            name: self.name,
1529            next_port: Rc::new(RefCell::new(10000)),
1530            next_external_port_id: Rc::new(RefCell::new(crate::Counter::default())),
1531            ports: Rc::new(RefCell::new(HashMap::new())),
1532            connection_info: Rc::new(RefCell::new(HashMap::new())),
1533        }
1534    }
1535}
hydro_lang/deploy/deploy_graph_containerized.rs

hydro_lang/deploy/
deploy_graph_containerized.rs
