Setting the file. One moment.
Di Status Assessment · AWS Observability · aws/agent-toolkit-for-aws · Skills Docs
Back to the top of the page 69
Creating Amazon Aurora Db Cluster With Instances
92
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Debugging Lambda Timeouts
scripts/cloudwatch/ di_status_assessment.py
Python · 144 lines · 5 KB
16 dispatches on, instead of leaking three different argument tuples to
17 three different renderers.
18
19 I/O lives in the caller. ``assess`` takes a ``check_status`` callable so
20 the policy can be tested without touching boto3.
21 """
22
23 from dataclasses import dataclass
24 from datetime import datetime, timezone
25 from typing import Callable, List, Optional, Tuple, Union
26
27 # A single check call: (status_label, start, end) → (has_events, events, error_or_None).
28 # Matches the existing ``_check_status_with_time_range`` shape.
29 CheckStatus = Callable[[ str , datetime, datetime], Tuple[ bool , List[ dict ], Optional[ str ]]]
30
31
32 @dataclass ( frozen = True )
33 class TimeWindow :
34 """The four ISO-formatted time strings every consolidated renderer needs."""
35
36 created_at: str
37 requested_start: str
38 active_query_start: str
39 query_end: str
40
41
42 @dataclass ( frozen = True )
43 class _StatusCheckResult :
44 has_events: bool
45 events: List[ dict ]
46 error: Optional[ str ]
47
48
49 @dataclass ( frozen = True )
50 class Active :
51 """ACTIVE events were found in the (clamped) ACTIVE window."""
52
53 active: _StatusCheckResult
54
55
56 @dataclass ( frozen = True )
57 class Ready :
58 """ACTIVE not confirmed, but READY events were found."""
59
60 active: _StatusCheckResult
61 ready: _StatusCheckResult
62
63
64 @dataclass ( frozen = True )
65 class ErrorOrPending :
66 """Neither ACTIVE nor READY confirmed.
67
68 The ERROR check decides between ERROR and PENDING based on whether
69 ``error.has_events`` is true.
70 """
71
72 active: _StatusCheckResult
73 ready: _StatusCheckResult
74 error: _StatusCheckResult
75
76
77 Verdict = Union[Active, Ready, ErrorOrPending]
78
79
80 def _check_result (
81 check: CheckStatus, status: str , start: datetime, end: datetime
82 ) -> _StatusCheckResult:
83 has_events, events, error = check(status, start, end)
84 return _StatusCheckResult( has_events = has_events, events = events, error = error)
85
86
87 def _format_iso (value: datetime) -> str :
88 return value.astimezone(timezone.utc).strftime( "%Y-%m- %d T%H:%M:%SZ" )
89
90
91 def assess (
92 * ,
93 created_at: datetime,
94 requested_start: datetime,
95 query_end: datetime,
96 check_status: CheckStatus,
97 ) -> Tuple[Verdict, TimeWindow]:
98 """Run the consolidated status assessment.
99
100 The caller is responsible for:
101
102 * Parsing ISO inputs into ``datetime`` objects (string parsing is an input
103 concern, not policy).
104 * Verifying ``query_end > requested_start`` before calling — that error
105 message is owned by the tool layer.
106 * Providing a ``check_status`` callable that issues the AWS query.
107
108 Returns a ``(verdict, time_window)`` pair. The renderer dispatches on
109 the verdict type; both verdict and time_window are passed to the
110 renderer.
111 """
112 requested_start_utc = requested_start.astimezone(timezone.utc)
113 query_end_utc = query_end.astimezone(timezone.utc)
114 created_at_utc = created_at.astimezone(timezone.utc)
115 active_query_start_utc = max (created_at_utc, requested_start_utc)
116
117 time_window = TimeWindow(
118 created_at = _format_iso(created_at_utc),
119 requested_start = _format_iso(requested_start_utc),
120 active_query_start = _format_iso(active_query_start_utc),
121 query_end = _format_iso(query_end_utc),
122 )
123
124 if query_end_utc > active_query_start_utc:
125 active = _check_result(check_status, "ACTIVE" , active_query_start_utc, query_end_utc)
126 else :
127 active = _StatusCheckResult(
128 has_events = False ,
129 events = [],
130 error = (
131 "Skipped: ACTIVE query window is empty after applying created_at clamp "
132 f "(start= { time_window.active_query_start } , end= { time_window.query_end } )"
133 ),
134 )
135
136 if active.has_events:
137 return Active( active = active), time_window
138
139 ready = _check_result(check_status, "READY" , requested_start_utc, query_end_utc)
140 if ready.has_events:
141 return Ready( active = active, ready = ready), time_window
142
143 error = _check_result(check_status, "ERROR" , requested_start_utc, query_end_utc)
144 return ErrorOrPending( active = active, ready = ready, error = error), time_window