Detecting Tag Drift with AWS Config Aggregators

The specific bottleneck this page solves is visibility, not evaluation. Tagging Policy Enforcement with AWS Config already covers writing the custom rule that returns NON_COMPLIANT when a resource is missing a required tag key. That rule runs per account, per region, against its own configuration recorder — which means a 40-account organization spread across three regions has 120 independent compliance views, none of which answer the question a FinOps engineer actually asks: how many resources org-wide are missing the cost-center tag right now? A configuration aggregator collects those per-account, per-region evaluations into one queryable dataset, but the query surface — SelectAggregateResourceConfig — is not SQL in any conventional sense. It has no FROM, no JOIN, dot-path field access into nested JSON, array-filtering behavior that returns more than you filtered for, and a hard 100-row page ceiling that forces every non-trivial query into a pagination loop. Get any of those wrong and you either under-report drift (missed pages) or double-count it (duplicate rows from array matches). This page builds the aggregator query correctly and turns it into a paginated drift report.

# Root Cause & Failure Modes

Four mechanics of the aggregator query API cause most broken drift reports:

• Fragmented source data, centralized late. An aggregator does not evaluate resources itself — it replicates the PutEvaluations results and configuration items that source accounts already produced. AWS publishes no fixed SLA for that replication, so a drift report is always “as of the aggregator’s last sync per source,” not “as of now.” A resource remediated five minutes ago in a member account can still show NON_COMPLIANT in the aggregator for a window after the fix.
• No FROM clause, and no error for a wrong field path. The advanced-query dialect is SELECT <fields> WHERE <predicate> over a fixed, implicit resource universe. To reach compliance data at all you filter resourceType = 'AWS::Config::ResourceCompliance' — a virtual resource type — and then dot-path into configuration.complianceType and configuration.configRuleList.configRuleName. Misspell a path and the query returns zero rows silently rather than raising a schema error.
• Array predicates match the parent, not the element. configuration.configRuleList is an array of every rule evaluated against a resource, not just the one you filtered on. Filtering configuration.configRuleList.configRuleName = 'required-tags' returns the whole resource row if any array element matches — including resources evaluated by several rules — so your application code, not the query, is responsible for confirming which rule actually drove the match.
• Limit caps at 100 rows per call. Any organization past a few hundred tagged resources needs NextToken pagination, and that token must survive retries: reissuing a fresh query instead of resuming the token restarts you from page one and can produce a different page boundary if the aggregator resynced between calls.

# Production Pipeline Architecture

The detection sweep is a four-phase read-only pipeline that sits downstream of the aggregator itself, which must already be provisioned as an organization-wide or account-based source under the parent resource tagging and validation pipelines architecture:

1. Query construction. Build the advanced-query Expression string once per rule name, filtering on the AWS::Config::ResourceCompliance virtual type and the target configRuleName. Keep the rule name as a controlled constant, not a value taken from user input — it is interpolated directly into the SQL-like expression.
2. Paginated retrieval. Call select_aggregate_resource_config with Limit=100, following NextToken until the API stops returning one. Wrap each page call in a retry that only fires on throttling and transient service errors — a malformed expression fails identically on every retry and should surface immediately.
3. Row-level confirmation and dedup. For every returned row, re-check configuration.complianceType and the matching entry inside configuration.configRuleList in application code, and key each drift record on account_id:aws_region:resource_id:rule_name to collapse any row that the array-match behavior surfaced more than once.
4. Structured report emission. Materialize the deduplicated records into a JSON drift report with a detection timestamp per record, ready to feed either an alert or the remediation path in Backfilling Tags with boto3 Batch Operations, which consumes exactly this kind of resource list to apply missing tags in batch.

This detection sweep intentionally stays read-only. It answers “what drifted” for a reporting cadence or an alert threshold; it does not write tags itself, which keeps the blast radius of a bad query at zero and lets you run it far more often than any remediation job — the same separation of concerns behind the retry and backoff conventions in Handling Billing API Rate Limits & Retries.

# Step-by-Step Python Implementation

import json
import logging
import os
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Iterator, Optional

import boto3
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")
logger = logging.getLogger(__name__)

AGGREGATOR_NAME = os.getenv("CONFIG_AGGREGATOR_NAME", "org-tag-aggregator")
RULE_NAME = os.getenv("CONFIG_RULE_NAME", "required-tags")  # controlled constant, never user input
PAGE_LIMIT = 100  # hard ceiling enforced by SelectAggregateResourceConfig

RETRYABLE_CODES = {"ThrottlingException", "InternalServerErrorException", "LimitExceededException"}


class RetryableConfigError(Exception):
    """Wraps a ClientError whose error code is safe to retry."""


def _raise_retryable(exc: ClientError) -> None:
    code = exc.response.get("Error", {}).get("Code", "")
    if code in RETRYABLE_CODES:
        raise RetryableConfigError(code) from exc
    raise exc  # malformed expression, auth failure, etc. — never retry these


@dataclass
class DriftRecord:
    resource_id: str
    resource_type: str
    account_id: str
    aws_region: str
    config_rule_name: str
    compliance_type: str
    detected_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

    def dedup_key(self) -> str:
        # Natural key: one drift entry per resource per rule per run, even if
        # the array-match behavior surfaces the parent row more than once.
        return f"{self.account_id}:{self.aws_region}:{self.resource_id}:{self.config_rule_name}"


@retry(
    retry=retry_if_exception_type(RetryableConfigError),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    stop=stop_after_attempt(5),
    reraise=True,
)
def _select_page(client, expression: str, next_token: Optional[str]) -> dict:
    kwargs = {
        "Expression": expression,
        "ConfigurationAggregatorName": AGGREGATOR_NAME,
        "Limit": PAGE_LIMIT,
    }
    if next_token:
        kwargs["NextToken"] = next_token  # resume the same query plan, never restart it
    try:
        return client.select_aggregate_resource_config(**kwargs)
    except ClientError as exc:
        _raise_retryable(exc)


def iter_non_compliant(client, rule_name: str) -> Iterator[DriftRecord]:
    """Yield one DriftRecord per resource currently NON_COMPLIANT on rule_name."""
    expression = (
        "SELECT resourceId, resourceType, accountId, awsRegion, "
        "configuration.complianceType, configuration.configRuleList.configRuleName "
        "WHERE resourceType = 'AWS::Config::ResourceCompliance' "
        "AND configuration.complianceType = 'NON_COMPLIANT' "
        f"AND configuration.configRuleList.configRuleName = '{rule_name}'"
    )
    next_token: Optional[str] = None
    seen: set[str] = set()
    pages = 0
    while True:
        pages += 1
        response = _select_page(client, expression, next_token)
        for raw in response.get("Results", []):
            row = json.loads(raw)
            record = DriftRecord(
                resource_id=row["resourceId"],
                resource_type=row["resourceType"],
                account_id=row["accountId"],
                aws_region=row["awsRegion"],
                config_rule_name=rule_name,
                compliance_type=row["configuration"]["complianceType"],
            )
            key = record.dedup_key()
            if key in seen:
                continue  # collapse the array-match duplicate described above
            seen.add(key)
            yield record
        next_token = response.get("NextToken")
        if not next_token:
            logger.info(
                "Aggregator query exhausted after %d page(s), %d unique drift record(s).",
                pages, len(seen),
            )
            return


def write_drift_report(records: list[DriftRecord], path: str) -> None:
    payload = [record.__dict__ for record in records]
    with open(path, "w") as fh:
        json.dump(payload, fh, indent=2)
    logger.info("Wrote %d drift record(s) to %s", len(payload), path)


def main() -> None:
    client = boto3.client("config")
    logger.info("Querying aggregator %s for rule %s ...", AGGREGATOR_NAME, RULE_NAME)
    records = list(iter_non_compliant(client, RULE_NAME))
    write_drift_report(records, "tag_drift_report.json")
    if records:
        logger.warning("%d resource(s) drifted from the %s policy.", len(records), RULE_NAME)


if __name__ == "__main__":
    main()

The seen set is the load-bearing line: without it, a resource evaluated by three rules where only one matches required-tags can still surface once cleanly, but a resource whose configRuleList contains the target rule listed under a re-evaluation entry will otherwise appear twice in the same page. Deduplicating on the natural key makes the report idempotent regardless of how many times the underlying array matched.

# Verification & Testing

• Cross-check against a single-account rule. Pick one account/region pair, run the native (non-aggregate) get_compliance_details_by_config_rule against it directly, and diff its NON_COMPLIANT resource IDs against the subset of the aggregator report scoped to that account and region. They should match exactly once the aggregator has finished its sync cycle for that source.
• Pagination completeness assertion. In a test account with more than 100 non-compliant resources (or a mocked select_aggregate_resource_config with a synthetic multi-page response), assert len(seen) equals the total row count summed across all mocked pages — a broken NextToken handoff typically shows up as an undercount, not an error.
• Dedup regression test. Feed iter_non_compliant a fixture row whose configRuleList contains the target rule twice (simulating a re-evaluation entry) and assert it yields exactly one DriftRecord.
• Dry-run the expression string. Before wiring a new rule name into production, run the Expression once interactively with a Limit=1 and inspect the raw JSON in Results[0] — a silently-empty result set almost always means a dot-path typo, not zero drift.

# Common Pitfalls Checklist

• Treating resourceType = 'AWS::Config::ResourceCompliance' as optional. Omit it and the query runs against the general resource inventory instead of compliance data, silently returning the wrong schema — always scope the WHERE clause to the compliance virtual type first.
• Restarting the query instead of resuming NextToken. Re-issuing the base Expression on a retry instead of reusing the last NextToken can shift page boundaries if the aggregator resynced between calls, producing gaps or repeats — always thread the token through retries.
• Assuming array-field filters return only the matching element. configuration.configRuleList.configRuleName = 'x' returns the whole resource, rule list and all — confirm the specific match in code, as this module does, rather than trusting the row shape.
• Reporting the aggregator timestamp as real-time. The report reflects each source’s last sync, not the current second — label the detected_at field and any downstream alert with that caveat instead of implying live state.
• Interpolating an unvalidated rule name into the expression. The rule name is embedded directly into the query string with an f-string; source it from a constant or an allow-list, never from unvalidated external input.

# Frequently Asked Questions

# Related

• Tagging Policy Enforcement with AWS Config — the per-account custom rule and PutEvaluations contract that produces the compliance data this page aggregates.
• Resource Tagging & Validation Pipelines — the four-stage tagging architecture this detection sweep sits inside, at the validation stage.
• Backfilling Tags with boto3 Batch Operations — the remediation job that consumes this drift report’s resource list to apply missing tags in batch.
• Enforcing Tag Policies in Terraform Plan Checks — the pre-deployment layer that prevents the drift this page detects from being introduced in the first place.
• Handling Billing API Rate Limits & Retries — the backoff conventions behind the retry decorator used in this module’s _select_page call.

Up: Tagging Policy Enforcement with AWS Config · Home: Cloud Cost Optimization & FinOps Automation