Setting the file. One moment.
Benchmark Model · Amazon DynamoDB · aws/agent-toolkit-for-aws · Skills Docs
Back to the top of the page Reference Architecture
93
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Creating API Gateway Stage
def _aggregate
— line 539
This file
Number 65.8
Position 8 of 14
Type Python
Size 46 KB
Lines 1,063 scripts/ benchmark_model.py
Python · 1,063 lines · 46 KB
16 4. Aggregate into perf_summary.json (per-pattern steady-state + cold-start +
17 coverage + seed verification).
18
19 Usage:
20 python3 benchmark_model.py \\
21 --model dynamodb_data_model.json \\
22 --config benchmark_config.json \\
23 --manifest created_resources.json \\
24 --raw-out perf_raw.jsonl \\
25 --summary-out perf_summary.json
26 """
27 from __future__ import annotations
28
29 import argparse
30 import json
31 import statistics
32 import sys
33 import time
34 from datetime import datetime, timezone
35 from pathlib import Path
36 from typing import Any, NoReturn
37
38 # Import the sibling calculator for exact per-op CU parity in the spend
39 # estimate — same module generate_perf_report.py uses, so the guardrail's
40 # numbers match the calculator the user already trusts.
41 _THIS = Path( __file__ ).resolve().parent
42 sys.path.insert( 0 , str ( _THIS ))
43 try :
44 import calculate_costs as cc # noqa: E402
45 except Exception : # pragma: no cover - calculator is a sibling; should import
46 cc = None # type: ignore[assignment]
47
48
49 # Write ops, used by the spend estimate when the calculator import is
50 # unavailable. Mirrors calculate_costs.WRITE_OPS (the source of truth when cc is
51 # importable); kept here so the estimate still works in a cc-less environment.
52 _FALLBACK_WRITE_OPS = {
53 "PutItem" ,
54 "UpdateItem" ,
55 "DeleteItem" ,
56 "BatchWriteItem" ,
57 "TransactWriteItems" ,
58 }
59
60
61 def _die (msg: str , code: int = 2 ) -> NoReturn:
62 """Exit with a message. Annotated NoReturn so callers narrow correctly.
63
64 Without NoReturn a type-checker cannot know `_die()` ends the program, so the
65 common `x = next(..., None); if x is None: _die(...)` guard leaves `x` optional
66 for everything after it. Matches iterate_design.py, which already declares it
67 this way.
68 """
69 print ( f "ERROR: { msg } " , file = sys.stderr)
70 sys.exit(code)
71
72
73 def _require_boto3 ():
74 try :
75 import boto3 # noqa: F401
76 from botocore.exceptions import ClientError # noqa: F401
77
78 return boto3
79 except ImportError :
80 _die( "boto3 not installed. Run: pip install boto3>=1.34" )
81
82
83 def _load_json (path: Path) -> dict :
84 if not path.exists():
85 _die( f "file not found: { path } " )
86 with path.open() as f:
87 return json.load(f)
88
89
90 _QUICK_PRESET = {
91 "table_settle_seconds" : 10 ,
92 "warmup_seconds" : 3 ,
93 "duration_seconds" : 15 ,
94 "ramp_seconds" : 3 ,
95 "seed_items_per_table" : 100 ,
96 }
97
98 # Representative mode: a proportionally-higher load tier that surfaces SCALE
99 # risk (hot partitions, throttle-under-load, GSI amplification at volume,
100 # Query-at-realistic-cardinality) at bounded cost. Distinct from quick/standard,
101 # which validate per-op UNIT cost at ~1% scale. scale_factor sits in the
102 # confirmed 0.10–0.25 band; zipf sampling concentrates load on a few partitions
103 # so one partition can approach the per-partition ceiling (Mechanics #3);
104 # items_per_partition makes partitions hold real collections so Query patterns
105 # read realistic multi-item pages. max_rps_per_pattern is the HONEST ceiling:
106 # one in-region Lambda with 32 I/O-bound threads tops out near 1500–2000
107 # RPS/pattern (raised from the prior 8-thread ~800–1000).
108 _REPRESENTATIVE_PRESET = {
109 "table_settle_seconds" : 45 ,
110 "warmup_seconds" : 15 ,
111 "duration_seconds" : 120 ,
112 "ramp_seconds" : 20 ,
113 "seed_items_per_table" : 2000 ,
114 "items_per_partition" : 40 ,
115 "scale_factor" : 0.15 ,
116 "max_rps_per_pattern" : 1800 ,
117 "min_rps_per_pattern" : 5 ,
118 "concurrency_per_pattern" : 32 ,
119 "read_pattern_key_sampling" : "zipf" ,
120 "zipf_s" : 1.1 ,
121 }
122
123 _PRESETS : dict[ str , dict[ str , Any]] = {
124 "quick" : _QUICK_PRESET ,
125 "representative" : _REPRESENTATIVE_PRESET ,
126 }
127
128
129 def _apply_mode_preset (cfg: dict ) -> dict :
130 """Overlay a mode preset (quick / representative) on cfg.
131
132 The preset only fills fields the user did NOT set explicitly — an explicit
133 value in benchmark_config.json always wins. Unknown modes are refused so a
134 typo like "quik" doesn't silently run a default-shaped benchmark.
135 """
136 mode = cfg.get( "mode" )
137 if mode is None or mode == "standard" :
138 return cfg
139 if mode not in _PRESETS :
140 _die(
141 f 'unknown benchmark mode { mode !r} . Valid values: "quick", '
142 '"representative", "standard" (default).'
143 )
144 merged = dict (cfg)
145 for k, v in _PRESETS [mode].items():
146 if k not in cfg:
147 merged[k] = v
148 if mode == "quick" :
149 print (
150 "Quick mode: running a short per-pattern window "
151 f "(settle= { merged[ 'table_settle_seconds' ] } s, "
152 f "warmup= { merged[ 'warmup_seconds' ] } s, "
153 f "duration= { merged[ 'duration_seconds' ] } s, "
154 f "seed= { merged[ 'seed_items_per_table' ] } items). "
155 "Percentiles and extrapolation are less stable than the standard "
156 "mode — treat this run as a smoke test, not a cost-validation result."
157 )
158 elif mode == "representative" :
159 print (
160 "Representative mode: scale ~"
161 f " { merged.get( 'scale_factor' , 0.15 ) } × declared peak, zipf hot-key "
162 "sampling ON, capped at "
163 f " { merged.get( 'max_rps_per_pattern' , 1800 ) } RPS/pattern, "
164 f " { merged.get( 'items_per_partition' , 40 ) } items/partition. "
165 "Surfaces hot-partition throttling, throttle-under-load, GSI "
166 "amplification at volume, and Query-at-realistic-cardinality. "
167 "ONE in-region Lambda (32 threads) tops out near 1500–2000 RPS/"
168 "pattern — representative mode is BOUNDED and does NOT prove the "
169 "design sustains declared peak RPS. Per-op cost extrapolation stays "
170 "linear and valid; throttle/latency numbers are load-risk signals, "
171 "not capacity-sustain proof — never extrapolate them linearly."
172 )
173 return merged
174
175
176 def _validate (model: dict ) -> None :
177 aps = model.get( "access_patterns" ) or []
178 if not aps:
179 _die( "access_patterns empty — refusing (Mechanics #2)." )
180 for ap in aps:
181 if not ap.get( "peak_rps" ):
182 _die(
183 f "pattern { ap.get( 'pattern_id' , '?' ) } has missing or zero "
184 "peak_rps — refusing per Mechanics #2."
185 )
186 # Structural-reference check. Runs even on the reuse path (where deploy_model
187 # is skipped), so a benchmark can never silently fail every call against a
188 # missing table or a Query on a non-existent GSI — it refuses up front with a
189 # clear message instead of reporting a "cheap" 0-CU run.
190 tables_by_name = {t.get( "table_name" ): t for t in model.get( "tables" , [])}
191 for ap in aps:
192 pid = ap.get( "pattern_id" , "?" )
193 tn = ap.get( "table" )
194 if not tn:
195 _die(
196 f 'pattern { pid } has no "table" — every pattern must name the '
197 "table it runs against."
198 )
199 td = tables_by_name.get(tn)
200 if td is None :
201 _die(
202 f "pattern { pid } references table { tn !r} , not defined in tables[]. "
203 f "Defined tables: { sorted (tables_by_name) } ."
204 )
205 idx = ap.get( "index" )
206 op = ap.get( "operation" )
207 gsi_names = {g.get( "index_name" ) for g in ((td or {}).get( "gsis" ) or [])}
208 vec_names = {v.get( "index_name" ) for v in ((td or {}).get( "vector_indexes" ) or [])}
209
210 # SearchVectors reads a VECTOR index, never a GSI, and Query/Scan cannot read a
211 # vector index at all. Checking each operation against the right index family
212 # matters more than usual here: a mismatch fails 100% of calls while the observed
213 # capacity reads as 0, which a naive report then presents as a free pattern.
214 if op == "SearchVectors" :
215 if not idx:
216 _die( f 'pattern { pid } is a SearchVectors but names no "index".' )
217 if idx not in vec_names:
218 _die(
219 f "pattern { pid } searches vector index { idx !r} on table { tn !r} , but "
220 f "that table defines no such vector index. Defined vector indexes: "
221 f " { sorted (n for n in vec_names if n) } . Fix the design JSON."
222 )
223 elif idx:
224 if idx in vec_names:
225 _die(
226 f "pattern { pid } targets { idx !r} on table { tn !r} with operation "
227 f " { op !r} , but { idx !r} is a VECTOR index. Query and Scan are rejected "
228 "against one ('Query operation not supported on this index type'). "
229 'Use operation "SearchVectors".'
230 )
231 if idx not in gsi_names:
232 _die(
233 f "pattern { pid } uses index { idx !r} on table { tn !r} , but that "
234 f "table defines no such GSI. Defined GSIs: "
235 f " { sorted (n for n in gsi_names if n) } . A Query/Scan against a "
236 "non-existent index fails every call — fix the design JSON."
237 )
238
239
240 def _compute_split (cfg: dict , n_patterns: int ) -> tuple[ int , float ]:
241 """Return (invocations_total, per_invocation_timeout_estimate).
242
243 The Lambda runs patterns serially inside a single invocation (see
244 run_warmup / run_measure in scripts/benchmark_lambda.py — they iterate
245 patterns one at a time). Real wall-clock per invocation is therefore
246 n_patterns × (warmup_seconds + duration_seconds_slice), NOT just
247 warmup + duration. A 31-pattern design at 90s each takes ~47 min — well
248 over Lambda's 15-min ceiling — so we must split into multiple
249 invocations when the total would exceed the usable budget.
250 """
251 settle = int (cfg.get( "table_settle_seconds" , 30 ))
252 # Seed wall-clock scales with total seed volume. run_seed seeds PER PATTERN
253 # (each pattern's keys embed its pattern_id, so they don't dedup), writing
254 # ~seed_items_per_table items per pattern in 25-item BatchWriteItem chunks at
255 # ~30ms/batch (serial within the seed phase): seed_items/25 × 0.03s ×
256 # n_patterns. Representative mode (2000 items) is ~2.4s per pattern. Scaling
257 # by n_patterns (not a flat 60s) keeps the split honest for many-pattern
258 # designs that seed a lot.
259 seed_items = int (cfg.get( "seed_items_per_table" , 500 ))
260 seed = int ( 30 + (seed_items / 25.0 ) * 0.03 * max ( 1 , n_patterns))
261 warmup = int (cfg.get( "warmup_seconds" , 10 ))
262 duration = int (cfg.get( "duration_seconds" , 90 ))
263 aggregation_margin = 20
264 lambda_timeout = int (cfg.get( "lambda_timeout_seconds" , 900 ))
265 usable = lambda_timeout * 0.9
266
267 # First invocation carries settle + seed + per-pattern warmup once.
268 first_overhead = settle + seed + n_patterns * warmup + aggregation_margin
269 # Subsequent invocations only measure; overhead is just the margin.
270 subsequent_overhead = aggregation_margin
271
272 total_measure_time = n_patterns * duration
273
274 # Can we fit everything into one invocation?
275 if first_overhead + total_measure_time <= usable:
276 return 1 , usable
277
278 # Need to split. Compute the per-invocation measurement budget for each
279 # regime (first vs subsequent) and size the slice conservatively.
280 first_slice_budget = max ( 0 , usable - first_overhead)
281 subsequent_slice_budget = max ( 0 , usable - subsequent_overhead)
282
283 # Use the smaller of the two so the same duration_per_pattern fits both.
284 per_invocation_measure_budget = min (first_slice_budget, subsequent_slice_budget)
285 if per_invocation_measure_budget <= 0 :
286 # First invocation's overhead alone exceeds usable — insufficient
287 # Lambda timeout for this design. Caller can surface this.
288 return - 1 , usable
289
290 slice_per_pattern = max ( 1 , int (per_invocation_measure_budget // n_patterns))
291 per_invocation_total_measure = slice_per_pattern * n_patterns
292 invocations = max ( 2 , int ( - ( - total_measure_time // per_invocation_total_measure)))
293 return invocations, usable
294
295
296 def _attach_vector_index_meta (model: dict ) -> list :
297 """Resolve each SearchVectors pattern's target vector index onto the pattern.
298
299 The Lambda driver needs three things the pattern alone does not carry: the index's
300 `dimensions` (a query vector of the wrong length fails every call with
301 `Input search vector dimension N does not match vector index dimension M`), its
302 SearchSchema partition key (mandatory in SearchConditionExpression when the index
303 defines one — otherwise every call fails with `SearchConditionExpression must be
304 provided when SearchSchema has a HASH key`), and that key's type.
305
306 Resolved here rather than in the Lambda so a mis-referenced index is a local error
307 instead of a run where 100% of calls fail and the observed cost reads as zero.
308 """
309 tables_by_name = {t.get( "table_name" ): t for t in model.get( "tables" ) or []}
310 out = []
311 for ap in model.get( "access_patterns" ) or []:
312 if ap.get( "operation" ) != "SearchVectors" :
313 out.append(ap)
314 continue
315 td = tables_by_name.get(ap.get( "table" )) or {}
316 vi = next (
317 (v for v in (td.get( "vector_indexes" ) or []) if v.get( "index_name" ) == ap.get( "index" )),
318 None ,
319 )
320 if vi is None :
321 _die(
322 f "access pattern { ap.get( 'pattern_id' ) } searches vector index "
323 f " { ap.get( 'index' ) !r} on table { ap.get( 'table' ) !r} , which declares no "
324 "such vector index. Fix the model before benchmarking — otherwise every "
325 "call fails and the run reports the pattern as free."
326 )
327 schema = vi.get( "search_schema" ) or {}
328 pk = schema.get( "partition_key" )
329 pk_type = "S"
330 for ent in td.get( "entities" ) or []:
331 for a in ent.get( "attributes" ) or []:
332 if a.get( "name" ) == pk:
333 pk_type = a.get( "type" , "S" )
334 for a in td.get( "attribute_definitions" ) or []:
335 name = a.get( "attribute_name" ) or a.get( "name" )
336 if name == pk:
337 pk_type = a.get( "attribute_type" ) or a.get( "type" ) or pk_type
338 enriched = dict (ap)
339 enriched[ "vector_index" ] = {
340 "index_name" : vi.get( "index_name" ),
341 "vector_attribute" : vi.get( "vector_attribute" ),
342 "dimensions" : vi.get( "dimensions" ),
343 "partition_key" : pk,
344 "pk_type" : pk_type,
345 "inline_filters" : list (schema.get( "inline_filters" ) or []),
346 }
347 out.append(enriched)
348 return out
349
350
351 def _invoke_lambda (lambda_client, function_name: str , payload: dict ):
352 raw = json.dumps(payload).encode()
353 resp = lambda_client.invoke(
354 FunctionName = function_name,
355 InvocationType = "RequestResponse" ,
356 Payload = raw,
357 )
358 status = resp.get( "StatusCode" , 0 )
359 body = resp[ "Payload" ].read()
360 try :
361 parsed = json.loads(body)
362 except json.JSONDecodeError:
363 _die(
364 f "Lambda returned non-JSON payload (status { status } ): "
365 f " { body[: 200 ].decode( errors = 'replace' ) } "
366 )
367 if resp.get( "FunctionError" ):
368 _die( f "Lambda function error ( { resp[ 'FunctionError' ] } ): " f " { json.dumps(parsed)[: 500 ] } " )
369 return parsed
370
371
372 def _bench_rps (pattern: dict , cfg: dict ) -> float :
373 declared = float (pattern.get( "peak_rps" , 0 ) or 0 )
374 scaled = declared * float (cfg.get( "scale_factor" , 0.01 ))
375 return max (
376 float (cfg.get( "min_rps_per_pattern" , 1 )),
377 min ( float (cfg.get( "max_rps_per_pattern" , 50 )), scaled),
378 )
379
380
381 def _estimate_bench_spend (model: dict , cfg: dict ) -> dict :
382 """Estimate the actual AWS charge a run will incur, BEFORE running it.
383
384 This is a PRE-SPEND gate distinct from the in-Lambda abort_on_throttle_rate
385 runtime guard. Representative runs drive far more traffic than the ~1% unit-
386 cost runs, so a cheap upper-bound estimate lets the orchestrator refuse (or
387 ask consent) before creating the bill. Two cost components:
388
389 driven load — Σ_patterns bench_rps × (warmup + duration) calls, each at the
390 calculator's expected per-op CU × on-demand unit price.
391 seeding — seed_items_per_table × items_per_partition writes per table
392 that has a pattern, each ⌈item_size/1KB⌉ WRU.
393
394 Uses the calculator's own per-op CU so the estimate is consistent with the
395 numbers the user already sees. Falls back to a coarse per-op CU when the
396 calculator import is unavailable. Conservative by construction (counts the
397 full warmup+duration window and ignores throttle-shed traffic)."""
398 aps = model.get( "access_patterns" ) or []
399 tables = model.get( "tables" ) or []
400 table_map = {t[ "table_name" ]: t for t in tables}
401 entity_attr_sizes = cc._build_entity_attr_sizes(tables) if cc else {}
402
403 warmup = float (cfg.get( "warmup_seconds" , 10 ))
404 duration = float (cfg.get( "duration_seconds" , 90 ))
405 window = warmup + duration
406
407 driven_cost = 0.0
408 vector_cost = 0.0
409 per_pattern = []
410 for ap in aps:
411 rps = _bench_rps(ap, cfg)
412 calls = rps * window
413 op = ap.get( "operation" , "GetItem" )
414 write_ops = cc. WRITE_OPS if cc else _FALLBACK_WRITE_OPS
415 is_write = op in write_ops
416 if cc:
417 td = table_map.get(ap.get( "table" , "" ))
418 try :
419 cap = cc.pattern_monthly_cost(ap, td, entity_attr_sizes)[ "cap" ]
420 cu_per_call = cap[ "rcus" ] + cap[ "wcus" ]
421 except Exception :
422 cu_per_call = 1.0
423 unit = cc. WRU_PRICE if is_write else cc. RRU_PRICE
424 else :
425 cu_per_call = 1.0
426 unit = 0.625 / 1_000_000 if is_write else 0.125 / 1_000_000
427 # Vector capacity is metered in bytes, so it contributes nothing to cu_per_call. A
428 # SearchVectors pattern would otherwise estimate at exactly $0 and the gate would
429 # wave through a run that does incur charges. Search uses the deliberately-high
430 # gate bound (the reported driver is a LOWER bound — wrong direction for a guard);
431 # writes use the same per-call figure the cost report uses, validated to 2.1%.
432 v_per_call = 0.0
433 if cc:
434 td = table_map.get(ap.get( "table" , "" ))
435 vidx = {v[ "index_name" ]: v for v in ((td or {}).get( "vector_indexes" ) or [])}
436 try :
437 if op == cc. VECTOR_SEARCH_OP :
438 vi = vidx.get(ap.get( "index" ))
439 if vi:
440 v_per_call = (
441 cc.vector_search_bytes_spend_gate_upper(ap, vi, td or {})
442 / cc. BYTES_PER_GB
443 * cc. VECTOR_SEARCH_PRICE_PER_GB
444 )
445 elif is_write and vidx:
446 wb, _ = cc.vector_write_bytes_per_call(ap, td or {})
447 v_per_call = wb / cc. BYTES_PER_GB * cc. VECTOR_WRITE_PRICE_PER_GB
448 except Exception :
449 v_per_call = 0.0
450 v_cost = calls * v_per_call
451 vector_cost += v_cost
452
453 c = calls * cu_per_call * unit + v_cost
454 driven_cost += c
455 entry = { "pattern_id" : ap.get( "pattern_id" ), "bench_rps" : rps, "calls" : calls, "cost" : c}
456 if v_cost:
457 entry[ "vector_cost" ] = v_cost
458 per_pattern.append(entry)
459
460 # Seeding cost: writes are billed ⌈item_size/1KB⌉ WRU each. run_seed seeds
461 # PER PATTERN, not per table — every seeded key embeds the pattern_id
462 # (_seed_pk(pid, …)), so patterns sharing a table do NOT dedup against each
463 # other. A table with K read/write patterns therefore gets ~K ×
464 # seed_items_per_table items. The seed item size run_seed uses is the LARGEST
465 # declared item size among the patterns on that pattern's table, so price
466 # each pattern at its table's max size. (n_partitions × items_per_partition
467 # still ≈ seed_items_per_table per pattern, so items_per_partition does not
468 # multiply the count.)
469 seed_items = int (cfg.get( "seed_items_per_table" , 500 ))
470 wru_price = cc. WRU_PRICE if cc else 0.625 / 1_000_000
471 # Per-table max item size (matches run_seed's choice of seed item size).
472 table_max_size: dict = {}
473 for ap in aps:
474 tn = ap.get( "table" )
475 if not tn:
476 continue
477 table_max_size[tn] = max (
478 table_max_size.get(tn, 0 ), int (ap.get( "estimated_item_size_bytes" , 1024 ))
479 )
480 seed_cost = 0.0
481 for ap in aps:
482 tn = ap.get( "table" )
483 if not tn:
484 continue
485 item_kb = max ( 1 , - ( - table_max_size[tn] // 1024 )) # ceil KB
486 seed_cost += seed_items * item_kb * wru_price
487 # Seeded items on a vector-index table carry the embedding, so every seed write
488 # also meters vector write bytes. On a 1024-dim KEYS_ONLY index that is ~4 KB per
489 # item at $0.52/GB — small, but it is the same order as the base-table seed cost
490 # and would otherwise be invisible to the gate.
491 td = table_map.get(tn)
492 vis = (td or {}).get( "vector_indexes" ) or []
493 if cc and vis:
494 try :
495 wb, _ = cc.vector_write_bytes_per_call(
496 {
497 "operation" : "PutItem" ,
498 "estimated_item_size_bytes" : table_max_size[tn],
499 "attributes_written" : [
500 v[ "vector_attribute" ] for v in vis if v.get( "vector_attribute" )
501 ],
502 },
503 td or {},
504 )
505 v = seed_items * wb / cc. BYTES_PER_GB * cc. VECTOR_WRITE_PRICE_PER_GB
506 seed_cost += v
507 vector_cost += v
508 except Exception :
509 pass
510
511 total = driven_cost + seed_cost
512 out = {
513 "total_usd" : total,
514 "driven_usd" : driven_cost,
515 "seed_usd" : seed_cost,
516 "per_pattern" : per_pattern,
517 "window_seconds" : window,
518 }
519 # Only surfaced for designs that actually use vectors, so the gate's output for every
520 # other design is unchanged.
521 if vector_cost:
522 out[ "vector_usd" ] = vector_cost
523 return out
524
525
526 def _percentile (values, q):
527 if not values:
528 return None
529 try :
530 if len (values) < 2 :
531 return values[ 0 ]
532 quantiles = statistics.quantiles(values, n = 100 , method = "inclusive" )
533 idx = max ( 0 , min ( len (quantiles) - 1 , q - 1 ))
534 return quantiles[idx]
535 except statistics.StatisticsError:
536 return values[ 0 ] if values else None
537
538
539 def _aggregate (
540 rows: list[ dict ],
541 patterns: list[ dict ],
542 tables: list[ dict ],
543 cfg: dict ,
544 exact_counts: dict | None = None ,
545 ) -> list[ dict ]:
546 """Per-pattern aggregation: steady-state + cold-start blocks.
547
548 `exact_counts` (keyed by (pattern_id, phase)) carries the Lambda's uncapped
549 call/throttle tallies. When present, the steady-state call_count and
550 throttle count come from it — not from the (per-key-capped) rows — so a
551 down-sampled measure window never under-reports throttles. Latency
552 percentiles still come from the recorded rows (a representative sample)."""
553 exact_counts = exact_counts or {}
554 gsi_index_names = {}
555 for t in tables:
556 for g in t.get( "gsis" ) or []:
557 gsi_index_names[g[ "index_name" ]] = g
558 vector_index_names = {}
559 for t in tables:
560 for vi in t.get( "vector_indexes" ) or []:
561 vector_index_names[vi[ "index_name" ]] = vi
562
563 out = []
564 for p in patterns:
565 pid = p[ "pattern_id" ]
566 p_rows = [r for r in rows if r.get( "pattern_id" ) == pid]
567 measure = [r for r in p_rows if r.get( "phase" ) == "measure" ]
568 warmup = [r for r in p_rows if r.get( "phase" ) == "warmup" ]
569
570 def _lat (rs):
571 return [r[ "latency_ms" ] for r in rs if r.get( "latency_ms" ) is not None ]
572
573 measure_lat = _lat(measure)
574 warmup_lat = _lat(warmup)
575
576 measure_cu = [r[ "consumed_cu" ] for r in measure if r.get( "consumed_cu" ) is not None ]
577 mean_cu = sum (measure_cu) / len (measure_cu) if measure_cu else 0.0
578
579 gsi_sum: dict = {}
580 base_wcu_sum = 0.0
581 gsi_wcu_sum = 0.0
582 for r in measure:
583 for name, v in (r.get( "gsi_cu" ) or {}).items():
584 gsi_sum[name] = gsi_sum.get(name, 0.0 ) + v
585 if p[ "operation" ] in (
586 "PutItem" ,
587 "UpdateItem" ,
588 "DeleteItem" ,
589 "BatchWriteItem" ,
590 "TransactWriteItems" ,
591 ):
592 gsi_wcu_sum += v
593 if p[ "operation" ] in (
594 "PutItem" ,
595 "UpdateItem" ,
596 "DeleteItem" ,
597 "BatchWriteItem" ,
598 "TransactWriteItems" ,
599 ):
600 base_wcu_sum += r.get( "consumed_cu" , 0.0 ) or 0.0
601 amp_ratio = (gsi_wcu_sum / base_wcu_sum) if base_wcu_sum > 0 else 0.0
602
603 # Prefer the Lambda's exact uncapped tallies for call/throttle/error
604 # counts; fall back to row-derived counts when exact_counts is absent
605 # (older Lambda or a phase that recorded no exact tally). Counting errors
606 # from the exact tally — not the capped rows — means a structurally broken
607 # pattern's true error count survives row down-sampling, so the report can
608 # raise a correctness finding rather than letting a "0 observed CU" delta
609 # look benign. The Lambda's `errors` tally is NON-throttle only (throttles
610 # are counted separately); the row fallback excludes throttled rows to
611 # match that definition.
612 ec = exact_counts.get((pid, "measure" ))
613 exact_calls = ec[ "calls" ] if ec else len (measure)
614 exact_throttles = ec[ "throttles" ] if ec else sum ( 1 for r in measure if r.get( "throttled" ))
615 if ec and "errors" in ec:
616 exact_errors = ec[ "errors" ]
617 error_codes = dict (ec.get( "error_codes" ) or {})
618 else :
619 exact_errors = sum ( 1 for r in measure if r.get( "error" ) and not r.get( "throttled" ))
620 error_codes = {}
621 for r in measure:
622 if r.get( "error" ) and not r.get( "throttled" ):
623 error_codes[r[ "error" ]] = error_codes.get(r[ "error" ], 0 ) + 1
624 error_rate = (exact_errors / exact_calls) if exact_calls else 0.0
625 # Per-item Transact* cancellation reasons (e.g. {"TransactionConflict":
626 # 13}) so the report can say WHY transactions cancelled instead of just
627 # "TransactionCanceledException". Older Lambdas don't emit it → {}.
628 cancellation_reason_codes = dict ((ec or {}).get( "cancellation_reason_codes" ) or {})
629
630 steady = {
631 "call_count" : exact_calls,
632 "rows_sampled" : len (measure),
633 "p50_ms" : _percentile(measure_lat, 50 ),
634 "p95_ms" : _percentile(measure_lat, 95 ),
635 "p99_ms" : _percentile(measure_lat, 99 ),
636 "mean_observed_cu" : mean_cu,
637 "gsi_cu_by_index" : gsi_sum,
638 "amplification_ratio" : amp_ratio,
639 "throttles" : exact_throttles,
640 "errors" : exact_errors,
641 "error_rate" : error_rate,
642 "error_codes" : error_codes,
643 "cancellation_reason_codes" : cancellation_reason_codes,
644 }
645
646 # Vector capacity is metered in BYTES, not capacity units, so it cannot ride the
647 # consumed_cu / gsi_cu fields — a SearchVectors pattern reports 0 CU and would
648 # otherwise look free. These are the figures calculate_costs.py deliberately
649 # REFUSES to model for search (the fraction of an index examined varies ~10x with
650 # configuration), so an observed per-call number is the only honest source.
651 #
652 # Emitted only when the design declares a vector index, so perf_summary.json for
653 # a non-vector model is unchanged.
654 #
655 # Means are over non-errored calls: a failed call reports 0 bytes, and averaging
656 # those in would understate the per-call cost used for extrapolation. The error
657 # count sits alongside in the same block, so a partly-failed pattern is still
658 # visible rather than silently flattered.
659 if vector_index_names:
660 vs_vals = [
661 r[ "vector_search_bytes" ]
662 for r in measure
663 if r.get( "vector_search_bytes" ) is not None and not r.get( "error" )
664 ]
665 vw_sum: dict = {}
666 vw_calls: dict = {}
667 for r in measure:
668 if r.get( "error" ):
669 continue
670 for name, v in (r.get( "vector_write_bytes" ) or {}).items():
671 vw_sum[name] = vw_sum.get(name, 0.0 ) + v
672 vw_calls[name] = vw_calls.get(name, 0 ) + 1
673 steady[ "mean_vector_search_bytes" ] = sum (vs_vals) / len (vs_vals) if vs_vals else 0.0
674 steady[ "vector_search_bytes_total" ] = sum (vs_vals)
675 steady[ "vector_write_bytes_by_index" ] = vw_sum
676 steady[ "mean_vector_write_bytes_by_index" ] = {
677 name: vw_sum[name] / vw_calls[name] for name in vw_sum if vw_calls[name]
678 }
679
680 # Key-distribution histogram → drives the key_skew_patterns signal
681 # (hot-partition risk, Mechanics #3). Emitted ONLY for skewed (zipf)
682 # sampling — i.e. representative mode or an explicit zipf config. Uniform
683 # runs (quick/standard) DO record key_idx on every row, but a uniform
684 # round-robin distribution is flat by construction, so its stddev_over_mean
685 # ≈ 0 and the key_skew signal could never fire usefully; omitting the field
686 # entirely keeps the documented guarantee ("uniform runs omit it") true and
687 # the signal strictly representative-mode. Gate on the config, not on the
688 # mere presence of key_idx.
689 key_sampling = (cfg.get( "read_pattern_key_sampling" ) or "uniform" ).lower()
690 measure_keyed = (
691 [r for r in measure if r.get( "key_idx" ) is not None ] if key_sampling == "zipf" else []
692 )
693 part_idxs = [r[ "key_idx" ] for r in measure_keyed]
694 if part_idxs:
695 counts: dict = {}
696 for k in part_idxs:
697 counts[k] = counts.get(k, 0 ) + 1
698 freqs = list (counts.values())
699 n = len (part_idxs)
700 mean_f = n / len (counts)
701 var = sum ((f - mean_f) ** 2 for f in freqs) / len (counts)
702 stddev = var ** 0.5
703 hot_part = max (counts, key =lambda k: counts[k])
704 # Baseline-free hot-partition latency signal: the hottest partition's
705 # tail latency vs every OTHER partition's, within this same pattern.
706 # If the hot partition is materially slower than the cold ones, that
707 # is a hot partition by definition — independent of other patterns or
708 # absolute thresholds. On on-demand tables this is how a hot key
709 # shows up (adaptive capacity absorbs it as latency, not throttles).
710 hot_lat = [
711 r[ "latency_ms" ]
712 for r in measure_keyed
713 if r[ "key_idx" ] == hot_part and r.get( "latency_ms" ) is not None
714 ]
715 cold_lat = [
716 r[ "latency_ms" ]
717 for r in measure_keyed
718 if r[ "key_idx" ] != hot_part and r.get( "latency_ms" ) is not None
719 ]
720 steady[ "key_distribution" ] = {
721 "n_distinct_keys" : len (counts),
722 "top_key_share" : max (freqs) / n,
723 "stddev_over_mean" : (stddev / mean_f) if mean_f > 0 else 0.0 ,
724 "hot_partition_p99_ms" : _percentile( sorted (hot_lat), 99 ),
725 "cold_partition_p99_ms" : _percentile( sorted (cold_lat), 99 ),
726 }
727
728 cold_start_elevated = False
729 w_p50 = _percentile(warmup_lat, 50 )
730 w_p99 = _percentile(warmup_lat, 99 )
731 if w_p99 is not None and steady[ "p99_ms" ] is not None and steady[ "p99_ms" ] > 0 :
732 cold_start_elevated = w_p99 > 2 * steady[ "p99_ms" ]
733
734 cold = {
735 "warmup_call_count" : len (warmup),
736 "warmup_p50_ms" : w_p50,
737 "warmup_p99_ms" : w_p99,
738 "warmup_throttles" : sum ( 1 for r in warmup if r.get( "throttled" )),
739 "cold_start_elevated" : cold_start_elevated,
740 }
741
742 out.append(
743 {
744 "pattern_id" : pid,
745 "bench_rps" : _bench_rps(p, cfg),
746 "steady_state" : steady,
747 "cold_start" : cold,
748 "measurement_tainted" : False , # set by caller if Lambda flagged it
749 }
750 )
751
752 return out
753
754
755 def main ():
756 p = argparse.ArgumentParser( description = __doc__ .splitlines()[ 0 ])
757 p.add_argument( "--model" , required = True , help = "path to dynamodb_data_model.json (the design)" )
758 p.add_argument(
759 "--config" ,
760 required = True ,
761 help = "path to benchmark_config.json (per-run knobs; mode, "
762 "scale, seeding — see references/performance-model-schema.md)" ,
763 )
764 p.add_argument(
765 "--manifest" ,
766 required = True ,
767 help = "path to created_resources.json written by deploy_model.py "
768 "(names the deployed tables + benchmark Lambda to invoke)" ,
769 )
770 p.add_argument(
771 "--raw-out" ,
772 required = True ,
773 help = "output path for per-call rows (JSONL, large; human/agent "
774 "do NOT read this — it feeds generate_perf_report.py)" ,
775 )
776 p.add_argument(
777 "--summary-out" ,
778 required = True ,
779 help = "output path for the aggregated perf_summary.json consumed "
780 "by generate_perf_report.py" ,
781 )
782 p.add_argument(
783 "--allow-spend" ,
784 action = "store_true" ,
785 help = "acknowledge the estimated AWS spend and skip the cost-guardrail "
786 "refusal (the orchestrator forwards this after user consent)." ,
787 )
788 args = p.parse_args()
789
790 boto3_mod = _require_boto3()
791 model = _load_json(Path(args.model))
792 cfg = _apply_mode_preset(_load_json(Path(args.config)))
793 manifest = _load_json(Path(args.manifest))
794
795 _validate(model)
796
797 if "lambda" not in manifest:
798 _die(
799 "manifest has no 'lambda' block — this benchmark_model.py "
800 "expects a Lambda-based run. Re-run deploy_model.py on the "
801 "current version of the skill."
802 )
803
804 session = boto3_mod.Session( profile_name = cfg[ "aws_profile" ], region_name = cfg[ "region" ])
805 # Sync Lambda invoke holds the HTTP connection for up to `lambda_timeout_seconds`.
806 # boto3 default read_timeout is 60s — way too short for a 90+s benchmark.
807 # Bump to the Lambda timeout plus a safety margin and disable boto's own
808 # invoke retries (Lambda surfaces handler errors via FunctionError we already
809 # parse; retrying would re-run the benchmark).
810 from botocore.config import Config as _BotoConfig
811
812 _invoke_timeout = int (cfg.get( "lambda_timeout_seconds" , 900 )) + 60
813 lam = session.client(
814 "lambda" ,
815 config = _BotoConfig(
816 retries = { "max_attempts" : 1 , "mode" : "standard" },
817 read_timeout = _invoke_timeout,
818 connect_timeout = 10 ,
819 ),
820 )
821 fn_name = manifest[ "lambda" ][ "function_name" ]
822
823 invocations_total, _ = _compute_split(cfg, len (model[ "access_patterns" ]))
824 if invocations_total < 0 :
825 _die(
826 "Even a single Lambda invocation's settle+seed+warmup overhead "
827 "exceeds the configured lambda_timeout_seconds. Either increase "
828 "lambda_timeout_seconds (max 900), reduce warmup_seconds, or "
829 "reduce the number of access patterns benchmarked per run."
830 )
831 print (
832 f "Benchmark budget: { len (model[ 'access_patterns' ]) } pattern(s), "
833 f "splitting measurement across { invocations_total } Lambda invocation(s)."
834 )
835
836 # Upfront wall-clock estimate + a loud foreground reminder. Patterns run
837 # SERIALLY inside the Lambda, so total time ≈ settle + seed + Σ_patterns
838 # (warmup + duration), plus a little per-invocation handoff. This is the most
839 # reliable nudge against the failure mode where the agent lets a long run get
840 # auto-backgrounded (when it exceeds a tool's default timeout) and then reads
841 # a stale prior summary. Seeing "~N min — run foreground and wait" BEFORE the
842 # blocking phase is what reliably triggers the right behavior.
843 _np = len (model[ "access_patterns" ])
844 _settle = int (cfg.get( "table_settle_seconds" , 30 ))
845 _seedw = int ( 30 + ( int (cfg.get( "seed_items_per_table" , 500 )) / 25.0 ) * 0.03 * max ( 1 , _np))
846 _warm = int (cfg.get( "warmup_seconds" , 10 ))
847 _dur = int (cfg.get( "duration_seconds" , 90 ))
848 _est_s = _settle + _seedw + _np * (_warm + _dur) + invocations_total * 20
849 _est_min = _est_s / 60.0
850 print (
851 f "Estimated wall-clock: ~ { _est_min :.0f} min "
852 f "( { _est_s } s: { _settle } s settle + ~ { _seedw } s seed + "
853 f " { _np } patterns x ( { _warm } s warmup + { _dur } s measure), serial)."
854 )
855 if _est_s > 110 :
856 print (
857 " ┌─ RUN THIS IN THE FOREGROUND AND WAIT ─────────────────────────┐ \n "
858 f " │ This run takes ~ { _est_min :.0f} min, longer than a default tool/shell │ \n "
859 " │ timeout. Do NOT background it: a backgrounded run can be │ \n "
860 " │ killed mid-flight, leaving a STALE perf_summary.json that │ \n "
861 " │ looks fresh. Raise your tool's timeout to exceed the estimate │ \n "
862 " │ above and let this command block to completion. Verify │ \n "
863 " │ benchmark_completed_at in the summary post-dates launch. │ \n "
864 " └────────────────────────────────────────────────────────────────┘" ,
865 flush = True ,
866 )
867
868 # Cost guardrail (pre-spend gate). Estimate the actual AWS charge BEFORE
869 # invoking the Lambda. Refuse if it exceeds cost_guardrail_usd unless the
870 # user explicitly acknowledged via --allow-spend. Distinct from the in-Lambda
871 # abort_on_throttle_rate runtime guard.
872 guardrail = float (cfg.get( "cost_guardrail_usd" , 0.50 ))
873 spend = _estimate_bench_spend(model, cfg)
874 print (
875 f "Estimated AWS spend for this run: ~$ { spend[ 'total_usd' ] :.3f} "
876 f "(driven load ~$ { spend[ 'driven_usd' ] :.3f} + seeding "
877 f "~$ { spend[ 'seed_usd' ] :.3f} ). Guardrail: $ { guardrail :.2f} ."
878 )
879 if spend[ "total_usd" ] > guardrail and not args.allow_spend:
880 _die(
881 f "estimated spend $ { spend[ 'total_usd' ] :.3f} exceeds the cost "
882 f "guardrail $ { guardrail :.2f} . Lower scale_factor / duration_seconds "
883 "/ max_rps_per_pattern, raise cost_guardrail_usd in "
884 "benchmark_config.json, or re-run with --allow-spend to proceed "
885 "after acknowledging the charge." ,
886 code = 3 ,
887 )
888
889 all_rows: list[ dict ] = []
890 tainted_overall: dict = {}
891 coverage_union_measured: set = set ()
892 seed_verification: dict = {}
893 first_invocation_errored = False
894 # Load-shape knobs the Lambda echoes back (mode, key_sampling,
895 # items_per_partition). Captured from the first invocation and surfaced at
896 # the summary top-level so generate_perf_report can branch its disclaimer /
897 # Load-risk section off the ACTUAL run, not just the config it was handed.
898 lambda_echo: dict = {}
899 # Accumulate exact (uncapped) call/throttle tallies across invocations,
900 # keyed by (pattern_id, phase). These override the row-derived counts in
901 # _aggregate so throttles are never under-reported when rows were capped.
902 exact_counts: dict = {}
903
904 # Wall-clock bookends for the whole invocation sequence. These let the
905 # report (and the agent) confirm the summary came from THIS run, not a
906 # stale prior one left on disk by a killed/backgrounded benchmark. The
907 # summary is written once, after all invocations return, so a run that
908 # never completes never stamps these — making them a reliable "completed"
909 # marker. UTC, ISO-8601.
910 run_started_at = datetime.now(timezone.utc)
911 _wall_start = time.monotonic()
912
913 for idx in range (invocations_total):
914 is_first = idx == 0
915 phase_plan = [ "settle" , "seed" , "warmup" , "measure" ] if is_first else [ "measure" ]
916 payload = {
917 "phase_plan" : phase_plan,
918 "invocation_index" : idx,
919 "invocations_total" : invocations_total,
920 "patterns" : _attach_vector_index_meta(model),
921 "tables" : model[ "tables" ],
922 "manifest" : manifest,
923 "config" : cfg,
924 }
925 print (
926 f "Invoking Lambda (invocation { idx + 1 } / { invocations_total } , " f "phases= { phase_plan } ) …"
927 )
928 t0 = time.monotonic()
929 resp = _invoke_lambda(lam, fn_name, payload)
930 dt = time.monotonic() - t0
931 print (
932 f " returned in { dt :.1f} s, "
933 f " { len (resp.get( 'raw_rows' , [])) } rows, "
934 f "phases_run= { resp.get( 'phases_run' ) } "
935 )
936
937 if is_first:
938 seed_verification = resp.get( "seed_verification" ) or {}
939 if resp.get( "seed_verification_failed" ):
940 print (
941 " ! seed verification failed — aborting further invocations." , file = sys.stderr
942 )
943 all_rows.extend(resp.get( "raw_rows" ) or [])
944 first_invocation_errored = True
945 break
946 if resp.get( "handler_error" ):
947 print ( f " ! Lambda handler error: { resp[ 'handler_error' ] } " , file = sys.stderr)
948 all_rows.extend(resp.get( "raw_rows" ) or [])
949 first_invocation_errored = True
950 break
951
952 if is_first:
953 for k in ( "mode" , "key_sampling" , "items_per_partition" ):
954 if k in resp:
955 lambda_echo[k] = resp[k]
956
957 all_rows.extend(resp.get( "raw_rows" ) or [])
958 tainted_overall.update(resp.get( "measurement_tainted" ) or {})
959 cov = resp.get( "coverage" ) or {}
960 for pid in cov.get( "measured_patterns" ) or []:
961 coverage_union_measured.add(pid)
962 for ec in resp.get( "exact_counts" ) or []:
963 key = (ec[ "pattern_id" ], ec[ "phase" ])
964 agg = exact_counts.setdefault(
965 key,
966 {
967 "calls" : 0 ,
968 "throttles" : 0 ,
969 "errors" : 0 ,
970 "error_codes" : {},
971 "cancellation_reason_codes" : {},
972 },
973 )
974 agg[ "calls" ] += ec.get( "calls" , 0 )
975 agg[ "throttles" ] += ec.get( "throttles" , 0 )
976 agg[ "errors" ] += ec.get( "errors" , 0 )
977 for code, cnt in (ec.get( "error_codes" ) or {}).items():
978 agg[ "error_codes" ][code] = agg[ "error_codes" ].get(code, 0 ) + cnt
979 for code, cnt in (ec.get( "cancellation_reason_codes" ) or {}).items():
980 agg[ "cancellation_reason_codes" ][code] = (
981 agg[ "cancellation_reason_codes" ].get(code, 0 ) + cnt
982 )
983
984 # Write raw.
985 raw_path = Path(args.raw_out)
986 with raw_path.open( "w" ) as f:
987 for r in all_rows:
988 f.write(json.dumps(r, default = str ))
989 f.write( " \n " )
990 print ( f "Raw rows: { raw_path } ( { len (all_rows) } rows)" )
991
992 # Aggregate.
993 patterns = model[ "access_patterns" ]
994 per_pattern = _aggregate(all_rows, patterns, model.get( "tables" ) or [], cfg, exact_counts)
995 for item in per_pattern:
996 if item[ "pattern_id" ] in tainted_overall:
997 item[ "measurement_tainted" ] = True
998 item[ "tainted_reason" ] = tainted_overall[item[ "pattern_id" ]]
999
1000 declared = {p[ "pattern_id" ] for p in patterns}
1001 missing = sorted (declared - coverage_union_measured)
1002
1003 summary = {
1004 "run_id" : manifest.get( "run_id" ),
1005 "account" : manifest.get( "account" ),
1006 "region" : manifest.get( "region" ),
1007 "prefix" : manifest.get( "prefix" ),
1008 "manifest" : {
1009 "account" : manifest.get( "account" ),
1010 "region" : manifest.get( "region" ),
1011 "prefix" : manifest.get( "prefix" ),
1012 "run_id" : manifest.get( "run_id" ),
1013 },
1014 "config" : cfg,
1015 # Freshness markers — written only here, after every invocation has
1016 # returned, so a killed/backgrounded run never produces them. The
1017 # report renders these and the agent checks benchmark_completed_at
1018 # against the time it launched the run; a stale summary (from a prior
1019 # run) will pre-date the launch and is caught instead of interpreted.
1020 "run_started_at" : run_started_at.isoformat(),
1021 "benchmark_completed_at" : datetime.now(timezone.utc).isoformat(),
1022 "benchmark_wall_seconds" : round (time.monotonic() - _wall_start, 1 ),
1023 "total_rows" : len (all_rows),
1024 # Load-shape knobs as the Lambda actually ran them (echoed back). The
1025 # report prefers summary["mode"] over config["mode"] so it branches on
1026 # what ran, not just what was requested. Falls back to cfg if an older
1027 # Lambda didn't echo.
1028 "mode" : lambda_echo.get( "mode" , cfg.get( "mode" , "standard" )),
1029 "key_sampling" : lambda_echo.get(
1030 "key_sampling" , cfg.get( "read_pattern_key_sampling" , "uniform" )
1031 ),
1032 "items_per_partition" : lambda_echo.get(
1033 "items_per_partition" , cfg.get( "items_per_partition" , 1 )
1034 ),
1035 "invocations_total" : invocations_total,
1036 "seed_verification" : seed_verification,
1037 "coverage" : {
1038 "measured_patterns" : sorted (coverage_union_measured),
1039 "missing_patterns" : missing,
1040 "coverage_incomplete" : bool (missing) or first_invocation_errored,
1041 },
1042 "patterns" : per_pattern,
1043 }
1044 sp = Path(args.summary_out)
1045 sp.write_text(json.dumps(summary, indent = 2 , default = str ))
1046 print ( f "Summary: { sp } " )
1047 print (
1048 f "Benchmark completed at { summary[ 'benchmark_completed_at' ] } "
1049 f "( { summary[ 'benchmark_wall_seconds' ] } s wall, "
1050 f " { summary[ 'total_rows' ] } rows) — verify this timestamp is newer "
1051 f "than when you launched the run before trusting the report."
1052 )
1053
1054 if summary[ "coverage" ][ "coverage_incomplete" ]:
1055 print (
1056 f " \n WARNING: coverage incomplete — missing: { missing } " ,
1057 file = sys.stderr,
1058 )
1059 sys.exit( 1 if first_invocation_errored else 0 )
1060
1061
1062 if __name__ == "__main__" :
1063 main()