Coverage for lambda / traffic-dial-controller / handler.py: 100.00%
202 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Capacity-driven Global Accelerator traffic-dial controller.
3Invoked on a schedule by an EventBridge rule (see
4``GCOGlobalStack._create_traffic_dial_controller`` in
5``gco/stacks/global_stack.py``). Converges each endpoint group's
6``TrafficDialPercentage`` toward the observed health of its region, using the
7per-cluster ``ClusterHealthy`` metric the health-monitor service already
8publishes to the ``GCO/HealthMonitor`` namespace in every workload region.
10This handler is self-contained (boto3 + stdlib only) and does not import the
11CLI/gco packages, matching the convention used by the other GCO Lambdas.
13Control flow is phased:
15 Phase 0 — accelerator readiness. The accelerator must be ``DEPLOYED``
16 before any decision is made. Endpoint-group updates submitted while a
17 previous change is still converging pile up and extend the window in
18 which the served configuration is unknown (the ga-registration Lambda
19 learned the equivalent lesson at deploy time), so a mid-deployment
20 cycle is skipped entirely.
21 Phase 1 — current state. ``ListEndpointGroups`` on the listener yields
22 each region's endpoint-group ARN and currently served dial.
23 Phase 2 — manual overrides. ``gco capacity traffic-dial set`` records an
24 override parameter per region; the controller never touches an
25 overridden region until ``gco capacity traffic-dial clear`` removes it.
26 Phase 3 — per-region decision. The region's healthy fraction over the
27 lookback window maps to a target dial: at or above
28 ``FULL_HEALTH_PERCENTage`` the target is 100, below it the target is
29 ``max(MIN_DIAL_PERCENTAGE, round(healthy_percent))``. The applied
30 change per run is bounded by ``MAX_STEP_PERCENTAGE`` in both
31 directions (gradual drain, gradual restore). Missing telemetry holds
32 the current dial: an absent signal must never look like ideal health,
33 and equally must never trigger a drain.
34 Phase 4 — last-healthy-region guard. If every non-overridden decision
35 lands below 100, the region with the best health signal is forced
36 back to 100 (bypassing the step limit — dialing *up* is safe). The
37 dial gates only first-choice traffic and redirects the remainder to
38 the next-closest group, and Global Accelerator does not document the
39 resulting distribution once *every* group sits below 100 — so the
40 guard keeps one fully dialed region as a deterministic absorber of
41 redirected traffic at all times.
42 Phase 5 — enforcement. In ``enforce`` mode changed dials are applied via
43 ``UpdateEndpointGroup`` carrying *only* ``TrafficDialPercentage``.
44 The API patches omitted fields, and omitting ``EndpointConfigurations``
45 is load-bearing: passing an empty list would detach the region's ALB.
46 ``monitor`` mode (the default) computes and publishes but never writes.
47 Phase 6 — publication. Every decision is emitted to the
48 ``GCO/TrafficDial`` CloudWatch namespace and the full run is stored in
49 the ``/{project}/traffic-dial/state`` SSM parameter for
50 ``gco capacity traffic-dial show``.
52Environment variables:
53 LISTENER_ARN Global Accelerator listener whose endpoint groups
54 are managed (required)
55 PROJECT_NAME deployment prefix for SSM paths and cluster names
56 (required)
57 MODE "monitor" (default) or "enforce"
58 REGIONS comma-separated workload regions to evaluate
59 LOOKBACK_MINUTES health window (default 15)
60 MIN_DIAL_PERCENTAGE floor for a degraded region's dial (default 10)
61 MAX_STEP_PERCENTAGE largest change one run may apply (default 20)
62 FULL_HEALTH_PERCENTAGE healthy percent at/above which a region returns
63 to 100 (default 95)
64"""
66from __future__ import annotations
68import json
69import logging
70import os
71from datetime import UTC, datetime, timedelta
72from typing import Any
74import boto3
75from botocore.exceptions import ClientError
77# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
78# Generated at (UTC): 2026-09-10T23:26:44Z
79# Generated from Git commit: 4c42b84d53d6cc01cd2b3c7e4011a43f850678b6
80# Flowchart(s) generated from this file:
81# * ``lambda_handler`` -> ``diagrams/code_diagrams/lambda/traffic-dial-controller/handler.lambda_handler.html``
82# (PNG: ``diagrams/code_diagrams/lambda/traffic-dial-controller/handler.lambda_handler.png``)
83# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
84# <pyflowchart-code-diagram> END
87logger = logging.getLogger()
88logger.setLevel(logging.INFO)
90#: The Global Accelerator control plane is homed in us-west-2 in the
91#: commercial partition (same convention as lambda/ga-registration).
92GA_CONTROL_PLANE_REGION = "us-west-2"
94#: Namespace the health-monitor service publishes ClusterHealthy to (see
95#: gco/services/metrics_publisher.py).
96HEALTH_METRIC_NAMESPACE = "GCO/HealthMonitor"
98#: Namespace this controller publishes its decisions to.
99DIAL_METRIC_NAMESPACE = "GCO/TrafficDial"
101#: CloudWatch PutMetricData batch bound (mirrors MetricsPublisher).
102METRIC_BATCH_SIZE = 20
104DEFAULT_LOOKBACK_MINUTES = 15
105DEFAULT_MIN_DIAL_PERCENTAGE = 10
106DEFAULT_MAX_STEP_PERCENTAGE = 20
107DEFAULT_FULL_HEALTH_PERCENTAGE = 95
110def _split_csv(raw: str | None) -> list[str]:
111 """Split a comma-separated env value into a clean list."""
112 if not raw:
113 return []
114 return [item.strip() for item in raw.split(",") if item.strip()]
117def _accelerator_arn_from_listener(listener_arn: str) -> str:
118 """Derive the accelerator ARN from one of its listener ARNs."""
119 return listener_arn.split("/listener/")[0]
122def _state_parameter_name(project_name: str) -> str:
123 """SSM parameter holding the last run's decisions."""
124 return f"/{project_name}/traffic-dial/state"
127def _override_prefix(project_name: str) -> str:
128 """SSM path under which per-region override parameters live."""
129 return f"/{project_name}/traffic-dial/"
132def list_endpoint_groups(ga_client: Any, listener_arn: str) -> dict[str, dict[str, Any]]:
133 """Return ``{region: {"arn", "traffic_dial"}}`` for the listener."""
134 groups: dict[str, dict[str, Any]] = {}
135 token: str | None = None
136 while True:
137 kwargs: dict[str, Any] = {"ListenerArn": listener_arn}
138 if token:
139 kwargs["NextToken"] = token
140 response = ga_client.list_endpoint_groups(**kwargs)
141 for group in response.get("EndpointGroups", []):
142 region = group.get("EndpointGroupRegion")
143 arn = group.get("EndpointGroupArn")
144 if not region or not arn:
145 continue
146 groups[str(region)] = {
147 "arn": str(arn),
148 "traffic_dial": int(round(float(group.get("TrafficDialPercentage", 100.0)))),
149 }
150 token = response.get("NextToken")
151 if not token:
152 return groups
155def read_overrides(ssm_client: Any, project_name: str) -> dict[str, str]:
156 """Return ``{region: raw_value}`` for every manual override parameter."""
157 overrides: dict[str, str] = {}
158 prefix = _override_prefix(project_name)
159 marker = f"{prefix}override-"
160 token: str | None = None
161 while True:
162 kwargs: dict[str, Any] = {"Path": prefix, "Recursive": False}
163 if token:
164 kwargs["NextToken"] = token
165 response = ssm_client.get_parameters_by_path(**kwargs)
166 for parameter in response.get("Parameters", []):
167 name = str(parameter.get("Name", ""))
168 if name.startswith(marker):
169 overrides[name.removeprefix(marker)] = str(parameter.get("Value", ""))
170 token = response.get("NextToken")
171 if not token:
172 return overrides
175def healthy_percent(
176 region: str,
177 cluster_name: str,
178 lookback_minutes: int,
179 *,
180 cloudwatch_client: Any | None = None,
181) -> float | None:
182 """Average ``ClusterHealthy`` (as a 0-100 percent) over the window.
184 Returns ``None`` when the metric produced no datapoints or the regional
185 CloudWatch call failed — the caller treats both as "hold the dial".
186 """
187 try:
188 cloudwatch = cloudwatch_client or boto3.client("cloudwatch", region_name=region)
189 end = datetime.now(UTC)
190 start = end - timedelta(minutes=lookback_minutes)
191 values: list[float] = []
192 token: str | None = None
193 while True:
194 kwargs: dict[str, Any] = {
195 "MetricDataQueries": [
196 {
197 "Id": "healthy",
198 "MetricStat": {
199 "Metric": {
200 "Namespace": HEALTH_METRIC_NAMESPACE,
201 "MetricName": "ClusterHealthy",
202 "Dimensions": [
203 {"Name": "ClusterName", "Value": cluster_name},
204 {"Name": "Region", "Value": region},
205 ],
206 },
207 "Period": 60,
208 "Stat": "Average",
209 },
210 }
211 ],
212 "StartTime": start,
213 "EndTime": end,
214 }
215 if token:
216 kwargs["NextToken"] = token
217 response = cloudwatch.get_metric_data(**kwargs)
218 for result in response.get("MetricDataResults", []):
219 values.extend(float(value) for value in result.get("Values", []))
220 token = response.get("NextToken")
221 if not token:
222 break
223 if not values:
224 return None
225 return 100.0 * sum(values) / len(values)
226 except Exception as exc: # noqa: BLE001 - missing telemetry means "hold"
227 logger.warning("Health signal unavailable for %s: %s", region, exc)
228 return None
231def target_dial(healthy: float, min_dial: int, full_health: int) -> int:
232 """Map a healthy percent to a target dial percentage."""
233 if healthy >= full_health:
234 return 100
235 return max(min_dial, int(round(healthy)))
238def step_limit(current: int, target: int, max_step: int) -> int:
239 """Bound the per-run dial change to ``max_step`` in either direction."""
240 if target > current:
241 return min(target, current + max_step)
242 if target < current:
243 return max(target, current - max_step)
244 return current
247def apply_last_healthy_region_guard(decisions: list[dict[str, Any]]) -> str | None:
248 """Force the best region to 100 when every computed dial fell below 100.
250 Only non-overridden decisions with an endpoint group participate: an
251 operator override is explicit intent the controller must not fight. The
252 forced restore deliberately bypasses the step limit — dialing up is safe,
253 and Global Accelerator health checks still protect against hard-down
254 endpoints. Returns the guarded region, or ``None`` when no guard applied.
255 """
256 candidates = [
257 decision
258 for decision in decisions
259 if decision["reason"] != "override" and decision["new_dial"] is not None
260 ]
261 if not candidates:
262 return None
263 if any(decision["new_dial"] >= 100 for decision in candidates):
264 return None
265 overridden_at_full = any(
266 decision["reason"] == "override" and (decision["current_dial"] or 0) >= 100
267 for decision in decisions
268 )
269 if overridden_at_full:
270 return None
272 best = max(
273 candidates,
274 key=lambda decision: (
275 decision["healthy_percent"] if decision["healthy_percent"] is not None else -1.0,
276 decision["region"],
277 ),
278 )
279 best["new_dial"] = 100
280 best["reason"] = "guard-last-healthy-region"
281 return str(best["region"])
284def publish_metrics(cloudwatch_client: Any, decisions: list[dict[str, Any]]) -> None:
285 """Emit per-region decision metrics to the GCO/TrafficDial namespace."""
286 metric_data: list[dict[str, Any]] = []
287 for decision in decisions:
288 dimensions = [{"Name": "Region", "Value": decision["region"]}]
289 if decision["new_dial"] is not None:
290 metric_data.append(
291 {
292 "MetricName": "TrafficDialPercentage",
293 "Value": float(decision["new_dial"]),
294 "Unit": "Percent",
295 "Dimensions": dimensions,
296 }
297 )
298 if decision["healthy_percent"] is not None:
299 metric_data.append(
300 {
301 "MetricName": "HealthyPercent",
302 "Value": float(decision["healthy_percent"]),
303 "Unit": "Percent",
304 "Dimensions": dimensions,
305 }
306 )
307 metric_data.append(
308 {
309 "MetricName": "HealthDataMissing",
310 "Value": 1.0 if decision["reason"] == "no-health-data" else 0.0,
311 "Unit": "Count",
312 "Dimensions": dimensions,
313 }
314 )
315 metric_data.append(
316 {
317 "MetricName": "DialApplied",
318 "Value": 1.0 if decision["applied"] else 0.0,
319 "Unit": "Count",
320 "Dimensions": dimensions,
321 }
322 )
323 try:
324 for index in range(0, len(metric_data), METRIC_BATCH_SIZE):
325 cloudwatch_client.put_metric_data(
326 Namespace=DIAL_METRIC_NAMESPACE,
327 MetricData=metric_data[index : index + METRIC_BATCH_SIZE],
328 )
329 except Exception as exc: # noqa: BLE001 - metrics are advisory
330 logger.warning("Failed to publish traffic-dial metrics: %s", exc)
333def store_state(ssm_client: Any, project_name: str, state: dict[str, Any]) -> None:
334 """Persist the run summary for `gco capacity traffic-dial show`."""
335 try:
336 ssm_client.put_parameter(
337 Name=_state_parameter_name(project_name),
338 Value=json.dumps(state, separators=(",", ":")),
339 Type="String",
340 Overwrite=True,
341 )
342 except Exception as exc: # noqa: BLE001 - state is advisory
343 logger.warning("Failed to store traffic-dial state: %s", exc)
346def lambda_handler(event: dict[str, Any], context: Any) -> dict[str, Any]:
347 """Evaluate every region's health and converge its traffic dial."""
348 listener_arn = os.environ.get("LISTENER_ARN")
349 project_name = os.environ.get("PROJECT_NAME")
350 if not listener_arn or not project_name:
351 raise ValueError("LISTENER_ARN and PROJECT_NAME environment variables are required")
353 mode = os.environ.get("MODE", "monitor").strip().lower()
354 regions = sorted(_split_csv(os.environ.get("REGIONS")))
355 lookback_minutes = int(os.environ.get("LOOKBACK_MINUTES", str(DEFAULT_LOOKBACK_MINUTES)))
356 min_dial = int(os.environ.get("MIN_DIAL_PERCENTAGE", str(DEFAULT_MIN_DIAL_PERCENTAGE)))
357 max_step = int(os.environ.get("MAX_STEP_PERCENTAGE", str(DEFAULT_MAX_STEP_PERCENTAGE)))
358 full_health = int(os.environ.get("FULL_HEALTH_PERCENTAGE", str(DEFAULT_FULL_HEALTH_PERCENTAGE)))
359 if not regions:
360 logger.warning("REGIONS is empty; nothing to evaluate")
362 now = datetime.now(UTC)
363 ga_client = boto3.client("globalaccelerator", region_name=GA_CONTROL_PLANE_REGION)
364 ssm_client = boto3.client("ssm")
366 # Phase 0 — never stack a change onto an accelerator that is still
367 # converging a previous one.
368 accelerator_arn = _accelerator_arn_from_listener(listener_arn)
369 accelerator = ga_client.describe_accelerator(AcceleratorArn=accelerator_arn)
370 accelerator_status = str(accelerator.get("Accelerator", {}).get("Status", "UNKNOWN"))
371 if accelerator_status != "DEPLOYED":
372 logger.info("Accelerator status is %s; skipping this cycle entirely", accelerator_status)
373 return {
374 "mode": mode,
375 "timestamp": now.isoformat(),
376 "accelerator_status": accelerator_status,
377 "skipped": "accelerator-not-deployed",
378 "decisions": [],
379 "updates_applied": 0,
380 "errors": 0,
381 }
383 # Phase 1 + 2 — current dials and operator overrides.
384 groups = list_endpoint_groups(ga_client, listener_arn)
385 overrides = read_overrides(ssm_client, project_name)
387 # Phase 3 — per-region decisions.
388 decisions: list[dict[str, Any]] = []
389 for region in regions:
390 group = groups.get(region)
391 decision: dict[str, Any] = {
392 "region": region,
393 "endpoint_group_arn": group["arn"] if group else None,
394 "current_dial": group["traffic_dial"] if group else None,
395 "healthy_percent": None,
396 "target_dial": None,
397 "new_dial": None,
398 "reason": "no-endpoint-group",
399 "applied": False,
400 }
401 if group is None:
402 logger.info("No endpoint group for %s yet; nothing to dial", region)
403 decisions.append(decision)
404 continue
406 current = int(group["traffic_dial"])
407 if region in overrides:
408 decision.update({"new_dial": current, "target_dial": current, "reason": "override"})
409 decisions.append(decision)
410 continue
412 health = healthy_percent(region, f"{project_name}-{region}", lookback_minutes)
413 decision["healthy_percent"] = None if health is None else round(health, 2)
414 if health is None:
415 # Hold: absent telemetry is neither health nor degradation.
416 decision.update(
417 {"new_dial": current, "target_dial": current, "reason": "no-health-data"}
418 )
419 decisions.append(decision)
420 continue
422 target = target_dial(health, min_dial, full_health)
423 decision["target_dial"] = target
424 decision["new_dial"] = step_limit(current, target, max_step)
425 decision["reason"] = "healthy" if target >= 100 else "degraded"
426 decisions.append(decision)
428 # Phase 4 — never leave every endpoint group dialed below 100.
429 guarded_region = apply_last_healthy_region_guard(decisions)
430 if guarded_region:
431 logger.warning(
432 "Every computed dial fell below 100; holding %s at 100 as the last fully dialed region",
433 guarded_region,
434 )
436 # Phase 5 — enforcement (monitor mode publishes without writing).
437 updates_applied = 0
438 errors = 0
439 for decision in decisions:
440 if decision["new_dial"] is None or decision["reason"] == "override":
441 continue
442 if decision["new_dial"] == decision["current_dial"]:
443 continue
444 if mode != "enforce":
445 continue
446 try:
447 # Only the dial: UpdateEndpointGroup patches omitted fields, and
448 # omitting EndpointConfigurations preserves the registered ALB.
449 ga_client.update_endpoint_group(
450 EndpointGroupArn=decision["endpoint_group_arn"],
451 TrafficDialPercentage=float(decision["new_dial"]),
452 )
453 decision["applied"] = True
454 updates_applied += 1
455 logger.info(
456 "Dialed %s from %s to %s (%s)",
457 decision["region"],
458 decision["current_dial"],
459 decision["new_dial"],
460 decision["reason"],
461 )
462 except ClientError as exc:
463 errors += 1
464 decision["error"] = str(exc)
465 logger.error("Failed to update traffic dial for %s: %s", decision["region"], exc)
467 # Phase 6 — publication.
468 summary: dict[str, Any] = {
469 "mode": mode,
470 "timestamp": now.isoformat(),
471 "accelerator_status": accelerator_status,
472 "decisions": decisions,
473 "updates_applied": updates_applied,
474 "errors": errors,
475 }
476 publish_metrics(boto3.client("cloudwatch"), decisions)
477 store_state(ssm_client, project_name, summary)
478 logger.info(
479 "traffic-dial cycle complete: mode=%s regions=%d updates_applied=%d errors=%d",
480 mode,
481 len(decisions),
482 updates_applied,
483 errors,
484 )
485 return summary