Coverage for app/services/terraform_executor.py: 98.45%
193 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-25 16:05 +0000
1"""
2Terraform execution utilities with comprehensive structured logging.
4Each long-running command (init, plan, apply, destroy) streams its
5combined stdout/stderr line-by-line through an optional ``output_callback``
6so a higher-level consumer (the deploy task's per-deployment logger) can
7forward each line onto the Celery event bus while the command is still
8running. The full output is also kept in memory and returned at the end
9in the same ``(success, stdout, stderr)`` tuple as before, so existing
10callers keep working unchanged.
11"""
13import contextlib
14import json
15import os
16import subprocess
17import threading
18from collections.abc import Callable
19from typing import Any
21from ..config import settings
22from ..utils.logger import LogCategory, get_logger
24logger = get_logger(__name__)
27# File written next to the cloned repo's terraform/ directory to force the
28# `pg` backend regardless of what the upstream module declares. Terraform
29# treats files ending in `_override.tf` as overrides and replaces the
30# `terraform { backend ... }` block in base config with this one.
31_PG_BACKEND_OVERRIDE_FILENAME = "pg_backend_override.tf"
34def _pg_backend_override_hcl(schema_name: str) -> str:
35 # schema_name is interpolated, not user-controlled (we generate it from
36 # the deployment UUID). Quoted as an HCL string literal.
37 safe = schema_name.replace('"', '\\"')
38 return f'terraform {{\n backend "pg" {{\n schema_name = "{safe}"\n }}\n}}\n'
41OutputCallback = Callable[[str, str], None]
42"""Signature: ``callback(tool_name, line) -> None``.
44Invoked once per line read from the subprocess. ``tool_name`` lets the
45callback distinguish ``terraform_init`` / ``terraform_plan`` / ... when one
46callback is shared across operations.
47"""
50def _stream_subprocess(
51 cmd: list[str],
52 *,
53 cwd: str,
54 env: dict[str, str],
55 timeout: int,
56 tool_name: str,
57 output_callback: OutputCallback | None,
58) -> tuple[int, str, str]:
59 """Run a subprocess and stream its output line-by-line.
61 stdout and stderr are merged onto stdout (``stderr=STDOUT``) so the
62 caller and the live consumer see the lines in the order Terraform
63 intended — Terraform interleaves progress and error messages across
64 the two streams and separating them would scramble the chronology.
65 The returned tuple still exposes the merged output as ``stdout`` and
66 keeps ``stderr`` as an empty string for ABI compatibility.
68 A timeout is enforced via ``Popen.wait(timeout)``; on expiry the
69 process group is killed (children inherit the same group via
70 ``start_new_session``) so terraform's child providers don't survive
71 as orphans.
72 """
73 process = subprocess.Popen(
74 cmd,
75 cwd=cwd,
76 stdout=subprocess.PIPE,
77 stderr=subprocess.STDOUT,
78 text=True,
79 bufsize=1, # line-buffered — without this, output is held until
80 # Terraform fills its 64 KiB pipe buffer
81 env=env,
82 start_new_session=True,
83 )
84 output_lines: list[str] = []
86 def _drain_stdout() -> None:
87 # Read in a thread so the parent can enforce a wall-clock timeout
88 # without blocking on readline.
89 try:
90 assert process.stdout is not None
91 for raw in process.stdout:
92 line = raw.rstrip("\n")
93 output_lines.append(line)
94 if output_callback is not None:
95 with contextlib.suppress(Exception):
96 # Never let a flaky live-stream callback break the
97 # actual deployment; swallow and keep draining.
98 output_callback(tool_name, line)
99 except Exception:
100 pass
102 reader = threading.Thread(target=_drain_stdout, name=f"{tool_name}-reader", daemon=True)
103 reader.start()
105 try:
106 returncode = process.wait(timeout=timeout)
107 except subprocess.TimeoutExpired:
108 # Kill the whole process group — terraform spawns provider plugins
109 # as children and a plain process.kill() would orphan them.
110 with contextlib.suppress(OSError, ProcessLookupError):
111 os.killpg(process.pid, 9)
112 reader.join(timeout=2)
113 return 124, "\n".join(output_lines), "Timeout"
115 reader.join(timeout=5)
116 return returncode, "\n".join(output_lines), ""
119class TerraformExecutor:
120 """Executor for Terraform operations with detailed logging.
122 When ``backend_conn_str`` and ``backend_schema_name`` are provided the
123 executor configures Terraform's ``pg`` backend so state is persisted
124 in a remote Postgres. The conn string lives only in the per-process
125 env (``PG_CONN_STR``), never on the command line.
127 ``output_callback`` is optional; if set, each line of subprocess output
128 is fed to it as it arrives. The deploy task uses this to forward lines
129 onto its per-deployment logger which then ships them to the backend
130 via Celery custom events.
131 """
133 def __init__(
134 self,
135 working_dir: str,
136 env_vars: dict[str, str] | None = None,
137 backend_conn_str: str | None = None,
138 backend_schema_name: str | None = None,
139 output_callback: OutputCallback | None = None,
140 ):
141 self.working_dir = working_dir
142 self.terraform_path = settings.TERRAFORM_PATH
143 self.env_vars = env_vars or {}
144 self.backend_conn_str = backend_conn_str
145 self.backend_schema_name = backend_schema_name
146 self.output_callback = output_callback
148 def _get_env(self, extra_env: dict[str, str] | None = None) -> dict[str, str]:
149 """Get environment variables including OpenStack credentials and Terraform debug logging."""
150 env = os.environ.copy()
151 env.update(self.env_vars)
152 if extra_env:
153 env.update(extra_env)
154 # ``TF_LOG`` is honoured by the terraform CLI and emits an enormous
155 # amount of provider/RPC trace to stderr — useful when debugging
156 # the worker itself, but it drowns the human-readable error block
157 # we forward to the user. Default to off; opt back in via
158 # ``WORKER_TF_LOG`` if you really want it.
159 tf_log = os.environ.get("WORKER_TF_LOG", "")
160 if tf_log:
161 env["TF_LOG"] = tf_log
162 else:
163 env.pop("TF_LOG", None)
164 # PG_CONN_STR is read by Terraform's pg backend. Putting it in env
165 # (not -backend-config="conn_str=...") keeps the password out of
166 # the process listing and command logs.
167 if self.backend_conn_str:
168 env["PG_CONN_STR"] = self.backend_conn_str
169 return env
171 def _write_pg_backend_override(self) -> None:
172 """Write ``pg_backend_override.tf`` so init configures the pg backend.
174 No-op if no schema is configured (legacy local-state mode).
175 """
176 if not self.backend_schema_name:
177 return
178 override_path = os.path.join(self.working_dir, _PG_BACKEND_OVERRIDE_FILENAME)
179 with open(override_path, "w") as f:
180 f.write(_pg_backend_override_hcl(self.backend_schema_name))
181 logger.debug(f"[TF] Wrote pg backend override at {override_path} (schema_name={self.backend_schema_name})")
183 # ------------------------------------------------------------------
184 # Long-running operations (streamed)
185 # ------------------------------------------------------------------
187 def _run_streamed(
188 self,
189 cmd: list[str],
190 *,
191 tool_name: str,
192 timeout: int,
193 ) -> tuple[bool, str, str]:
194 logger.debug(f"[TF] Running command: {' '.join(cmd)}")
195 returncode, stdout, stderr = _stream_subprocess(
196 cmd,
197 cwd=self.working_dir,
198 env=self._get_env(),
199 timeout=timeout,
200 tool_name=tool_name,
201 output_callback=self.output_callback,
202 )
203 return returncode == 0, stdout, stderr
205 def init(self) -> tuple[bool, str, str]:
206 """Initialize Terraform in the working directory.
208 Returns:
209 tuple: (success, stdout, stderr) — stderr merged into stdout
210 """
211 logger.operation_start("terraform_init", working_dir=self.working_dir)
212 try:
213 # Write the backend override BEFORE init runs. If we leave it
214 # to a later step Terraform will already have committed to
215 # whatever backend the upstream module declares (or to ``local``).
216 self._write_pg_backend_override()
218 cmd = [self.terraform_path, "init", "-input=false"]
219 # ``-reconfigure`` makes init idempotent across deploy/update/destroy
220 # runs on the same dir — the schema_name comes from the override
221 # file we just wrote, not from a stale ``.terraform/terraform.tfstate``.
222 if self.backend_schema_name:
223 cmd.append("-reconfigure")
225 success, stdout, stderr = self._run_streamed(cmd, tool_name="terraform_init", timeout=300)
227 if stdout:
228 logger.command_output("terraform_init", stdout, 0 if success else 1)
230 if not success:
231 logger.error(
232 "Terraform init failed",
233 category=LogCategory.ERROR,
234 stderr=stderr[:1000] if stderr else None,
235 )
236 else:
237 logger.success("Terraform init completed", category=LogCategory.STATUS)
239 logger.operation_end("terraform_init", success)
240 return success, stdout, stderr
241 except Exception as e:
242 logger.exception("Terraform init failed with exception", exception=e)
243 logger.operation_end("terraform_init", success=False)
244 return False, "", str(e)
246 def plan(self, var_file: str | None = None, variables: dict[str, Any] | None = None) -> tuple[bool, str, str]:
247 """Run terraform plan."""
248 logger.operation_start("terraform_plan", var_file=var_file, var_count=len(variables or {}))
249 try:
250 # ``-lock=false``: each deployment has its own pg schema (see
251 # ``_tfstate_schema_name`` in the worker), so state is isolated
252 # per deployment. A shared state lock would serialize unrelated
253 # deployments; real concurrency on the same state is already
254 # prevented by the backend.
255 cmd = [self.terraform_path, "plan", "-input=false", "-lock=false"]
256 if var_file:
257 cmd.extend(["-var-file", var_file])
258 if variables:
259 for key, value in variables.items():
260 cmd.extend(["-var", f"{key}={value}"])
262 logger.info("Analyzing Terraform configuration...", category=LogCategory.STATUS)
263 logger.debug("plan variable keys", category=LogCategory.OPERATION, keys=list((variables or {}).keys()))
265 success, stdout, stderr = self._run_streamed(cmd, tool_name="terraform_plan", timeout=300)
267 if stdout:
268 logger.command_output("terraform_plan", stdout, 0 if success else 1)
270 if not success:
271 logger.error("Terraform plan failed", category=LogCategory.ERROR)
272 else:
273 logger.success("Terraform plan completed", category=LogCategory.STATUS)
275 logger.operation_end("terraform_plan", success)
276 return success, stdout, stderr
277 except Exception as e:
278 logger.exception("Terraform plan failed with exception", exception=e)
279 logger.operation_end("terraform_plan", success=False)
280 return False, "", str(e)
282 def apply(
283 self,
284 var_file: str | None = None,
285 variables: dict[str, Any] | None = None,
286 targets: list[str] | None = None,
287 replace: list[str] | None = None,
288 ) -> tuple[bool, str, str]:
289 """Run terraform apply.
291 Args:
292 var_file: Optional ``-var-file`` value.
293 variables: Optional ``-var KEY=VALUE`` map.
294 targets: Optional list of resource addresses to pass via
295 ``-target=…``. Used by the per-VM redeploy task to scope
296 an apply to ONE compute instance — leaves the rest of
297 the deployment untouched. Each entry MUST be a single
298 terraform state address (``type.name[index]``); passing
299 a malformed value would expand into a different CLI
300 flag, so the worker is the line of defense (callers in
301 the backend additionally whitelist against the state).
302 replace: Optional list of resource addresses to pass via
303 ``-replace=…``. Combined with ``targets`` for the
304 redeploy contract: the targeted resource gets
305 explicitly tainted so terraform destroys + recreates
306 it instead of detecting "no changes" and short-
307 circuiting. Same validation rules as ``targets``.
308 """
309 logger.operation_start(
310 "terraform_apply",
311 var_file=var_file,
312 var_count=len(variables or {}),
313 target_count=len(targets or []),
314 replace_count=len(replace or []),
315 )
316 try:
317 # ``-lock=false``: see the rationale in ``plan()`` — the
318 # per-deployment schema isolates state.
319 cmd = [self.terraform_path, "apply", "-auto-approve", "-input=false", "-lock=false"]
320 if var_file:
321 cmd.extend(["-var-file", var_file])
322 if variables:
323 for key, value in variables.items():
324 cmd.extend(["-var", f"{key}={value}"])
325 # Targets / replaces must be a plain list of addresses; we
326 # don't validate the shape here (the caller has already
327 # whitelisted against the cached TF state). Passing them as
328 # separate ``[flag, value]`` pairs to subprocess avoids
329 # shell quoting concerns — double-quotes inside the address
330 # (``team_ide["Team-A"]``) reach Terraform unmodified.
331 for tgt in targets or []:
332 cmd.extend(["-target", tgt])
333 for repl in replace or []:
334 cmd.extend(["-replace", repl])
336 logger.info("Applying Terraform configuration (this may take minutes)...", category=LogCategory.STATUS)
338 success, stdout, stderr = self._run_streamed(cmd, tool_name="terraform_apply", timeout=1800)
340 if stdout:
341 logger.command_output("terraform_apply", stdout, 0 if success else 1)
343 if not success:
344 logger.error("Terraform apply failed", category=LogCategory.ERROR)
345 else:
346 logger.success("Terraform apply completed successfully", category=LogCategory.STATUS)
348 logger.operation_end("terraform_apply", success)
349 return success, stdout, stderr
350 except Exception as e:
351 logger.exception("Terraform apply failed with exception", exception=e)
352 logger.operation_end("terraform_apply", success=False)
353 return False, "", str(e)
355 def destroy(
356 self,
357 var_file: str | None = None,
358 variables: dict[str, Any] | None = None,
359 refresh: bool = True,
360 ) -> tuple[bool, str, str]:
361 """Run terraform destroy.
363 ``refresh=False`` adds ``-refresh=false`` so Terraform tears down
364 purely from state without re-reading data sources. Only used as a
365 targeted fallback by the worker when a stale data source (e.g. a
366 Glance image deleted out-of-band) blocks the refresh-based destroy.
367 """
368 logger.operation_start("terraform_destroy", var_file=var_file, var_count=len(variables or {}))
369 try:
370 # ``-lock=false``: see the rationale in ``plan()`` — the
371 # per-deployment schema isolates state.
372 cmd = [self.terraform_path, "destroy", "-auto-approve", "-input=false", "-lock=false"]
373 if not refresh:
374 cmd.append("-refresh=false")
375 if var_file:
376 cmd.extend(["-var-file", var_file])
377 if variables:
378 for key, value in variables.items():
379 cmd.extend(["-var", f"{key}={value}"])
381 logger.info("Destroying Terraform resources (this may take minutes)...", category=LogCategory.STATUS)
383 success, stdout, stderr = self._run_streamed(cmd, tool_name="terraform_destroy", timeout=1800)
385 if stdout:
386 logger.command_output("terraform_destroy", stdout, 0 if success else 1)
388 if not success:
389 logger.error("Terraform destroy failed", category=LogCategory.ERROR)
390 else:
391 logger.success("Terraform destroy completed successfully", category=LogCategory.STATUS)
393 logger.operation_end("terraform_destroy", success)
394 return success, stdout, stderr
395 except Exception as e:
396 logger.exception("Terraform destroy failed with exception", exception=e)
397 logger.operation_end("terraform_destroy", success=False)
398 return False, "", str(e)
400 # ------------------------------------------------------------------
401 # Short read-only operations (still buffered — fast, no live consumer
402 # waits on these)
403 # ------------------------------------------------------------------
405 def output(self) -> dict[str, Any] | None:
406 """Get terraform outputs as JSON."""
407 logger.operation_start("terraform_output")
408 try:
409 cmd = [self.terraform_path, "output", "-json"]
410 logger.debug(f"[TF] Running command: {' '.join(cmd)}")
411 result = subprocess.run(
412 cmd, cwd=self.working_dir, capture_output=True, text=True, timeout=60, env=self._get_env()
413 )
414 if result.returncode != 0:
415 logger.warning(
416 "Terraform output retrieval failed", category=LogCategory.OPERATION, returncode=result.returncode
417 )
418 logger.operation_end("terraform_output", success=False)
419 return None
421 outputs = json.loads(result.stdout)
422 logger.success(f"Terraform outputs retrieved ({len(outputs)} outputs)", category=LogCategory.STATUS)
423 logger.operation_end("terraform_output", success=True)
424 return outputs
425 except Exception as e:
426 logger.exception("Terraform output extraction failed with exception", exception=e)
427 logger.operation_end("terraform_output", success=False)
428 return None
430 def state_pull(self) -> str | None:
431 """Return the current state JSON via ``terraform state pull``.
433 Works for both the local backend (reads ``terraform.tfstate``) and
434 the pg backend (reads from Postgres). Used by the worker to
435 snapshot state onto the task row for debugging — the canonical
436 copy lives in the pg backend.
437 """
438 try:
439 cmd = [self.terraform_path, "state", "pull"]
440 result = subprocess.run(
441 cmd,
442 cwd=self.working_dir,
443 capture_output=True,
444 text=True,
445 timeout=60,
446 env=self._get_env(),
447 )
448 if result.returncode != 0:
449 logger.warning(
450 "Terraform state pull failed",
451 category=LogCategory.OPERATION,
452 returncode=result.returncode,
453 )
454 return None
455 return result.stdout
456 except Exception as e:
457 logger.warning(f"Terraform state pull raised: {e}", category=LogCategory.OPERATION)
458 return None