Building a Per-Pull-Request Ephemeral “Cost Guardrail” With Terraform and OPA

I got tired of the same loop in our platform team: a developer opens a pull request, CI spins up preview infrastructure, and—somewhere between “it should cost nothing” and “why is that account on fire?”—we discover the change could trigger expensive cloud behavior. The most annoying part was that we only found out after the preview already existed.

This is a niche pattern I ended up implementing: an ephemeral cost guardrail environment per pull request that runs a Terraform plan, evaluates it with OPA (Open Policy Agent—an engine for declarative policy checks), and then either allows the preview to proceed or blocks it with a precise explanation.

Below is exactly what I built: a small toolchain that

Generates a Terraform plan for a PR branch
Converts that plan into structured input
Runs OPA policies that block “dangerous spend behaviors” (like large instance families or unbounded egress)
Posts a failing status with a human-readable reason

The problem I targeted: “Preview” infrastructure that can silently explode cost

In our environment, PR previews were created from Terraform modules. Some changes were harmless (like updating a container image), but others could accidentally enable expensive features, for example:

Switching an autoscaled group from a cheap instance class to a large one
Enabling public egress paths that multiply network charges
Creating NAT gateways or load balancers in every preview (even if they should be shared)

To stop the loop, I wanted a mechanism that fails fast, deterministically, and with useful error messages.

Architecture (the parts that matter)

Components

Terraform plan step
Produces an execution plan without applying changes.
Plan parsing into JSON
Turns the plan output into structured data so policies can reason about it.
OPA policy evaluation
Uses Rego (OPA’s policy language) to reject plans based on rules.
A PR gating script
Runs the whole thing and exits non-zero when blocked.

I kept the system intentionally small: one executable script, one plan parser, and a few policy rules.

Working implementation

1) Terraform plan in CI (produce JSON plan)

Terraform can produce a JSON representation of a plan using terraform show -json.

Here’s the script I used in CI. It:

runs terraform init and terraform plan
converts the plan to JSON
stores it as plan.json for policy evaluation

#!/usr/bin/env bash
set -euo pipefail

# Assumes Terraform code is already checked out at the PR revision.
WORKDIR="${1:-.}"

cd "$WORKDIR"

terraform init -input=false

# Create a plan file (binary) so we have an exact snapshot.
terraform plan -input=false -no-color -out=plan.tfplan

# Convert binary plan to machine-readable JSON.
terraform show -json plan.tfplan > plan.json

echo "Wrote plan.json"

Why this matters: policies should evaluate a stable artifact. Evaluating the raw text plan is brittle; evaluating JSON is consistent.

2) Convert Terraform plan JSON into a simpler “facts” model

OPA policies are easiest when they don’t have to traverse the full Terraform plan schema.

I wrote a parser that extracts the resource changes and pulls out a few fields we care about (instance type, NAT gateway count, etc.). The parser is intentionally defensive: if a field doesn’t exist, it just won’t be checked.

Create extract_facts.py:

import json
import sys

def extract_changes(plan):
    changes = []
    # Terraform JSON plan structure includes:
    # plan["resource_changes"] for planned changes.
    for rc in plan.get("resource_changes", []):
        addr = rc.get("address")
        rtype = rc.get("type")
        name = rc.get("name")
        actions = rc.get("change", {}).get("actions", [])

        before = rc.get("change", {}).get("before", {}) or {}
        after = rc.get("change", {}).get("after", {}) or {}

        changes.append({
            "address": addr,
            "type": rtype,
            "name": name,
            "actions": actions,
            "before": before,
            "after": after,
        })
    return changes

def main():
    if len(sys.argv) != 3:
        print("Usage: extract_facts.py plan.json facts.json", file=sys.stderr)
        sys.exit(2)

    plan_path = sys.argv[1]
    out_path = sys.argv[2]

    with open(plan_path, "r", encoding="utf-8") as f:
        plan = json.load(f)

    facts = {
        "resource_changes": extract_changes(plan)
    }

    with open(out_path, "w", encoding="utf-8") as f:
        json.dump(facts, f, indent=2)

    print(f"Wrote {out_path}")

if __name__ == "__main__":
    main()

Run it like:

python3 extract_facts.py plan.json facts.json

3) OPA policies that block risky spend behaviors

OPA needs:

a policy file in Rego
an input JSON file (facts.json)
a query that returns violations

I created policy.rego with three rules:

Block instance types above a threshold (example: disallow large families)
Block NAT gateway creation in previews
Block “public load balancer” creation without a shared-by-tag guard (toy example)

Create policy.rego:

package costguard

# Helper: check whether a value matches a prefix.
starts_with(value, prefix) {
  startswith(value, prefix)
}

# Collect all violations into an array.
violations[violation] {
  some c
  c := input.resource_changes[_]

  # Only evaluate creates/updates that affect the "after" state.
  actions := c.actions
  (contains(actions, "create") or contains(actions, "update"))

  # Example: AWS EC2 Launch Template / Auto Scaling resources often contain instance_type.
  # Exact schema differs by provider and resource, so this policy targets common shapes.
  it := c.after.instance_type
  it != ""

  # Example rule: block anything starting with "m6i.16" or "m6i.24" (large spend risk).
  # In real life, you'd probably map to a curated allowlist of instance types.
  (starts_with(it, "m6i.16") or starts_with(it, "m6i.24") or starts_with(it, "m6i.32"))

  violation := {
    "type": "INSTANCE_TYPE_BLOCKED",
    "address": c.address,
    "instance_type": it,
    "message": sprintf("Instance type %v is blocked for PR previews due to cost risk.", [it]),
  }
}

violations[violation] {
  some c
  c := input.resource_changes[_]

  actions := c.actions
  (contains(actions, "create") or contains(actions, "update"))

  # Block NAT gateway creation. Often resource type: "aws_nat_gateway"
  c.type == "aws_nat_gateway"

  # If it exists after, it implies a NAT gateway will be created/kept.
  after_exists := c.after != null
  after_exists

  violation := {
    "type": "NAT_GATEWAY_BLOCKED",
    "address": c.address,
    "message": "NAT gateways are blocked in PR previews to prevent runaway network charges.",
  }
}

violations[violation] {
  some c
  c := input.resource_changes[_]

  actions := c.actions
  (contains(actions, "create") or contains(actions, "update"))

  # Example: block internet-facing load balancers.
  c.type == "aws_lb"

  scheme := c.after.scheme
  scheme == "internet-facing"

  # Toy guard: require a tag "PreviewShared" to be true.
  # If tag missing or false, block.
  tags := c.after.tags
  # tags could be null or missing keys
  not tags.previewshared == "true"

  violation := {
    "type": "PUBLIC_LB_BLOCKED",
    "address": c.address,
    "scheme": scheme,
    "message": "Public load balancers are blocked in PR previews unless explicitly shared via tags.",
  }
}

How the policies work (in plain terms)

Each violations[violation] rule contributes one or more objects.
OPA evaluates all rules and aggregates matches.
The final output is just the list of violations.

4) Run OPA and fail the build with a useful message

Create cost_guard.py. It loads:

facts.json (extracted plan info)
runs OPA with the policy
prints a clear error and exits non-zero if violations exist

import json
import subprocess
import sys
from pathlib import Path

def main():
    if len(sys.argv) != 4:
        print("Usage: cost_guard.py facts.json policy.rego data.json", file=sys.stderr)
        sys.exit(2)

    facts_path = Path(sys.argv[1])
    policy_path = Path(sys.argv[2])
    # data.json is unused in this approach, but kept to match common CI layouts.
    _ = sys.argv[3]

    with facts_path.open("r", encoding="utf-8") as f:
        facts = json.load(f)

    # Write temp input file for OPA.
    input_path = Path("opa_input.json")
    input_path.write_text(json.dumps(facts), encoding="utf-8")

    # Query the violations.
    # We use 'opa eval' to print JSON results.
    cmd = [
        "opa", "eval",
        "--format", "json",
        "--data", str(policy_path),
        "--input", str(input_path),
        "data.costguard.violations"
    ]

    proc = subprocess.run(cmd, capture_output=True, text=True)

    if proc.returncode != 0:
        print(proc.stdout)
        print(proc.stderr, file=sys.stderr)
        sys.exit(proc.returncode)

    # Parse OPA output.
    out = json.loads(proc.stdout)
    # OPA result structure:
    # out["result"][0]["expressions"][0]["value"]
    violations = out["result"][0]["expressions"][0]["value"]

    if violations:
        print("❌ Cost guardrail blocked this PR preview plan:")
        for v in violations:
            print(f" - [{v['type']}] {v['message']} (at {v.get('address','unknown')})")
        sys.exit(1)

    print("✅ Cost guardrail passed. Preview plan allowed.")

if __name__ == "__main__":
    main()

Run it locally:

python3 cost_guard.py facts.json policy.rego /dev/null

5) End-to-end: how CI wires it together

Here’s a simple CI job script I used (adapt paths to your repo). It assumes you’ve installed:

Terraform
opa binary
Python 3

#!/usr/bin/env bash
set -euo pipefail

WORKDIR="${1:-.}"

cd "$WORKDIR"

terraform init -input=false
terraform plan -input=false -no-color -out=plan.tfplan
terraform show -json plan.tfplan > plan.json

python3 extract_facts.py plan.json facts.json

# Run OPA-based guardrails.
python3 cost_guard.py facts.json policy.rego /dev/null

What happens when you run this for a PR:

CI generates a deterministic plan snapshot for the PR branch
Facts extraction reduces the plan to the fields the policies care about
OPA evaluates the policy rules against those facts
If any “cost-risk” conditions match, CI fails before preview infrastructure is created

Practical notes from what I learned building this

1) JSON plans beat text parsing every time

The first time I tried this, I grepped for strings like instance_type = "m6i.24xlarge". It worked until it didn’t. Terraform’s text formatting changed, and the grep either missed changes or false-positive’d.

2) Policies need to handle missing fields gracefully

Terraform plans vary by resource type and by how modules structure attributes. My extractor and policies treat absent fields as “not checked,” which is safer than crashing OPA.

3) Keep policies small and opinionated

The most effective policies were the ones that block a small number of high-cost failure modes (NAT gateways, public LBs, oversized instance types). Everything else can be handled with allowlists and human review later.

Conclusion

I built an ephemeral per-PR “cost guardrail” pipeline by generating a Terraform plan snapshot, extracting key facts into JSON, and using OPA Rego policies to block expensive spend behaviors before any preview infrastructure is applied. The result was a deterministic, debuggable failure mode that turned “surprise costs” into actionable, plan-time feedback—exactly what I needed for mission-critical platform engineering.