Setting the file. One moment. 69
Creating Amazon Aurora Db Cluster With Instances
92
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Creating API Gateway Stage
scripts/cloudwatch/di_status_rendering.py
Python·354 lines·12 KB
(
12 data: Dict[str, Any],
13 normalized_type: str,
14 service: str,
15 environment: str,
16 requested_status: str,
17) -> str:
18 """Render the explicit status-history response."""
19 events = data.get("Events", [])
20
21 output = f"""INSTRUMENTATION STATUS
22
23TYPE: {normalized_type}
24SERVICE: {data.get('Service', service)}
25ENVIRONMENT: {data.get('Environment', environment)}
26SIGNAL TYPE: {data.get('SignalType', SNAPSHOT_SIGNAL_TYPE)}
27REQUESTED STATUS FILTER: {requested_status}
28CURRENT STATUS: {data.get('Status', 'N/A')}
29
30LOCATION:
31"""
32 output += render_location_block(
33 location=location_from_response(data.get("Location", {})),
34 location_hash=data.get("LocationHash"),
35 )
36 output += f"- Events Returned: {len(events)}\n"
37
38 if events:
39 output += f"- Status Confirmation: CONFIRMED ({requested_status} events present)\n"
40 else:
41 output += f"- Status Confirmation: NOT CONFIRMED (no {requested_status} events)\n"
42
43 output += (
44 "- Interpretation Rule: Do not treat CURRENT STATUS as confirmed unless "
45 "STATUS EVENTS contain entries.\n"
46 )
47
48 if requested_status == "ACTIVE" and not events:
49 output += (
50 "- ACTIVE Clarification: Breakpoint is not confirmed as hit yet. "
51 "If READY is not yet confirmed, check READY first. "
52 "Otherwise wait for traffic and poll ACTIVE again.\n"
53 )
54
55 output += "\nSTATUS EVENTS:\n"
56 if not events:
57 output += f"- No {requested_status} status events found\n"
58 else:
59 for index, event in enumerate(events, 1):
60 event_time = format_timestamp(event.get("Time"))
61 error_cause = event.get("ErrorCause")
62 output += f"- Event {index}: {event_time}"
63 if error_cause:
64 output += f" | ErrorCause: {error_cause}"
65 output += "\n"
66
67 next_token_response = data.get("NextToken")
68 if next_token_response:
69 output += (
70 f'\nPAGINATION: More results available. Use next_token="{next_token_response}" '
71 "to retrieve next page."
72 )
73
74 return output
75
76
77def _render_status_section(
78 title: str,
79 start_time: str,
80 end_time: str,
81 has_events: bool,
82 events: List[dict],
83 error: Optional[str],
84 include_error_cause: bool = False,
85) -> str:
86 output = f"{title} STATUS:\n"
87 output += f"- Time Window: {start_time} to {end_time}\n"
88 if error:
89 if error.startswith("Skipped:"):
90 output += f"- Check Skipped: {error}\n"
91 else:
92 output += f"- Check Failed: {error}\n"
93 return output
94
95 if has_events:
96 output += f"- Confirmed: YES ({len(events)} event(s))\n"
97 for index, event in enumerate(events[:3], 1):
98 output += f' - Event {index}: {format_timestamp(event.get("Time"))}'
99 if include_error_cause:
100 output += f' | ErrorCause: {event.get("ErrorCause", "Unknown")}'
101 output += "\n"
102 if len(events) > 3:
103 output += f" - ... and {len(events) - 3} more\n"
104 else:
105 output += f"- Confirmed: NO (no {title} events found)\n"
106 return output
107
108
109def render_consolidated_active_status_output(
110 location_hash: str,
111 service: str,
112 environment: str,
113 normalized_type: str,
114 created_at: str,
115 requested_start_str: str,
116 active_query_start_str: str,
117 query_end_str: str,
118 active_has_events: bool,
119 active_events: List[dict],
120 active_error: Optional[str],
121) -> str:
122 """Render a consolidated status response when ACTIVE is confirmed or checked first."""
123 output = f"""CONSOLIDATED STATUS CHECK
124
125INSTRUMENTATION INFO:
126- LocationHash: {location_hash}
127- Service: {service}
128- Environment: {environment}
129- Type: {normalized_type}
130
131TIME RANGE:
132- Created At: {created_at}
133- Requested Start: {requested_start_str}
134- ACTIVE Query Start: {active_query_start_str}
135- Query End: {query_end_str}
136
137"""
138 output += _render_status_section(
139 title="ACTIVE",
140 start_time=active_query_start_str,
141 end_time=query_end_str,
142 has_events=active_has_events,
143 events=active_events,
144 error=active_error,
145 )
146 output += "\n"
147
148 if active_has_events:
149 output += (
150 "SNAPSHOT QUERY TIP: Try these timestamps with search_snapshots_for_status_event\n"
151 f' (log group: "{resolve_snapshot_log_group(service)}")\n'
152 " Oldest first — older events are more likely to have snapshots ingested:\n"
153 )
154 for idx, event in enumerate(reversed(active_events[:5])):
155 label = " (oldest, try first)" if idx == 0 else ""
156 if idx == len(active_events[:5]) - 1 and idx > 0:
157 label = " (most recent)"
158 output += (
159 f' - status_timestamp="{format_timestamp(event.get("Time"), default="")}"{label}\n'
160 )
161 output += "\n"
162 output += "OVERALL STATUS: ACTIVE ✓ (breakpoint is being hit)\n"
163 return output
164
165 output += "OVERALL STATUS: ACTIVE not confirmed yet\n"
166 return output
167
168
169def render_consolidated_ready_status_output(
170 location_hash: str,
171 service: str,
172 environment: str,
173 normalized_type: str,
174 created_at: str,
175 requested_start_str: str,
176 active_query_start_str: str,
177 query_end_str: str,
178 active_has_events: bool,
179 active_events: List[dict],
180 active_error: Optional[str],
181 ready_has_events: bool,
182 ready_events: List[dict],
183 ready_error: Optional[str],
184) -> str:
185 """Render a consolidated status response when READY is the best confirmed state."""
186 output = render_consolidated_active_status_output(
187 location_hash=location_hash,
188 service=service,
189 environment=environment,
190 normalized_type=normalized_type,
191 created_at=created_at,
192 requested_start_str=requested_start_str,
193 active_query_start_str=active_query_start_str,
194 query_end_str=query_end_str,
195 active_has_events=active_has_events,
196 active_events=active_events,
197 active_error=active_error,
198 )
199 if output.endswith("OVERALL STATUS: ACTIVE not confirmed yet\n"):
200 output = output[: -len("OVERALL STATUS: ACTIVE not confirmed yet\n")]
201
202 output += _render_status_section(
203 title="READY",
204 start_time=requested_start_str,
205 end_time=query_end_str,
206 has_events=ready_has_events,
207 events=ready_events,
208 error=ready_error,
209 )
210 output += "\nOVERALL STATUS: READY (waiting for traffic)\n"
211 return output
212
213
214def render_consolidated_error_or_pending_status_output(
215 location_hash: str,
216 service: str,
217 environment: str,
218 normalized_type: str,
219 created_at: str,
220 requested_start_str: str,
221 active_query_start_str: str,
222 query_end_str: str,
223 active_has_events: bool,
224 active_events: List[dict],
225 active_error: Optional[str],
226 ready_has_events: bool,
227 ready_events: List[dict],
228 ready_error: Optional[str],
229 error_has_events: bool,
230 error_events: List[dict],
231 error_error: Optional[str],
232) -> str:
233 """Render a consolidated status response for ERROR or PENDING outcomes."""
234 output = render_consolidated_active_status_output(
235 location_hash=location_hash,
236 service=service,
237 environment=environment,
238 normalized_type=normalized_type,
239 created_at=created_at,
240 requested_start_str=requested_start_str,
241 active_query_start_str=active_query_start_str,
242 query_end_str=query_end_str,
243 active_has_events=active_has_events,
244 active_events=active_events,
245 active_error=active_error,
246 )
247 if output.endswith("OVERALL STATUS: ACTIVE not confirmed yet\n"):
248 output = output[: -len("OVERALL STATUS: ACTIVE not confirmed yet\n")]
249
250 output += "\n"
251 output += _render_status_section(
252 title="READY",
253 start_time=requested_start_str,
254 end_time=query_end_str,
255 has_events=ready_has_events,
256 events=ready_events,
257 error=ready_error,
258 )
259 output += "\n"
260 output += _render_status_section(
261 title="ERROR",
262 start_time=requested_start_str,
263 end_time=query_end_str,
264 has_events=error_has_events,
265 events=error_events,
266 error=error_error,
267 include_error_cause=True,
268 )
269
270 output += "\nOVERALL STATUS: "
271 if error_has_events:
272 error_cause = error_events[0].get("ErrorCause", "Unknown") if error_events else "Unknown"
273 output += f"ERROR ({error_cause})\n"
274 output += "\nTROUBLESHOOTING:\n"
275 if error_cause == "FILE_NOT_FOUND":
276 output += "- Verify file_path is correct\n"
277 elif error_cause == "METHOD_NOT_FOUND":
278 output += "- Verify method_name and code_unit are correct\n"
279 output += "- Check if the function is loaded at runtime\n"
280 elif error_cause == "LINE_NOT_EXECUTABLE":
281 output += (
282 "- Verify line_number points to executable code (not comment/blank/declaration)\n"
283 )
284 else:
285 output += f"- Check instrumentation configuration for {error_cause}\n"
286 else:
287 output += (
288 "PENDING (no ACTIVE, READY, or ERROR events yet - wait longer or check configuration)\n"
289 )
290 output += "\nNOTE: Status events can take 1-2 minutes to appear after creation.\n"
291
292 return output
293
294
295def render_status_assessment(
296 verdict: Verdict,
297 *,
298 location_hash: str,
299 service: str,
300 environment: str,
301 normalized_type: str,
302 time_window: TimeWindow,
303) -> str:
304 """Dispatch a ``Verdict`` to the appropriate consolidated-status renderer.
305
306 Each existing renderer keeps its own prose contract; this function only
307 routes. New renderers should be added as ``Verdict`` variants gain
308 distinct presentation.
309 """
310 common = {
311 "location_hash": location_hash,
312 "service": service,
313 "environment": environment,
314 "normalized_type": normalized_type,
315 "created_at": time_window.created_at,
316 "requested_start_str": time_window.requested_start,
317 "active_query_start_str": time_window.active_query_start,
318 "query_end_str": time_window.query_end,
319 }
320
321 if isinstance(verdict, Active):
322 return render_consolidated_active_status_output(
323 **common,
324 active_has_events=verdict.active.has_events,
325 active_events=verdict.active.events,
326 active_error=verdict.active.error,
327 )
328
329 if isinstance(verdict, Ready):
330 return render_consolidated_ready_status_output(
331 **common,
332 active_has_events=verdict.active.has_events,
333 active_events=verdict.active.events,
334 active_error=verdict.active.error,
335 ready_has_events=verdict.ready.has_events,
336 ready_events=verdict.ready.events,
337 ready_error=verdict.ready.error,
338 )
339
340 if isinstance(verdict, ErrorOrPending):
341 return render_consolidated_error_or_pending_status_output(
342 **common,
343 active_has_events=verdict.active.has_events,
344 active_events=verdict.active.events,
345 active_error=verdict.active.error,
346 ready_has_events=verdict.ready.has_events,
347 ready_events=verdict.ready.events,
348 ready_error=verdict.ready.error,
349 error_has_events=verdict.error.has_events,
350 error_events=verdict.error.events,
351 error_error=verdict.error.error,
352 )
353
354 raise TypeError(f"Unknown Verdict variant: {type(verdict).__name__}")