hydro_deploy/
terraform.rs

1use std::collections::HashMap;
2use std::ffi::OsStr;
3use std::io::{BufRead, BufReader};
4#[cfg(unix)]
5use std::os::unix::process::CommandExt;
6use std::path::PathBuf;
7use std::process::{Child, ChildStdout, Command};
8use std::sync::{Arc, OnceLock, RwLock};
9
10use anyhow::{Context, Result, bail};
11use async_process::Stdio;
12use serde::{Deserialize, Serialize};
13use tempfile::TempDir;
14
15use super::progress::ProgressTracker;
16
17pub static TERRAFORM_ALPHABET: [char; 16] = [
18    '1', '2', '3', '4', '5', '6', '7', '8', '9', '0', 'a', 'b', 'c', 'd', 'e', 'f',
19];
20
21static TERRAFORM_PATH: OnceLock<PathBuf> = OnceLock::new();
22
23/// Returns the path to the terraform-compatible CLI (tofu or terraform).
24/// Prefers `tofu` if available, otherwise falls back to `terraform`.
25/// The result is cached in a `OnceLock` for subsequent calls.
26fn get_terraform_path() -> &'static PathBuf {
27    TERRAFORM_PATH.get_or_init(|| {
28        which::which("tofu")
29            .or_else(|_| which::which("terraform"))
30            .expect("Neither `tofu` nor `terraform` found in PATH. Please install one of them.")
31    })
32}
33
34fn terraform_command() -> Command {
35    Command::new(get_terraform_path())
36}
37
38/// Returns the name of the terraform-compatible CLI being used (for display purposes).
39pub fn terraform_name() -> &'static str {
40    get_terraform_path()
41        .file_name()
42        .and_then(OsStr::to_str)
43        .unwrap_or("terraform")
44}
45
46/// Keeps track of resources which may need to be cleaned up.
47#[derive(Default)]
48pub struct TerraformPool {
49    counter: u32,
50    active_applies: HashMap<u32, Arc<tokio::sync::RwLock<TerraformApply>>>,
51}
52
53impl TerraformPool {
54    fn create_apply(
55        &mut self,
56        deployment_folder: TempDir,
57    ) -> Result<(u32, Arc<tokio::sync::RwLock<TerraformApply>>)> {
58        let next_counter = self.counter;
59        self.counter += 1;
60
61        let mut apply_command = terraform_command();
62
63        apply_command
64            .current_dir(deployment_folder.path())
65            .arg("apply")
66            .arg("-auto-approve")
67            .arg("-no-color")
68            .arg("-parallelism=128");
69
70        #[cfg(unix)]
71        {
72            apply_command.process_group(0);
73        }
74
75        let spawned_child = apply_command
76            .stdout(Stdio::piped())
77            .stderr(Stdio::piped())
78            .spawn()
79            .with_context(|| format!("Failed to spawn `{}`. Is it installed?", terraform_name()))?;
80
81        let spawned_id = spawned_child.id();
82
83        let deployment = Arc::new(tokio::sync::RwLock::new(TerraformApply {
84            child: Some((spawned_id, Arc::new(RwLock::new(spawned_child)))),
85            deployment_folder: Some(deployment_folder),
86        }));
87
88        self.active_applies.insert(next_counter, deployment.clone());
89
90        Ok((next_counter, deployment))
91    }
92
93    fn drop_apply(&mut self, counter: u32) {
94        self.active_applies.remove(&counter);
95    }
96}
97
98impl Drop for TerraformPool {
99    fn drop(&mut self) {
100        #[expect(
101            clippy::disallowed_methods,
102            reason = "nondeterministic iteration order, fine for assertions"
103        )]
104        for (_, apply) in self.active_applies.drain() {
105            debug_assert_eq!(Arc::strong_count(&apply), 1);
106        }
107    }
108}
109
110#[derive(Serialize, Deserialize)]
111pub struct TerraformBatch {
112    pub terraform: TerraformConfig,
113    #[serde(skip_serializing_if = "HashMap::is_empty")]
114    pub provider: HashMap<String, serde_json::Value>,
115    #[serde(skip_serializing_if = "HashMap::is_empty")]
116    pub data: HashMap<String, HashMap<String, serde_json::Value>>,
117    pub resource: HashMap<String, HashMap<String, serde_json::Value>>,
118    pub output: HashMap<String, TerraformOutput>,
119}
120
121impl Default for TerraformBatch {
122    fn default() -> TerraformBatch {
123        TerraformBatch {
124            terraform: TerraformConfig {
125                required_providers: HashMap::new(),
126            },
127            provider: HashMap::new(),
128            data: HashMap::new(),
129            resource: HashMap::new(),
130            output: HashMap::new(),
131        }
132    }
133}
134
135impl TerraformBatch {
136    pub async fn provision(self, pool: &mut TerraformPool) -> Result<TerraformResult> {
137        if self.terraform.required_providers.is_empty()
138            && self.resource.is_empty()
139            && self.data.is_empty()
140            && self.output.is_empty()
141        {
142            return Ok(TerraformResult {
143                outputs: HashMap::new(),
144                deployment_folder: None,
145            });
146        }
147
148        ProgressTracker::with_group(terraform_name(), Some(1), || async {
149            let dothydro_folder = std::env::current_dir().unwrap().join(".hydro");
150            std::fs::create_dir_all(&dothydro_folder).unwrap();
151            let deployment_folder = tempfile::tempdir_in(dothydro_folder).unwrap();
152
153            std::fs::write(
154                deployment_folder.path().join("main.tf.json"),
155                serde_json::to_string(&self).unwrap(),
156            )
157            .unwrap();
158
159            if !terraform_command()
160                .current_dir(deployment_folder.path())
161                .arg("init")
162                .stdout(Stdio::null())
163                .spawn()
164                .with_context(|| {
165                    format!("Failed to spawn `{}`. Is it installed?", terraform_name())
166                })?
167                .wait()
168                .with_context(|| format!("Failed to launch {} init command", terraform_name()))?
169                .success()
170            {
171                bail!("Failed to initialize {}", terraform_name());
172            }
173
174            let (apply_id, apply) = pool.create_apply(deployment_folder)?;
175
176            let output = ProgressTracker::with_group(
177                "apply",
178                #[expect(
179                    clippy::disallowed_methods,
180                    reason = "nondeterministic iteration order, fine for addition"
181                )]
182                Some(self.resource.values().map(|r| r.len()).sum()),
183                || async { apply.write().await.output().await },
184            )
185            .await;
186            pool.drop_apply(apply_id);
187            output
188        })
189        .await
190    }
191}
192
193struct TerraformApply {
194    child: Option<(u32, Arc<RwLock<Child>>)>,
195    deployment_folder: Option<TempDir>,
196}
197
198async fn display_apply_outputs(stdout: &mut ChildStdout) {
199    let lines = BufReader::new(stdout).lines();
200    let mut waiting_for_result = HashMap::new();
201
202    for line in lines {
203        if let Ok(line) = line {
204            let mut split = line.split(':');
205            if let Some(first) = split.next()
206                && first.chars().all(|c| c != ' ')
207                && split.next().is_some()
208                && split.next().is_none()
209            {
210                if line.starts_with("Plan:")
211                    || line.starts_with("Outputs:")
212                    || line.contains(": Still creating...")
213                    || line.contains(": Reading...")
214                    || line.contains(": Still reading...")
215                    || line.contains(": Read complete after")
216                {
217                } else if line.ends_with(": Creating...") {
218                    let id = line.split(':').next().unwrap().trim();
219                    let (channel_send, channel_recv) = tokio::sync::oneshot::channel();
220                    waiting_for_result.insert(
221                        id.to_owned(),
222                        (
223                            channel_send,
224                            tokio::task::spawn(ProgressTracker::leaf(id.to_owned(), async move {
225                                // `Err(RecvError)` means send side was dropped due to another error.
226                                // Ignore here to prevent spurious panic stack traces.
227                                let _result = channel_recv.await;
228                            })),
229                        ),
230                    );
231                } else if line.contains(": Creation complete after") {
232                    let id = line.split(':').next().unwrap().trim();
233                    let (sender, to_await) = waiting_for_result.remove(id).unwrap();
234                    let _ = sender.send(());
235                    to_await.await.unwrap();
236                } else {
237                    panic!("Unexpected from {}: {}", terraform_name(), line);
238                }
239            }
240        } else {
241            break;
242        }
243    }
244}
245
246fn filter_terraform_logs(child: &mut Child) {
247    let lines = BufReader::new(child.stdout.take().unwrap()).lines();
248    let tf_name = terraform_name();
249    for line in lines {
250        if let Ok(line) = line {
251            let mut split = line.split(':');
252            if let Some(first) = split.next()
253                && first.chars().all(|c| c != ' ')
254                && split.next().is_some()
255                && split.next().is_none()
256            {
257                eprintln!("[{}] {}", tf_name, line);
258            }
259        } else {
260            break;
261        }
262    }
263}
264
265impl TerraformApply {
266    async fn output(&mut self) -> Result<TerraformResult> {
267        let (_, child) = self.child.as_ref().unwrap().clone();
268        let mut stdout = child.write().unwrap().stdout.take().unwrap();
269        let stderr = child.write().unwrap().stderr.take().unwrap();
270
271        let status = tokio::task::spawn_blocking(move || {
272            // it is okay for this thread to keep running even if the future is cancelled
273            child.write().unwrap().wait().unwrap()
274        });
275
276        let display_apply = display_apply_outputs(&mut stdout);
277        let tf_name = terraform_name();
278        let stderr_loop = tokio::task::spawn_blocking(move || {
279            let mut lines = BufReader::new(stderr).lines();
280            while let Some(Ok(line)) = lines.next() {
281                ProgressTracker::println(format!("[{}] {}", tf_name, line));
282            }
283        });
284
285        let _ = futures::join!(display_apply, stderr_loop);
286
287        let status = status.await;
288
289        self.child = None;
290
291        if !status.unwrap().success() {
292            bail!(
293                "{tf} deployment failed, see `[{tf}]` logs above.",
294                tf = terraform_name(),
295            );
296        }
297
298        let mut output_command = terraform_command();
299        output_command
300            .current_dir(self.deployment_folder.as_ref().unwrap().path())
301            .arg("output")
302            .arg("-json");
303
304        #[cfg(unix)]
305        {
306            output_command.process_group(0);
307        }
308
309        let output = output_command
310            .output()
311            .with_context(|| format!("Failed to read {} outputs", terraform_name()))?;
312
313        Ok(TerraformResult {
314            outputs: serde_json::from_slice(&output.stdout).unwrap(),
315            deployment_folder: self.deployment_folder.take(),
316        })
317    }
318}
319
320fn destroy_deployment(deployment_folder: TempDir) {
321    println!(
322        "Destroying {} deployment at {}",
323        terraform_name(),
324        deployment_folder.path().display()
325    );
326
327    let mut destroy_command = terraform_command();
328    destroy_command
329        .current_dir(deployment_folder.path())
330        .arg("destroy")
331        .arg("-auto-approve")
332        .arg("-no-color")
333        .arg("-parallelism=128")
334        .stdout(Stdio::piped());
335
336    #[cfg(unix)]
337    {
338        destroy_command.process_group(0);
339    }
340
341    let mut destroy_child = destroy_command
342        .spawn()
343        .unwrap_or_else(|_| panic!("Failed to spawn {} destroy command", terraform_name()));
344
345    filter_terraform_logs(&mut destroy_child);
346
347    if !destroy_child
348        .wait()
349        .unwrap_or_else(|_| panic!("Failed to destroy {} deployment", terraform_name()))
350        .success()
351    {
352        // prevent the folder from being deleted
353        let _ = deployment_folder.keep();
354        eprintln!("WARNING: failed to destroy {} deployment", terraform_name());
355    }
356}
357
358impl Drop for TerraformApply {
359    fn drop(&mut self) {
360        if let Some((pid, child)) = self.child.take() {
361            #[cfg(unix)]
362            nix::sys::signal::kill(
363                nix::unistd::Pid::from_raw(pid as i32),
364                nix::sys::signal::Signal::SIGINT,
365            )
366            .unwrap();
367            #[cfg(not(unix))]
368            let _ = pid;
369
370            let mut child_write = child.write().unwrap();
371            if child_write.try_wait().unwrap().is_none() {
372                println!("Waiting for Terraform apply to finish...");
373                child_write.wait().unwrap();
374            }
375        }
376
377        if let Some(deployment_folder) = self.deployment_folder.take() {
378            destroy_deployment(deployment_folder);
379        }
380    }
381}
382
383#[derive(Serialize, Deserialize)]
384pub struct TerraformConfig {
385    pub required_providers: HashMap<String, TerraformProvider>,
386}
387
388#[derive(Serialize, Deserialize)]
389pub struct TerraformProvider {
390    pub source: String,
391    pub version: String,
392}
393
394#[derive(Serialize, Deserialize, Debug)]
395pub struct TerraformOutput {
396    pub value: String,
397}
398
399#[derive(Debug)]
400pub struct TerraformResult {
401    pub outputs: HashMap<String, TerraformOutput>,
402    /// `None` if no deployment was performed
403    pub deployment_folder: Option<TempDir>,
404}
405
406impl Drop for TerraformResult {
407    fn drop(&mut self) {
408        if let Some(deployment_folder) = self.deployment_folder.take() {
409            destroy_deployment(deployment_folder);
410        }
411    }
412}
413
414#[derive(Serialize, Deserialize)]
415pub struct TerraformResultOutput {
416    value: String,
417}
hydro_deploy/terraform.rs

hydro_deploy/
terraform.rs
