data-plane-controller: skip DNS await if nothing changed in pulumi up

estuary · Feb 24, 2025 · 65112fb · 65112fb
1 parent 9882f75
commit 65112fb
Show file tree

Hide file tree

Showing 2 changed files with 104 additions and 22 deletions.
diff --git a/crates/data-plane-controller/src/controller.rs b/crates/data-plane-controller/src/controller.rs
@@ -529,19 +529,32 @@ impl Controller {
         )
         .await?;
 
-        self.logs_tx
-            .send(logs::Line {
-                token: state.logs_token,
-                stream: "controller".to_string(),
-                line: format!("Waiting {DNS_TTL:?} for DNS propagation before continuing."),
-            })
-            .await
-            .context("failed to send to logs sink")?;
+        let stack::PulumiStackHistory { resource_changes } = self.last_pulumi_run(&state, &checkout).await?;
 
-        state.status = Status::AwaitDNS1;
         state.last_pulumi_up = chrono::Utc::now();
-
-        Ok(DNS_TTL)
+        if resource_changes.changed() {
+            self.logs_tx
+                .send(logs::Line {
+                    token: state.logs_token,
+                    stream: "controller".to_string(),
+                    line: format!("Waiting {DNS_TTL:?} for DNS propagation before continuing."),
+                })
+                .await
+                .context("failed to send to logs sink")?;
+            state.status = Status::AwaitDNS1;
+            Ok(DNS_TTL)
+        } else {
+            self.logs_tx
+                .send(logs::Line {
+                    token: state.logs_token,
+                    stream: "controller".to_string(),
+                    line: format!("No changes detected, continuing to Ansible."),
+                })
+                .await
+                .context("failed to send to logs sink")?;
+            state.status = Status::Ansible;
+            Ok(POLL_AGAIN)
+        }
     }
 
     #[tracing::instrument(
@@ -694,19 +707,32 @@ impl Controller {
         )
         .await?;
 
-        self.logs_tx
-            .send(logs::Line {
-                token: state.logs_token,
-                stream: "controller".to_string(),
-                line: format!("Waiting {DNS_TTL:?} for DNS propagation before continuing."),
-            })
-            .await
-            .context("failed to send to logs sink")?;
+        let stack::PulumiStackHistory { resource_changes } = self.last_pulumi_run(&state, &checkout).await?;
 
-        state.status = Status::AwaitDNS2;
         state.last_pulumi_up = chrono::Utc::now();
-
-        Ok(DNS_TTL)
+        if resource_changes.changed() {
+            self.logs_tx
+                .send(logs::Line {
+                    token: state.logs_token,
+                    stream: "controller".to_string(),
+                    line: format!("Waiting {DNS_TTL:?} for DNS propagation before continuing."),
+                })
+                .await
+                .context("failed to send to logs sink")?;
+            state.status = Status::AwaitDNS2;
+            Ok(DNS_TTL)
+        } else {
+            self.logs_tx
+                .send(logs::Line {
+                    token: state.logs_token,
+                    stream: "controller".to_string(),
+                    line: format!("No changes detected, done."),
+                })
+                .await
+                .context("failed to send to logs sink")?;
+            state.status = Status::Idle;
+            Ok(POLL_AGAIN)
+        }
     }
 
     #[tracing::instrument(
@@ -810,6 +836,39 @@ impl Controller {
 
         Ok(checkout)
     }
+
+    async fn last_pulumi_run(&self, state: &State, checkout: &repo::Checkout) -> anyhow::Result<stack::PulumiStackHistory>  {
+        // Check if any resources changed
+        let output = async_process::output(
+            async_process::Command::new("pulumi")
+                .arg("stack")
+                .arg("history")
+                .arg("--stack")
+                .arg(&state.stack_name)
+                .arg("--json")
+                .arg("--page-size")
+                .arg("1")
+                .arg("--cwd")
+                .arg(&checkout.path())
+                .envs(self.pulumi_secret_envs())
+                .env("PULUMI_BACKEND_URL", self.state_backend.as_str())
+                .env("VIRTUAL_ENV", checkout.path().join("venv")),
+        )
+        .await?;
+
+        if !output.status.success() {
+            anyhow::bail!(
+                "pulumi stack history output failed: {}",
+                String::from_utf8_lossy(&output.stderr),
+            );
+        }
+
+        let mut out: Vec<stack::PulumiStackHistory> = serde_json::from_slice(&output.stdout).context("failed to parse pulumi stack history output")?;
+
+        let result = out.pop().expect("at least one pulumi stack history record after a pulumi run");
+
+        return Ok(result)
+    }
 }
 
 impl automations::Outcome for Outcome {

diff --git a/crates/data-plane-controller/src/stack.rs b/crates/data-plane-controller/src/stack.rs
@@ -121,4 +121,27 @@ pub struct ControlExports {
     pub gcp_service_account_email: String,
     pub hmac_keys: Vec<String>,
     pub ssh_key: String,
+
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
+pub struct PulumiStackResourceChanges {
+    #[serde(default)]
+    pub same: usize,
+    #[serde(default)]
+    pub update: usize,
+    #[serde(default)]
+    pub delete: usize,
+    #[serde(default)]
+    pub create: usize,
+}
+
+impl PulumiStackResourceChanges {
+    pub fn changed(&self) -> bool {
+        return self.update > 0 || self.delete > 0 || self.create > 0
+    }
+}
+
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PulumiStackHistory {
+    pub resource_changes: PulumiStackResourceChanges,
 }