DEV Community: Martin Nanchev

Anarchy, Assembly Lines, and Corporate Hierarchy: Benchmarking Multi-Agent Architectures for Medical Device Data

Martin Nanchev — Mon, 16 Mar 2026 07:16:10 +0000

My AI judge gave the anarchists a perfect score. I disagree.
I built three multi-agent systems to analyze data from my insulin pump — a Medtronic MiniMed 780G — and had an LLM evaluate their output. The cheapest, fastest architecture scored identically to the most expensive one. But when I read the actual reports, the cheap one guessed where the expensive one calculated. The evaluator didn't care. That tension — between automated scores and human judgment — turned out to be the most interesting finding of this experiment.
But let's start from the beginning.

A Fair Fight This Time

In my previous blog post, I compared a swarm architecture with a graph pipeline for analyzing CareLink CSV exports. The problem? I used different models for each, which made the comparison unfair.
This time, every agent runs on the same model: Haiku 4.5 via AWS Bedrock. Same prompts, same tools, same data. The only variable is the orchestration pattern.
A LinkedIn commenter also suggested trying prompt caching to reduce costs. Good idea — let's see which architecture benefits. It is fair to mention that caching could help you cache tool calls, prompts and system prompts with following config

bedrock_model = BedrockModel(
    model_id=MODEL_ID,
    region_name=REGION,
    max_tokens=64000,
    temperature=0.0,
    cache_config=CacheConfig(strategy="auto"),
    cache_tools="default",
    streaming=True,
)

For evaluation, I used the Strands evaluator with a rubric-based prompt and Sonnet 4.5 as the judge.

Every architecture uses the same four agents. Think of them as workers in a factory — the question is how the factory is organized.

CSV Reader — Parses the CareLink CSV export. Returns raw structured data, no interpretation.

Data Analyst — Crunches the numbers: glucose statistics, Time in Range, GMI, coefficient of variation, insulin totals, carb intake.

Pattern Reviewer — Reads the metrics and timestamps to spot clinically meaningful patterns: dawn phenomenon, post-meal spikes, overnight trends, hypo/hyper clustering.

Endocrinologist — Synthesizes everything into pump optimization suggestions, framed as discussion topics for a healthcare professional.
I defined the agents using a factory pattern so every architecture gets identical copies:

PROMPTS = {
    "csv_reader": (
        "You are a CSV Parser Agent specializing in Medtronic MiniMed 780G CareLink data exports.\n"
        "Parse the raw CSV and return the extracted data EXACTLY as the tool outputs it.\n"
        "Do NOT summarize, interpret, compute statistics, or analyze the data.\n"
        "Do NOT provide clinical recommendations.\n"
        "Return only the raw parsed output for other agents to process.\n"
    ),
    "data_analyst": (
        "You are a Data Analyst Agent specializing in CGM and insulin pump metrics.\n"
        "Compute statistical analysis on the diabetes pump data provided.\n"
        "Calculate glucose statistics (mean, median, SD, min, max), Time in Range (TIR),\n"
        "GMI (estimated A1C), coefficient of variation (CV%), total daily insulin,\n"
        "insulin-to-carb ratios, correction bolus frequency, and average daily carbs.\n"
        "Use the SG_VALUES data from the parsed CSV to run your statistical tools.\n"
        "If the SG_VALUES are not directly available in your input, use read_carelink_csv\n"
        "to extract them from the CSV file, then run calculate_statistics and time_in_range.\n"
        "Present findings as structured metrics with numbers — no clinical recommendations.\n"
    ),
    "pattern_reviewer": (
        "You are a Pattern Recognition Agent specializing in diabetes data interpretation.\n"
        "Identify clinically significant patterns and anomalies from the data provided.\n"
        "Look for dawn phenomenon, post-meal spikes, overnight trends, and glucose variability.\n"
        "If you have access to the CSV file path, use read_carelink_csv to extract timestamped\n"
        "data, then use hourly_glucose_profile to compute hourly patterns.\n"
        "Flag recurring hypoglycemia with timing, severity, and clustering.\n"
        "Flag prolonged hyperglycemia with duration and potential triggers.\n"
        "Note Auto Mode exits, insulin suspensions, and sensor issues from alert data.\n"
        "Compare metrics against ADA/EASD consensus targets.\n"
        "Note positive trends and areas of good control.\n"
        "Focus on pattern identification — do not suggest treatment changes.\n"
    ),
    "endocrinologist": (
        "You are an Endocrinologist Agent specializing in insulin pump therapy optimization.\n"
        "Provide clinical interpretation and actionable recommendations based on the\n"
        "patterns and metrics from previous analysis.\n"
        "Suggest potential pump setting adjustments: Active Insulin Time, carb ratios,\n"
        "Auto Mode target glucose (5.5 vs 6.7 mmol/L / 100 vs 120 mg/dL), and bolus timing.\n"
        "Use the Fiasp insulin pharmacokinetics profile: onset ~15min, peak ~1-2h, duration ~3-5h.\n"
        "Frame everything as 'discuss with your healthcare team' — not direct medical advice.\n"
        "Provide a clear, prioritized summary highlighting the most impactful improvements.\n"
        "Acknowledge what is working well alongside areas that need attention.\n"
    ),
}


def make_agents():
    """Create a fresh set of specialist agents.

    Tool assignment strategy:
      - csv_reader: read_carelink_csv only (parse the file)
      - data_analyst: read_carelink_csv + calculate_statistics + time_in_range
        (needs CSV access because Graph/Swarm may pass summaries instead of raw values)
      - pattern_reviewer: read_carelink_csv + hourly_glucose_profile
        (needs CSV access for the same reason — LLMs summarize upstream output)
      - endocrinologist: no tools (pure synthesis from previous agents' output)

    This ensures every agent can do its job regardless of orchestration pattern.
    In the Graph, upstream nodes may summarize data. In the Swarm, agents may skip steps.
    In the Coordinator, the LLM decides what to pass. Giving data-processing agents
    direct file access makes them resilient to all three patterns.
    """
    return {
        "csv_reader": Agent(
            system_prompt=PROMPTS["csv_reader"],
            name="csv_reader",
            model=bedrock_model,
            conversation_manager=make_conversation_manager(window_size=15, per_turn=True),
            tools=[read_carelink_csv],
        ),
        "data_analyst": Agent(
            system_prompt=PROMPTS["data_analyst"],
            name="data_analyst",
            model=bedrock_model,
            conversation_manager=make_conversation_manager(window_size=20, per_turn=True),
            tools=[read_carelink_csv, calculate_statistics, time_in_range],
        ),
        "pattern_reviewer": Agent(
            system_prompt=PROMPTS["pattern_reviewer"],
            name="pattern_reviewer",
            model=bedrock_model,
            conversation_manager=make_conversation_manager(window_size=20, per_turn=True),
            tools=[read_carelink_csv, hourly_glucose_profile],
        ),
        "endocrinologist": Agent(
            system_prompt=PROMPTS["endocrinologist"],
            name="endocrinologist",
            model=bedrock_model,
            conversation_manager=make_conversation_manager(window_size=20),
            tools=[],
        ),
    }

Shared constants across all runs :)

CSV_PATH = "input_carelink.csv"
MODEL_ID = "global.anthropic.claude-haiku-4-5-20251001-v1:0"
REGION = "us-east-1"

PROMPT = (
    f"Analyze the Medtronic MiniMed 780G CareLink CSV export at '{CSV_PATH}'. "
    "Parse the data, compute glucose and insulin metrics, identify patterns, "
    "and provide clinical interpretation with actionable recommendations."
)

Now lets try out the 3 political systems

The Commune: Swarm

The swarm is anarchy by design. No central authority, no predefined order. Agents self-organize, decide when to hand off work, and collectively arrive at an answer. Think of a commune where everyone has a specialty but nobody has a boss — the CSV reader finishes and announces "data's ready," and whoever feels qualified picks it up next.
The promise: emergent intelligence. The reality: emergent shortcuts. The swarm ran only two nodes — csv_reader → endocrinologist — skipping the data analyst and pattern reviewer entirely. Without anyone enforcing the pipeline, the endocrinologist never got computed statistics. It estimated Time in Range as "likely <70%" when the actual value was 92.3%. No hourly glucose profiles were generated, no hypoglycemia clustering by timestamp.
And yet — the clinical insights were sharp. The swarm correctly identified carb miscounting, flagged insulin timing issues, and produced a well-prioritized five-tier recommendation system. The anarchists were sloppy with numbers but wise in judgment.
I took inspiration from the Strands agents samples repo for structuring the swarm roles. Although I decided to avoid being exact in my instructions -> anarchy

strands agents GitHub repo

agents = make_agents()

swarm = Swarm(
    [agents["csv_reader"], agents["data_analyst"], agents["pattern_reviewer"], agents["endocrinologist"]],
    entry_point=agents["csv_reader"],
    max_handoffs=10,
    max_iterations=15,
    execution_timeout=600.0,
    node_timeout=180.0,
    repetitive_handoff_detection_window=6,
    repetitive_handoff_min_unique_agents=3,
)

start = time.time()
swarm_result = swarm(PROMPT)
swarm_elapsed = time.time() - start

swarm_output = str(swarm_result)
swarm_metrics = extract_metrics(swarm_result)
swarm_metrics["wall_time_s"] = round(swarm_elapsed, 1)

print(swarm_output)
print_metrics(swarm_elapsed, swarm_metrics, "SWARM")

benchmark["swarm"] = {"output": swarm_output, "metrics": swarm_metrics}

Assembly line: Graph Pipeline

The graph pipeline is the Soviet factory model — a rigid, sequential process where each station does exactly one job and passes the product forward. No loops, no improvisation. Parse → Analyse → Review → Recommend, in that order, every time.
What the assembly line lacks in flexibility it makes up in thoroughness. Because each agent receives the full output of the previous one, nothing gets lost. The analyst computed exact TIR (92.3%), precise CV% (23.9%), and correct GMI (6.5%). The pattern reviewer clustered hypoglycemia events by timestamp. Every number was grounded in the actual data.
The downside? Every agent processes everything sequentially, which means 4x the token throughput. And the rigid structure means you can't easily skip steps or parallelize work. The factory runs at the speed of its slowest station.

agents = make_agents()

builder = GraphBuilder()
builder.add_node(agents["csv_reader"], "parse")
builder.add_node(agents["data_analyst"], "analyse")
builder.add_node(agents["pattern_reviewer"], "review")
builder.add_node(agents["endocrinologist"], "recommend")

builder.add_edge("parse", "analyse")
builder.add_edge("analyse", "review")
builder.add_edge("review", "recommend")
builder.set_entry_point("parse")

graph = builder.build()

start = time.time()
graph_result = graph(PROMPT)
graph_elapsed = time.time() - start


graph_output_parts = []
for node_id, node_result in graph_result.results.items():
    text = str(node_result.result) if hasattr(node_result, "result") else str(node_result)
    print(f"\n--- {node_id} ---")
    print(text)
    graph_output_parts.append(f"[{node_id}]\n{text}")

graph_output = "\n\n".join(graph_output_parts)
graph_metrics = extract_metrics(graph_result)
graph_metrics["wall_time_s"] = round(graph_elapsed, 1)

print_metrics(graph_elapsed, graph_metrics, "GRAPH")

benchmark["graph"] = {"output": graph_output, "metrics": graph_metrics}

Coordinator

The coordinator is the capitalist org chart. One manager, four direct reports. The manager decides who works on what, in what order, and synthesizes the final deliverable. The specialists are demoted to tools — they don't talk to each other, they report up.
Caching works differently across architectures. Both the swarm and coordinator benefit from prompt caching — the swarm's csv_reader cached 73K tokens across its internal conversation cycles, and the coordinator cached 57K tokens across its four sequential tool calls. The graph creates fresh agent contexts per node, so it gets zero cache hits. My extract_metrics function reported the swarm's cache as 0 at the top level, but the nested AgentInvocation metrics tell the real story.
The result: the coordinator combined the statistical precision of the graph with the clinical nuance of the swarm. It added measurable targets (e.g., "post-lunch glucose <9.0 mmol/L") and flagged a battery failure event as an "unacceptable safety risk." The graph also caught the battery failure in its pattern review, but the swarm — having skipped the pattern reviewer — missed it entirely.
An interesting self-correction happened inside the coordinator: the pattern reviewer estimated TIR as "~70-75%" (wrong), but the data analyst computed 92.3% (correct), and the final synthesized report used the right numbers. The corporate hierarchy's redundancy — multiple specialists touching the same data — caught errors that a single-pass architecture would miss.

agents = make_agents()

# Wrap each specialist as a tool for the coordinator
@tool
def parse_csv(file_path: str) -> str:
    """Parse a Medtronic CareLink CSV export and extract raw diabetes data."""
    return str(agents["csv_reader"](f"Parse the CareLink CSV at '{file_path}'"))

@tool
def analyse_data(parsed_data: str) -> str:
    """Compute glucose and insulin statistics on parsed CareLink data."""
    return str(agents["data_analyst"](parsed_data))

@tool
def review_patterns(metrics: str) -> str:
    """Identify clinically significant patterns and anomalies in diabetes metrics."""
    return str(agents["pattern_reviewer"](metrics))

@tool
def clinical_assessment(patterns: str) -> str:
    """Provide endocrinology clinical interpretation and pump setting recommendations."""
    return str(agents["endocrinologist"](patterns))

coordinator = Agent(
    system_prompt=(
        "You are a coordinator analyzing Medtronic MiniMed 780G insulin pump data.\n"
        "You have access to four specialist tools:\n"
        "  - parse_csv: extracts raw structured data from the CareLink CSV\n"
        "  - analyse_data: computes glucose statistics, TIR, GMI, CV%\n"
        "  - review_patterns: identifies glucose patterns and anomalies\n"
        "  - clinical_assessment: provides endocrinology interpretation\n"
        "Call them in order: parse_csv first, then analyse_data with the parsed output,\n"
        "then review_patterns with the analysis, then clinical_assessment with the patterns.\n"
        "After all specialists have contributed, synthesize their outputs into a final report.\n"
    ),
    name="coordinator",
    model=bedrock_model,
    conversation_manager=make_conversation_manager(window_size=25, per_turn=True),
    tools=[parse_csv, analyse_data, review_patterns, clinical_assessment],
)

start = time.time()
coordinator_result = coordinator(PROMPT)
coordinator_elapsed = time.time() - start

coordinator_output = str(coordinator_result)
coordinator_metrics = extract_metrics(coordinator_result)
coordinator_metrics["wall_time_s"] = round(coordinator_elapsed, 1)

print(coordinator_output)
print_metrics(coordinator_elapsed, coordinator_metrics, "COORDINATOR")

benchmark["coordinator"] = {"output": coordinator_output, "metrics": coordinator_metrics}

The Judge: LLM-as-Evaluator

To score each architecture's output, I used the Strands evaluator framework with Sonnet 4.5 as the judging model. The rubric scores five criteria, each weighted equally at 20%:

Data Completeness — Are glucose, insulin, carbs, and device status all covered?
Statistical Accuracy — Are TIR, GMI, CV%, and glucose stats correctly calculated and compared to clinical targets?
Pattern Identification — Are meaningful patterns (dawn phenomenon, post-meal spikes, hypo clustering) identified with specifics?
Clinical Recommendations — Are pump optimization suggestions specific, balanced, and actionable?
Safety & Framing — Are severe hypos flagged, and is the report framed as informational rather than medical advice?
In code this prompt is called rubric :

from strands_evals import Case, Experiment
from strands_evals.evaluators import OutputEvaluator

RUBRIC = """
You are an expert endocrinologist and diabetes data analyst evaluating
AI-generated reports on Medtronic MiniMed 780G insulin pump data.

Score each criterion from 0.0 to 1.0:

1. DATA COMPLETENESS (weight: 0.20)
   - Were Sensor Glucose values extracted and summarized?
   - Were insulin delivery records (basal, bolus, auto-correction) covered?
   - Were carbohydrate entries mentioned?
   - Were device settings or Auto Mode status addressed?
   Score 1.0 if all four are present. Deduct 0.25 per missing category.

2. STATISTICAL ACCURACY (weight: 0.20)
   - Were Time in Range (TIR) percentages reported?
   - Were glucose statistics (mean, median, SD, CV%) included?
   - Was GMI / estimated A1C calculated?
   - Were values compared against ADA/EASD consensus targets?
   Score 1.0 if all metrics present and correctly interpreted.

3. PATTERN IDENTIFICATION (weight: 0.20)
   - Were hypoglycemia patterns identified (timing, severity, frequency)?
   - Were post-meal spikes addressed?
   - Was dawn phenomenon or overnight trends discussed?
   - Were Auto Mode exits or insulin suspensions noted?
   Score 1.0 if clinically significant patterns are identified with specifics.

4. CLINICAL RECOMMENDATIONS (weight: 0.20)
   - Were pump setting adjustments suggested (AIT, carb ratios, target)?
   - Were recommendations specific and actionable (not generic)?
   - Were both problem areas and positive aspects acknowledged?
   - Was pre-bolusing timing discussed if relevant?
   Score 1.0 if recommendations are specific, balanced, and actionable.

5. SAFETY & FRAMING (weight: 0.20)
   - Were severe hypos (<54 mg/dL) flagged as safety concerns?
   - Was the report framed as informational, not medical advice?
   - Was the patient directed to discuss changes with their healthcare team?
   - Was Fiasp pharmacokinetics referenced appropriately if relevant?
   Score 1.0 if safety concerns are prominently flagged and framing is appropriate.

OVERALL SCORE: Weighted average of all five criteria.
Provide the overall score and a brief explanation for each criterion.
"""

evaluator = OutputEvaluator(
    rubric=RUBRIC,
    include_inputs=True,
    model="global.anthropic.claude-sonnet-4-5-20250929-v1:0"
)

# Build cases — one per pattern
pattern_names = list(benchmark.keys())
cases = []
for pattern_name in pattern_names:
    data = benchmark[pattern_name]
    cases.append(
        Case[str, str](
            name=pattern_name,
            input=PROMPT,
            expected_output=data["output"],
            metadata={"pattern": pattern_name},
        )
    )

def task_fn(case: Case) -> str:
    """Return the pre-computed output for evaluation."""
    return case.expected_output

experiment = Experiment[str, str](cases=cases, evaluators=[evaluator])
reports = experiment.run_evaluations(task_fn)

# reports is a list per evaluator — we have 1 evaluator
# report.scores is a list per case, report.reasons is a list per case
report = reports[0]

print(f"Overall score across all patterns: {report.overall_score}")
print()

for idx, pattern_name in enumerate(pattern_names):
    score = report.scores[idx] if idx < len(report.scores) else None
    reason = report.reasons[idx] if idx < len(report.reasons) else None
    benchmark[pattern_name]["eval_score"] = score
    benchmark[pattern_name]["eval_reasoning"] = reason
    print(f"{pattern_name}: {score}")

Results

	Score	Cost	Time	Cache	Verdict
Swarm anarchist	1.00	$0.057	56s	0	Anarchists on assembly line don't work
Graph	0.98	$0.315	337s	0	Expensive perfectionist
Coordinator	1.00	$0.071	395s	57K	Slow but right

Swarm cache hits were internal to the csv_reader's conversation cycles — extract_metrics reported 0 at the top level, but nested AgentInvocation metrics show 73K tokens read from cache.
The anarchists (swarm) and the corporate manager both scored 1.00. The factory scored lowest. And yet — the factory and coordinator both computed Time in Range (92.3%) from raw data. The swarm wrote "likely <70%" — off by 22 percentage points — and the judge didn't blink.
Swarm, passed the rubric. That's the headline.
The swarm skipped two agents entirely — csv_reader handed off straight to the endocrinologist, bypassing the data analyst and pattern reviewer. The endocrinologist improvised statistics from alert counts and data summaries instead of computing them from raw glucose values. Maybe the swarm architecture is more appropriate for creative non-deterministic work. Here we have a process, that required agents to talk to each other in specific way.

The graph was rigorous but couldn't surface trade-offs the upstream agent didn't frame.

*The coordinator * had an interesting self-correction: its pattern reviewer also estimated TIR wrong (~70-75%), but the data analyst computed it correctly, and the final synthesis used the right numbers. Redundancy caught what a single pass missed.

Both the swarm and coordinator got cache hits. The graph paid full price for every node — fresh agent context each time, zero caching.
Pick Your Politics

Swarm — cheap and fast, don't trust the numbers if you skip and agent or you don't have explicit handoff especially for deterministic use-cases. Graph — when precision justifies 5× the cost. Coordinator — the default.

The Real Takeaway
My rubric checked whether metrics appeared, not how they were derived. Maybe here we can put more work here. Add a provenance criterion — can each claim trace to a tool call? — and the swarm drops where it belongs.

Optimizing Multi-Agent Costs on Bedrock: From ~$18 to ~$7 per Diabetes Report Run (Graph vs Swarm Comparison)

Martin Nanchev — Thu, 12 Mar 2026 17:38:51 +0000

The Problem With 5,000 Rows of Blood Sugar Data

I've been living with Type 1 diabetes for over 17 years. My mother had it too, along with some of its complications. The disease hasn't changed much — but the tech around it has.

I use a MiniMed 780G insulin pump with a Guardian 4 CGM sensor running in SmartGuard auto mode. Every 14 days it produces a report: a ~5,000-row CSV of pump events and CGM readings plus pdf data.

I wanted something better — a structured clinical summary that's actually useful for medical staff and for people, that makes sense of patterns. And because I'm a DevOps engineer who can't resist over-engineering things, I decided to benchmark two multi-agent architectures against each other:

Graph (sequential pipeline)
Swarm (autonomous handoffs)

This article covers Graph and Swarm.

Architecture 1: The Graph Pipeline

Four agents, one after another. Each does its job and passes results to the next:

Reader — Ingests the raw CareLink CSV. Extracts CGM glucose readings, insulin delivery (basal/bolus), timestamps, sensor metadata. Flags data quality issues like gaps and sensor warmup periods.

Analyser — Crunches the numbers with Python: Time in Range (TIR), GMI, CV%, time-block patterns. Cross-validates against the PDF report.

Reviewer — The sceptic. Checks the analysis for statistical validity, flags confounders like compression lows and sensor first-day artefacts, separates validated findings from questionable ones.

Endocrinologist — Takes everything and writes a clinical consultation report with pump setting recommendations and discussion points for the next endo visit.

The Code

Setup — imports, PDF tool, and model config

import logging
from pypdf import PdfReader
from strands.models import BedrockModel
from strands import Agent, tool
from strands.multiagent import GraphBuilder
from strands_tools import python_repl, calculator, file_read
import os
from strands.agent.conversation_manager import SlidingWindowConversationManager

os.environ["BYPASS_TOOL_CONSENT"] = "true"

@tool
def read_pdf(filename: str) -&gt; str:
    """Reads the content of a pdf file and returns it as a string list"""
    try:
        reader = PdfReader(filename)
        texts = []
        for page in reader.pages:
            texts.append(page.extract_text())
        return '\n'.join(texts)
    except Exception as e:
        return f"Error reading file {filename}: {str(e)}"

model = BedrockModel(
    model_id="global.anthropic.claude-opus-4-6",
    region_name="us-east-1"
)

Agent definitions — four specialists, each with a clear handoff

reader = Agent(
    name="reader",
    system_prompt="""You extract and structure diabetes management data from files 
    in the current directory. Files follow the pattern 'X DD-MM-YYYY.csv' and 
    'X DD-MM-YYYY.pdf'.

    Extract: CGM glucose readings, insulin delivery (basal/bolus), timestamps, 
    sensor events. Flag any data quality issues (gaps, sensor warmup, anomalies).

    HANDOFF: Summarize date range, data completeness %, readings extracted.""",
    tools=[file_read, read_pdf],
    model=model
)

analyser = Agent(
    name="analyser",
    system_prompt="""You perform quantitative analysis on structured diabetes data.

    Calculate: TIR (3.9-10.0), hypo/hyper percentages, GMI, CV%, average glucose, 
    basal/bolus ratio, patterns by time block (overnight/morning/afternoon/evening).

    HANDOFF: All computed metrics, identified patterns with confidence levels, 
    data quality caveats.""",
    tools=[calculator, python_repl, file_read, read_pdf],
    model=model
)

reviewer = Agent(
    name="reviewer",
    system_prompt="""You critically review diabetes data analysis.

    Assess: data sufficiency, pattern validity, confounders (weekend vs weekday, 
    sensor first-day inaccuracy, compression lows), risk prioritization.

    HANDOFF: Validated findings, disputed findings, risk-prioritized concerns.""",
    tools=[],
    model=model
)

endocrinologist = Agent(
    name="endocrinologist",
    system_prompt="""You are a virtual endocrinology consultant. The patient uses 
    a MiniMed 780G with Guardian 4 CGM in SmartGuard auto mode.

    Produce: executive summary, wins, priority concerns, actionable recommendations 
    (active insulin time, carb ratios, targets), discussion points for next visit.""",
    tools=[],
    model=model
)

Wiring the graph — four nodes, three edges, linear flow

builder = GraphBuilder()
builder.add_node(reader, "reader")
builder.add_node(analyser, "analyser")
builder.add_node(reviewer, "reviewer")
builder.add_node(endocrinologist, "endocrinologist")

builder.add_edge("reader", "analyser")
builder.add_edge("analyser", "reviewer")
builder.add_edge("reviewer", "endocrinologist")
builder.set_entry_point("reader")

graph = builder.build()
result = graph("Analyze the diabetes data files and produce a clinical report.")

What Came Out

Here's the executive summary the system produced, condensed:

Excellent Type 1 diabetes control — 92.5% Time in Range, GMI 6.4%, low glucose variability — placing among the top 5–10% of international T1D outcomes. The MiniMed 780G is performing particularly well overnight and fasting. Main safety concern: three severe hypoglycaemic episodes (<3.0 mmol/L) within 14 days. Secondary optimisation: afternoon post-meal hyperglycaemia (14:00–17:00). Recommended approach: reduce hypo risk first (adjust active insulin time and glucose target), then address lunch-related spikes through earlier pre-bolusing and improved carb counting.

Clinically useful? I think so.

What impressed me most was how deep the review went. The reviewer noticed that the reader’s rough guess of TIR (55–65%) was actually way off — when they calculated it from all 3,716 data points, it was 92.5%. They also spotted things the analyzer completely missed, like differences between weekends and weekdays, first-day sensor issues, and compression lows. And the endocrinologist summed it up perfectly by saying it’s about “fine-tuning, not an overhaul” — which is exactly how you’d approach someone who’s already at 92% TIR.

The $18 Problem

Here's where it gets uncomfortable. The unoptimised graph: 18 minutes, ~3.3 million tokens:

Node	Input Tokens	Output Tokens	Total Cost
reader	1,623,805	19,252	$8.60
analyser	1,667,440	40,461	$9.35
reviewer	5,810	7,780	$0.22
endocrinologist	~3,331	~4,242	~$0.25
Total	~3.3M	~67K	$18.42

$18.42 for a single report. The reader and analyser are the culprits — they're passing the entire conversation history (including all tool call results) as context to each subsequent model invocation. The 5,000-row CSV gets re-read and re-sent multiple times.

My agents were going through tokens like people go through rakia (alcohol drink) at a village wedding.

Cutting It With a Sliding Window

One-liner fix per agent. Strands has a SlidingWindowConversationManager that caps how much conversation history gets sent to the model:

Sliding window configuration — add to reader and analyser

reader = Agent(
    name="reader",
    system_prompt="...",  # same as before
    tools=[file_read, read_pdf],
    model=model,
    conversation_manager=SlidingWindowConversationManager(
        window_size=10,
        should_truncate_results=True,
        per_turn=True,
    ),
)

analyser = Agent(
    name="analyser",
    system_prompt="...",  # same as before
    tools=[calculator, python_repl, file_read, read_pdf],
    model=model,
    conversation_manager=SlidingWindowConversationManager(
        window_size=10,
        should_truncate_results=True,
        per_turn=True,
    ),
)

The reviewer and endocrinologist don't need it — their input is already small.

Token reduction by node:

Node	Previous (no manager)	With Sliding Window	Savings
reader	1,623,805 input	1,054,919 input	-35%
analyser	1,667,440 input	135,770 input	-92%
reviewer	5,810 input	1,260 input	-78%
endocrinologist	~3,331 input	~3,331 input	same

Cost after optimisation:

Node	Input Cost	Output Cost	Total
reader	$5.27	$0.31	$5.58
analyser	$0.68	$0.39	$1.07
reviewer	$0.01	$0.08	$0.09
endocrinologist	$0.02	$0.13	$0.15
Total			$6.89

From $18.42 to $6.89 — a 63% cost reduction with no meaningful loss in output quality. The analyser benefited most because it was the worst offender: every python_repl tool call was accumulating in context.

Architecture 2: The Swarm

Swarm is the wildest of the three. No predefined sequence — agents share context and decide for themselves who to talk to next. Less assembly line, more group chat where everyone's an expert. A life without a manager, basically.

Doesn't mean it's cheaper though. Shared context means every agent sees what every other agent said, and token counts snowball. I hit ~3.6M tokens and some failed runs trying this on Opus. So I got pragmatic: Sonnet 4.6 as the coordinator, three worker agents on Sonnet 4. Costs came down, the thing actually finished.The Sonnet 4 is actually the default model for the agent swarm tool

Swarm setup — Sonnet coordinator, Sonnet workers

from strands.multiagent import Swarm

agent = Agent(
    name="Diabetes analyser",
    model=BedrockModel(model_id="global.anthropic.claude-sonnet-4-6"),
    tools=[swarm, python_repl, file_read, calculator]
)

result = agent(
    "Create three agents to analyse 'X 10-03-2026.csv' in current folder"
)

Three swarm agents — glucose analyst, insulin analyst, clinical advisor — each working their own angle on the data.

What Came Out

The Swarm got the numbers right. TIR 92.5%, afternoon variability, the hypo events — all there. The clinical advisor generated standard recommendations: pre-bolusing, carb counting, alert management.

But the report read more like a template than a consultation. Generic advice about rotating sensor sites and keeping firmware updated sat next to the actual data-specific findings. The coordinator's summary was basically "here's what each agent did" rather than a unified clinical document. I think I was wrong a bit, setting the coordinator to the larger model, while keeping the workers smaller or older model. I think another approach will be using Haiku for coordinator, while leaving each agent to use Opus or Sonnet. This will lead to more depth, but I was careful with the tokens or rakia as well, so i decided to use the default Sonnet

The Showdown: Graph vs Swarm

Same 14-day CareLink export, both architectures. Here's what happened.

Tokens & Cost

Metric	Graph (Opus, optimised)	Swarm (Opus coord + Sonnet agents)
Total tokens	~900K	~175K
Cost	~$6.89	~$2.50
Time	~11 min	~2.5 min
Worker model	Opus (all 4 nodes)	Sonnet 4 (3 agents)
Coordinator	N/A (sequential)	Opus

Swarm is 3x cheaper and 4x faster. But this comparison is again a bit unfair — the Graph runs everything on the most expensive model, while the Swarm pushes the heavy lifting to Sonnet.

Where the Graph pulled ahead

Self-correction. The reviewer noticed the reader's TIR estimate (55–65%) was way off and flagged 92.5% as the real number. The Swarm had no mechanism for one agent to challenge another.

Confounders. The reviewer flagged that the Level 2 hypo events weren't checked against the sensor change date (Mar 5). If any of them fell on that day, they could be first-day sensor artefacts, not real hypos or low blood glucose. Swarm didn't consider this.

Compression lows. Overnight readings below range could include compression lows — you roll onto the sensor and it reads artificially low. Graph flagged it. Swarm didn't.

Weekend vs weekday. The reviewer noted this wasn't assessed at all. The afternoon variability might look completely different on weekends vs workdays. Swarm just generated recommendations without thinking about it.

Pattern prioritisation. The reviewer took the "12.6 bolus entries/day" grazing pattern, upgraded it from moderate to high confidence, and called it the single most actionable finding. Swarm saw the same number but didn't do anything with it.

Specific recommendations. The endocrinologist said "weaken afternoon CR by 10–15%", "verify AIT is 3.0–3.5 hours", "do NOT lower SmartGuard target until hypos are addressed." The Swarm said "rotate sensor sites" and "keep firmware updated." One of these I can take to my doctor. The other I already know.

Where the Swarm held up

The core metrics were correct — TIR, GMI, CV% all matched. Parallel analysis was efficient. And for a quick "how's my last two weeks looking?" check, it's perfectly fine.

The thing I need to be honest about

I can't cleanly separate architecture quality from model quality here. And that bugs me.

The Graph runs all four nodes on Opus. The Swarm runs the workers on Sonnet. So when the Graph's reviewer catches confounders that the Swarm misses — is that the sequential pipeline being better, or is it just Opus being smarter than Sonnet?

Probably both. My gut says ~60% model, ~40% architecture. The confounder analysis — compression lows, sensor artefacts, weekend splits — that's Opus-level reasoning that Sonnet doesn't typically do unprompted. But the architecture gave Opus a dedicated step to do that reasoning. The Swarm doesn't have a reviewer step even if you ran it all on Opus.

The fair test would be:

Graph with Sonnet everywhere vs Swarm with Sonnet (isolate architecture)
Graph with Opus everywhere vs Swarm with Opus (same thing, higher tier)

The Verdict

Dimension	Graph	Swarm
Output quality	★★★★★	★★★☆☆
Clinical depth	Deep, specific	Competent, generic
Cost (optimised)	$6.89	~$2.50
Speed	~11 min	~2.5 min
Error correction	Built-in (reviewer)	None
Best for	Clinic-ready reports	Weekly trend checks

Graph wins on quality. I'd hand the Graph report to my endocrinologist. The Swarm report I'd keep for myself.

Swarm wins on speed and cost. For a quick "how's my last two weeks" check, it gives 80% of the insight at 30% of the cost.

As we say in Bulgaria: — "A good word opens even an iron door." The Graph doesn't just give you more words — it gives you the right ones. When you're handing a report to the person managing your chronic condition, that matters.

Key Takeaways

1. Context management is everything. Sliding window: $18 → $7. 63% cut. Zero quality loss. If you're building multi-agent pipelines, do this first. Watch your analyser nodes — any agent with tool use will balloon context because every tool call/result pair stays in history.

2. Review steps are worth it. The reviewer cost $0.09 and caught a wrong TIR estimate, four unassessed confounders, and upgraded the most actionable finding. Best nine cents I've ever spent.

3. Model and architecture are tangled. I used Opus for Graph and Sonnet for Swarm workers. That means I can't say for sure whether the Graph's better output is architecture or just Opus being Opus. Probably both — 60/40 model/architecture is my guess. Need to test properly.

4. Pick the right tool. Graph for depth, Swarm for speed. "Which is better?" → "Better at what?"

5. The output is actually useful. Whether $7 per report is worth it — or at least I think so. My endo agreed with the recommendations. That's the test that matters.

6. Start with the smallest model. It could do magic and be really useful.

How much did the over - engineering costs? Any additional costs?

I always thought this will cost a couple of bucks max.
Well I ran these agents a couple of times - 20 for development and there is a catch about the pricing. There are additional costs for long context. Luckily I am Community Builder, so I don't need to pay the $500 for using the models unresponsive. Use the models responsibly and start with the smallest possible model

The smart way of centralizing VPC endpoints with Route 53 Profiles

Martin Nanchev — Wed, 27 Aug 2025 15:27:46 +0000

How to Centralize Endpoints Smartly

A few months ago in April, AWS introduced a new feature to the Route 53 arsenal: Route 53 Profiles. One would think—ah, another AWS feature to manage DNS centrally. But there's much more to it than that.

Basically, Amazon realized that managing DNS across multiple environments is about as organized as a toddler's toy box or socks after laundry day (one of them falls victim to the sock-eating monster). So they created profiles—separate rule sets you can share between AWS accounts. It's like having one proper manual that everyone follows, instead of each department making up their own rules and calling it "innovation." When it goes wrong—which it will—you know exactly which profile to blame and how to fix it.

At a high level, the design allows you to, for example, share VPC endpoints and centralize them.

You create a central VPC with all of the VPC interface endpoints. Why interface? Because they are transitive compared to gateway endpoints and can be reused.
You can keep private DNS enabled on your centralized endpoints. No need to manually create hosted zones or resolver rules.
Create a Route 53 profile.
Simply associate endpoints with a Route 53 profile in the hub account and the hub VPC.
Share the profile via AWS Resource Access Manager.
Associate spoke VPCs to the profile—done.

Reality Check

This will save you between 84-87% of costs per hour by centralizing the endpoints according to my previous article. In reality, after one month living with this centralization technique, the savings were around 70%, mainly because I forgot to account for the attachment costs between accounts and the traffic.

Other Benefits

Remember that DNS Firewall feature in VPC, where you can define a rule group and domain list of whitelisted domains? Now you can associate them with the Route 53 profile as well and share them between accounts via Resource Access Manager.
One last—but not least—thing is that you can associate the profile with a private hosted zone. Afterward, we share this Route 53 profile with RAM again and associate it with each and every VPC that needs it.

Summary

Route 53 Profiles simplify the life of DevOps engineers and Solutions Architects by giving you the ability to share firewall rule groups, hosted zones, and VPC endpoints between accounts. Why does this matter? Besides being easy and removing the heavy lifting, we now have some saved money in our pockets and we're more sustainable—fewer network interfaces are better for the environment. And don't try to argue; I know your VPC endpoints are already underutilized.

Now you would ask: "Is there Terraform for it?"
Of course there is!

Terraform Code

resource "aws_route53profiles_profile" "primorsko" {
  name = "dtpl-r53-profiles"
  tags = {
    Environment = "ProbkoTestov testva surfa na primorsko"
  }
}

resource "aws_route53profiles_association" "spoke" {
  name        = "spoke"
  profile_id  = aws_route53profiles_profile.primorsko.id
  resource_id = var.hub_vpc_id
}

resource "aws_route53profiles_association" "hub" {
  name        = "hub"
  profile_id  = aws_route53profiles_profile.primorsko.id
  resource_id = var.spoke_vpc_id
}

resource "aws_route53profiles_resource_association" "ssm" {
  name         = "ssm"
  profile_id   = aws_route53profiles_profile.primorsko.id
  resource_arn = var.vpc_endpoint_ec2
}

resource "aws_route53profiles_resource_association" "ssmmessages" {
  name         = "ssmmessages"
  profile_id   = aws_route53profiles_profile.primorsko.id
  resource_arn = var.vpc_endpoint_ssmmessages
}

resource "aws_route53profiles_resource_association" "ec2messages" {
  name         = "ec2messages"
  profile_id   = aws_route53profiles_profile.primorsko.id
  resource_arn = var.vpc_endpoint_ec2messages # This is legacy; you can go without it
}

Build Games Challenge: Build Tower of Hanoi with Amazon Q Developer CLI

Martin Nanchev — Wed, 02 Jul 2025 17:44:54 +0000

The Frog and the Ox: Why I Started This Project
Why Tower of Hanoi?
Getting Started with Amazon Q Developer CLI
The Zero-Shot Miracle
- The First Success
The Design Challenge
The Material Design Evolution
The Final Polish: Advanced Zero-Shot
Technical Insights
Code Architecture Highlights
Performance and User Experience
Reflections on AI-Assisted Development
Conclusion: The Frog's Success
Try It Yourself

Building Tower of Hanoi with Amazon Q Developer CLI: A Journey from Zero-Shot to Polished Game

The Frog and the Ox: Why I Started This Project

A few days ago, I stumbled upon an AWS Community post about the Build Games Challenge using Amazon Q Developer CLI. It reminded me of a Bulgarian proverb: "The frog saw the ox being shod and lifted her leg too." This saying captures the essence of someone copying others blindly, trying to be part of something they clearly don't belong to—like a frog thinking she's an ox just because she saw the ox getting horseshoes.

That's exactly why I decided to build the Tower of Hanoi game and participate in the challenge. Part nostalgia, part curiosity, and maybe a little bit of that frog mentality.

Why Tower of Hanoi?

Tower of Hanoi holds a special place in my programming journey. I first coded it back in university, and it's stuck with me ever since. The game's elegant simplicity masks its mathematical complexity—moving a set of disks from one pole to another while following just a few rules:

Three rods/poles total
Move only one disk at a time
Never place a larger disk on top of a smaller one

The mathematical beauty lies in its solution: the minimum number of moves needed is 2^n - 1, where n is the number of disks.

This time, I wanted to bring it to life in a new way—powered by Amazon Q Developer CLI, without the hard memories of Java classes that haunted my university days.

Getting Started with Amazon Q Developer CLI

Setting up Amazon Q Developer CLI is refreshingly simple. If you're on macOS with Homebrew:

brew install amazon-q

Once installed, just type q in your terminal to start chatting with your AI coding companion.

The Zero-Shot Miracle

My first prompt was deliberately naive:

help me build tower of hanoi using pygame framework

The results absolutely stunned me. Without any examples, rules, or detailed specifications, Amazon Q Developer CLI generated a fully functional Tower of Hanoi game using pygame. Zero-shot prompting at its finest.

The First Success

The initial game was surprisingly complete:

✅ Interactive gameplay with mouse controls
✅ Visual feedback and animations
✅ Rules enforcement
✅ Move counter
✅ Auto-solve functionality
✅ Variable disk counts

Here's what that first iteration looked like:

Clean, functional, but lacking visual polish

The Design Challenge

Emboldened by this success, I decided to push further. If zero-shot prompting could create a working game, surely it could handle design improvements, right?

My next prompt:

make it more polished using modern ui design patterns like material design

Plot twist: This didn't work as smoothly. The generated code was incomplete, cutting off mid-function. After several attempts and follow-up prompts like "the last function draw_rounded_rect was not finished," I finally got a working solution—but it required manual debugging.

Lessons Learned

Zero-shot works best for complete, well-defined tasks
Incremental changes can be trickier than starting fresh
Always review and test AI-generated code
Be prepared to debug and iterate

The Material Design Evolution

After some back-and-forth, I achieved a much more polished version with Material Design principles:

Modern Material Design aesthetic with elevated cards, shadows, and proper color palette

Key Material Design Elements Added:

Elevated surfaces with subtle shadows
Material color palette (Blue 500, Orange 500, etc.)
Rounded corners and proper spacing
Card-based UI for controls
Visual hierarchy with proper typography
Hover states and interactive feedback

The Final Polish: Advanced Zero-Shot

For my final iteration, I crafted a comprehensive prompt that specified exactly what I wanted:

Create a complete, polished Tower of Hanoi game using Python Pygame framework, styled according to Google's Material Design principles. Include:

Visually appealing, responsive UI with Material Design elements

Smooth animations and visual feedback

Drag-and-drop or click-to-move functionality

Rules enforcement and user-friendly error handling

Move counter, elapsed timer, and reset functionality

Level selection (3-7 disks)

Light/dark theme toggle

Modular, well-documented OOP code

Material Design color palettes and shadows

This comprehensive prompt yielded the most impressive result:

The final version with enhanced responsiveness and refined interactions

New Features in the Final Version:

Enhanced mouse interactions with better responsiveness
Improved visual feedback for user actions
Smoother animations and transitions
Better error handling and edge case management
More intuitive UI with clearer visual hierarchy

Technical Insights

What Amazon Q Developer CLI Excelled At:

Complete game logic implementation
Pygame framework integration
Object-oriented design patterns
Mathematical algorithm implementation (recursive Hanoi solver)
Event handling and game state management

Where It Struggled:

Incremental design changes to existing code
Complex prompt continuation when code was cut off
Fine-tuning visual details without explicit guidance

The Sweet Spot:

The most effective approach was comprehensive, specific prompts that clearly defined the entire scope of what I wanted. This worked better than trying to modify existing code piecemeal.

Code Architecture Highlights

The final implementation showcased excellent software engineering practices:

Key architectural decisions:

Separation of concerns between game logic and rendering
Event-driven architecture for user interactions
State management for game progression
Modular design for easy extensibility

Performance and User Experience

The final game delivers on multiple fronts:

Responsive controls with immediate visual feedback
Intuitive interface following Material Design principles
Accessibility considerations with clear visual hierarchy
Error prevention through smart UI design

Reflections on AI-Assisted Development

This project revealed fascinating insights about working with AI coding assistants:

The Good:

Rapid prototyping from concept to working game
Best practices implementation without explicit instruction
Complex algorithm generation (recursive Hanoi solver)
Framework expertise beyond what I could have written alone

The Challenging:

Iteration difficulties when making incremental changes
Need for human oversight and debugging
Prompt engineering importance for optimal results

The Surprising:

Zero-shot capability exceeded expectations for complete tasks
Code quality was consistently high and well-structured
Documentation and comments were thoughtfully included

Conclusion: The Frog's Success

Looking back at that Bulgarian proverb, maybe sometimes it's okay to be the frog lifting her leg when she sees the ox being shod. In this case, the "copying" led to genuine learning and a surprisingly sophisticated result.

Amazon Q Developer CLI proved to be an impressive coding companion, especially for:

Complete project generation from clear specifications
Framework-specific implementations with best practices
Algorithm implementation with proper optimization

The key lesson? Be specific, be comprehensive, and be prepared to iterate. The most successful prompts were those that painted a complete picture of the desired outcome rather than asking for incremental modifications.

Try It Yourself

Want to experiment with Amazon Q Developer CLI? Start with a clear, comprehensive prompt for a complete project rather than trying to modify existing code. You might be surprised by what you can achieve with the right approach to AI-assisted development.

The Tower of Hanoi might be an ancient puzzle, but building it with modern AI tools offers fresh insights into both game development and the evolving landscape of programming assistance.

Have you tried building games with AI coding assistants? What was your experience? Share your own "frog and ox" moments in the comments below.

Code initial version:

import pygame
import sys
import time

# Initialize pygame
pygame.init()

# Constants
WIDTH, HEIGHT = 800, 600
DISK_HEIGHT = 20
MAX_DISKS = 5
ANIMATION_SPEED = 5

# Colors
BACKGROUND = (50, 50, 50)
TOWER_COLOR = (139, 69, 19)
DISK_COLORS = [
    (255, 0, 0),    # Red
    (255, 165, 0),  # Orange
    (255, 255, 0),  # Yellow
    (0, 255, 0),    # Green
    (0, 0, 255),    # Blue
]

# Set up the display
screen = pygame.display.set_mode((WIDTH, HEIGHT))
pygame.display.set_caption("Tower of Hanoi")
clock = pygame.time.Clock()

class Disk:
    def __init__(self, size, color):
        self.size = size
        self.color = color
        self.x = 0
        self.y = 0
        self.moving = False
        self.target_x = 0
        self.target_y = 0

    def draw(self):
        width = (self.size + 1) * 30
        pygame.draw.rect(screen, self.color, (self.x - width // 2, self.y - DISK_HEIGHT // 2, width, DISK_HEIGHT), 0, 5)

    def move_towards_target(self):
        dx = self.target_x - self.x
        dy = self.target_y - self.y

        if abs(dx) < ANIMATION_SPEED and abs(dy) < ANIMATION_SPEED:
            self.x = self.target_x
            self.y = self.target_y
            self.moving = False
            return True

        if abs(dx) > 0:
            self.x += ANIMATION_SPEED if dx > 0 else -ANIMATION_SPEED

        if abs(dy) > 0:
            self.y += ANIMATION_SPEED if dy > 0 else -ANIMATION_SPEED

        return False

class Tower:
    def __init__(self, x):
        self.x = x
        self.y = HEIGHT - 100
        self.disks = []

    def draw(self):
        # Draw tower base
        pygame.draw.rect(screen, TOWER_COLOR, (self.x - 10, self.y, 20, 20))
        pygame.draw.rect(screen, TOWER_COLOR, (self.x - 50, self.y + 20, 100, 10))

        # Draw tower pole
        pygame.draw.rect(screen, TOWER_COLOR, (self.x - 5, self.y - 200, 10, 200))

    def add_disk(self, disk):
        disk_y = self.y - len(self.disks) * DISK_HEIGHT - DISK_HEIGHT // 2
        disk.x = self.x
        disk.y = disk_y
        self.disks.append(disk)

    def remove_top_disk(self):
        if self.disks:
            return self.disks.pop()
        return None

    def can_add_disk(self, disk):
        if not self.disks:
            return True
        return disk.size < self.disks[-1].size

class Game:
    def __init__(self, num_disks=3):
        self.num_disks = min(num_disks, MAX_DISKS)
        self.towers = [
            Tower(WIDTH // 4),
            Tower(WIDTH // 2),
            Tower(3 * WIDTH // 4)
        ]
        self.moves = 0
        self.selected_tower = None
        self.selected_disk = None
        self.moving_disk = None
        self.auto_solving = False
        self.solution_steps = []
        self.solution_index = 0
        self.last_move_time = 0
        self.game_won = False

        # Initialize the first tower with disks
        for i in range(self.num_disks, 0, -1):
            disk = Disk(i - 1, DISK_COLORS[(i - 1) % len(DISK_COLORS)])
            self.towers[0].add_disk(disk)

    def draw(self):
        screen.fill(BACKGROUND)

        # Draw towers
        for tower in self.towers:
            tower.draw()

        # Draw disks
        for tower in self.towers:
            for disk in tower.disks:
                disk.draw()

        # Draw moving disk
        if self.moving_disk:
            self.moving_disk.draw()

        # Draw move counter
        font = pygame.font.SysFont('Arial', 24)
        moves_text = font.render(f"Moves: {self.moves}", True, (255, 255, 255))
        screen.blit(moves_text, (20, 20))

        # Draw win message
        if self.game_won:
            win_font = pygame.font.SysFont('Arial', 48)
            win_text = win_font.render("You Win!", True, (255, 215, 0))
            screen.blit(win_text, (WIDTH // 2 - win_text.get_width() // 2, 50))

        # Draw buttons
        self.draw_buttons()

        pygame.display.flip()

    def draw_buttons(self):
        # Reset button
        pygame.draw.rect(screen, (200, 200, 200), (20, HEIGHT - 60, 100, 40), 0, 5)
        font = pygame.font.SysFont('Arial', 20)
        reset_text = font.render("Reset", True, (0, 0, 0))
        screen.blit(reset_text, (45, HEIGHT - 50))

        # Auto Solve button
        pygame.draw.rect(screen, (200, 200, 200), (140, HEIGHT - 60, 120, 40), 0, 5)
        solve_text = font.render("Auto Solve", True, (0, 0, 0))
        screen.blit(solve_text, (155, HEIGHT - 50))

        # Disk count buttons
        for i in range(1, MAX_DISKS + 1):
            pygame.draw.rect(screen, (200, 200, 200), (280 + (i-1)*60, HEIGHT - 60, 50, 40), 0, 5)
            disk_text = font.render(str(i), True, (0, 0, 0))
            screen.blit(disk_text, (300 + (i-1)*60, HEIGHT - 50))

    def handle_click(self, pos):
        x, y = pos

        # Check if a tower was clicked
        if not self.auto_solving and y < HEIGHT - 70:
            for i, tower in enumerate(self.towers):
                if abs(x - tower.x) < 50:
                    self.handle_tower_click(i)
                    return

        # Check if reset button was clicked
        if 20 <= x <= 120 and HEIGHT - 60 <= y <= HEIGHT - 20:
            self.reset()
            return

        # Check if auto solve button was clicked
        if 140 <= x <= 260 and HEIGHT - 60 <= y <= HEIGHT - 20:
            self.start_auto_solve()
            return

        # Check if disk count buttons were clicked
        for i in range(1, MAX_DISKS + 1):
            if 280 + (i-1)*60 <= x <= 330 + (i-1)*60 and HEIGHT - 60 <= y <= HEIGHT - 20:
                self.reset(i)
                return

    def handle_tower_click(self, tower_index):
        if self.selected_tower is None:
            # No tower selected yet, try to select this one
            if self.towers[tower_index].disks:
                self.selected_tower = tower_index
                self.selected_disk = self.towers[tower_index].remove_top_disk()
                self.moving_disk = self.selected_disk
                self.moving_disk.moving = True
                self.moving_disk.target_x = self.towers[tower_index].x
                self.moving_disk.target_y = 100  # Move up
        else:
            # A tower was already selected, try to move the disk
            if tower_index == self.selected_tower:
                # Put the disk back
                self.towers[tower_index].add_disk(self.selected_disk)
            elif self.towers[tower_index].can_add_disk(self.selected_disk):
                # Move the disk to the new tower
                self.moving_disk.target_x = self.towers[tower_index].x
                self.moving_disk.target_y = self.towers[tower_index].y - len(self.towers[tower_index].disks) * DISK_HEIGHT - DISK_HEIGHT // 2
                self.towers[tower_index].add_disk(self.selected_disk)
                self.moves += 1

                # Check if the game is won
                if len(self.towers[2].disks) == self.num_disks:
                    self.game_won = True
            else:
                # Invalid move, put the disk back
                self.towers[self.selected_tower].add_disk(self.selected_disk)

            self.selected_tower = None
            self.selected_disk = None
            self.moving_disk = None

    def reset(self, num_disks=None):
        if num_disks is not None:
            self.num_disks = min(num_disks, MAX_DISKS)

        self.towers = [
            Tower(WIDTH // 4),
            Tower(WIDTH // 2),
            Tower(3 * WIDTH // 4)
        ]

        for i in range(self.num_disks, 0, -1):
            disk = Disk(i - 1, DISK_COLORS[(i - 1) % len(DISK_COLORS)])
            self.towers[0].add_disk(disk)

        self.moves = 0
        self.selected_tower = None
        self.selected_disk = None
        self.moving_disk = None
        self.auto_solving = False
        self.solution_steps = []
        self.solution_index = 0
        self.game_won = False

    def solve_hanoi(self, n, source, target, auxiliary, steps):
        if n > 0:
            self.solve_hanoi(n-1, source, auxiliary, target, steps)
            steps.append((source, target))
            self.solve_hanoi(n-1, auxiliary, target, source, steps)

    def start_auto_solve(self):
        if self.auto_solving:
            return

        # Reset the game first
        self.reset(self.num_disks)

        # Generate solution steps
        self.solution_steps = []
        self.solve_hanoi(self.num_disks, 0, 2, 1, self.solution_steps)
        self.solution_index = 0
        self.auto_solving = True
        self.last_move_time = time.time()

    def update_auto_solve(self):
        if not self.auto_solving or self.solution_index >= len(self.solution_steps):
            return

        current_time = time.time()
        if current_time - self.last_move_time < 1.0:  # Wait 1 second between moves
            return

        source, target = self.solution_steps[self.solution_index]

        # Make the move
        if self.towers[source].disks:
            disk = self.towers[source].remove_top_disk()
            self.towers[target].add_disk(disk)
            self.moves += 1

            # Check if the game is won
            if len(self.towers[2].disks) == self.num_disks:
                self.game_won = True
                self.auto_solving = False

        self.solution_index += 1
        self.last_move_time = current_time

        if self.solution_index >= len(self.solution_steps):
            self.auto_solving = False

def main():
    game = Game(3)

    running = True
    while running:
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                running = False
            elif event.type == pygame.MOUSEBUTTONDOWN:
                if event.button == 1:  # Left mouse button
                    game.handle_click(event.pos)

        game.update_auto_solve()
        game.draw()
        clock.tick(60)

    pygame.quit()
    sys.exit()

if __name__ == "__main__":
    main()

Code result material non polished:

import pygame
import sys
import time

# Initialize pygame
pygame.init()

# Constants
WIDTH, HEIGHT = 800, 600
DISK_HEIGHT = 24
MAX_DISKS = 5
ANIMATION_SPEED = 5

# Material Design Colors
BACKGROUND = (245, 245, 245)    # Grey 100
PRIMARY_COLOR = (33, 150, 243)   # Blue 500
ACCENT_COLOR = (255, 152, 0)     # Orange 500
TOWER_COLOR = (96, 125, 139)     # Blue Grey 500
TEXT_PRIMARY = (33, 33, 33)      # Grey 900
TEXT_SECONDARY = (117, 117, 117) # Grey 600
BUTTON_COLOR = (255, 255, 255)   # White
BUTTON_HOVER = (238, 238, 238)   # Grey 200
WIN_COLOR = (76, 175, 80)        # Green 500

# Material Design Disk Colors
DISK_COLORS = [
    (244, 67, 54),   # Red 500
    (156, 39, 176),  # Purple 500
    (76, 175, 80),   # Green 500
    (3, 169, 244),   # Light Blue 500
    (255, 193, 7),   # Amber 500
]

# Set up the display
screen = pygame.display.set_mode((WIDTH, HEIGHT))
pygame.display.set_caption("Tower of Hanoi - Material Design")
clock = pygame.time.Clock()

# Helper function for drawing rounded rectangles with shadow effect
def draw_material_rect(surface, color, rect, radius=8, elevation=2):
    x, y, width, height = rect

    # Draw shadow (if elevation > 0)
    if elevation > 0:
        shadow_rect = (x, y + elevation, width, height)
        pygame.draw.rect(surface, (0, 0, 0, 50), shadow_rect, 0, radius)

    # Draw the rounded rectangle
    pygame.draw.rect(surface, color, (x, y, width, height), 0, radius)

# Helper function for drawing material buttons
def draw_material_button(surface, rect, text, font, color=BUTTON_COLOR, text_color=TEXT_PRIMARY, elevation=2):
    draw_material_rect(surface, color, rect, 4, elevation)
    text_surf = font.render(text, True, text_color)
    text_rect = text_surf.get_rect(center=(rect[0] + rect[2]//2, rect[1] + rect[3]//2))
    surface.blit(text_surf, text_rect)

class Disk:
    def __init__(self, size, color):
        self.size = size
        self.color = color
        self.x = 0
        self.y = 0
        self.moving = False
        self.target_x = 0
        self.target_y = 0

    def draw(self):
        width = (self.size + 1) * 30
        # Draw disk with elevation effect
        shadow_rect = (self.x - width // 2, self.y - DISK_HEIGHT // 2 + 2, width, DISK_HEIGHT)
        pygame.draw.rect(screen, (0, 0, 0, 30), shadow_rect, 0, DISK_HEIGHT // 2)

        # Draw main disk
        disk_rect = (self.x - width // 2, self.y - DISK_HEIGHT // 2, width, DISK_HEIGHT)
        pygame.draw.rect(screen, self.color, disk_rect, 0, DISK_HEIGHT // 2)

        # Add a subtle highlight on top
        highlight_rect = (self.x - width // 2, self.y - DISK_HEIGHT // 2, width, DISK_HEIGHT // 4)
        highlight_color = tuple(min(c + 30, 255) for c in self.color[:3])
        pygame.draw.rect(screen, highlight_color, highlight_rect, 0, DISK_HEIGHT // 2)

    def move_towards_target(self):
        dx = self.target_x - self.x
        dy = self.target_y - self.y

        if abs(dx) < ANIMATION_SPEED and abs(dy) < ANIMATION_SPEED:
            self.x = self.target_x
            self.y = self.target_y
            self.moving = False
            return True

        if abs(dx) > 0:
            self.x += ANIMATION_SPEED if dx > 0 else -ANIMATION_SPEED

        if abs(dy) > 0:
            self.y += ANIMATION_SPEED if dy > 0 else -ANIMATION_SPEED

        return False

class Tower:
    def __init__(self, x):
        self.x = x
        self.y = HEIGHT - 100
        self.disks = []

    def draw(self):
        # Draw tower base with material design
        base_width = 120

        # Draw base shadow
        pygame.draw.rect(screen, (0, 0, 0, 30), (self.x - base_width//2, self.y + 22, base_width, 8), 0, 4)

        # Draw base
        pygame.draw.rect(screen, TOWER_COLOR, (self.x - base_width//2, self.y + 20, base_width, 8), 0, 4)

        # Draw tower pole with subtle gradient
        pole_height = 220
        for i in range(pole_height):
            # Create subtle gradient effect
            shade = max(0, min(20, i // 10))
            color = tuple(max(0, min(255, c + shade)) for c in TOWER_COLOR[:3])
            pygame.draw.rect(screen, color, (self.x - 4, self.y - pole_height + i, 8, 1))

    def add_disk(self, disk):
        disk_y = self.y - len(self.disks) * DISK_HEIGHT - DISK_HEIGHT // 2
        disk.x = self.x
        disk.y = disk_y
        self.disks.append(disk)

    def remove_top_disk(self):
        if self.disks:
            return self.disks.pop()
        return None

    def can_add_disk(self, disk):
        if not self.disks:
            return True
        return disk.size < self.disks[-1].size

class Game:
    def __init__(self, num_disks=3):
        self.num_disks = min(num_disks, MAX_DISKS)
        self.towers = [
            Tower(WIDTH // 4),
            Tower(WIDTH // 2),
            Tower(3 * WIDTH // 4)
        ]
        self.moves = 0
        self.selected_tower = None
        self.selected_disk = None
        self.moving_disk = None
        self.auto_solving = False
        self.solution_steps = []
        self.solution_index = 0
        self.last_move_time = 0
        self.game_won = False

        # Load fonts with fallbacks
        try:
            self.title_font = pygame.font.SysFont('Roboto', 36)
            self.main_font = pygame.font.SysFont('Roboto', 24)
            self.button_font = pygame.font.SysFont('Roboto', 18)
        except:
            self.title_font = pygame.font.SysFont(None, 36)
            self.main_font = pygame.font.SysFont(None, 24)
            self.button_font = pygame.font.SysFont(None, 18)

        # Initialize the first tower with disks
        for i in range(self.num_disks, 0, -1):
            disk = Disk(i - 1, DISK_COLORS[(i - 1) % len(DISK_COLORS)])
            self.towers[0].add_disk(disk)

    def draw(self):
        screen.fill(BACKGROUND)

        # Draw app bar
        pygame.draw.rect(screen, PRIMARY_COLOR, (0, 0, WIDTH, 60))
        title_text = self.title_font.render("Tower of Hanoi", True, (255, 255, 255))
        screen.blit(title_text, (20, 15))

        # Draw towers
        for tower in self.towers:
            tower.draw()

        # Draw disks
        for tower in self.towers:
            for disk in tower.disks:
                disk.draw()

        # Draw moving disk
        if self.moving_disk:
            self.moving_disk.draw()

        # Draw move counter with card style
        counter_rect = (WIDTH - 150, 70, 130, 50)
        draw_material_rect(screen, (255, 255, 255), counter_rect, 4, 2)
        moves_text = self.main_font.render(f"Moves: {self.moves}", True, TEXT_PRIMARY)
        screen.blit(moves_text, (WIDTH - 140, 85))

        # Draw win message with material card
        if self.game_won:
            win_rect = (WIDTH // 2 - 150, 70, 300, 60)
            draw_material_rect(screen, WIN_COLOR, win_rect, 4, 3)
            win_text = self.title_font.render("You Win!", True, (255, 255, 255))
            screen.blit(win_text, (WIDTH // 2 - win_text.get_width() // 2, 85))

        # Draw buttons
        self.draw_buttons()

        pygame.display.flip()

    def draw_buttons(self):
        # Create a card for the controls
        control_card_rect = (20, HEIGHT - 80, WIDTH - 40, 60)
        draw_material_rect(screen, (255, 255, 255), control_card_rect, 4, 3)

        # Reset button
        reset_rect = (40, HEIGHT - 70, 100, 40)
        draw_material_button(screen, reset_rect, "Reset", self.button_font, PRIMARY_COLOR, (255, 255, 255))

        # Auto Solve button
        solve_rect = (160, HEIGHT - 70, 120, 40)
        draw_material_button(screen, solve_rect, "Auto Solve", self.button_font, ACCENT_COLOR, (255, 255, 255))

        # Disk count buttons
        for i in range(1, MAX_DISKS + 1):
            disk_rect = (300 + (i-1)*80, HEIGHT - 70, 60, 40)
            draw_material_button(screen, disk_rect, str(i), self.button_font,
                                BUTTON_COLOR if i != self.num_disks else PRIMARY_COLOR,
                                TEXT_PRIMARY if i != self.num_disks else (255, 255, 255))

    def handle_click(self, pos):
        x, y = pos

        # Check if a tower was clicked
        if not self.auto_solving and y < HEIGHT - 90 and y > 60:
            for i, tower in enumerate(self.towers):
                if abs(x - tower.x) < 60:
                    self.handle_tower_click(i)
                    return

        # Check if reset button was clicked
        if 40 <= x <= 140 and HEIGHT - 70 <= y <= HEIGHT - 30:
            self.reset()
            return

        # Check if auto solve button was clicked
        if 160 <= x <= 280 and HEIGHT - 70 <= y <= HEIGHT - 30:
            self.start_auto_solve()
            return

        # Check if disk count buttons were clicked
        for i in range(1, MAX_DISKS + 1):
            if 300 + (i-1)*80 <= x <= 360 + (i-1)*80 and HEIGHT - 70 <= y <= HEIGHT - 30:
                self.reset(i)
                return

    def handle_tower_click(self, tower_index):
        if self.selected_tower is None:
            # No tower selected yet, try to select this one
            if self.towers[tower_index].disks:
                self.selected_tower = tower_index
                self.selected_disk = self.towers[tower_index].remove_top_disk()
                self.moving_disk = self.selected_disk
                self.moving_disk.moving = True
                self.moving_disk.target_x = self.towers[tower_index].x
                self.moving_disk.target_y = 100  # Move up
        else:
            # A tower was already selected, try to move the disk
            if tower_index == self.selected_tower:
                # Put the disk back
                self.towers[tower_index].add_disk(self.selected_disk)
            elif self.towers[tower_index].can_add_disk(self.selected_disk):
                # Move the disk to the new tower
                self.moving_disk.target_x = self.towers[tower_index].x
                self.moving_disk.target_y = self.towers[tower_index].y - len(self.towers[tower_index].disks) * DISK_HEIGHT - DISK_HEIGHT // 2
                self.towers[tower_index].add_disk(self.selected_disk)
                self.moves += 1

                # Check if the game is won
                if len(self.towers[2].disks) == self.num_disks:
                    self.game_won = True
            else:
                # Invalid move, put the disk back
                self.towers[self.selected_tower].add_disk(self.selected_disk)

            self.selected_tower = None
            self.selected_disk = None
            self.moving_disk = None

    def reset(self, num_disks=None):
        if num_disks is not None:
            self.num_disks = min(num_disks, MAX_DISKS)

        self.towers = [
            Tower(WIDTH // 4),
            Tower(WIDTH // 2),
            Tower(3 * WIDTH // 4)
        ]

        for i in range(self.num_disks, 0, -1):
            disk = Disk(i - 1, DISK_COLORS[(i - 1) % len(DISK_COLORS)])
            self.towers[0].add_disk(disk)

        self.moves = 0
        self.selected_tower = None
        self.selected_disk = None
        self.moving_disk = None
        self.auto_solving = False
        self.solution_steps = []
        self.solution_index = 0
        self.game_won = False

    def solve_hanoi(self, n, source, target, auxiliary, steps):
        if n > 0:
            self.solve_hanoi(n-1, source, auxiliary, target, steps)
            steps.append((source, target))
            self.solve_hanoi(n-1, auxiliary, target, source, steps)

    def start_auto_solve(self):
        if self.auto_solving:
            return

        # Reset the game first
        self.reset(self.num_disks)

        # Generate solution steps
        self.solution_steps = []
        self.solve_hanoi(self.num_disks, 0, 2, 1, self.solution_steps)
        self.solution_index = 0
        self.auto_solving = True
        self.last_move_time = time.time()

    def update_auto_solve(self):
        if not self.auto_solving or self.solution_index >= len(self.solution_steps):
            return

        current_time = time.time()
        if current_time - self.last_move_time < 1.0:  # Wait 1 second between moves
            return

        source, target = self.solution_steps[self.solution_index]

        # Make the move
        if self.towers[source].disks:
            disk = self.towers[source].remove_top_disk()
            self.towers[target].add_disk(disk)
            self.moves += 1

            # Check if the game is won
            if len(self.towers[2].disks) == self.num_disks:
                self.game_won = True
                self.auto_solving = False

        self.solution_index += 1
        self.last_move_time = current_time

        if self.solution_index >= len(self.solution_steps):
            self.auto_solving = False

def main():
    game = Game(3)

    running = True
    while running:
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                running = False
            elif event.type == pygame.MOUSEBUTTONDOWN:
                if event.button == 1:  # Left mouse button
                    game.handle_click(event.pos)

        game.update_auto_solve()
        game.draw()
        clock.tick(60)

    pygame.quit()
    sys.exit()

if __name__ == "__main__":
    main()

Code result material polished:

import random

import pygame
import sys
import time
import math

# Initialize pygame
pygame.init()

# Constants
WIDTH, HEIGHT = 800, 600
FPS = 60
MIN_DISKS = 3
MAX_DISKS = 7

# Material Design Colors
LIGHT_THEME = {
    "background": (245, 245, 245),  # Grey 100
    "surface": (255, 255, 255),     # White
    "primary": (33, 150, 243),      # Blue 500
    "primary_dark": (25, 118, 210), # Blue 700
    "primary_light": (100, 181, 246), # Blue 300
    "secondary": (255, 152, 0),     # Orange 500
    "text_primary": (33, 33, 33),   # Grey 900
    "text_secondary": (117, 117, 117), # Grey 600
    "tower": (96, 125, 139),        # Blue Grey 500
    "error": (244, 67, 54),         # Red 500
    "success": (76, 175, 80),       # Green 500
}

DARK_THEME = {
    "background": (48, 48, 48),     # Grey 900
    "surface": (66, 66, 66),        # Grey 800
    "primary": (33, 150, 243),      # Blue 500
    "primary_dark": (25, 118, 210), # Blue 700
    "primary_light": (100, 181, 246), # Blue 300
    "secondary": (255, 152, 0),     # Orange 500
    "text_primary": (255, 255, 255),# White
    "text_secondary": (189, 189, 189), # Grey 400
    "tower": (176, 190, 197),       # Blue Grey 300
    "error": (244, 67, 54),         # Red 500
    "success": (76, 175, 80),       # Green 500
}

# Material Design Disk Colors - Light Theme
DISK_COLORS_LIGHT = [
    (244, 67, 54),    # Red 500
    (156, 39, 176),   # Purple 500
    (33, 150, 243),   # Blue 500
    (76, 175, 80),    # Green 500
    (255, 193, 7),    # Amber 500
    (255, 87, 34),    # Deep Orange 500
    (0, 188, 212),    # Cyan 500
]

# Material Design Disk Colors - Dark Theme
DISK_COLORS_DARK = [
    (229, 115, 115),  # Red 300
    (186, 104, 200),  # Purple 300
    (100, 181, 246),  # Blue 300
    (129, 199, 132),  # Green 300
    (255, 213, 79),   # Amber 300
    (255, 138, 101),  # Deep Orange 300
    (77, 208, 225),   # Cyan 300
]

# Set up the display
screen = pygame.display.set_mode((WIDTH, HEIGHT))
pygame.display.set_caption("Tower of Hanoi - Material Design")
clock = pygame.time.Clock()

# Helper function for drawing rounded rectangles
def draw_rounded_rect(surface, color, rect, radius=10, border=0, border_color=(0, 0, 0)):
    """Draw a rounded rectangle with optional border"""
    x, y, width, height = rect

    if border > 0:
        # Draw border first (as a slightly larger rectangle)
        pygame.draw.rect(surface, border_color,
                        (x-border, y-border, width+2*border, height+2*border),
                        0, radius+border)

    # Draw the main rectangle
    pygame.draw.rect(surface, color, (x, y, width, height), 0, radius)

# Helper function for drawing material shadows
def draw_shadow(surface, rect, radius=10, alpha=30, offset=(0, 4), blur=4):
    """Draw a soft shadow under a rectangle"""
    shadow_surf = pygame.Surface((rect[2] + blur * 2, rect[3] + blur * 2), pygame.SRCALPHA)
    pygame.draw.rect(shadow_surf, (0, 0, 0, alpha),
                    (blur, blur, rect[2], rect[3]), 0, radius)

    # Apply simple blur by scaling down and up
    scale_factor = 0.5
    small_surf = pygame.transform.smoothscale(shadow_surf,
                                            (int(shadow_surf.get_width() * scale_factor),
                                             int(shadow_surf.get_height() * scale_factor)))
    blurred = pygame.transform.smoothscale(small_surf, shadow_surf.get_size())

    # Blit the shadow
    surface.blit(blurred, (rect[0] - blur + offset[0], rect[1] - blur + offset[1]))

# Helper function for drawing material buttons
def draw_material_button(surface, rect, text, font, colors, is_hover=False, is_active=False):
    """Draw a material design button with hover and active states"""
    base_color = colors["primary"] if is_active else colors["surface"]
    text_color = colors["text_primary"] if not is_active else (255, 255, 255)

    # Draw shadow
    if not is_hover:
        draw_shadow(surface, rect, radius=4, offset=(0, 2), blur=4)
    else:
        draw_shadow(surface, rect, radius=4, offset=(0, 4), blur=6)

    # Draw button
    draw_rounded_rect(surface, base_color, rect, radius=4)

    # Draw ripple effect if hovering
    if is_hover and not is_active:
        hover_color = (*base_color[:3], 20)  # Semi-transparent overlay
        hover_surf = pygame.Surface((rect[2], rect[3]), pygame.SRCALPHA)
        pygame.draw.rect(hover_surf, hover_color, (0, 0, rect[2], rect[3]), 0, 4)
        surface.blit(hover_surf, (rect[0], rect[1]))

    # Draw text
    text_surf = font.render(text, True, text_color)
    text_rect = text_surf.get_rect(center=(rect[0] + rect[2]//2, rect[1] + rect[3]//2))
    surface.blit(text_surf, text_rect)

# Helper function for drawing material cards
def draw_material_card(surface, rect, colors, elevation=2):
    """Draw a material design card with elevation"""
    # Draw shadow
    draw_shadow(surface, rect, radius=8, offset=(0, elevation), blur=elevation*2)

    # Draw card
    draw_rounded_rect(surface, colors["surface"], rect, radius=8)

class Disk:
    def __init__(self, size, color, theme_colors):
        self.size = size
        self.base_color = color
        self.theme_colors = theme_colors
        self.x = 0
        self.y = 0
        self.target_x = 0
        self.target_y = 0
        self.moving = False
        self.dragging = False
        self.drag_offset_x = 0
        self.drag_offset_y = 0
        self.width = (size + 1) * 30
        self.height = 24
        self.animation_progress = 0
        self.animation_duration = 0.3  # seconds
        self.animation_start_time = 0
        self.animation_start_pos = (0, 0)

    def draw(self, surface):
        # Calculate disk rectangle
        rect = (self.x - self.width // 2, self.y - self.height // 2, self.width, self.height)

        # Draw shadow
        if not self.dragging:
            shadow_offset = 2
            draw_shadow(surface, rect, radius=self.height//2, offset=(0, shadow_offset), blur=4)
        else:
            # Larger shadow when dragging
            shadow_offset = 6
            draw_shadow(surface, rect, radius=self.height//2, offset=(0, shadow_offset), blur=8, alpha=40)

        # Draw disk with rounded corners
        draw_rounded_rect(surface, self.base_color, rect, radius=self.height//2)

        # Add a subtle highlight on top for 3D effect
        highlight_rect = (self.x - self.width // 2, self.y - self.height // 2, self.width, self.height // 3)
        highlight_color = tuple(min(c + 30, 255) for c in self.base_color[:3])
        draw_rounded_rect(surface, highlight_color, highlight_rect, radius=self.height//2)

    def contains_point(self, point):
        """Check if the disk contains the given point"""
        return (abs(point[0] - self.x) <= self.width // 2 and
                abs(point[1] - self.y) <= self.height // 2)

    def start_drag(self, mouse_pos):
        """Start dragging the disk"""
        self.dragging = True
        self.drag_offset_x = self.x - mouse_pos[0]
        self.drag_offset_y = self.y - mouse_pos[1]

    def update_drag(self, mouse_pos):
        """Update the disk position while dragging"""
        if self.dragging:
            self.x = mouse_pos[0] + self.drag_offset_x
            self.y = mouse_pos[1] + self.drag_offset_y

    def end_drag(self):
        """End dragging the disk"""
        self.dragging = False

    def start_animation(self, target_x, target_y):
        """Start animating the disk to a new position"""
        self.moving = True
        self.animation_start_time = time.time()
        self.animation_progress = 0
        self.animation_start_pos = (self.x, self.y)
        self.target_x = target_x
        self.target_y = target_y

    def update_animation(self):
        """Update the disk animation"""
        if not self.moving:
            return False

        current_time = time.time()
        elapsed = current_time - self.animation_start_time
        self.animation_progress = min(elapsed / self.animation_duration, 1.0)

        # Use easeOutCubic easing function for smooth animation
        progress = 1 - (1 - self.animation_progress) ** 3

        # Update position
        self.x = self.animation_start_pos[0] + (self.target_x - self.animation_start_pos[0]) * progress
        self.y = self.animation_start_pos[1] + (self.target_y - self.animation_start_pos[1]) * progress

        # Check if animation is complete
        if self.animation_progress >= 1.0:
            self.x = self.target_x
            self.y = self.target_y
            self.moving = False
            return True

        return False

class Tower:
    def __init__(self, x, y, theme_colors):
        self.x = x
        self.y = y
        self.theme_colors = theme_colors
        self.disks = []
        self.base_width = 120
        self.pole_height = 220
        self.pole_width = 8
        self.highlight = False
        self.highlight_alpha = 0
        self.highlight_direction = 1

    def draw(self, surface):
        tower_color = self.theme_colors["tower"]

        # Draw base shadow
        base_rect = (self.x - self.base_width//2, self.y, self.base_width, 10)
        draw_shadow(surface, base_rect, radius=5, offset=(0, 2), blur=4)

        # Draw base
        draw_rounded_rect(surface, tower_color, base_rect, radius=5)

        # Draw tower pole with subtle gradient
        for i in range(self.pole_height):
            # Create subtle gradient effect
            shade = max(0, min(20, i // 10))
            color = tuple(max(0, min(255, c + shade)) for c in tower_color[:3])
            pygame.draw.rect(surface, color,
                            (self.x - self.pole_width//2, self.y - self.pole_height + i,
                             self.pole_width, 1))

        # Draw highlight if this tower is a valid drop target
        if self.highlight:
            self.highlight_alpha += self.highlight_direction * 5
            if self.highlight_alpha >= 60:
                self.highlight_alpha = 60
                self.highlight_direction = -1
            elif self.highlight_alpha <= 20:
                self.highlight_alpha = 20
                self.highlight_direction = 1

            highlight_color = (*self.theme_colors["primary"][:3], self.highlight_alpha)
            highlight_surf = pygame.Surface((self.base_width + 20, self.pole_height + 10), pygame.SRCALPHA)
            pygame.draw.rect(highlight_surf, highlight_color,
                            (0, 0, self.base_width + 20, self.pole_height + 10), 0, 10)
            surface.blit(highlight_surf,
                        (self.x - (self.base_width + 20)//2, self.y - self.pole_height - 5))

    def add_disk(self, disk):
        """Add a disk to this tower"""
        disk_y = self.y - len(self.disks) * disk.height - disk.height // 2
        disk.x = self.x
        disk.y = disk_y
        self.disks.append(disk)

    def remove_top_disk(self):
        """Remove and return the top disk from this tower"""
        if self.disks:
            return self.disks.pop()
        return None

    def can_add_disk(self, disk):
        """Check if a disk can be added to this tower"""
        if not self.disks:
            return True
        return disk.size < self.disks[-1].size

    def get_top_disk(self):
        """Get the top disk without removing it"""
        if self.disks:
            return self.disks[-1]
        return None

    def contains_point(self, point):
        """Check if the tower contains the given point for dropping"""
        return abs(point[0] - self.x) < self.base_width // 2

    def get_top_position(self):
        """Get the position for a new disk at the top of the tower"""
        disk_y = self.y - len(self.disks) * 24 - 24 // 2
        return self.x, disk_y

    def set_highlight(self, highlight):
        """Set whether this tower should be highlighted as a valid drop target"""
        self.highlight = highlight

class Button:
    def __init__(self, x, y, width, height, text, theme_colors, action=None):
        self.rect = (x, y, width, height)
        self.text = text
        self.theme_colors = theme_colors
        self.action = action
        self.hover = False
        self.active = False

        # Load font
        try:
            self.font = pygame.font.SysFont('Roboto', 18)
        except:
            self.font = pygame.font.SysFont(None, 18)

    def draw(self, surface):
        draw_material_button(surface, self.rect, self.text, self.font,
                            self.theme_colors, self.hover, self.active)

    def contains_point(self, point):
        x, y, width, height = self.rect
        return (x <= point[0] <= x + width and y <= point[1] <= y + height)

    def set_hover(self, hover):
        self.hover = hover

    def set_active(self, active):
        self.active = active

    def click(self):
        if self.action:
            self.action()

class Game:
    def __init__(self, num_disks=3, dark_mode=False):
        self.num_disks = min(max(num_disks, MIN_DISKS), MAX_DISKS)
        self.dark_mode = dark_mode
        self.theme = DARK_THEME if dark_mode else LIGHT_THEME
        self.disk_colors = DISK_COLORS_DARK if dark_mode else DISK_COLORS_LIGHT

        # Game state
        self.moves = 0
        self.start_time = time.time()
        self.elapsed_time = 0
        self.game_won = False
        self.show_win_animation = False
        self.win_animation_start = 0
        self.win_particles = []

        # Tower setup
        tower_y = HEIGHT - 100
        self.towers = [
            Tower(WIDTH // 4, tower_y, self.theme),
            Tower(WIDTH // 2, tower_y, self.theme),
            Tower(3 * WIDTH // 4, tower_y, self.theme)
        ]

        # Disk interaction state
        self.selected_disk = None
        self.source_tower = None
        self.last_valid_position = (0, 0)

        # Initialize the first tower with disks
        for i in range(self.num_disks, 0, -1):
            disk = Disk(i - 1, self.disk_colors[(i - 1) % len(self.disk_colors)], self.theme)
            self.towers[0].add_disk(disk)

        # Load fonts
        try:
            self.title_font = pygame.font.SysFont('Roboto', 36)
            self.main_font = pygame.font.SysFont('Roboto', 24)
            self.button_font = pygame.font.SysFont('Roboto', 18)
        except:
            self.title_font = pygame.font.SysFont(None, 36)
            self.main_font = pygame.font.SysFont(None, 24)
            self.button_font = pygame.font.SysFont(None, 18)

        # Create UI buttons
        self.create_buttons()

        # Feedback message
        self.feedback_message = ""
        self.feedback_color = self.theme["text_primary"]
        self.feedback_timer = 0

    def create_buttons(self):
        """Create all UI buttons"""
        self.buttons = []

        # Reset button
        self.buttons.append(Button(20, HEIGHT - 60, 100, 40, "Reset", self.theme,
                                  action=lambda: self.reset()))

        # Theme toggle button
        theme_text = "Light Mode" if self.dark_mode else "Dark Mode"
        self.buttons.append(Button(140, HEIGHT - 60, 120, 40, theme_text, self.theme,
                                  action=lambda: self.toggle_theme()))

        # Disk count buttons
        for i in range(MIN_DISKS, MAX_DISKS + 1):
            x_pos = 280 + (i - MIN_DISKS) * 70
            self.buttons.append(Button(x_pos, HEIGHT - 60, 60, 40, str(i), self.theme,
                                      action=lambda i=i: self.reset(i)))

    def toggle_theme(self):
        """Toggle between light and dark themes"""
        self.dark_mode = not self.dark_mode
        self.theme = DARK_THEME if self.dark_mode else LIGHT_THEME
        self.disk_colors = DISK_COLORS_DARK if self.dark_mode else DISK_COLORS_LIGHT

        # Update tower colors
        for tower in self.towers:
            tower.theme_colors = self.theme

        # Update disk colors
        for tower in self.towers:
            for i, disk in enumerate(tower.disks):
                disk.theme_colors = self.theme
                disk.base_color = self.disk_colors[(disk.size) % len(self.disk_colors)]

        if self.selected_disk:
            self.selected_disk.theme_colors = self.theme
            self.selected_disk.base_color = self.disk_colors[(self.selected_disk.size) % len(self.disk_colors)]

        # Recreate buttons with new theme
        self.create_buttons()

    def draw(self, surface):
        # Fill background
        surface.fill(self.theme["background"])

        # Draw app bar
        pygame.draw.rect(surface, self.theme["primary"], (0, 0, WIDTH, 60))
        title_text = self.title_font.render("Tower of Hanoi", True, (255, 255, 255))
        surface.blit(title_text, (20, 15))

        # Draw towers
        for tower in self.towers:
            tower.draw(surface)

        # Draw disks on towers
        for tower in self.towers:
            for disk in tower.disks:
                disk.draw(surface)

        # Draw selected disk (being dragged) on top
        if self.selected_disk:
            self.selected_disk.draw(surface)

        # Draw move counter and timer
        self.draw_stats(surface)

        # Draw buttons
        for button in self.buttons:
            button.draw(surface)

        # Draw disk count indicator
        self.draw_disk_count_indicator(surface)

        # Draw feedback message
        if self.feedback_message and time.time() < self.feedback_timer:
            self.draw_feedback(surface)

        # Draw win animation
        if self.show_win_animation:
            self.draw_win_animation(surface)

        # Draw win message
        if self.game_won:
            self.draw_win_message(surface)

    def draw_stats(self, surface):
        # Create a card for stats
        stats_rect = (WIDTH - 200, 70, 180, 80)
        draw_material_card(surface, stats_rect, self.theme)

        # Draw move counter
        moves_text = self.main_font.render(f"Moves: {self.moves}", True, self.theme["text_primary"])
        surface.blit(moves_text, (WIDTH - 180, 85))

        # Draw timer
        minutes = int(self.elapsed_time) // 60
        seconds = int(self.elapsed_time) % 60
        time_text = self.main_font.render(f"Time: {minutes:02d}:{seconds:02d}", True,
                                         self.theme["text_primary"])
        surface.blit(time_text, (WIDTH - 180, 115))

    def draw_disk_count_indicator(self, surface):
        # Draw text above disk count buttons
        text = self.button_font.render("Number of Disks:", True, self.theme["text_primary"])
        surface.blit(text, (280, HEIGHT - 90))

        # Highlight the current disk count button
        for button in self.buttons[2:]:  # Skip Reset and Theme buttons
            button.set_active(button.text == str(self.num_disks))

    def draw_feedback(self, surface):
        text = self.main_font.render(self.feedback_message, True, self.feedback_color)
        text_rect = text.get_rect(center=(WIDTH // 2, HEIGHT - 100))
        surface.blit(text, text_rect)

    def draw_win_message(self, surface):
        # Create a card for the win message
        win_rect = (WIDTH // 2 - 150, 70, 300, 60)
        draw_material_card(surface, win_rect, self.theme, elevation=4)

        # Draw win text
        win_text = self.title_font.render("You Win!", True, self.theme["success"])
        text_rect = win_text.get_rect(center=(WIDTH // 2, 100))
        surface.blit(win_text, text_rect)

    def draw_win_animation(self, surface):
        # Generate particles
        if time.time() - self.win_animation_start < 2.0:
            if len(self.win_particles) < 100 and random.random() < 0.3:
                self.win_particles.append({
                    'x': random.randint(0, WIDTH),
                    'y': random.randint(0, HEIGHT // 2),
                    'size': random.randint(5, 15),
                    'color': random.choice([self.theme["primary"], self.theme["secondary"],
                                           self.theme["success"]]),
                    'speed': random.uniform(1, 3),
                    'angle': random.uniform(0, 2 * math.pi)
                })

        # Update and draw particles
        particles_to_keep = []
        for particle in self.win_particles:
            # Update position
            particle['y'] += particle['speed']
            particle['x'] += math.sin(particle['angle']) * 0.5

            # Draw particle
            pygame.draw.circle(surface, particle['color'],
                              (int(particle['x']), int(particle['y'])),
                              particle['size'])

            # Keep particles that are still on screen
            if particle['y'] < HEIGHT:
                particles_to_keep.append(particle)

        self.win_particles = particles_to_keep

        # End animation after a while
        if time.time() - self.win_animation_start > 3.0:
            self.show_win_animation = False

    def update(self):
        # Update elapsed time if game is not won
        if not self.game_won:
            self.elapsed_time = time.time() - self.start_time

        # Update disk animations
        for tower in self.towers:
            for disk in tower.disks:
                disk.update_animation()

    def handle_click(self, pos):
        # Check if a button was clicked
        for button in self.buttons:
            if button.contains_point(pos):
                button.click()
                return

        # Don't allow moves if game is won
        if self.game_won:
            return

        # Check if a tower was clicked
        if not self.selected_disk:
            # Try to select a disk
            for i, tower in enumerate(self.towers):
                if tower.disks and tower.get_top_disk().contains_point(pos):
                    self.source_tower = i
                    self.selected_disk = tower.remove_top_disk()
                    self.selected_disk.start_drag(pos)
                    self.last_valid_position = (self.selected_disk.x, self.selected_disk.y)
                    return
        else:
            # Try to place the disk
            self.end_disk_drag()

    def handle_mouse_motion(self, pos):
        # Update button hover states
        for button in self.buttons:
            button.set_hover(button.contains_point(pos))

        # Update selected disk position
        if self.selected_disk:
            self.selected_disk.update_drag(pos)

            # Highlight valid drop towers
            for i, tower in enumerate(self.towers):
                can_drop = tower.contains_point(pos) and tower.can_add_disk(self.selected_disk)
                tower.set_highlight(can_drop)

    def handle_mouse_up(self, pos):
        if self.selected_disk:
            self.end_disk_drag()

    def end_disk_drag(self):
        """End dragging the selected disk and place it on a tower if valid"""
        if not self.selected_disk:
            return

        valid_drop = False

        # Check if the disk is over a valid tower
        for i, tower in enumerate(self.towers):
            if tower.contains_point((self.selected_disk.x, self.selected_disk.y)):
                if tower.can_add_disk(self.selected_disk):
                    # Valid move
                    target_x, target_y = tower.get_top_position()
                    self.selected_disk.start_animation(target_x, target_y)
                    tower.add_disk(self.selected_disk)
                    self.selected_disk = None

                    # Update move counter
                    if i != self.source_tower:
                        self.moves += 1

                    # Check if the game is won
                    if len(self.towers[2].disks) == self.num_disks:
                        self.game_won = True
                        self.show_win_animation = True
                        self.win_animation_start = time.time()
                        import random
                        self.win_particles = []

                    valid_drop = True
                    break
                else:
                    # Invalid move - show feedback
                    self.show_feedback("Invalid move: Can't place larger disk on smaller disk",
                                      self.theme["error"])

        # If no valid drop, return the disk to its original tower
        if not valid_drop:
            self.towers[self.source_tower].add_disk(self.selected_disk)
            target_x, target_y = self.last_valid_position
            self.selected_disk.start_animation(target_x, target_y)
            self.selected_disk = None

        # Clear tower highlights
        for tower in self.towers:
            tower.set_highlight(False)

    def show_feedback(self, message, color=None):
        """Show a feedback message for a short time"""
        self.feedback_message = message
        self.feedback_color = color if color else self.theme["text_primary"]
        self.feedback_timer = time.time() + 2.0  # Show for 2 seconds

    def reset(self, num_disks=None):
        """Reset the game with the specified number of disks"""
        if num_disks is not None:
            self.num_disks = min(max(num_disks, MIN_DISKS), MAX_DISKS)

        # Reset game state
        self.moves = 0
        self.start_time = time.time()
        self.elapsed_time = 0
        self.game_won = False
        self.show_win_animation = False
        self.selected_disk = None
        self.source_tower = None

        # Reset towers
        self.towers = [
            Tower(WIDTH // 4, HEIGHT - 100, self.theme),
            Tower(WIDTH // 2, HEIGHT - 100, self.theme),
            Tower(3 * WIDTH // 4, HEIGHT - 100, self.theme)
        ]

        # Initialize the first tower with disks
        for i in range(self.num_disks, 0, -1):
            disk = Disk(i - 1, self.disk_colors[(i - 1) % len(self.disk_colors)], self.theme)
            self.towers[0].add_disk(disk)

        # Update disk count indicator
        for button in self.buttons[2:]:  # Skip Reset and Theme buttons
            button.set_active(button.text == str(self.num_disks))



def main():
    # Import needed only for win animation
    import random

    # Initialize game
    game = Game(3)

    # Main game loop
    running = True
    while running:
        # Handle events
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                running = False
            elif event.type == pygame.MOUSEBUTTONDOWN:
                if event.button == 1:  # Left mouse button
                    game.handle_click(event.pos)
            elif event.type == pygame.MOUSEMOTION:
                game.handle_mouse_motion(event.pos)
            elif event.type == pygame.MOUSEBUTTONUP:
                if event.button == 1:  # Left mouse button
                    game.handle_mouse_up(event.pos)

        # Update game state
        game.update()

        # Draw everything
        game.draw(screen)

        # Update display
        pygame.display.flip()

        # Cap the frame rate
        clock.tick(FPS)

    # Clean up
    pygame.quit()
    sys.exit()

if __name__ == "__main__":
    main()

How to burn money on VPC Endpoints

Martin Nanchev — Tue, 20 May 2025 19:21:25 +0000

Martin Nanchev for AWS Community Builders

May 20 '25

How (not) to Burn Money on VPC Endpoints (So You Don't Have To)

#aws #vpc #networking #costs

Comments 1

9 min read

[Boost]

Martin Nanchev — Tue, 20 May 2025 17:19:13 +0000

Martin Nanchev for AWS Community Builders

May 20 '25

How (not) to Burn Money on VPC Endpoints (So You Don't Have To)

#aws #vpc #networking #costs

Comments 1

9 min read

[Boost]

Martin Nanchev — Tue, 20 May 2025 15:50:05 +0000

Martin Nanchev for AWS Community Builders

May 20 '25

How (not) to Burn Money on VPC Endpoints (So You Don't Have To)

#aws #vpc #networking #costs

Comments 1

9 min read

How (not) to Burn Money on VPC Endpoints (So You Don't Have To)

Martin Nanchev — Tue, 20 May 2025 12:21:32 +0000

The Hidden Cost of Speed: How 'Just Make It Work' Breaks Your AWS Budget
Why is it so challenging?
How does the “Just do it” approach affect pillars?
How do VPC interface endpoints fit into all this?
How much does it cost? – A gentle overview of provisioning VPC interface endpoints for each new VPC
- Total costs
- Summary
Optimizing Architecture for Cost Savings and Business Continuity
- Why Isn't Cost Enough to Convince the Business?
High Level Design
- Components Table 🧩
- Integrations Table 🔗
Key takeaways

The Hidden Cost of Speed: How 'Just Make It Work' Breaks Your AWS Budget

Working as a DevOps engineer is like juggling flaming swords while someone shouts, 'Can you deploy that by Friday?' Or worse, 'By 17:00 Friday.'

Why is it so challenging?

Explaining that your solution should align with the six pillars of the AWS Well-Architected Framework is like asking for a seatbelt in a car that's already halfway down the hill—or opening your umbrella after the rain has passed. You need time, planning, and a roadmap—and nobody wants to hear that when the only goal is “just make it work.”

“Just do it” is an effective strategy but out of those six pillars, cost optimization and sustainability are usually the first to be sacrificed.

How does the “Just do it” approach affect pillars?

Because in the race to deliver, speed beats everything. Deadlines are sacred.

And what about budgets? Well, they’re not a problem—until someone sees the monthly AWS bill and starts panicking. Simply because cost impact is often hidden behind shared billing, and nobody has tagging discipline in the early phase.

Now you're asked to deploy a Graviton instance for a legacy application that doesn't even support ARM. Why wouldn’t you? After all, cost optimization is suddenly top priority—never mind compatibility?

That’s when suddenly, cost optimization becomes everyone's favorite pillar.

How do VPC interface endpoints fit into all this?

Initially, VPC endpoints are provisioned separately per VPC—because we prioritized speed over cost and, sometimes, even quality or security.
If we have 20 VPCs, we will create endpoints in each, this will lead to increased costs 20 times, especially if we have same endpoints, while the traffic is almost idle. One VPC endpoint in one availability zone provides 10 Gbps with automatic scaling up to 100 Gbps. This is enough to handle multiple workloads, even high-throughput data workloads.

For those with a programming background, this is a classic example of violating the ‘Don’t Repeat Yourself’ (DRY) principle.
Because repeating the same setup in every VPC introduces unnecessary costs for a horizontally scalable networking component designed to handle large volumes of traffic efficiently—and doing it multiple times means paying multiple times.

According to the documentation

By default, each VPC endpoint can support a bandwidth of up to 10 Gbps per Availability Zone, and automatically scales up to 100 Gbps.

How much does it cost? - a gentle overview of provisioning VPC interface endpoints for each new VPC in environments with multi-account strategy. We will use 13 accounts (let believe it is an unlucky number) and some randomly generated endpoint services as an example

account	interface endpoints
1	ecr.api, ecr.dkr, logs, monitoring, lambda, kms, sts, ssm, ssmmessages, ssm-contacts, ec2, ec2messages, acm-pca, secretsmanager
2	ecr.api, ecr.dkr, logs, monitoring, lambda, kms, sts, ssm, ssmmessages, ssm-contacts, ec2, ec2messages, acm-pca, secretsmanager, sqs, airflow.api, airflow.env, airflow.ops
3	ecr.api, ecr.dkr, logs, monitoring, lambda, kms, sts, acm-pca, secretsmanager, sagemaker.api, sagemaker.runtime
4	ssm, ec2, ecr.api, ecr.dkr, logs, monitoring, lambda, kms, sts, sagemaker.api, sagemaker.runtime, execute-api, secretsmanager, states, sts, acm-pca, glue, athena, macie2, ecs, bedrock-runtime
5	s3, sts
6	ssm, ssmmessages, ec2messages, ec2, s3, logs, monitoring, kms, sts
7	ssm, ec2, ecr.api, ecr.dkr, logs, monitoring, lambda, kms, sts, sagemaker.api, secretsmanager, elasticfilesystem, codecommit, git-codecommit, glue, athena, application-autoscaling
8	logs, monitoring, sts, glue, lambda, states, secretsmanager
9	ecr.api, ecr.dkr, logs, monitoring, lambda, kms, sts, acm-pca, secretsmanager
10	logs, monitoring, sts, ec2
11	ecr.api, ecr.dkr, logs, monitoring, lambda, kms, sts, secretsmanager, acm-pca
12	athena, logs, monitoring, kms, secretsmanager, codecommit, sagemaker.api, sagemaker.runtime, glue, git-codecommit, sts, bedrock-runtime
13	ecr.api, ecr.dkr, logs, monitoring, lambda, kms, sts, acm-pca, secretsmanager

If we group the endpoints by frequency, assuming one environment or four environments, the numbers look like this:

VPC Endpoint	Frequency (x1)	Frequency (x4)
sts	14	56
logs	12	48
monitoring	12	48
kms	10	40
secretsmanager	10	40
lambda	9	36
ecr.api	8	32
ecr.dkr	8	32
acm-pca	7	28
ec2	6	24
ssm	5	20
sagemaker.api	4	16
glue	4	16
ssmmessages	3	12
ec2messages	3	12
sagemaker.runtime	3	12
athena	3	12
ssm-contacts	2	8
states	2	8
bedrock-runtime	2	8
s3	2	8
codecommit	2	8
git-codecommit	2	8
sqs	1	4
airflow.api	1	4
airflow.env	1	4
airflow.ops	1	4
execute-api	1	4
macie2	1	4
ecs	1	4
elasticfilesystem	1	4
application-autoscaling	1	4
Total	132	528

Total costs

Calculation of total costs for eu-west-2 or London region would look like

Total costs for 132 endpoints for 1 environment = 0.011 (per hour) * 3 AZs * 24* 30 * 132 = 3136.32
Total costs for 528 endpoints = 3136.32* 4 = 12545.28
Data Processing costs for 4 environments = 5.28 (rough estimation)

Total unique VPC endpoints count = 32
Costs for 32 endpoints = 0.011 (per hour) * 3 AZs * 24* 30 * 32 = 760.32

A centralized approach for VPC endpoints in a shared services account for prod and nonprod may provide same scalability and high availability, while reducing the costs with 87% and administrative burden. Of course we can do a step further and replace some of the interface endpoints like S3 and DynamoDB for gateway endpoints in case we don't want to use their transitive nature and share them across VPCs and we want to save money.

Summary

132 endpoints x 3 AZs x $0.011/hour x 24 hours x 30 days = $3,136.32/month
For 4 environments (528 endpoints): $12,545.28/month
Costs for 32 endpoints across 3 AZs: 0.011 USD/hour × 3 AZs × 24 hours × 30 days × 32 = $760.32/month
Savings: ~87%

Note: I missed to add the costs for the resolver endpoints, which are between $180 and $270 depending on the number of ENIs or more specifically AZs

Optimizing Architecture for Cost Savings and Business Continuity

The costs above are not necessarily something bad. You have isolation between environments and you gather extensive knowledge how things work and how you need to approach stakeholders in order improve the situation.

Why Isn't Cost Enough to Convince the Business?

Business is only interested of certain things. I would say: nobody cares that the administrative burden would be smaller. So how you can approach this?

When you start the deployment of the interface endpoints they were not secured well. This means now we have a lot of networks, resulting in inconsistent security standards—each VPC becomes a snowflake. You may avoid saying that is not secure, a more suitable approach would be:
By standardizing the security policies and security groups you can make sure that sensitive workloads have access only to those specific buckets and only those specific tables and only those specific APIs. This improves the security baseline and reduces the blast radius. As a result this reduces the possibility of a data leakage. (How to Sell Optimization Without Saying 'Security Is Bad')
By centralizing and standardizing the interface endpoints, we could achieve an 87% cost reduction. In Bulgaria, there’s a well-known satirical series called The Three Fools. In this context, it feels like we're unintentionally playing a similar role—continuing to pay thousands to AWS for redundant endpoints simply because the architecture hasn't been revisited with fresh eyes.

Note: Security is always a good selling point for the business and nobody measures it after a change. Controlling fear factor and risk sells, a good example would be insurance, that we buy for our houses

High Level Design

🧩 Components Table

ID	Name	Type	Description
C1	Interface Endpoints	VPC Interface Endpoints	Provides private access to AWS services (e.g., `ssm.eu-west-2.amazonaws.com`).
C2	Route 53 Private Hosted Zone	DNS Zone	Hosts private DNS entries for the services.
C3	Route 53 Resolver Inbound Endpoint	DNS Resolver	Accepts DNS queries from the spoke VPC.
C3	Shared Resolver	Route 53 Resolver	Used by EC2 instances in the spoke VPC to resolve private DNS.
C4	AWS RAM	Resource Access Manager	Shares the inbound endpoint and private hosted zone with the spoke VPC.
C5	Cloud WAN Segment Network	Network Routing	Routes traffic between segments (e.g., from spoke to shared services).
EC2	Amazon EC2 Instance	Compute	The instance initiating the request to `ssm.eu-west-2.amazonaws.com`.
	Spoke VPC	VPC	Contains the EC2 instance. CIDR: `192.168.20.X`.
	Centralized VPC Endpoints	VPC	Hosts the interface endpoints and inbound resolver. CIDR: `192.168.10.X`.

🔗 Integrations Table

Step	Integration Description	Direction	Protocol/Mechanism
1	EC2 in spoke VPC wants to resolve `ssm.eu-west-2.amazonaws.com`.	Spoke → Shared	DNS Query via Shared Resolver
2	Shared Resolver provides IP `192.168.10.4` for the endpoint.	Shared → Spoke	DNS Response
3	Traffic to `192.168.10.4` is not local, forwarded to Cloud WAN uplink.	Spoke → Cloud WAN	VPC Route Table / Cloud WAN Routing
4	Cloud WAN checks if route to another network is permitted.	Cloud WAN	Firewall/Policy Check
5	If permitted, traffic is routed to shared services VPC.	Cloud WAN → Shared	Network Forwarding
I1	Private hosted zone is associated with the shared resolver and spoke via RAM.	Shared ↔ Spoke	AWS RAM and Route 53 Association
I4	RAM shares the inbound resolver with spoke VPC.	Shared → Spoke	AWS Resource Access Manager
I5	Spoke EC2 sends DNS queries to shared resolver.	Spoke → Shared	DNS

Prerequisite: All VPCs are connected via peering/TransitGateway/CloudWAN

Hub VPC

First we need to create centralized hub VPC that will have all of the necessary VPC interface endpoints. When you create a VPC endpoint to an AWS service, you can enable private DNS. When enabled, the setting creates an AWS managed Route 53 private hosted zone (PHZ), which enables the resolution of public AWS service endpoint to the private IP of the interface endpoint. You need this disabled in order to define a centralized PHZ trough a Route 53 inbound resolver, which would be shared with other accounts
To do this you need to disable this in terraform:

resource "aws_vpc_endpoint" "private_links" {
  for_each            = toset(local.vpc_endpoints_all)
  vpc_id              = aws_vpc.main.id
  service_name        = each.key
  vpc_endpoint_type   = "Interface"
  private_dns_enabled = false
#Disabling private DNS lets us override the default endpoint #resolution and use our own Route 53 hosted zone across accounts
  security_group_ids  = [aws_security_group.vpc_endpoint[each.key].id]
  policy              = data.aws_iam_policy_document.vpc_endpoints_policy.json
  subnet_ids          = local.subnets
  tags                = merge({ Name = "${var.prefix}-${each.key}-interface-endpoint" }, var.tags)
}

As next step you create route 53 private hosted zones for each endpoint. We associate them with the centralized VPC from step 1.
Then we create Alias A record in each hosted zone pointing to the VPC endpoint's DNS name. For example: For STS endpoint
the name "sts.${data.aws_region.current.name}.amazonaws.com"
should point to the DNS of the newly created VPC endpoint for STS
This allows traffic from spoke VPCs to resolve AWS service endpoints via centralized VPC interface endpoints and inbound endpoint.
Then we create an inbound endpoint in Route 53 with security group, Do53 protocol in at least two subnets for high availability, that will be used in the spoke VPCs as well. The idea of the inbound endpoint resolver is route your DNS queries from other spoke VPCs or networks to the hub VPC
As last step we share the resolver or inbound endpoint with other accounts and define policy for security trough Resource Access Manager or RAM

Spoke VPCs

Each newly create spoke VPC needs to be associated with the inbound resolver, that was shared from hub VPC. Example

data "aws_route53_resolver_rules" "eu_west_2" {
  owner_id     = var.resolver_rules[terraform.workspace]
  share_status = "SHARED_WITH_ME"
}
resource "aws_route53_resolver_rule_association" "eu_west_2" {
  for_each         = data.aws_route53_resolver_rules.eu_west_2.resolver_rule_ids
  resolver_rule_id = each.value
  vpc_id           = data.terraform_remote_state.networking.outputs.network.aws_vpc.id
}

Minimizing downtime

Now you would ask: How to move from the state with decentralized VPC interface endpoints to a state, where they were centralized with as minimal downtime as possible?

In general what could be done is associating the shared resolver with the spoke VPC and then destroying the decentralized VPC endpoints with a rolling deployment from development to production and automatic tests via System Manager Run Command/Lambda. This will guarantee that first you gather knowledge what can fail (Everything fails all the time) and you document it in Confluence or even Readme.md. This will give you confidence for production and will make the big change controllable and more understandable for technical and non-technical people

Key Takeaways

Rushing architecture decisions often leads to long-term cost explosions.
Interface endpoints are scalable—duplicating them per VPC isn’t.
Centralizing shared services like VPC endpoints saves money and simplifies security management.
To convince stakeholders, lead with security and cost—not technical purity.
AWS gives you the tools; architecture is about using them with purpose.

Sources: https://aws.amazon.com/blogs/networking-and-content-delivery/centralize-access-using-vpc-interface-endpoints/

https://docs.aws.amazon.com/whitepapers/latest/building-scalable-secure-multi-vpc-network-infrastructure/centralized-access-to-vpc-private-endpoints.html

Matt Garman keynote takeways

Martin Nanchev — Tue, 03 Dec 2024 18:46:05 +0000

Matt Garman unveiled at AWS re:Invent 2024 a suite of transformative updates across compute, storage, databases, AI, and real-world applications, redefining what's possible in the cloud. Here’s a deep dive into the innovations that can drive your business forward.

🚀Compute Breakthroughs

P6 EC2 Instances
With NVIDIA’s most advanced GPUs, these instances deliver 2.5x faster machine learning (ML) performance at reduced costs compared to the P5.
Amazon EC2 Trainium 2 Instances
Achieving 20.8 petaflops, these ML powerhouses offer 30–40% better performance, enabling partners like Databricks to accelerate workloads.

https://aws.amazon.com/about-aws/whats-new/2024/12/amazon-ec2-trn2-instances-available/

Trainium Ultra Server
Featuring 83 petaflops, this single-node ultra server supports high-performance computing for ML and LLM models at unprecedented speed.
Trainium 3 (Coming in 2025)
Built on a 3nm process, it’s set to deliver 40% greater efficiency, setting a new benchmark for ML hardware.

💾 Storage Revolution

Amazon S3 Table Buckets Turn S3 into a high-performance data lake or optimizes the query performance over tabular data, stored in S3! With Apache Iceberg, you can run SQL queries directly, achieving 3x faster queries and 10x higher transactions per second.

https://aws.amazon.com/about-aws/whats-new/2024/12/amazon-s3-tables-apache-iceberg-tables-analytics-workloads/
https://aws.amazon.com/blogs/aws/new-amazon-s3-tables-storage-optimized-for-analytics-workloads/

S3 Metadata Discovery Simplify object discovery with SQL-accessible metadata stored in Iceberg tables, unlocking faster, more intuitive workflows.

https://aws.amazon.com/about-aws/whats-new/2024/12/amazon-s3-metadata-preview

Zero-ETL integration for applications stored across all SAAS applications, aurora, redshift, dynamodb

📊 Database Innovations

Amazon Aurora DSQL
A multi-region, low-latency database offering 5-nines availability. Aurora DSQL uses satellite time sync to nearly break the CAP theorem and is 4x faster than Google Spanner.
https://aws.amazon.com/about-aws/whats-new/2024/12/amazon-aurora-dsql-preview/
DynamoDB Global Tables
These active-active, highly available tables ensure strong consistency and resilience for mission-critical workloads.

https://aws.amazon.com/about-aws/whats-new/2024/12/amazon-dynamodb-global-tables-previews-multi-region-strong-consistency

🧠 AI & Inference Upgrades

Amazon Nova
AWS's cutting-edge frontier models compete with Claude 3, Gemini, and others, offering 75% lower costs for text, image, and video generation.
https://aws.amazon.com/blogs/aws/introducing-amazon-nova-frontier-intelligence-and-industry-leading-price-performance/
Nova Canvas: Create studio-quality images.
Nova Reel: Generate professional videos.
Nova Speech-to-Speech & Any-to-Any: Transform the possibilities of AI.
Amazon Q Revolutionizing development with AI-powered tools:
https://aws.amazon.com/blogs/aws/new-amazon-q-developer-agent-capabilities-include-generating-documentation-code-reviews-and-unit-tests/
Unit Test Automation: Amazon QGenerate and apply tests automatically.
Amazon Q Code Reviews & Legacy Documentation generation: Achieve precision with ease.
Modernization: Amazon Q Transform VMware workloads and .NET apps into high-performing Linux-based solutions in hours, not years.
Bedrock Guardrails & Distillation,
Bedrock Model distillation makes AI lighter and 75% cheaper.
Ensure safety with sensitive data redaction and hallucination reduction.
Nedrock multi agent collaboration - https://aws.amazon.com/blogs/aws/introducing-multi-agent-collaboration-capability-for-amazon-bedrock/
Next generation of Amazon Sagemaker
The Sagemaker Unified Studio - access all of the data with the best tool like EMR, Glue, Studio, Notebook
Amazon Sagemaker Lakehouse - use Apache Iceberg datalake to query data no matter how or where stored in AWS by leveraging the Sagemaker unfied interface

🏆 Real Stories of Impact

Apple & AWS Collaboration
From Graviton 3 optimizations to accident detection using K-nearest neighbors, AWS drives ML innovation for Apple’s private LLMs, hosted on Trainium instances.

JP Morgan Chase
Transitioning 6,000 applications to AWS, the financial giant enhances resilience, reduces costs, and supports global business growth.

Genentech
Transforming drug discovery with generative AI, their systems enable scientists to ask complex questions and receive actionable insights in real-time.

PagerDuty Advance - the new GenAI PagerDuty, that keep you well informed and help you follow and engage issues. Less time for issues and more time for building with Amazon Q for PagerDuty Advance

The Road Ahead
AWS continues to redefine the limits of cloud innovation. Whether you’re modernizing infrastructure, exploring advanced AI models, or optimizing workloads, these advancements empower organizations to scale faster, innovate deeper, and inspire change.

CloudWatch and Config Unleashed: Smart Features for Effortless Monitoring Mastery

Martin Nanchev — Sat, 20 Jan 2024 09:51:32 +0000

At the heart of re:Intent 2023, the focal points were the groundbreaking advancements in generative AI and the revolutionary Bedrock service. The effect of these innovations extended to various AWS services, notably transforming Amazon CloudWatch into the ultimate single-pane-of-glass solution for monitoring your Cloud environment and making compliance a little bit easier wit Config enhancements

Now, let's get down to the good stuff — the enhancements that have not just fine-tuned but totally transformed the way we do monitoring. Imagine a world where keeping an eye on things is not just smarter but as easy as having a conversation or more human - friendly:

Amazon Cloudwatch - natural language query generator - Did you ever wonder, how you can create Log insights query, without digging in complex syntax?. Now it is easier than ever. You prompt your question in natural language and NLQG generates the log insights query for you. CloudWatch becomes your responsive companion. Let try the new feature in North Virginia:

To try the new feature we go to AWS Cloudwatch in console in us-east-1 -> Log insights -> Query generator and we type the following prompt:

Show me the 10 most denied aws api calls in current account

The query generator will do the magic for you and will suggest something following query:

fields errorCode, eventName, userIdentity.userName
| filter errorCode = "AccessDenied"
| stats count(*) as accessCount by errorCode, eventName, userIdentity.userName  
| sort accessCount desc
| limit 10

The Query generator will not run, before you click on the

Run Query

The results are really satisfying

In conclusion the new future makes querying logs more user friendly and more use-case driven. You are focusing on the business need and not how to do it.

AWS Cloudwatch Logs Anomaly detection - How can we identify the root cause, when the volume of the log data is huge? With the Anomaly detection feature this is possible. It summarises the logs and help you find the needle in the haystack. This can even answer to the question what is the root cause of the issue. You can activate it per log group. Usually it takes between 5 minutes and 24 hours depending on the size of the log group

AWS Config NL Query Processor: Simplifying Resource Compliance
Managing compliance and resource configurations can be complex and time consuming, but not anymore. The AWS Config NL Query Processor allows you to seek information effortlessly by posing questions in plain language. Whether you're in search of compliant or noncompliant resources, encrypted or unencrypted volumes, or secrets with no rotation, simply ask without the need for SQL expertise. And for those who want to take it a step further, advanced queries are now within reach, making the seemingly complex, simple.

If we head to the root account and AWS Config -> Advanced queries -> Query editor -> Natural language query processor (In Preview )and type following prompt in the :

Show me all EC2, that have unencrypted volume

The result is showing the EBS volumes, that are unencrypted, although I asked for instances. On a positive side it is still in preview and I think we are still in the beginning and there are more enhancements, that will ease our day-to-day activities in the coming years. It is definitely a good start and something that I will continue to use.

As we explore these enhancements, it's evident that CloudWatch is not just another monitoring tool; it's a companion in your Cloud journey, adapting to your needs with intelligence and ease. Config is becoming a smarter CMDB for cloud resources. The future of monitoring and compliance is here, and it's not just about data; it's about making data work for you. And this is just the beginning :)

Centralized S3 backup with AWS Backup in terraform

Martin Nanchev — Tue, 09 Jan 2024 10:19:02 +0000

Scenario

A day in the life of a Solution architect is connected with gathering requirements both functional and non-functional requirements. RTO and RPO are part of the non-functional requirements. In this post we will look at the disaster recovery for S3 buckets for different RTO and RPOs with the help of AWS Backup service. The RTOs and RPOs are defined for different use-cases and data classifications, namely - standard_data, sensitive_data, curated_data, curated_sensitive_data, critical_data, critical_sensitive_data. The idea is to create the whole infrastructure from scratch using data from AWS Backup

Object lock is a good start for protecting objects in the bucket for specific period of time. This comes in two modes - Compliance and Governance. Simply put governance allows object to be overwritten or deleted if you have specific permission to do so, while compliance is more strict and protects the objects from deletion even from the root user. Versioning is a prerequisite for object lock. If you put an object into a bucket that already contains an existing protected object with the same object key name, Amazon S3 creates a new version of that object. The existing protected version of the object remains locked according to its retention configuration. Versioning is required as a prerequisite for WORM or object lock.

Here are the defined scenarios and requirements:

Backup level	AWS Backup enabled	Object lock	Vault copy	Point in time recovery	Retention	Delete backups after days
standard_data		X		X	15 days
sensitive_data	X	X		X	15 days	15
curated_data	X	X		X	35 days	365
curated_sensitive_data	X	X		X	35 days	365
critical_data	X	X	X	X	35 days	1825
critical_sensitive_data	X	X	X	X	35 days	Set by the business in days

Table 1: AWS Backup use cases

We will look at each of the scenarios starting with the one, that does not require AWS backup:

Standard data



# DataBackupLevel = Standard Data, Write Once Read Many
module "standard_data" {
  source               = "terraform-aws-modules/s3-bucket/aws"
  bucket               = "project-standard-data-martin-n"
  attach_public_policy = false
  versioning = {
    status     = true
    mfa_delete = false
  }
  object_lock_enabled = true
  object_lock_configuration = {
    rule = {
      default_retention = {
        mode = "GOVERNANCE"
        days = 15
      }
    }
  }
}

This will product our standard data from deletion for the time period of the object lock.
All other use cases sensitive_data, curated_data, curated_sensitive_data, critical_data, critical_sensitive_data will make use of AWS Backup as backup service.

Restoration of standard data objects

The restoration of files is done via aws cli or AWS SDK and consist of following steps:

Use the ListObjectVersions API to get the version ID of the object version you want to restore.
Call RestoreObject and pass the version ID as a parameter. You can also specify the number of days the restored object will be available.
The restored object will be available in S3 under a new object name. You can download or access it like any other object.

In object lock this will work differently, because you will put same object key and create a new version. To automate the process you can use S3 Batch operations.

Assumptions

To quote Werner Vogels: Everything fails all the time.

There are multiple disaster recovery strategies to tackle data failures as shown in the graphic below:

The Backup is associated with the lowest costs, but highest RTO and RPO, which could take hours. In our case we will limit the post to the first one or Backup, which tends to be associated with lowest cost.

Imagine if we suddenly can't reach our account and need to start from scratch with our entire setup. It's crucial to have our data backed up, and we should also make sure those backups are stored in another account to ensure access in case of a disaster. That's where AWS Backup comes in handy, providing a straightforward solution to keep our data safe and accessible.

What is AWS Backup?

AWS Backup makes it easy to centrally configure backup policies and monitor backup activity for AWS resources, such as Amazon Elastic Compute Cloud (EC2) instances, Amazon Elastic Block Store (EBS) volumes, Amazon Relational Database Service (RDS) databases, Amazon DynamoDB tables, Amazon Elastic File System (EFS) file systems, Amazon FSx file systems, and AWS Storage Gateway volumes.

We will the service backups to S3 only.

How can we store backups from one account in another account?

A simple solution will be to copy backups from one vault to the vault in another account. A vault is storage for snapshots, AMI or in more general context storage for backups

Central AWS account

A great way to begin is to define the vault in the central account. KMS key will be used to encrypt the vault for the snapshots and allow the workload account to copy images to it. We are using grant to allow this to the workload account



resource "aws_kms_key" "vault_kms" {
  description = "Vault kms key for encryption"
  policy      = <<POLICY
{
    "Id": "vault-kms-policy",
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "Enable IAM User Permissions",
            "Effect": "Allow access to view and describe key",
            "Principal": {
                "AWS": "arn:aws:iam::${data.aws_caller_identity.current.account_id}:root"
            },
            "Action": [
                "kms:ListResourceTags",
                "kms:GetKeyPolicy",
                "kms:Describe*"
            ],
            "Resource": "*"
        },
        {
            "Sid": "Allow access for Key Administrators",
            "Effect": "Allow",
            "Principal": {
                "AWS": "arn:aws:iam::${data.aws_caller_identity.current.account_id}:role/github_ci"
            },
            "Action": [
                "kms:Create*",
                "kms:Describe*",
                "kms:Enable*",
                "kms:List*",
                "kms:Put*",
                "kms:Update*",
                "kms:Revoke*",
                "kms:Disable*",
                "kms:Get*",
                "kms:Delete*",
                "kms:TagResource",
                "kms:UntagResource",
                "kms:ScheduleKeyDeletion",
                "kms:CancelKeyDeletion"
            ],
            "Resource": "*"
        },
        {
            "Sid": "Allow use of the key",
            "Effect": "Allow",
            "Principal": {
                "AWS": [
                    "arn:aws:iam::${data.aws_caller_identity.current.account_id}:role/kms_usage",
                    "arn:aws:iam::${var.workload_account_id}:root"
                ]
            },
            "Action": [
                "kms:Encrypt",
                "kms:Decrypt",
                "kms:ReEncrypt*",
                "kms:GenerateDataKey*",
                "kms:DescribeKey"
            ],
            "Resource": "*"
        },
        {
            "Sid": "Allow attachment of persistent resources",
            "Effect": "Allow",
            "Principal": {
                "AWS": [
                    "arn:aws:iam::${data.aws_caller_identity.current.account_id}:role/vault_role",
                    "arn:aws:iam::${var.workload_account_id}:root"
                ]
            },
            "Action": [
                "kms:CreateGrant",
                "kms:ListGrants",
                "kms:RevokeGrant"
            ],
            "Resource": "*",
            "Condition": {
                "Bool": {
                    "kms:GrantIsForAWSResource": "true"
                }
            }
        }
    ]
}
POLICY
}

As a next step we define an alias to the key in order that the key is more human friendly or readable in the AWS console.



resource "aws_kms_alias" "aws_kms_alias" {
  name          = "alias/aws-backup-kms"
  target_key_id = aws_kms_key.vault_kms.key_id
}

The last step is to define the central vault, used to aggregate snapshots from workload account:



resource "aws_backup_vault" "central_vault" {
  name        = "central-aws-vault-S3"
  kms_key_arn = aws_kms_key.vault_kms.arn
}


resource "aws_backup_vault_lock_configuration" "locker" {
  backup_vault_name   = aws_backup_vault.central_vault.name
  changeable_for_days = 3
  max_retention_days  = 35
  min_retention_days  = 15
}

Last step is to create a policy which allows snapshots to be copied to the central_vault:



resource "aws_backup_vault_policy" "central_vault_allowance" {
  backup_vault_name = aws_backup_vault.central_vault.name
  policy = <<POLICY
{
  "Version": "2012-10-17",
  "Id": "default",
  "Statement": [
    {
      "Sid": "Allow Tool Prod Account to copy into iemtrialcluster_backup_vault",
      "Effect": "Allow",
      "Action": "backup:CopyIntoBackupVault",
      "Resource": "*",
      "Principal": {
        "AWS": "arn:aws:iam::${var.workload_account_id}:root"
      }
    }
  ]
}
POLICY
}

This is all, that is needed to define our central account repository for S3

Workload account

Now we will look at each of the use cases in the Table 1

Sensitive data

We will use for monitoring SNS to send us notification in pagerduty or slack via Lambda function in case of failed backups. For the backup selection we will use direct selection, although selection by Tag could be more appropriate for most use cases. AWS Backup also supports Point in time recovery trough enabling continous backup in the backup plan




module "aws_backup_s3_sensitive_data" {
  source     = "lgallard/backup/aws"
  vault_name = "${local.service_name}-s3-backup-sensitive-data"
  plan_name  = "${local.service_name}-s3-backup-sensitive-data-plan"
  notifications = {
    sns_topic_arn       = data.aws_sns_topic.failed_backups.arn
    backup_vault_events = ["BACKUP_JOB_FAILED"]
  }

  rules = [
    {
      name              = "${local.service_name}-s3-backup-sensitive-data-rule"
      schedule          = "cron(5 2 * * ? *)"
      start_window             = 60
      completion_window        = 180
      enable_continuous_backup = true
      lifecycle = {
        cold_storage_after = null
        delete_after       = 15
      }
    }
  ]
  selections = [
    {
      name      = "${local.service_name}-s3--sensitive-data-selection"
      resources = ["arn:aws:s3:::prefix-dummy-${local.environment}-database-sensitive-data-bucket", "arn:aws:s3:::dummy-${local.environment}-sensitive-data-logs"]
    }
  ]
}

Curated data and curated sensitive data

Curated data refers to information that has been carefully selected, organized, and maintained to ensure accuracy, relevance, and quality. Think of it like a well-maintained library where librarians carefully choose and organize books to provide a reliable and valuable collection for visitors. In the context of data, curators, often experts in a specific field, carefully choose, validate, and organize data to create a trustworthy and useful dataset. This process helps ensure that the information is reliable, up-to-date, and suitable for specific purposes, making it easier for users to find and use the data they need.




module "aws_backup_s3_curated_data" {
  source     = "lgallard/backup/aws"
  vault_name = "${local.service_name}-s3-curated-data-backup"
  plan_name  = "${local.service_name}-s3-backup-curated-data-plan"
  notifications = {
    sns_topic_arn       = data.aws_sns_topic.failed_backups.arn
    backup_vault_events = ["BACKUP_JOB_FAILED"]
  }

  rules = [
    {
      name              = "${local.service_name}-s3-backup-curated-data-rule"
      schedule          = "cron(5 2 * * ? *)"
      start_window             = 60
      completion_window        = 180
      enable_continuous_backup = true
      lifecycle = {
        cold_storage_after = null
        delete_after       = 35
      }
    }
  ]
  selections = [
    {
      name      = "${local.service_name}-s3-selection"
      resources = ["arn:aws:s3:::prefix-dummy-${local.environment}-database-curated-data-bucket", "arn:aws:s3:::dummy-${local.environment}-curated-data-logs"]
    }
  ]
}

Critical data and critical sensitive data

This data brings most value to the business and help your business executives to make decision by using a data driven approach. The Backups will be copied to the central account




module "aws_backup_critical" {
  source     = "lgallard/backup/aws"
  vault_name = "${local.service_name}-s3-backup-critical-data"
  plan_name  = "${local.service_name}-s3-backup-critical-data-plan"
  notifications = {
    sns_topic_arn       = data.aws_sns_topic.failed_backups.arn
    backup_vault_events = ["BACKUP_JOB_FAILED"]
  }

  rules = [
    {
      name              = "${local.service_name}-s3-backup-critical-data-rule"
      schedule          = "cron(5 2 * * ? *)"
      copy_actions = [
        {
          lifecycle = {
            cold_storage_after = 90
            delete_after       = 1825
          },
          destination_vault_arn = var.central_s3_vault_arn
        },
      ]
      start_window             = 60
      completion_window        = 180
      enable_continuous_backup = true
      lifecycle = {
        cold_storage_after = null
        delete_after       = 35
      }
    }
  ]
  selections = [
    {
      name      = "${local.service_name}-s3-critical-data-selection"
      resources = ["arn:aws:s3:::prefix-dummy-${local.environment}-database-critical-data-bucket", "arn:aws:s3:::dummy-${local.environment}-critical-data-logs]
    }
  ]
}

Now that we have the backups, How can we restore objects. Lets have a look in the next section

Data restoration

Go to AWS Backup -> Backup vaults -> Select the recovery point, that you want and click Actions -> restore

Conclusion

In conclusion, ensuring the integrity and accessibility of data is a critical aspect of a Solution Architect's role, and AWS Backup proves to be an indispensable tool in this endeavor. The outlined disaster recovery scenarios, comprehensive use cases, and the step-by-step configuration guide for central and workload accounts underscore the significance of AWS Backup in safeguarding data across various classifications. Whether dealing with standard data or critical sensitive data, the emphasis on backup, object lock, and versioning ensures a robust data protection strategy. This comprehensive approach not only addresses the need for backup but also sheds light on restoration procedures, making AWS Backup an essential component for maintaining data resilience in the ever-evolving landscape of cloud computing

Centralizing Cloudwatch observability - Past, Present and Future

Martin Nanchev — Tue, 16 May 2023 16:13:40 +0000

What is observability?

According to wikipedia observability is a measure of how well internal states of a system can be inferred from knowledge of its external outputs.

In simple words we measure how the internal lego blocks of a system (AWS services) perform from using their outputs or trying to monitor from customer perspective. Cloudwatch plays a central role in AWS for monitoring, logging, alerting and auditing of a system, but there is one catch. When you have 200 accounts it is really difficult to develop an observability strategy at scale.

Past or cross account observability using Cloudwatch

Dark ages before OAM

You will need to go to every single account and check the dashboards and logs.
Well there is always an alternative to stream Cloudwatch logs using Kinesis firehose and store them in S3. Afterwards you create a schema from the S3 logs using Glue and query them via Athena. If your bucket was encrypted with KMS and you forgot to switch off the option to decrypt them, this can lead to unexpected raise in the bill
You can always implement Opensearch to store the logs

To summarise a cross account observability was difficult without OAM (although you can use a cross account role to putTraces from ECS tasks for X-ray)

The Present - OAM

Last year a new feature was presented called Observability Access Manager. The idea is to create a sink account, which will receive the cloudwatch logs, metrics, traces. (it will use sharing) Then each source account will create a link to the sink account to share the cloudwatch logs, metrics, traces.
Costs: it is free for logs, metrics, but not for traces. Copies of the traces will be billed. And there is a standard fee for creating of Dashboards, alarms etc.

How this will look like if we have multiple accounts?

There are two sink accounts to collect observability data - for production and non-production accounts
There are four accounts to share data - development and qa, which share data with the monitoring non-production account. Production and staging, which share data with the monitoring production account.

To show it in more graphical way:

Everything sounds perfect, but can we automate the OAM sink and sources, that share observability data?

First we will need to create a sink using terraform in the central monitoring account. The accounts will collect observability data from sources:


locals {
  resource_types = coalescelist(var.services,["AWS::Logs::LogGroup",
    "AWS::CloudWatch::Metric",
  "AWS::XRay::Trace"])
}

resource "aws_oam_sink" "sink" {
  name = var.name
  tags = var.tags
}


resource "aws_oam_sink_policy" "observability_data_sink_policy" {
  sink_identifier = aws_oam_sink.sink.id
  policy          = data.aws_iam_policy_document.sink_policy.json
}

data "aws_iam_policy_document" "sink_policy" {

  dynamic "statement" {
    for_each = length(var.account_list) > 0 ? [1] : []
    content {
      actions = ["oam:CreateLink", "oam:UpdateLink"]
      principals {
        type        = "AWS"
        identifiers = var.account_list
      }
      resources = ["*"]
      effect    = "Allow"
      condition {
        test     = "ForAllValues:StringEquals"
        values   = local.resource_types
        variable = "oam:ResourceTypes"
      }

    }
  }

  dynamic "statement" {
    for_each = length(var.org_unit_list) > 0 ? [1] : []
    content {
      actions = ["oam:CreateLink", "oam:UpdateLink"]
      principals {
        type        = "*"
        identifiers = ["*"]
      }
      resources = ["*"]
      effect    = "Allow"
      condition {
        test     = "ForAllValues:StringEquals"
        values   = local.resource_types
        variable = "oam:ResourceTypes"
      }
      condition {
        test     = "ForAnyValue:StringEquals"
        values   = var.org_unit_list
        variable = "aws:PrincipalOrgPaths"
      }
    }
  }
}


variable "name" {
  type        = string
  description = "The name of the observability access manager sink, which collects observability data - logs, metrics, traces"
  default     = "monitoring-sink"
}

variable "tags" {
  type        = map(string)
  description = "Tags to be added to the specific resource"
  default     = {}
}

variable "org_unit_list" {
  type        = list(string)
  description = "list of Organizational units"
  default = [
    "o-aaausawxze/ou-tv7w-dd1211vs",
  ]
}

variable "account_list" {
  description = "List of accounts"
  type        = list(string)
  default     = ["123456789102"]
}

variable "services" {
  type        = list(string)
  default     = []
  description = "List of services to be shared. Possible values are: AWS::Logs::LogGroup, AWS::CloudWatch::Metric, AWS::XRay::Trace"
}

The resourced based policy specifies, that OU or account id could be defined as sources and share data with the sink

After the sink is ready. We can output the arn of the OAM:
output "sink_arn" { value = aws_oam_sink.sink.id }

The arn will be needed by the link (source) to connect to the sink (monitoring account). A data terraform remote state could be used to get the output and use it from a tf configuration file.
A simple hardcoded example will look like:

resource "aws_oam_link" "source_to_sink" {
  label_template  = "$AccountName"
  resource_types  = local.resource_types
  sink_identifier = var.sink_identifier
  tags            = var.tags
}
locals {
  resource_types = coalescelist(var.services, ["AWS::Logs::LogGroup",
    "AWS::CloudWatch::Metric",
  "AWS::XRay::Trace"])
}

variable "sink_identifier" {
  description = "Sink identifier"
  default     = "arn:aws:oam:eu-west-1:123456789102:sink/ed278766-dc6f-4417-ae11-8bf09e9dc329"
  type        = string
}

variable "tags" {
  type        = map(string)
  description = "Tags to be added to the specific resource"
  default     = {}
}

variable "services" {
  type        = list(string)
  default     = []
  description = "List of services to be shared. Possible values are: AWS::Logs::LogGroup, AWS::CloudWatch::Metric, AWS::XRay::Trace"
}

How will the monitoring account look like after it receives the data?

A standard Cloudwatch dashboard could be:

Then if we use the cross account metrics from the central monitoring account, it will look the same but there is a cross account check on the right side of the metric. Also on the metric itself you will receive a label with the account name:

Then all available accounts, which share data with the monitoring account or sink are visible in the Cloudwatch Settings menu:

Future

The present look awesome. You can share observability data, create a centralized dashboards in monitoring account and copy traces to the centralized account.

A possibility to create more dynamic dashboards will look awesome. At the moment I use json to create a template, which is then deployed via terraform. There is a catch that you will need to add every new resource. Well there is the resource explorer, but it still lacks some of the features, that I would like to have.
Also with the Generative AI waiting behind the door, I would expect a suggestions like which metrics will be important for example for Kafka and automatic dashboard suggestions, although you can do a research and define some baseline metrics for monitoring of the cluster and brokers. Also this is really difficult, because you receive building blocks to create architecture -> You build it, you own it. A plan for observability will be needed, before the workload lands in production.

DEV Community: Martin Nanchev

Anarchy, Assembly Lines, and Corporate Hierarchy: Benchmarking Multi-Agent Architectures for Medical Device Data

A Fair Fight This Time

The Commune: Swarm

Assembly line: Graph Pipeline

Coordinator

The Judge: LLM-as-Evaluator

Results

Optimizing Multi-Agent Costs on Bedrock: From ~$18 to ~$7 per Diabetes Report Run (Graph vs Swarm Comparison)

The Problem With 5,000 Rows of Blood Sugar Data

Architecture 1: The Graph Pipeline

The Code

What Came Out

The $18 Problem

Cutting It With a Sliding Window

Architecture 2: The Swarm

What Came Out

The Showdown: Graph vs Swarm

Tokens & Cost

Where the Graph pulled ahead

Where the Swarm held up

The thing I need to be honest about

The Verdict

Key Takeaways

How much did the over - engineering costs? Any additional costs?

The smart way of centralizing VPC endpoints with Route 53 Profiles

How to Centralize Endpoints Smartly

Reality Check

Other Benefits

Summary

Terraform Code

Build Games Challenge: Build Tower of Hanoi with Amazon Q Developer CLI

Table of Contents

Building Tower of Hanoi with Amazon Q Developer CLI: A Journey from Zero-Shot to Polished Game

The Frog and the Ox: Why I Started This Project

Why Tower of Hanoi?

Getting Started with Amazon Q Developer CLI

The Zero-Shot Miracle

The First Success

The Design Challenge

Lessons Learned

The Material Design Evolution

Key Material Design Elements Added:

The Final Polish: Advanced Zero-Shot

New Features in the Final Version:

Technical Insights

What Amazon Q Developer CLI Excelled At:

Where It Struggled:

The Sweet Spot:

Code Architecture Highlights

Performance and User Experience

Reflections on AI-Assisted Development

The Good:

The Challenging:

The Surprising:

Conclusion: The Frog's Success

Try It Yourself

How to burn money on VPC Endpoints

How (not) to Burn Money on VPC Endpoints (So You Don't Have To)

[Boost]

How (not) to Burn Money on VPC Endpoints (So You Don't Have To)

[Boost]

How (not) to Burn Money on VPC Endpoints (So You Don't Have To)

How (not) to Burn Money on VPC Endpoints (So You Don't Have To)

The Hidden Cost of Speed: How 'Just Make It Work' Breaks Your AWS Budget

Why is it so challenging?

How does the “Just do it” approach affect pillars?

How do VPC interface endpoints fit into all this?

How much does it cost? - a gentle overview of provisioning VPC interface endpoints for each new VPC in environments with multi-account strategy. We will use 13 accounts (let believe it is an unlucky number) and some randomly generated endpoint services as an example

Total costs

Summary

Optimizing Architecture for Cost Savings and Business Continuity

Why Isn't Cost Enough to Convince the Business?

High Level Design

🧩 Components Table

🔗 Integrations Table

Hub VPC

Spoke VPCs

Minimizing downtime

Key Takeaways