Automated Tag Remediation Workflows

A validation verdict does not fix anything by itself — it is a label, not a mutation. Somewhere between the moment a resource is marked NON_COMPLIANT and the moment its tags are actually correct sits an asynchronous correction layer: a queue, a worker, and a set of safety guards that decide whether it is safe to write. This page covers that layer in isolation — the event-driven consumer that drains verdicts from EventBridge or AWS Config into tag:TagResources calls, without racing deployments, without clobbering Terraform state, and without double-applying the same fix twice. It is the remediation and persistence stage of the Resource Tagging & Validation Pipelines architecture, and the discipline here is narrower than validation: given a verdict you already trust, apply the minimum safe mutation, exactly once, with a full audit trail.

# Architecture Context & Data-Flow Position

Remediation is the fourth stage of the four-stage pipeline (acquisition → normalization → allocation → persistence) defined by the parent Resource Tagging & Validation Pipelines reference. By the time a message reaches this layer, discovery has already enumerated the resource and validation has already computed a verdict — either through the real-time evaluation loop in Tagging Policy Enforcement with AWS Config or through a scheduled batch sweep. Remediation’s only job is to turn a trusted NON_COMPLIANT (or policy-violation) verdict into a safe write, and to do that asynchronously so a throttled tagging API or a slow write never blocks the validation path that produced the verdict.

The decoupling is deliberate. EventBridge (or an AWS Config compliance-change notification) publishes the verdict; an SQS queue absorbs the burst and gives the consumer backpressure control; a worker pool pulls messages at a rate the tagging API can actually sustain. This buys three properties that a direct Lambda-to-TagResources call cannot: retry without re-evaluating the verdict, natural batching for the tagging API’s per-call resource limits, and a dead-letter path for anything that fails repeatedly. The same fault-tolerant consumer shape — bounded concurrency, exponential backoff, explicit acknowledgment — is the one used for building fault-tolerant billing ingestion with Celery; the transport differs (SQS vs. a broker-backed task queue) but the reliability contract is identical.

Not every verdict should become a tag write. A production remediation layer routes each message to one of four actions, gated by different triggers and different safety guards:

# Core Implementation Patterns

# 1. Least-privilege IAM for the remediation role

The remediation worker’s IAM role is the single largest blast-radius risk in the whole pipeline — it is the one role in the system with write access to production tags. Scope it tightly: read access for the pre-write check, write access for the actual mutation, and nothing else.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ConsumeRemediationQueue",
      "Effect": "Allow",
      "Action": ["sqs:ReceiveMessage", "sqs:DeleteMessage", "sqs:GetQueueAttributes"],
      "Resource": "arn:aws:sqs:*:*:tag-remediation-queue"
    },
    {
      "Sid": "ReadCurrentTagsBeforeWrite",
      "Effect": "Allow",
      "Action": ["tag:GetResources"],
      "Resource": "*"
    },
    {
      "Sid": "ApplyMissingTagsOnly",
      "Effect": "Allow",
      "Action": ["tag:TagResources"],
      "Resource": "*",
      "Condition": {
        "StringEquals": {"aws:RequestedRegion": ["us-east-1", "eu-west-1"]}
      }
    },
    {
      "Sid": "NoDeletePathByDefault",
      "Effect": "Deny",
      "Action": ["tag:UntagResources"],
      "Resource": "*"
    }
  ]
}

tag:UntagResources is denied explicitly rather than simply omitted — an explicit deny survives a future policy merge that accidentally grants it elsewhere. If your remediation program does need a delete path (removing a deprecated tag key across an estate), issue that from a separate, more tightly audited role that requires a change ticket, never from the same role that runs the always-on apply-missing loop. Backfilling absent tags at estate scale, as opposed to reactive per-message remediation, is a distinct batch workload covered in Backfilling Tags with boto3 Batch Operations.

# 2. Read-before-write and never overwriting IaC or human values

The single rule that makes remediation safe to run unattended is this: always read current tag state immediately before mutating, and apply only the keys that are actually absent. A verdict computed a few seconds or a few minutes ago can be stale — an operator may have fixed the resource by hand, or a Terraform apply may have landed in between. Blind-applying the verdict’s fix list risks stomping a value someone just set correctly to something else.

current = tagging_client.get_resources(ResourceARNList=[resource_arn])
existing_keys = {
    tag["Key"]
    for mapping in current.get("ResourceTagMappingList", [])
    for tag in mapping.get("Tags", [])
}
to_apply = {k: v for k, v in proposed_tags.items() if k not in existing_keys}
if not to_apply:
    return "noop"  # someone already fixed it; do not re-write

This is also why remediation must never treat “resource carries a Terraform-managed marker tag” as license to overwrite everything else — a resource can be under IaC management for some keys and still missing a cost_center that only the remediation layer will ever supply. The full decision tree for distinguishing an IaC-owned key from a remediation-owned key, including how to detect drift the next terraform plan would otherwise fight, is worked through in Safe Tag Remediation Without Overwriting IaC State.

# 3. Content-addressed idempotency keys

Every message the worker processes must resolve to the same idempotency key on every replay, so an at-least-once queue (SQS’s default delivery guarantee) never produces a double-apply. The key is a hash of the resource identity, the schema version, and the sorted set of proposed changes — the same construction used by the upstream validation engine’s remediation_key in the parent pipeline.

def idempotency_key(resource_arn: str, schema_version: str, proposed: dict) -> str:
    canonical = f"{resource_arn}|{schema_version}|{'|'.join(sorted(proposed))}"
    return hashlib.sha256(canonical.encode()).hexdigest()[:32]

Store applied keys in a small dedup table (DynamoDB with a TTL is the usual choice) and check it before attempting a write. If the key was already applied, the message is a no-op — this is what lets you safely redrive a dead-letter queue or replay an EventBridge archive without fear of re-applying a fix that already succeeded.

# 4. Change-window awareness and backoff

A remediation write during an active deployment is worse than a delayed one — it creates a race with the deployment’s own tag writes and a drift signal the next terraform plan or CloudFormation drift-detect run will flag as unexplained. The worker checks a change-window signal (a deployment-lock table, a CI pipeline’s in-flight-deployments API, or a simple TTL key written by the CI/CD system on apply start) before writing, and requeues with a visibility-timeout delay if a window is active rather than dropping the message.

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    retry=retry_if_exception_type(ClientError),
)
def apply_tags(client, resource_arn: str, tags: dict) -> None:
    client.tag_resources(ResourceARNList=[resource_arn], Tags=tags)

Exponential backoff with a ceiling matters here for a second reason beyond throttling: it naturally spaces out retries so a resource caught mid-deployment on attempt one is very likely past its change window by attempt three or four, without the worker needing to poll the deployment system directly.

# Production-Grade Python Remediation Worker

The module below is a complete SQS consumer. It pulls remediation messages, checks the idempotency dedup store, evaluates the change window, reads current tags, routes to one of the four actions from the table above, applies the mutation with tenacity backoff, writes an audit record, and dead-letters anything that exhausts its retry budget. A --dry-run flag lets you replay a queue’s worth of messages and see exactly what would be written without mutating anything.

import argparse
import hashlib
import json
import logging
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional

import boto3
from botocore.config import Config as BotoConfig
from botocore.exceptions import ClientError
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential,
)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%SZ",
)
logger = logging.getLogger("finops.tag_remediation_worker")

_BOTO_CONFIG = BotoConfig(retries={"max_attempts": 3, "mode": "adaptive"})

QUEUE_URL = "https://sqs.us-east-1.amazonaws.com/123456789012/tag-remediation-queue"
DLQ_URL = "https://sqs.us-east-1.amazonaws.com/123456789012/tag-remediation-dlq"
AUDIT_TABLE = "tag-remediation-audit"
DEDUP_TABLE = "tag-remediation-dedup"
MAX_MUTATION_ATTEMPTS = 5


@dataclass
class RemediationMessage:
    """One verdict pulled off the queue, ready to be actioned."""

    resource_arn: str
    account_id: str
    schema_version: str
    proposed_tags: Dict[str, str]
    violation_type: str  # missing_tags | forbidden_value | restricted_classification
    receipt_handle: str

    @classmethod
    def from_sqs(cls, record: Dict[str, Any]) -> "RemediationMessage":
        body = json.loads(record["Body"])
        return cls(
            resource_arn=body["resource_arn"],
            account_id=body["account_id"],
            schema_version=body["schema_version"],
            proposed_tags=body["proposed_tags"],
            violation_type=body.get("violation_type", "missing_tags"),
            receipt_handle=record["ReceiptHandle"],
        )


@dataclass
class RemediationResult:
    action: str  # apply-missing | quarantine | notify | block
    resource_arn: str
    applied_tags: Dict[str, str] = field(default_factory=dict)
    reason: str = ""
    idempotency_key: str = ""
    evaluated_at: str = ""

    def __post_init__(self) -> None:
        if not self.evaluated_at:
            self.evaluated_at = datetime.now(timezone.utc).isoformat()


def idempotency_key(resource_arn: str, schema_version: str, proposed: Dict[str, str]) -> str:
    """Hash resource identity + schema version + sorted proposed keys so a
    replayed message always resolves to the same key. Content, not the
    message envelope, determines identity — this is what makes at-least-once
    SQS delivery safe to consume without a double-apply."""
    canonical = f"{resource_arn}|{schema_version}|{'|'.join(sorted(proposed))}"
    return hashlib.sha256(canonical.encode()).hexdigest()[:32]


class RemediationWorker:
    def __init__(self, dry_run: bool = False) -> None:
        self.dry_run = dry_run
        self.sqs = boto3.client("sqs", config=_BOTO_CONFIG)
        self.tagging = boto3.client("resourcegroupstaggingapi", config=_BOTO_CONFIG)
        self.dynamodb = boto3.resource("dynamodb", config=_BOTO_CONFIG)
        self.dedup = self.dynamodb.Table(DEDUP_TABLE)
        self.audit = self.dynamodb.Table(AUDIT_TABLE)

    def already_applied(self, key: str) -> bool:
        resp = self.dedup.get_item(Key={"idempotency_key": key})
        return "Item" in resp

    def mark_applied(self, key: str, resource_arn: str) -> None:
        if self.dry_run:
            return
        self.dedup.put_item(
            Item={
                "idempotency_key": key,
                "resource_arn": resource_arn,
                "applied_at": datetime.now(timezone.utc).isoformat(),
                "ttl": int(time.time()) + 30 * 86400,  # 30-day dedup window
            }
        )

    def in_change_window(self, resource_arn: str) -> bool:
        """Deployment-lock check. In production this queries a table written
        by the CI/CD system's apply-start / apply-end hooks. Fail safe: if the
        lock table is unreachable, assume a window is active rather than risk
        racing a live deployment."""
        try:
            locks = self.dynamodb.Table("deployment-locks")
            resp = locks.get_item(Key={"resource_arn": resource_arn})
            return "Item" in resp
        except ClientError as exc:
            logger.warning("Lock check failed for %s, assuming active window: %s", resource_arn, exc)
            return True

    def current_tags(self, resource_arn: str) -> set:
        resp = self.tagging.get_resources(ResourceARNList=[resource_arn])
        return {
            tag["Key"]
            for mapping in resp.get("ResourceTagMappingList", [])
            for tag in mapping.get("Tags", [])
        }

    @retry(
        stop=stop_after_attempt(MAX_MUTATION_ATTEMPTS),
        wait=wait_exponential(multiplier=1, min=2, max=60),
        retry=retry_if_exception_type(ClientError),
        reraise=True,
    )
    def _write_tags(self, resource_arn: str, tags: Dict[str, str]) -> None:
        self.tagging.tag_resources(ResourceARNList=[resource_arn], Tags=tags)

    def route(self, msg: RemediationMessage) -> RemediationResult:
        key = idempotency_key(msg.resource_arn, msg.schema_version, msg.proposed_tags)

        if self.already_applied(key):
            return RemediationResult("apply-missing", msg.resource_arn, {}, "noop: already applied", key)

        if msg.violation_type in ("forbidden_value", "restricted_classification"):
            return RemediationResult("quarantine", msg.resource_arn, {}, f"quarantined: {msg.violation_type}", key)

        if self.in_change_window(msg.resource_arn):
            return RemediationResult("block", msg.resource_arn, {}, "deployment window active, requeued", key)

        existing = self.current_tags(msg.resource_arn)
        to_apply = {k: v for k, v in msg.proposed_tags.items() if k not in existing}
        if not to_apply:
            self.mark_applied(key, msg.resource_arn)
            return RemediationResult("apply-missing", msg.resource_arn, {}, "noop: no gap remaining", key)

        if not to_apply.get("owner") and "owner" not in existing:
            return RemediationResult("notify", msg.resource_arn, {}, "owner unresolved, notifying only", key)

        if not self.dry_run:
            self._write_tags(msg.resource_arn, to_apply)
        self.mark_applied(key, msg.resource_arn)
        return RemediationResult("apply-missing", msg.resource_arn, to_apply, "applied missing keys", key)

    def record_audit(self, result: RemediationResult) -> None:
        logger.info(json.dumps({"event": "remediation_outcome", **result.__dict__}))
        if self.dry_run:
            return
        self.audit.put_item(Item={**result.__dict__, "audit_id": f"{result.idempotency_key}:{result.evaluated_at}"})

    def dead_letter(self, record: Dict[str, Any], error: str) -> None:
        self.sqs.send_message(
            QueueUrl=DLQ_URL,
            MessageBody=record["Body"],
            MessageAttributes={"failure_reason": {"DataType": "String", "StringValue": error[:256]}},
        )
        logger.error("Dead-lettered message after exhausted retries: %s", error)

    def poll(self, max_messages: int = 10, wait_seconds: int = 15) -> int:
        resp = self.sqs.receive_message(
            QueueUrl=QUEUE_URL,
            MaxNumberOfMessages=max_messages,
            WaitTimeSeconds=wait_seconds,  # long polling, avoids empty-receive cost
            MessageAttributeNames=["All"],
        )
        records = resp.get("Messages", [])
        processed = 0
        for record in records:
            try:
                msg = RemediationMessage.from_sqs(record)
                result = self.route(msg)
                self.record_audit(result)
                if not self.dry_run:
                    self.sqs.delete_message(QueueUrl=QUEUE_URL, ReceiptHandle=msg.receipt_handle)
                processed += 1
            except ClientError as exc:
                self.dead_letter(record, str(exc))
            except Exception as exc:  # noqa: BLE001 — never let a bad message wedge the poll loop
                self.dead_letter(record, f"unexpected: {exc}")
        return processed


def main() -> None:
    parser = argparse.ArgumentParser(description="Tag remediation SQS worker")
    parser.add_argument("--dry-run", action="store_true", help="Route and log without mutating or acking")
    parser.add_argument("--once", action="store_true", help="Poll a single batch and exit (for cron/local testing)")
    args = parser.parse_args()

    worker = RemediationWorker(dry_run=args.dry_run)
    logger.info("Starting remediation worker (dry_run=%s)", args.dry_run)

    while True:
        count = worker.poll()
        logger.info("Processed %d message(s)", count)
        if args.once:
            break


if __name__ == "__main__":
    main()

The design choices that matter in production: route() is pure decision logic separated from _write_tags()'s side effect, so the routing rules are unit-testable without a live AWS account; the dedup check happens before the change-window check so a replayed already-applied message never even queries the lock table; and dead_letter() preserves the original message body plus a failure reason, so triaging the DLQ tells you exactly what to fix without re-deriving the verdict from scratch.

# Schema Reference Table

The remediation message schema is deliberately narrow — it carries only what the worker needs to make a safe decision, not the full validation context.

# Operational Considerations

• Tagging API rate limits. TagResources and GetResources on the Resource Groups Tagging API are throttled per account; treat sustained throughput above roughly 5–10 requests/second per account as a reliable trigger for ThrottlingException under concurrent load, and cap worker concurrency per account accordingly rather than relying solely on SDK-level adaptive retry.
• SQS visibility timeout. Set the queue’s visibility timeout to at least 6x the worker’s expected processing time (including the read-before-write round trip and retry backoff) — a timeout that expires mid-processing causes a second worker to pick up the same message before the first one acks, which is a correctness bug even with idempotency keys in place, since two concurrent writers can still race the read-then-write window.
• Dead-letter queue redrive policy. Set maxReceiveCount to 5 on the source queue so a poison message (malformed JSON, a resource that no longer exists) does not loop indefinitely; redrive the DLQ manually after a fix, never automatically, since automatic redrive of a systemic failure just re-floods the same error.
• Change-window lock staleness. A deployment-lock entry that is never cleared (a CI job crashes before its apply-end hook fires) permanently blocks remediation for that resource. Attach a short TTL (30–60 minutes) to lock entries so a crashed deployment does not create a silent, permanent remediation freeze.
• Audit retention. Keep the audit table’s full outcome history — including noop and block — for at least the length of your compliance reporting cycle; a block history is what proves the worker respected change windows during an incident review, not just what it eventually applied.
• Monitoring hooks. Alarm on DLQ depth greater than zero, on a rising ratio of block/quarantine outcomes to apply-missing (a signal that upstream validation is producing more edge cases than the routing table handles), and on ApproximateAgeOfOldestMessage on the primary queue climbing past your SLA for time-to-remediate.

# Troubleshooting

Remediation queue backs up and ApproximateAgeOfOldestMessage keeps climbing. Root cause: worker concurrency is capped below the arrival rate, or every message is hitting the change-window lock-table fail-safe and looping through requeue delays. Detection: CloudWatch shows steady message arrival but flat or falling NumberOfMessagesDeleted. Remediation: scale worker concurrency up to the tagging API’s sustainable rate first, then check whether the lock table itself is throttled or unreachable, which would explain every message failing safe into a block.

The same resource gets tagged twice in the audit log with different idempotency_key values. Root cause: the schema_version changed between the two verdicts, so the content-addressed key legitimately differs — this is by design, not a bug, but it means a schema bump can look like duplicate remediation in a naive audit query. Detection: two audit records for the same resource_arn with different schema_version. Remediation: group audit queries by resource_arn and treat a schema_version change as an expected re-remediation event, not drift.

Worker applies tags that Terraform reverts on the next apply. Root cause: the proposed tag key is actually managed by IaC and just hadn’t been applied yet when validation ran, so remediation and the pending terraform apply raced. Detection: the audit log shows an apply-missing outcome immediately followed by a Config or CloudTrail event reverting the same key. Remediation: extend the change-window check to cover “IaC apply queued” states, not just “in progress,” and see the drift-avoidance patterns in Safe Tag Remediation Without Overwriting IaC State.

Dead-letter queue fills with AccessDenied errors from one specific account. Root cause: the remediation role’s cross-account assume-role trust policy was not updated when a new account joined the organization. Detection: DLQ messages all share one account_id and the same ClientError code. Remediation: confirm the target account’s trust policy permits the central remediation role, then redrive the DLQ once the fix is confirmed with a --dry-run pass.

notify outcomes never actually reach an owner. Root cause: the notify action logs and audits correctly but the webhook/Slack integration silently fails, and because no tag mutation occurs there is no downstream signal (like a reverted tag) to surface the gap. Detection: audit table shows a steady stream of notify outcomes with no corresponding drop in NON_COMPLIANT counts over subsequent sweeps. Remediation: alarm on notify outcome volume with no compliance-rate improvement over N sweeps, and add a delivery-confirmation check to the notification call itself rather than treating queueing the notification as success.

# Frequently Asked Questions

# Related

• Resource Tagging & Validation Pipelines — the parent reference defining the four-stage pipeline this worker implements at the remediation and persistence stage.
• Tagging Policy Enforcement with AWS Config — the validation layer whose NON_COMPLIANT verdicts feed this queue.
• Backfilling Tags with boto3 Batch Operations — the estate-wide batch counterpart to this per-message reactive worker.
• Safe Tag Remediation Without Overwriting IaC State — the deeper decision tree behind the read-before-write and change-window guards used here.
• Building Fault-Tolerant Billing Ingestion with Celery — the analogous asynchronous consumer pattern (bounded concurrency, backoff, dead-lettering) applied to billing ingestion instead of tag remediation.

Up: Resource Tagging & Validation Pipelines