DEV Community: Takashi Iwamoto

Test your Step Functions workflows locally with pytest

Takashi Iwamoto — Wed, 23 Sep 2026 13:47:08 +0000

Before you deploy a Step Functions workflow, it is hard to check what happens when a service fails: which errors get retried, which Catch handles them, and what the next Task receives. AWS documents Step Functions Local as unsupported. The recommended alternative, the TestState API, supports mocks but tests one state per call. To test a complete path, you have to chain those calls yourself.

sfnx includes a local runner for testing an entire workflow with pytest. You provide mock service responses, the runner follows the definition's Retry and Catch rules, and your assertions check the calls and the result. It works with existing JSONata definitions, so there's no need to rewrite the workflow or add states to it.

sfnx also has a compiler if you'd prefer to maintain the workflow itself in Python. I'll cover that option in the second half.

A test can run the ASL you already have

The runner reads the same JSON that Step Functions runs. A workflow is a JSON document in the Amazon States Language (ASL), whether you write it by hand or build it in Workflow Studio, the console's visual editor. In JSONata mode, fields such as a Task's Arguments can hold expressions inside {% ... %}.

For this example, I'll use AWS's coding-agent sample. It looks up a clinical term with Lambda, calls an AI agent if there's no direct match, routes the answer by confidence, and writes back the result. The agent runs in an Amazon Bedrock AgentCore harness, which Step Functions can invoke directly as a Task.

Each Task has a Retry policy for transient errors. If the lookup, agent call or initial write-back fails with an error handled by its Catch, the workflow moves to a fallback Task that marks the term as open and records the reason. If the fallback also fails after its retries, the execution fails.

The test below asks one question: if the agent is throttled on every attempt, how many times is it called, and what does the fallback Task receive?

Set up a project

You need uv and Python 3.11 or later. In a new directory:

uv init
uv add --dev "sfnx[testing]==2.1.0" pytest

These examples were tested with sfnx v2.1.0. Omit ==2.1.0 to install the latest version instead.

Save the JSON form of the sample's definition from sfnx v2.1.0 as coding-workflow.asl.json in the project root. It is AWS's definition, converted from the sample's YAML, with nothing added.

Write the test

One mock function handles the lookup, agent call and write-back. Save this test as test_coding_workflow.py in the project root:

import json
from pathlib import Path

from sfnx.testing import Call, Failure, run

DEFINITION = json.loads(Path("coding-workflow.asl.json").read_text())
TERM = {
    "record_id": "r1",
    "verbatim": "migrane",
    "encoding_dictionary": "MedDRA",
    "encoding_dictionary_version": "v27.0",
    "source_study": "ONCO-2024-01",
}
HARNESS = "arn:aws:states:::bedrockagentcore:invokeHarness"


def throttled_agent(call: Call) -> object:
    """No direct match, the agent is throttled, and the write-back succeeds."""
    if call.resource == HARNESS:
        raise Failure("BedrockAgentCore.ThrottlingException", "slow down")
    assert isinstance(call.arguments, dict)
    if call.arguments["FunctionName"] == "${CheckDirectFunctionArn}":
        return {"Payload": {"blocked": False, "matched": False}}
    return {"Payload": call.arguments["Payload"]}


def test_a_throttled_agent_is_retried_then_the_term_is_left_open():
    execution = run(DEFINITION, TERM, throttled_agent)
    harness_calls = [c for c in execution.calls if c.resource == HARNESS]
    assert len(harness_calls) == 4  # the first attempt and 3 retries
    last = execution.calls[-1].arguments
    assert last["Payload"]["target_status"] == "open"
    assert last["Payload"]["failure_reason"] == "BedrockAgentCore.ThrottlingException"
    assert execution.output == {
        "record_id": "r1",
        "target_status": "open",
        "failure_reason": "BedrockAgentCore.ThrottlingException",
        "rationale": None,
    }

Run it from the project root:

uv run pytest test_coding_workflow.py

What the runner did

This test runs locally without AWS credentials or calls to AWS services. run follows the definition state by state. At each Task, it passes the Resource and evaluated Arguments to throttled_agent and uses the return value as the service response.

When the mock raises Failure, the runner applies the definition's error handling. Here, the agent gets one initial attempt and three retries before the Catch routes execution to the fallback Task.

The assertions check the four agent attempts, the fallback Task's arguments, and the final output. If the execution fails, accessing execution.output raises the error, which also fails the test.

${...} placeholders are left unchanged. The mock receives ${CheckDirectFunctionArn} as written, so you can test the definition before substituting the real ARNs.

The runner skips delays: retries don't wait between attempts, and Wait states return immediately. Testing a retry policy doesn't mean waiting through its backoff intervals.

Testing a definition from Workflow Studio

To test your own JSONata workflow, export its definition from the console as JSON. Use each Task's Resource and Arguments to decide what your mock should return, or raise Failure to test an error path. Then add assertions for the calls and output you expect.

I also tested a small JSONata workflow built in Workflow Studio and exported without edits. It had a Lambda call with the retry policy the editor adds, a Choice, Succeed and Fail states, and a Catch.

It followed the expected path in all four cases I tried: success, the Choice's other branch, a throttled call caught after retries, and an error caught without retries. I haven't tested every state type or integration available in the editor.

The runner checks for unsupported states and fields before execution. It accepts JSONata mode only; a JSONPath state or an unsupported field produces an error identifying the state. See the testing guide for the supported states and fields.

A local run checks control flow against your mocks, not the services. It does not reproduce elapsed time, concurrency, real service behavior, or every detail of JSONata on AWS. The testing guide lists where a local run differs.

If you want to maintain the workflow in Python, sfnx can compile it

The compiler is optional; nothing above used it. You write the workflow's intent in Python, sfnx compiles it to readable ASL in JSONata mode, and you deploy the generated definition with your usual tools.

Here's how the same coding-agent workflow looks in Python. You can find the full source and the generated definition in the repository.

Define a shared retry policy once

The original definition repeats the same Lambda retry policy on four Tasks. In Python, it is one module-level constant (excerpt; the full constant has a second retrier for timeouts and database errors):

LAMBDA_RETRY = [
    {
        "ErrorEquals": [Lambda.TooManyRequestsException, Lambda.ServiceException],
        "IntervalSeconds": 2,
        "MaxAttempts": 3,
        "BackoffRate": 2.0,
    },
    ...
]

Each Lambda Task uses the same constant:

direct = task(
    "arn:aws:states:::lambda:invoke",
    {
        "FunctionName": "${CheckDirectFunctionArn}",
        "Payload": {
            "verbatim": verbatim,
            "encoding_dictionary": dictionary,
            "encoding_dictionary_version": version,
        },
    },
    retry=LAMBDA_RETRY,
)["Payload"]

To change the backoff, you edit the constant once. The compiler includes the full policy in each Task that uses it.

Write branches with if/elif

The original uses a Choice state to select the next Task based on the agent's score:

"ScoreThreshold": {
  "Type": "Choice",
  "Choices": [
    { "Condition": "{% $agentScore >= 0.9 %}", "Next": "WriteBack" },
    { "Condition": "{% $agentScore >= 0.7 %}", "Next": "MarkForReview" }
  ],
  "Default": "MarkOpen"
}

In Python, the same decision is an if / elif / else:

if score >= 0.9:
    status = "autocoded"
elif score >= 0.7:
    status = "approval_required"
else:
    status = "open"

This compiles to a single Choice state. Each matching rule sets the status variable; the default path sets it to open:

"if_3": {
  "Type": "Choice",
  "Choices": [
    {
      "Condition": "{% $score >= 0.9 %}",
      "Assign": { "status": "autocoded" },
      "Next": "if_4"
    },
    {
      "Condition": "{% $score >= 0.7 %}",
      "Assign": { "status": "approval_required" },
      "Next": "if_4"
    }
  ],
  "Assign": { "status": "open" },
  "Default": "if_4"
}

The Python version's graph has a different shape from the original: it uses one write-back Task for the accepted and review cases, passing status as target_status.

Handle errors with try/except: only Task, Parallel and Map states catch

There is an important difference from Python's try / except: the compiler adds Catch clauses to Task, Parallel and Map states in the try block, but Pass states cannot catch errors.

The example therefore parses the agent's reply in the same statement as the task() call. If parsing fails, the Task's Catch handles the error. Moving the parsing to a separate statement would put it in a Pass state, where a parsing error would fail the entire execution. The language reference covers the details.

Stay within the supported Python subset

The compiler supports a subset of Python, so arbitrary library calls won't compile. Unsupported constructs produce an error with a suggested alternative. You can also use jsonata() to write an expression directly in JSONata.

Deployment stays with your existing tools. Resources, IAM and substitutions go through CDK, SAM, CloudFormation or the AWS CLI; the deployment guide has examples.

The Python version behaves like the original in 12 mocked scenarios

To check the Python version against the original, this test runs both definitions through 12 scenarios with the same mocks. It uses AWS's definition from commit ab76ef3 (MIT-0) and compares every call's resource and arguments, along with the final output or error.

The scenarios cover direct matches, blocked terms, each confidence band, replies without a score, a rationale or JSON, and each service failing. To reproduce them:

git clone https://github.com/iwamot/sfnx.git
cd sfnx
git checkout v2.1.0
uv run pytest tests/test_samples.py

Both definitions produce the same calls and results for these scenarios in the local runner. That doesn't establish that they behave identically on AWS with real Lambda and AgentCore responses.

The compiler's trade-off: it can add states

In this example, the compiled workflow has more states than the original. This affects workflows you compile with sfnx; testing existing ASL with the runner adds no states. The compiler emits Choice states for branches and separate states for assignments it cannot combine with another state.

Some assignments are combined with an existing state: a Wait can hold the assignments after it, a Task can assign its result, and a Choice rule can set a variable. The compiler keeps states separate when combining them could change the behavior.

Coding agent	AWS original	Python version
States in the definition	8	17
States entered, agent confident	6	12
States entered, lookup fails	3	7
States entered, 12 scenarios	3 to 6	6 to 13

Across the 12 scenarios, the Python version enters 1.5 to 2.3 times as many states on the same path.

The 12 scenarios

Scenario	AWS original	Python version
blocked	4	7
direct match	4	6
agent confident / unsure	6	12
agent doubtful	6	13
agent without a score / without a rationale	6	13
agent without JSON / with broken JSON	5	10
lookup fails	3	7
agent fails	5	10
write-back fails	5	9

These are local execution counts, not billing estimates. The runner records each state entry but excludes retry attempts from this count. Standard workflows are billed per state transition, including retries.

What you pay depends on the path each execution takes and how many executions you run. For a Standard workflow that runs often, look at the generated graph before you adopt the compiler.

For a smaller example, see the Python version of the console's Hello World template. It shows how if, wait(), parallel() and raise compile. The generated definition has 11 states compared with the template's 9; a successful run enters 9 states compared with 8.

Start with one failure path

Try adding a test for one failure path in a JSONata workflow you already have. Export the definition, mock its Tasks, and check that the workflow handles the error and passes the expected data to the next Task.

If you'd also like to maintain the workflow in Python, start with a small example and inspect the generated ASL before deciding whether to use the compiler.

sfnx is on GitHub. If you try it on a real workflow, I'd like to hear which paths you tested, and where the runner fell short.

Connect Your Bedrock AgentCore Agent to Slack — Without Writing the Slack Part

Takashi Iwamoto — Sun, 12 Jul 2026 14:16:03 +0000

Want to use the agent you deployed on Amazon Bedrock AgentCore from Slack? You're not alone — there are already quite a few samples and blog posts that connect AgentCore agents to Slack.

Reading through them, I noticed something: on the Slack side, everyone is solving the same problems. Receiving events, fetching thread history and converting it into the model's input format, rendering replies. The agents themselves all differ, but the Slack work is the same.

Welt factors that common part out, so it doesn't have to be rebuilt for every agent.

What Is Welt

Welt is a general-purpose Slack frontend for agents on AgentCore. It sits between Slack and your agent:

Slack
  ⇅
Welt
  ⇅  plain JSON
AgentCore Runtime
  └── your agent, wired through an adapter

The division of labor is clear — Welt owns everything Slack, you own everything agent:

Side	Owns
Welt	Slack tokens, event intake, thread-history fetch, streaming rendering of replies, file upload/download, approval buttons
Your agent	Model, tools, MCP, memory

One Wire, Any Framework

The two sides exchange plain JSON, specified in the Wire Contract, so nothing ties Welt to any particular agent framework. An adapter maps the wire to one framework's types:

Adapter	Language	Framework	Package
welt-io-strands	Python	Strands Agents	welt-io-strands
welt-io-strands-ts	TypeScript	Strands Agents	@welt-io/strands
welt-io-mastra	TypeScript	Mastra	@welt-io/mastra

This article uses the Python Strands adapter; the others work the same way. And since the wire is plain JSON, an adapter can be built for any stack — if there's one you'd like, let me know in an issue.

The Agent-Side Changes Are Small

If you already have a Strands agent, connecting it to Slack takes nothing more than wiring in two welt-io-strands functions. This is Welt's biggest selling point.

from bedrock_agentcore.runtime import BedrockAgentCoreApp
from strands import Agent
from welt_io_strands import decode_messages, renderable_events

app = BedrockAgentCoreApp()

@app.entrypoint
async def invoke(payload: dict):
    messages = decode_messages(payload["messages"])
    # your agent, unchanged
    agent = Agent(tools=[...])
    # yield the events Welt renders into the Slack thread
    async for event in renderable_events(agent.stream_async(messages)):
        yield event

app.run()

The two functions cover the two directions:

Function	Direction	What it does
`decode_messages()`	Inbound	Restores the Converse-shaped messages Welt sends (files from Slack uploads arrive base64-encoded) to the form Strands expects
`renderable_events()`	Outbound	Reduces the raw `stream_async` events to the JSON-serializable events Welt can render

Your agent's logic — model choice, tools, system prompt — stays untouched. And you never write a single line of Slack API code.

What You Get on the Slack Side

With just the code above, the Slack thread gets history, streaming, and files — with zero extra work on the agent:

Feature	How it works
Thread history	On every turn, Welt fetches the whole thread, converts it into Converse-shaped messages, and hands it to your agent — the agent needs no conversation store of its own (if you'd rather manage history on the agent side, there's a mode that sends only the new messages)
Streaming	The agent's reply streams into the thread in real time, with an indicator while a tool is running
Files, both ways	Files uploaded to Slack go to the agent, and images or documents the agent (or its tools) generate are uploaded back into the thread — strands-tools' `generate_image` works with zero extra code

Human-in-the-Loop Becomes Slack Buttons

Welt also supports human-in-the-loop. Among the AgentCore-to-Slack integrations out there, I haven't seen many that go this far.

With Strands interrupts, a tool can pause the run and wait for human approval. Welt renders that pause as buttons and a text field right in the Slack thread.

You describe the buttons and the text field with welt-io-strands's interrupt_reason():

answer = tool_context.interrupt(
    "deploy-approval",
    reason=interrupt_reason(
        "Deploy to prod?",
        [
            {"value": "y", "label": "Deploy", "style": "primary"},
            {"value": "n", "label": "Cancel"},
        ],
        input={"label": "Or tell me what to do instead"},
    ),
)

Pressing a button (or typing into the field) resumes the run. Strands' stock HumanInTheLoop intervention also works over Welt as-is — the stdio approval prompt simply becomes Slack buttons.

The complete code, including the resume handling, is in welt-io-strands's example agent.

Try It

The Quick Start runs everything on your machine — nothing is deployed, and the only AWS dependency is the Bedrock model the agent calls:

Create a Slack app from the manifest.yml bundled in the Welt repository
Clone Welt and put the two tokens in a .env file
uv run --env-file .env main.py
In another terminal, run welt-io-strands's example agent — with no AGENT_ARN set, Welt talks to it at localhost:8080

By default, Welt connects to Slack over Socket Mode. There's no public endpoint to stand up, so the local uv run just works.

Moving to production is a change of address, not code. Deploy the agent to AgentCore Runtime with the AgentCore CLI and point AGENT_ARN at it; for Welt itself, put the container image (ghcr.io/iwamot/welt) on ECS or similar, and it works over Socket Mode the same way.

If you'd rather not pay for an always-on process, you can also run Welt on AWS Lambda. In that case it receives HTTP at a Function URL instead of using Socket Mode (Welt handles the signature verification), and the idle cost drops to zero.

From there, make it yours. The Slack part is no longer your problem — swap the example agent for your own model, tools, MCP servers, and memory, and you have an agent that's actually useful to you and your team, living in a Slack thread everyone can talk to.

Welt and its adapters are MIT-licensed open source. If you've been wanting to use your AgentCore agent from Slack, give it a try.

How I Made RDS Private Without Extra Cost Using Egress-Only IGW

Takashi Iwamoto — Sun, 01 Feb 2026 06:12:32 +0000

I had an Amazon RDS DB instance with public access enabled due to certain requirements. When public access was no longer needed, I migrated it to a private subnet.

By using an Egress-Only Internet Gateway (Egress-Only IGW), I completed the migration without adding paid resources like NAT Gateways or VPC endpoints.

In this article, I'll share the background and the actual migration steps.

Background

The reason I wanted to move the DB to a private subnet was that public access was no longer required.

Public access had been enabled because Lambda functions in multiple AWS accounts needed to access this DB. While VPC peering was an option, I chose simple public access since IAM authentication alone seemed sufficient.

Later, the requirements changed—only Lambda functions in the same account as the DB needed access. This meant I could finally migrate to a private subnet.

The Challenge: Connecting to AWS Service Endpoints

However, there was one hurdle to overcome: "How do Lambda functions connect to AWS service endpoints?"

The Lambda functions needed to access not only the DB but also AWS service endpoints—specifically Athena, STS, and CloudWatch.

Attaching a VPC to the Lambda function would enable access to the DB in the private subnet. But it would also cut off the route to service endpoints.

My first thought was to use paid NAT Gateways or VPC endpoints. But I wanted to complete the migration without adding costs if possible.

Egress-Only IGW: A Cost-Free Solution

After some thought, I came up with the solution: Egress-Only IGW. It has the following characteristics:

Used only for IPv6 traffic
Allows outbound communication from VPC to the internet (inbound connections cannot be initiated from the internet)
No charge

This means if the AWS service endpoints support IPv6, Lambda functions with VPC attached can connect to them for free.

You can check IPv6 support for service endpoints in this documentation.

Athena, STS, and CloudWatch all support IPv6 (note: the STS global endpoint does not support IPv6; regional endpoints are required), so the following migration should be possible:

Component	Before	After
DB instance	Public subnet	Private subnet
Lambda → DB	Internet	Within VPC
Lambda → AWS services	IPv4	IPv6 (via Egress-Only IGW)

Migration Steps

With the goal set, I proceeded with the following steps:

Add IPv6 support to the VPC
Update Lambda function configuration
Migrate DB to private subnet
Delete unnecessary resources

1. Add IPv6 Support to the VPC

First, I added IPv6 support to the VPC, including associating CIDR blocks and updating route tables. See the documentation for details.

2. Update Lambda Function Configuration

Next, I updated the Lambda function configuration. Key points:

Allow IPv6 egress in the security group
Set the environment variable AWS_USE_DUALSTACK_ENDPOINT=true (reference)
Attach VPC and enable "Allow IPv6 traffic for dual-stack subnets" (reference)

The environment variable tells boto3 (used by the Lambda function) to send requests to IPv4/IPv6 dual-stack service endpoints.

3. Migrate DB to Private Subnet

Then, I migrated the DB to the private subnet. This was the main task.

I expected it to be simple, but you cannot directly migrate a DB from a public subnet to a private subnet. It required some extra steps.

Specifically, you need to follow steps like "disable Multi-AZ → edit subnet group → enable Multi-AZ → ..." See this article for details.

After migration, I edited the DB's security group:

Before: Allow access from 0.0.0.0/0
After: Allow access from the Lambda function's security group

4. Delete Unnecessary Resources

Finally, I deleted the now-unnecessary public subnet and the regular Internet Gateway. Migration complete.

Conclusion

That's how I used Egress-Only IGW to make RDS private without extra cost. It was a great opportunity to work with this feature in a real-world scenario.

This example used Lambda, but the approach should apply to other services like ECS as well—assuming the target endpoints support IPv6.

If you're choosing a public subnet due to cost-security trade-offs, Egress-Only IGW might be your solution. Give it a try!

Migrating SOCI Index Manifest from v1 to v2 for ECS/Fargate

Takashi Iwamoto — Sun, 01 Feb 2026 04:40:31 +0000

At my current workplace, we use Seekable OCI (SOCI) to speed up Amazon ECS task launches on AWS Fargate.

Recently, we migrated the SOCI Index Manifest version from v1 to v2. In this article, I'll explain why and how we did it.

Why We Migrated

The reason is simple: v1 support is ending. AWS sent out the following notification:

After careful consideration, we have made the decision to discontinue the support for SOCI Index Manifest v1 for Amazon ECS customers using AWS Fargate launch mode, effective February 9, 2026. Until this date, AWS Fargate customers using Index Manifest v1 will be able to leverage existing indexes and create new indexes using Index Manifest v1 to deploy applications.

Without migrating to v2, task launches would become slower.

How We Migrated

The migration method is also straightforward. In short, we "deployed the new SOCI Index Builder that supports v2." We had been using a different solution with the same name, but it was deprecated without v2 support, so we switched.

Here are the general steps:

1. Prepare the Tool

Clone the SOCI Index Builder Git repository:

git clone https://github.com/awslabs/cfn-ecr-aws-soci-index-builder.git
cd cfn-ecr-aws-soci-index-builder

2. Edit `.taskcat.yml`

Next, edit .taskcat.yml. This mainly involves changing the deployment region. For example, if you only want to accelerate images in the Tokyo region ECR repository:

--- a/.taskcat.yml
+++ b/.taskcat.yml
@@ -12,11 +12,5 @@ tests:
     parameters:
       SociRepositoryImageTagFilters: "*:*"
     regions:
-      - us-east-1
-      - us-east-2
-      - us-west-1
-      - us-west-2
-      - eu-west-1
-      - eu-west-2
-      - eu-west-3
+      - ap-northeast-1
     template: templates/SociIndexBuilder.yml

3. Run `taskcat upload`

Then run taskcat upload. This uploads the CloudFormation templates and Lambda function packages to S3.

I recommend running taskcat via Docker rather than installing it locally, as it doesn't support Python 3.14 yet and you probably won't use it very often anyway.

Here's an example of a dry run. Remove --dry-run to actually upload:

docker run --rm \
  -v $(pwd):/mnt \
  -v /var/run/docker.sock:/var/run/docker.sock \
  -e AWS_ACCESS_KEY_ID \
  -e AWS_SECRET_ACCESS_KEY \
  -e AWS_SESSION_TOKEN \
  -e GIT_PYTHON_REFRESH=quiet \
  taskcat/taskcat taskcat upload --dry-run

4. Create the CloudFormation Stack

Finally, create a stack using the uploaded CloudFormation template. As documented, the easiest way is through the Management Console.

For reference, here's an example of creating a stack in a single region using Terraform:

locals {
  repository_image_tag_filters = ["*:*"]
}

data "aws_s3_object" "template" {
  bucket = "<your-s3-bucket-name>"
  key    = "cfn-ecr-aws-soci-index-builder/templates/SociIndexBuilder.yml"
}

resource "aws_cloudformation_stack" "this" {
  capabilities     = ["CAPABILITY_IAM"]
  name             = "cfn-ecr-aws-soci-index-builder"
  parameters = {
    IamPermissionsBoundaryArn     = "none"
    QSS3BucketName                = data.aws_s3_object.template.bucket
    QSS3KeyPrefix                 = "${split("/", data.aws_s3_object.template.key)[0]}/"
    SociIndexVersion              = "V2"
    SociRepositoryImageTagFilters = join(",", local.repository_image_tag_filters)
  }
  template_url = format(
    "https://%s.s3.%s.amazonaws.com/%s",
    data.aws_s3_object.template.bucket,
    data.aws_s3_object.template.region,
    data.aws_s3_object.template.key
  )
}

Conclusion

That's how we migrated the SOCI Index Manifest from v1 to v2 for ECS/Fargate.

If you've been using v1 and haven't migrated yet, I recommend doing so; it's not that much work.

Haven't adopted SOCI at all? You can easily speed up your task launches with these four steps. Give it a try!

AWS ChatOps Without Lambda? Yes, You Can!

Takashi Iwamoto — Sat, 17 May 2025 09:42:18 +0000

Hi, I'm Takashi Iwamoto (@iwamot), an AWS Community Builder (Cloud Operations).

When I looked for prior examples of building ChatOps on AWS, most of them used Lambda.

However, I had a hunch that we could get rid of Lambda by using the Custom Actions feature of Amazon Q Developer in chat applications (formerly AWS Chatbot), which was announced in November 2023.

After trying it out, I was able to build a fairly practical architecture.

In this post, I'll walk you through how I built AWS ChatOps without Lambda.

Why Avoid Lambda

Before we dive in, let me touch on why I want to avoid Lambda.

In a word: "Because it makes operations easier."

If you don't use Lambda, you never have to worry about runtime end‑of‑life. The more ChatOps targets you have, the more the cost of those EOL updates will hurt.

Services Used & Overall Architecture

Here's what I actually built.

The main AWS services I used are these three:

Amazon EventBridge – event detection & routing
Amazon SNS – notification delivery
Amazon Q Developer in chat applications – Slack notifications & Custom Actions

And the overall architecture looks like this:

An EventBridge rule detects the CodeDeploy deployment state‑change event.
The event is transformed with an input transformer and routed to an SNS topic.
The message is delivered to an Amazon Q Developer Slack channel.
The notification message and Custom Action buttons appear in Slack.
A user picks one of the Custom Actions.
The corresponding CodeDeploy command is executed (e.g. continue-deployment, stop-deployment).

Creating & Associating Custom Actions

To show Custom Action buttons in chat apps like Slack, you first need to create the Custom Actions themselves.

There are two ways to define Custom Actions:

Create them directly in the chat application
Create them with the Amazon Q Developer in chat applications API

I chose the API method because I wanted to manage everything as IaC. If you create the action in the chat UI, it seems that the resource is owned by AWS and not created inside your own AWS account.

If you go the API route, you'll also need to associate the Custom Actions with the chat channel yourself.

The steps are:

Call the CreateCustomAction API to create the action.
Call the AssociateToConfiguration API to attach it to the chat channel.

Below is how I implemented it in Terraform. (It's a bit long, so I've collapsed it.)

Definition of Custom Actions

locals {
  custom_actions = {
    SwitchTasks = {
      button_text = "🔁 Switch task sets"
      criteria = [
        {
          operator      = "EQUALS"
          value         = "blue-green-deployment"
          variable_name = "ActionGroup"
        },
      ]
      variables = {
        "ActionGroup"  = "event.metadata.additionalContext.ActionGroup"
        "DeploymentId" = "event.metadata.additionalContext.DeploymentId"
        "Region"       = "event.metadata.additionalContext.Region"
      }
      command_text = "codedeploy continue-deployment --deployment-id $DeploymentId --region $Region --deployment-wait-type READY_WAIT"
    }

    RollbackDeployment = {
      button_text = "🔙 Rollback"
      criteria = [
        {
          operator      = "EQUALS"
          value         = "blue-green-deployment"
          variable_name = "ActionGroup"
        },
      ]
      variables = {
        "ActionGroup"  = "event.metadata.additionalContext.ActionGroup"
        "DeploymentId" = "event.metadata.additionalContext.DeploymentId"
        "Region"       = "event.metadata.additionalContext.Region"
      }
      command_text = "codedeploy stop-deployment --deployment-id $DeploymentId --region $Region --auto-rollback-enabled true"
    }

    TerminateOldTask = {
      button_text = "🧹 Terminate old task set"
      criteria = [
        {
          operator      = "EQUALS"
          value         = "blue-green-deployment"
          variable_name = "ActionGroup"
        },
      ]
      variables = {
        "ActionGroup"  = "event.metadata.additionalContext.ActionGroup"
        "DeploymentId" = "event.metadata.additionalContext.DeploymentId"
        "Region"       = "event.metadata.additionalContext.Region"
      }
      command_text = "codedeploy continue-deployment --deployment-id $DeploymentId --region $Region --deployment-wait-type TERMINATION_WAIT"
    }
  }
}

resource "awscc_chatbot_custom_action" "this" {
  for_each = local.custom_actions

  action_name = each.key
  attachments = [
    {
      button_text       = each.value.button_text
      criteria          = each.value.criteria
      notification_type = "Custom"
      variables         = each.value.variables
    },
  ]
  definition = {
    command_text = each.value.command_text
  }
  tags = [
    for key, value in merge(data.aws_default_tags.this.tags, { "Name" = each.key }) : {
      key   = key
      value = value
    }
  ]
}

Associating the Custom Actions with the Chat Channel (implemented with null_resource)

# In this example we set custom_action_groups = ["blue-green-deployment"]
variable "custom_action_groups" {
  description = "A list of Custom Action group names to associate specific AWS Chatbot actions with the specified Slack channel configuration."
  type        = list(string)
  default     = []
}

locals {
  action_group_map = {
    "blue-green-deployment" = ["SwitchTasks", "RollbackDeployment", "TerminateOldTask"]
  }

  selected_actions = distinct(flatten([
    for group in var.custom_action_groups :
    lookup(local.action_group_map, group, [])
  ]))

  action_arns = {
    for k, v in {
      for action in local.selected_actions :
      action => try(
        one([
          for id in data.awscc_chatbot_custom_actions.this.ids : id
          if endswith(id, "/${action}")
        ]),
        null
      )
    } : k => v if v != null
  }
}

resource "null_resource" "associate_actions" {
  for_each = local.action_arns

  triggers = {
    action_arn  = each.value
    chatbot_arn = aws_chatbot_slack_channel_configuration.this.chat_configuration_arn
  }

  provisioner "local-exec" {
    command = "aws chatbot associate-to-configuration --resource ${self.triggers.action_arn} --chat-configuration ${self.triggers.chatbot_arn} --region us-west-2"
  }

  provisioner "local-exec" {
    when    = destroy
    command = "aws chatbot disassociate-from-configuration --resource ${self.triggers.action_arn} --chat-configuration ${self.triggers.chatbot_arn} --region us-west-2"
  }
}

EventBridge Rule

variable "sns_topic_arn" {
  description = "The SNS topic to notify when the blue/green deployment is ready for validation."
  type        = string
  nullable    = false
}

variable "deployment_group_name" {
  description = "The name of the deployment group targeted by ChatOps."
  type        = string
  nullable    = false
}

variable "green_url" {
  description = "The URL for validating the green environment."
  type        = string
  nullable    = false
}

resource "aws_cloudwatch_event_rule" "this" {
  description    = null
  event_bus_name = "default"
  event_pattern = jsonencode(
    {
      detail = {
        deploymentGroup = [
          var.deployment_group_name,
        ]
        state = [
          "READY",
        ]
      }
      detail-type = [
        "CodeDeploy Deployment State-change Notification",
      ]
      source = [
        "aws.codedeploy",
      ]
    }
  )
  force_destroy       = false
  name                = "${var.deployment_group_name}-deployment-ready"
  name_prefix         = null
  role_arn            = null
  schedule_expression = null
  state               = "ENABLED"
  tags = {
    "Name" = "${var.deployment_group_name}-deployment-ready"
  }
}

resource "aws_cloudwatch_event_target" "this" {
  arn            = var.sns_topic_arn
  event_bus_name = "default"
  force_destroy  = false
  input          = null
  input_path     = null
  role_arn       = aws_iam_role.this.arn
  rule           = aws_cloudwatch_event_rule.this.name
  target_id      = "sns"

  input_transformer {
    input_paths = {
      "account"         = "$.account"
      "deploymentGroup" = "$.detail.deploymentGroup"
      "deploymentId"    = "$.detail.deploymentId"
      "region"          = "$.region"
      "time"            = "$.time"
    }
    input_template = replace(replace(
      jsonencode(
        {
          content = {
            description = trimspace(
              <<-EOT
              *Deployment group:* <deploymentGroup>
              *Deployment ID:* <deploymentId>
              *Account:* <account>
              *Region:* <region>
              *Time:* <time>
            EOT
            )
            nextSteps = [
              "Check if the app works correctly at ${var.green_url}",
              "If all looks good, switch task sets. If not, trigger a rollback",
              "Once validated, terminate the original task set",
            ]
            textType = "client-markdown"
            title    = ":large_blue_circle::large_green_circle: Deployment is now ready for validation"
          }
          metadata = {
            additionalContext = {
              ActionGroup  = "blue-green-deployment"
              DeploymentId = "<deploymentId>"
              Region       = "<region>"
            }
            enableCustomActions = true
          }
          source  = "custom"
          version = "1.0"
        }
      ),
    "\u003c", "<"), "\u003e", ">")
  }
}

With this in place, Slack now shows the following notification, and the entire Blue/Green deployment can be operated using just the Custom Action buttons.

Remaining Challenge

As you can see, this solution works quite well, but one issue is that we can't control the order of the Custom Action buttons.

Sometimes they show up as "🧹 Terminate old task set", "🔙 Rollback", "🔁 Switch task sets", in that order. The screenshot above just happened to be in the desired order.

If we could control that order, I'm sure this approach would be solid enough to become a go-to ChatOps pattern. I've already submitted a feature request to AWS Support, and I'm hopeful it will be addressed in the future.

Conclusion: ChatOps Without Lambda Is Possible

That was an example of ChatOps built with Amazon Q Developer in chat applications Custom Actions.

When you build it without Lambda, operations get easier. The unresolved button‑ordering issue remains, but if you can live with that, it's already perfectly usable.

I plan to expand our ChatOps coverage with this method, and I hope it inspires you to try it out yourself.

How to Change Network Configurations for Blue/Green Amazon ECS Services

Takashi Iwamoto — Sat, 19 Oct 2024 15:10:57 +0000

If you want to change the network settings for an Amazon ECS service using blue/green deployment with AWS CodeDeploy, you need to edit the AppSpec file and trigger a new deployment.

Without knowing this method, you'd either have to recreate the entire ECS service or give up on changing the network settings.

Background

For ECS services using blue/green deployment with AWS CodeDeploy, you cannot directly change the network settings.

When you try to make changes by calling the ECS UpdateService API, you will encounter the following error:

InvalidParameterException: Unable to update network parameters on services with a CODE_DEPLOY deployment controller. Use AWS CodeDeploy to trigger a new deployment.

This error message indicates that you need to trigger a new deployment with CodeDeploy, as noted in the documentation.

For services using the blue/green (CODE_DEPLOY) deployment controller, only the desired count, deployment configuration, health check grace period, task placement constraints and strategies, enable ECS managed tags option, and propagate tags can be updated using this API. If the network configuration, platform version, task definition, or load balancer need to be updated, create a new AWS CodeDeploy deployment. For more information, see CreateDeployment in the AWS CodeDeploy API Reference.

UpdateService - Amazon Elastic Container Service

Solution

Now that we know we need to trigger a new deployment, the question remains: how exactly should we do that?

The answer is to "edit the AppSpec file and then trigger a new deployment." If you trigger it without editing the AppSpec file, the network settings won't change.

In the case of appspec.yml, edit it as shown below.

        NetworkConfiguration:
          AwsvpcConfiguration:
            Subnets: ["subnet-1234abcd","subnet-5678abcd"]
            SecurityGroups: ["sg-12345678"]
            AssignPublicIp: "ENABLED"

AppSpec File example - AWS CodeDeploy

Specify the subnets, security groups, and whether or not to assign a public IP address for the ECS tasks in the AppSpec file.

When doing so, you need to specify all three: Subnets, SecurityGroups, and AssignPublicIp.

All or none of the settings under NetworkConfiguration must be specified. For example, if you want to specify Subnets, you must also specify SecurityGroups and AssignPublicIp. If none is specified, CodeDeploy uses the current network Amazon ECS settings.

AppSpec 'resources' section (Amazon ECS and AWS Lambda deployments only) - AWS CodeDeploy

After editing the AppSpec file, trigger a new deployment, and the network settings will be applied. The new ECS tasks in the green environment should start with the specified network configuration.

Once the deployment is complete, you can safely remove the NetworkConfiguration from the AppSpec file.

Conclusion

As described, even for Amazon ECS services using blue/green deployment with CodeDeploy, you can change the network settings by editing the AppSpec file and triggering a new deployment.

I almost ended up recreating the ECS service because I wasn't aware of this method. I hope this article helps those in a similar situation.

Implementing SLO Error Budget Monitoring with AWS Services Only

Takashi Iwamoto — Sun, 08 Sep 2024 14:32:28 +0000

In this article, I’ll introduce an SLO error budget monitoring system that I built entirely using AWS services.

System Overview

This system is based on the multi-window, multi-burn-rate alerting method proposed in the Site Reliability Workbook. It sends alerts to Slack when the SLO error budget burn rate exceeds a certain threshold.

expr: (
        job:slo_errors_per_request:ratio_rate1h{job="myjob"} > (14.4*0.001)
      and
        job:slo_errors_per_request:ratio_rate5m{job="myjob"} > (14.4*0.001)
      )
    or
      (
        job:slo_errors_per_request:ratio_rate6h{job="myjob"} > (6*0.001)
      and
        job:slo_errors_per_request:ratio_rate30m{job="myjob"} > (6*0.001)
      )
severity: page

expr: (
        job:slo_errors_per_request:ratio_rate24h{job="myjob"} > (3*0.001)
      and
        job:slo_errors_per_request:ratio_rate2h{job="myjob"} > (3*0.001)
      )
    or
      (
        job:slo_errors_per_request:ratio_rate3d{job="myjob"} > 0.001
      and
        job:slo_errors_per_request:ratio_rate6h{job="myjob"} > 0.001
      )
severity: ticket

https://sre.google/workbook/alerting-on-slos/

The Site Reliability Workbook recommends combining multiple time windows and burn rates to create more accurate alerts. For example, if you only alert based on a short time window, you may get too many false positives. On the other hand, if you rely on a longer time window, it may take too long for an alert to trigger or recover. By using both short and long windows, the system can generate more precise alerts.

For more detailed explanations of this alerting method, I encourage you to check out the Site Reliability Workbook, which is available for free. Personally, I believe that strictly following these guidelines is the best way to monitor service reliability.

Why I Implemented This

The primary motivation behind this monitoring system was to solve the problem of "alert fatigue." As an operations engineer, I was constantly bombarded with alerts that didn’t actually require any action.

Previously, we monitored CPU and memory usage metrics directly, which led to alerts even when the service itself was running smoothly. For example, we would receive a Slack alert if the CPU usage of a web server exceeded 90%, regardless of whether there were any issues with availability or latency.

However, even if the CPU usage was high, if the service remained available and responsive, there was no need for intervention. Over time, receiving alerts for issues that didn't require action caused fatigue and increased the risk of missing critical alerts. Missing an important alert could damage service reliability and lead to a loss of users.

To address this, I decided to focus only on critical indicators—such as availability and latency—and set up an SLO-based monitoring system that tracks the error budget burn rate.

Why I Used AWS Services Only

At first, I considered using third-party monitoring tools like New Relic or Datadog. However, after investigating these tools in early 2023, I found that it was either impossible or very difficult to implement the kind of SLO monitoring I needed with their existing features. I worked with support teams from both platforms, but we couldn't come up with a viable solution.

On the other hand, by combining AWS services such as Amazon CloudWatch, I could implement the necessary monitoring and at a much lower cost—around $20 per month.

Since most of our workloads are hosted on AWS, building the monitoring system directly on AWS made it easier to integrate everything into a single console. AWS also offers a high degree of flexibility in terms of combining services, giving us more room to scale and customize compared to third-party tools like New Relic or Datadog.

Architecture

Here’s an overview of the system’s architecture:

Although it may look a bit complex at first glance, the core workflow is simple:

EventBridge Scheduler triggers a Lambda function at regular intervals.
The Lambda function runs an Athena query to aggregate ALB access logs.
The results are stored in both CloudWatch custom metrics and RDS.
CloudWatch composite alarms use these metrics to trigger alerts based on the SLO conditions.

The Lambda function is triggered by EventBridge instead of an S3 file save event. This is because when there is heavy traffic, multiple log files are saved to S3, which could result in the aggregation process being run multiple times unnecessarily.

We use RDS to store the aggregated results in order to reduce Athena scan costs. For more details on these optimizations and custom metric definitions, please refer to my presentation at JAWS PANKRATION 2024.

Results of the Implementation

Since we introduced this system, alert fatigue has significantly decreased. I personally noticed a reduction in the number of unnecessary alerts that required no action.

Additionally, the development team has become more invested in the reliability of the services. Now, when an alert is triggered, the team takes the initiative to tune performance or even revise the SLOs when necessary.

As of September 2023, this system is monitoring 7 services and 12 critical user journeys.

Operational Tips

To streamline operations, I developed an internal Terraform module that allows us to quickly accommodate additional monitoring requests from the development team. While the module itself is private, this is an actual example of its interface, which could be useful as a reference if you plan to implement a similar module.

module "slomon" {
  source = "git@github.com:enechange/terraform-modules.git//slomon-for-alb?ref=v0.53.0"

  environment_name              = "prod-enechange"
  alb_access_logs_s3_url        = local.alb_access_logs_s3_url
  sns_topic_names_for_paging    = ["cto-incident-enechange"]
  sns_topic_names_for_ticketing = ["cto-alert-enechange"]

  critical_user_journeys = {
    input1_submit = {
      http_method     = "POST"
      path            = "/try/input1_submit"
      dashboard_order = 1

      slo = {
        availability_target   = 95.0
        latency_p95_threshold = 4.0
        latency_p50_threshold = 3.0
      }
    }
  }
}

This Terraform module also includes Athena queries for analyzing historical data, allowing us to quickly set SLOs based on past performance.

Future Outlook

At present, this monitoring system meets all of our needs. However, if CloudWatch's Application Signals feature evolves to support the required conditions for SLO monitoring, this custom system may no longer be necessary. In that case, we would likely transition to a fully managed solution.

Likewise, if New Relic or Datadog introduce features that support the kind of SLO monitoring we need, we may consider switching to them. These APM tools generally offer better capabilities for diagnosing root causes of performance degradation compared to CloudWatch.

Ultimately, we will continue to assess the best solution for our needs and adapt our system as new tools and services emerge.

Conclusion

In this article, I introduced an SLO error budget monitoring system that I built using AWS services only.

By implementing this system, we were able to reduce alert fatigue and increase the development team’s focus on service reliability.

I hope this article provides useful insights for those dealing with similar challenges around alert fatigue.

DEV Community: Takashi Iwamoto

Test your Step Functions workflows locally with pytest

A test can run the ASL you already have

Set up a project

Write the test

What the runner did

Testing a definition from Workflow Studio

If you want to maintain the workflow in Python, sfnx can compile it

Define a shared retry policy once

Write branches with if/elif

Handle errors with try/except: only Task, Parallel and Map states catch

Stay within the supported Python subset

The Python version behaves like the original in 12 mocked scenarios

The compiler's trade-off: it can add states

Start with one failure path

Connect Your Bedrock AgentCore Agent to Slack — Without Writing the Slack Part

What Is Welt

One Wire, Any Framework

The Agent-Side Changes Are Small

What You Get on the Slack Side

Human-in-the-Loop Becomes Slack Buttons

Try It

How I Made RDS Private Without Extra Cost Using Egress-Only IGW

Background

The Challenge: Connecting to AWS Service Endpoints

Egress-Only IGW: A Cost-Free Solution

Migration Steps

1. Add IPv6 Support to the VPC

2. Update Lambda Function Configuration

3. Migrate DB to Private Subnet

4. Delete Unnecessary Resources

Conclusion

Migrating SOCI Index Manifest from v1 to v2 for ECS/Fargate

Why We Migrated

How We Migrated

1. Prepare the Tool

2. Edit .taskcat.yml

3. Run taskcat upload

4. Create the CloudFormation Stack

Conclusion

AWS ChatOps Without Lambda? Yes, You Can!

Why Avoid Lambda

Services Used & Overall Architecture

Creating & Associating Custom Actions

Remaining Challenge

Conclusion: ChatOps Without Lambda Is Possible

How to Change Network Configurations for Blue/Green Amazon ECS Services

Background

Solution

Conclusion

Implementing SLO Error Budget Monitoring with AWS Services Only

System Overview

Why I Implemented This

Why I Used AWS Services Only

Architecture

Results of the Implementation

Operational Tips

Future Outlook

Conclusion

2. Edit `.taskcat.yml`

3. Run `taskcat upload`