AWS AI Automation · Chapter W14

Observability & cost

You can't operate what you can't see. Model-invocation logging, CloudWatch alarms, cost controls, governance, and an incident runbook with a kill switch — O4, in AWS primitives.

⏱️ ~1.5 hours🧪 2 labs🎯 Advanced→Expert

⚙️ To run this for realThe code here is complete and correct as written. To actually execute it you'll need:

AWS credentials (aws configure) + Bedrock model access enabled in your region + pip install boto3
AWS credentials (aws configure) + pip install boto3

Reading and learning works without any of this — run when you're ready. IDs/ARNs in examples are placeholders; swap in your own.

Learning objectives

Enable Bedrock model-invocation logging to CloudWatch/S3.
Build CloudWatch metrics, alarms, and a cost view for AI usage.
Apply governance: least-privilege IAM, data residency, and audit trails.
Assemble the AI incident runbook with a kill switch.

▶ Runnable companionEvery code block in this lesson is also a standalone file under code/aws14-observability-cost-governance/ in the course, with a README on how to run it. Read here, run there — no need to copy-paste.

You can't operate what you can't see advanced

This is the AWS-specific version of your O4 chapter. A live AI system needs the four signals — latency, errors, cost, quality — plus an audit trail of every model call. AWS gives you the primitives; you wire them.

Model-invocation logging advanced

Example code for learning — review, test, and adapt it before running against real or production systems. Commands can create, change, or delete resources. See the Terms & Disclaimer.

Lab W14.1

enable_logging.pyimport boto3
bedrock = boto3.client("bedrock", region_name="us-east-1")

# Log every Bedrock invocation (prompt, response, token counts) to CloudWatch + S3.
bedrock.put_model_invocation_logging_configuration(
    loggingConfig={
        "cloudWatchConfig": {
            "logGroupName": "/bedrock/invocations",
            "roleArn": "arn:aws:iam::123456789012:role/bedrock-logging",
        },
        "s3Config": {"bucketName": "bedrock-logs"},
        "textDataDeliveryEnabled": True,
    }
)
print("invocation logging enabled")

▶ How this works

This turns on an audit trail: it tells AWS Bedrock (Amazon's service for calling AI models) to write down a record of every model call — the prompt sent, the answer returned, and how many tokens were used. Without this, an AI call leaves no trace; with it, you can later ask "what did we send, and what came back?". The code doesn't send any prompts itself — it changes a setting on your account.

boto3 is the official Python library for talking to AWS. boto3.client("bedrock", region_name="us-east-1") opens a connection to the Bedrock control API in the US-East-1 region — the part that manages settings (not the part that runs models). That connection object is named bedrock.
put_model_invocation_logging_configuration(...) is the one command that flips logging on. "Put" here means "set/replace the configuration". Everything inside loggingConfig={...} describes where the logs should go.
cloudWatchConfig sends logs to CloudWatch (AWS's monitoring and log service). logGroupName is the named bucket of logs (/bedrock/invocations), and roleArn is the IAM identity AWS assumes to write them — an ARN is just AWS's unique ID for a resource.
s3Config also drops a copy into an S3 bucket (AWS file storage) called bedrock-logs — handy for long-term retention. textDataDeliveryEnabled: True means the actual prompt and response text is included, not just the numbers.

What the output means: It prints invocation logging enabled. From then on, every Bedrock call your account makes gets logged automatically — you don't add logging code to each call. Nothing is logged retroactively; only calls made after this runs.

Try this: The warning box below is the real lesson: those logs contain the raw prompt and answer text, which can be sensitive. Before enabling this for real, decide who can read the bedrock-logs bucket and how long you keep the data.

Logs contain promptsModel-invocation logs include prompt and completion text — potentially sensitive. Lock the S3 bucket and log group down with least-privilege IAM and a retention policy. This is a governance requirement, not an option.

Metrics, alarms & cost expert

Bedrock publishes CloudWatch metrics (InvocationLatency, Invocations, input/output token counts). Alarm on latency and error spikes; build a cost view from token metrics × price. Set AWS Budgets alerts so runaway spend pages you, not the invoice.

Lab W14.2

alarm.pyimport boto3
cw = boto3.client("cloudwatch", region_name="us-east-1")

cw.put_metric_alarm(
    AlarmName="bedrock-latency-high",
    Namespace="AWS/Bedrock",
    MetricName="InvocationLatency",
    Statistic="Average",
    Period=300, EvaluationPeriods=2,
    Threshold=5000.0,                       # ms
    ComparisonOperator="GreaterThanThreshold",
    AlarmActions=["arn:aws:sns:us-east-1:123456789012:oncall"],
)
print("alarm set")

▶ How this works

This creates a CloudWatch alarm — an automatic watchdog. Bedrock already publishes performance numbers ("metrics") to CloudWatch on its own; here we ask CloudWatch to watch one of those numbers and page a human if it gets too high. The metric we watch is InvocationLatency: how long, in milliseconds, model calls are taking. Slow calls usually mean something is wrong.

cw = boto3.client("cloudwatch", ...) opens a connection to the CloudWatch service. put_metric_alarm(...) creates (or overwrites) an alarm named by AlarmName.
Namespace="AWS/Bedrock" and MetricName="InvocationLatency" together point at the exact number to watch — think of Namespace as the folder and MetricName as the file. Statistic="Average" means look at the average latency (not the max or the total).
Period=300, EvaluationPeriods=2 defines the timing: measure over 300-second (5-minute) windows, and only alarm if two windows in a row breach the limit. Requiring two in a row avoids false alarms from a single blip.
Threshold=5000.0 with ComparisonOperator="GreaterThanThreshold" is the rule: fire when average latency goes above 5000 ms (5 seconds). The # ms comment reminds you the unit is milliseconds.
AlarmActions=[...] is what happens when it fires: it notifies an SNS topic (AWS's message/notification hub) named oncall, which can then email, text, or page whoever is on call.

What the output means: It prints alarm set. The alarm now lives in your account and checks latency continuously — you don't keep this script running. It sits OK until latency crosses the threshold for two periods, then flips to ALARM and fires the notification.

Try this: Lower Threshold to something tiny like 1.0 and you'd get paged almost immediately — a quick way to test that the paging path actually works before you rely on it in a real incident.

Governance & the incident runbook expert

Governance is least-privilege IAM per service, data-residency via region choice, guardrails (W6) as policy, and the invocation log as the audit trail. The incident runbook needs a kill switch: the fastest way to stop all model calls (disable the agent alias, revoke the role, or flip a feature flag) when something goes wrong.

Concern	AWS mechanism
Who can call models	least-privilege IAM policies
Where data lives	region selection + VPC endpoints
What was asked/answered	model-invocation logging
Policy enforcement	Guardrails (W6)
Stop everything now	disable agent alias / revoke role

Exercise W14.1 — Operate your project

Context: Operating an AI service is more than deploying it: you must be able to see it, bound its cost, and stop it. This capstone makes an existing project observable and governable, then proves the alarm actually fires.

Your task: For any AWS AI project you built, enable invocation logging, add a latency alarm and a budget alert, and write a one-page incident runbook that names the kill switch — then trigger the alarm with load and confirm it pages.

Requirements:

Enable Bedrock model-invocation logging to CloudWatch/S3 on the project
Add a CloudWatch latency alarm wired to a real notification (SNS/on-call)
Add an AWS Budgets alert so cost overruns are detected, not discovered on the bill
Write a one-page incident runbook that names a concrete kill switch (disable alias / revoke role / feature flag)
Drive load until the latency alarm breaches and confirm it actually pages
Tie it together: logging attributes the cause, the alarm/budget detects, the runbook stops it

💡 Hint: An alarm you have never seen fire is a guess; generate enough load to cross the threshold and watch the page arrive, then check the runbook's kill switch is one you could actually execute at 2am.

🪜 Practice ladder beginner → industry

Six graded exercises, easy to real-world. Try each before opening its solution.

Exercise 1 · Enable model-invocation loggingBeginner

Context: You cannot audit what you do not record. Bedrock's model-invocation logging captures every prompt, response, and token count to CloudWatch and S3 — the foundational audit trail behind cost, quality, and compliance work.

Your task: Use put_model_invocation_logging_configuration to log Bedrock prompts, responses, and token counts to CloudWatch and S3.

Requirements:

Call put_model_invocation_logging_configuration on a boto3 bedrock client (the control plane, not bedrock-runtime)
Configure a cloudWatchConfig with a log group and the roleArn Bedrock uses to write logs
Configure an s3Config destination bucket for the log data
Enable text data delivery so prompts/responses (not just metadata) are captured
Recognize this is what later powers token/cost accounting and auditing

💡 Hint: This is a one-time account/region configuration on the bedrock control plane; give it both a CloudWatch and an S3 destination plus a role that can write them.

Show solution

Model-invocation logging is the audit trail: it captures prompt, response, and token counts to CloudWatch and S3.

import boto3

bedrock = boto3.client("bedrock", region_name="us-east-1")
bedrock.put_model_invocation_logging_configuration(
    loggingConfig={
        "cloudWatchConfig": {"logGroupName": "/bedrock/invocations",
                             "roleArn": "arn:aws:iam::123:role/bedrock-logs"},
        "s3Config": {"bucketName": "bedrock-logs"},
        "textDataDeliveryEnabled": True,
    },
)
print("logging enabled")

Exercise 2 · Alarm on high latencyIntermediate

Context: A single latency spike shouldn't wake anyone, but sustained slowness should. CloudWatch alarms let you page on InvocationLatency only after it stays high across several evaluation windows, cutting false positives.

Your task: Create a CloudWatch put_metric_alarm on InvocationLatency that fires after two 5-minute windows above 5 seconds.

Requirements:

Call put_metric_alarm on a boto3 cloudwatch client for the AWS/Bedrock namespace, metric InvocationLatency
Set Period=300 (5 min) and EvaluationPeriods=2 so it needs two sustained windows
Set the Threshold in milliseconds (5s = 5000) with ComparisonOperator="GreaterThanThreshold"
Wire AlarmActions to an SNS topic (e.g. on-call) so it actually pages
Understand two evaluation periods suppress single-spike noise

💡 Hint: Latency is reported in milliseconds, so 5s is 5000.0; the two evaluation periods are what turn a blip into an alarm-worthy trend.

Show solution

Two evaluation periods avoid false positives from a single spike; latency is in milliseconds.

import boto3

cw = boto3.client("cloudwatch", region_name="us-east-1")
cw.put_metric_alarm(
    AlarmName="bedrock-latency-high",
    Namespace="AWS/Bedrock", MetricName="InvocationLatency",
    Statistic="Average", Period=300, EvaluationPeriods=2,
    Threshold=5000.0, ComparisonOperator="GreaterThanThreshold",
    AlarmActions=["arn:aws:sns:us-east-1:123:oncall"],
)
print("alarm created")

Exercise 3 · Cost from logged token countsAdvanced

Context: Invocation logs give you exact input/output token counts; model pricing is quoted per 1K tokens. Multiply and sum and you have real spend for a batch — the basis of any cost dashboard or chargeback.

Your task: Given per-invocation input/output token counts pulled from logs, compute the total spend for a batch at a model's prices. Runs offline.

Requirements:

The function takes the log records plus per-1K input and output prices
For each record, cost the input and output sides separately at their own per-1K rates
Convert token counts to thousands before multiplying by the per-1K price
Sum across all records into one total and round sensibly
Demonstrate it offline on a small list of {in, out} records

💡 Hint: Input and output are priced differently, so keep the two rates apart; divide each count by 1000 before multiplying, then add every record's cost.

Show solution

Logs give tokens; multiply by per-1K prices and sum to get spend.

def batch_cost(records, in_per_k, out_per_k):
    total = 0.0
    for r in records:
        total += r["in"]/1000*in_per_k + r["out"]/1000*out_per_k
    return round(total, 6)

logs = [{"in": 1200, "out": 300},
        {"in": 800,  "out": 500},
        {"in": 1500, "out": 200}]
print(batch_cost(logs, 0.003, 0.015))   # 0.0261

Exercise 4 · A four-signal health checkExpert

Context: No single metric tells you if a service is healthy. Operators watch latency, error rate, cost/hour, and a quality score together, and roll them into one status where the worst signal dominates — a familiar red/yellow/green rollup.

Your task: Combine latency, error rate, cost/hour, and a quality score into a single status (green/yellow/red) against thresholds. Runs offline.

Requirements:

Grade each of the four signals independently into green/yellow/red against its own thresholds
Remember quality is inverted — lower is worse — unlike latency, errors, and cost
Compute the overall status as the worst of the four signals
Return both the overall status and the per-signal breakdown so the cause is visible
Demonstrate a green case and a case that goes red on a single bad signal, offline

💡 Hint: The overall status is a max-of-severity: any red makes it red, else any yellow makes it yellow; return the per-signal map so you can see which one tripped.

Show solution

Operate on the four signals together; the worst signal sets the overall status.

def health(latency_ms, error_rate, cost_per_hr, quality):
    checks = {
        "latency": "red" if latency_ms > 5000 else "yellow" if latency_ms > 3000 else "green",
        "errors":  "red" if error_rate > 0.05 else "yellow" if error_rate > 0.01 else "green",
        "cost":    "red" if cost_per_hr > 50 else "yellow" if cost_per_hr > 25 else "green",
        "quality": "red" if quality < 0.7 else "yellow" if quality < 0.85 else "green",
    }
    worst = "red" if "red" in checks.values() else \
            "yellow" if "yellow" in checks.values() else "green"
    return {"status": worst, "signals": checks}

print(health(2000, 0.002, 12, 0.9))     # green
print(health(6000, 0.002, 12, 0.9))     # red (latency)

Exercise 5 · Governance readiness checklistProfessional

Context: Governance turns into concrete controls: who can call (IAM), where data lives (region/VPC), what was asked (logging), what's allowed (guardrails), and how to stop it (kill switch). A gap in any one is a compliance risk, so encode the checklist.

Your task: Encode governance_gaps(controls) that checks the required controls — least-privilege IAM, region/VPC, invocation logging, guardrails, kill switch — and returns the ones that are missing. Runs offline.

Requirements:

Define the set of required controls in one place
Treat a control as satisfied only when its flag is truthy
Return the sorted set difference — the required controls not present/enabled
An empty result means ready; a non-empty result names exactly what to fix
Demonstrate offline with a controls dict where some are False

💡 Hint: Model it as set arithmetic: required minus the ones that are enabled leaves the gaps; sort the result so the report is stable.

Show solution

Each governance question maps to a concrete control; a gap in any one is a compliance risk.

REQUIRED = {"least_privilege_iam", "region_vpc", "invocation_logging",
            "guardrails", "kill_switch"}

def governance_gaps(controls):
    have = {k for k, v in controls.items() if v}
    return sorted(REQUIRED - have)

controls = {"least_privilege_iam": True, "region_vpc": True,
            "invocation_logging": True, "guardrails": False,
            "kill_switch": False}
gaps = governance_gaps(controls)
print(gaps)   # ['guardrails', 'kill_switch']
print("NOT ready" if gaps else "ready")

Exercise 6 · A cost-runaway incident runbook with a kill switchIndustry scenario

Context: Spend spikes 10× at 2am. A mature response is a runbook, not a scramble: detect the breach via a budget/alarm, attribute it to the offending caller from logs, then trip a concrete kill switch to stop the bleeding.

Your task: Build the cost-runaway incident response: detect via budget/alarm breach, identify the offending caller from logs, and trip a kill switch (disable the agent alias / revoke the role / flip a feature flag). Model the decision offline.

Requirements:

Compute how far over budget you are (spend vs. budget) as the trigger
Below budget returns a monitor decision — no action
Over budget selects a concrete kill switch, e.g. disable_agent_alias when one caller dominates the spend, else revoke_role
Attach the follow-up steps (flip the feature flag off, page on-call, review logs)
Demonstrate the 10× over-budget, one-caller case offline

💡 Hint: Attribution drives the switch: if a single caller's share dominates, disabling that alias is the surgical stop; otherwise revoke the role, then do the standard follow-ups.

Show solution

Observability detects, logs attribute, and a concrete kill switch stops the bleeding.

def incident(spend_now, budget, top_caller_share):
    over = spend_now / budget
    if over < 1.0:
        return {"action": "monitor", "over_budget_x": round(over, 2)}
    action = "disable_agent_alias" if top_caller_share > 0.8 else "revoke_role"
    return {"action": action,               # the kill switch
            "over_budget_x": round(over, 2),
            "then": "flip feature flag off, page on-call, review logs"}

print(incident(spend_now=1000, budget=100, top_caller_share=0.95))
# over_budget_x 10.0 -> disable_agent_alias (one caller dominates), then flag off

✓ Checkpoint — you can move on when you can…

Enable and secure Bedrock model-invocation logging.
Create a CloudWatch alarm and a cost/budget alert for AI usage.
Name the AWS mechanism for each governance concern.
Write an incident runbook with a concrete kill switch.

Knowledge check check yourself

✓ Knowledge check

What does Bedrock model-invocation logging capture, and why is enabling it a governance concern rather than a free win?

Show answer

It records every invocation -- prompt, response, and token counts -- to CloudWatch and/or S3, giving you an audit trail (only for calls made after it's enabled). Because those logs contain raw prompt and completion text, which may be sensitive, you must lock down the bucket/log group with least-privilege IAM and set a retention policy.

✓ Knowledge check

The CloudWatch latency alarm uses Period=300 with EvaluationPeriods=2. What behavior does that combination produce, and what is a 'kill switch' in the incident runbook?

Show answer

It measures average InvocationLatency over 5-minute windows and only fires after two consecutive windows breach the threshold, avoiding false alarms from a single blip. A kill switch is the fastest way to stop all model calls in an incident -- e.g. disable the agent alias, revoke the role, or flip a feature flag.

Educational content, provided as-is and without warranty. Code samples are examples — review, test, and adapt them before using in production. See the Terms of Use & Disclaimer. Use at your own risk.

Learning objectives

You can't operate what you can't see advanced

Model-invocation logging advanced

Metrics, alarms & cost expert

Governance & the incident runbook expert

Exercise W14.1 — Operate your project

🪜 Practice ladder beginner → industry

✓ Checkpoint — you can move on when you can…

Knowledge check check yourself

More in AWS AI automation