Setting the file. One moment. 69
Creating Amazon Aurora Db Cluster With Instances
92
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Debugging Lambda Timeouts
scripts/cloudwatch/di_crud_rendering.py
Python·372 lines·13 KB
11
12def _render_create_capture_limits(
13 max_hits: Optional[int],
14 max_string_length: Optional[int],
15 max_collection_width: Optional[int],
16 max_collection_depth: Optional[int],
17 max_stack_frames: Optional[int],
18 max_stack_trace_size: Optional[int],
19 max_object_depth: Optional[int],
20 max_fields_per_object: Optional[int],
21) -> str:
22 if not any(
23 v is not None
24 for v in [
25 max_hits,
26 max_string_length,
27 max_collection_width,
28 max_collection_depth,
29 max_stack_frames,
30 max_stack_trace_size,
31 max_object_depth,
32 max_fields_per_object,
33 ]
34 ):
35 return ""
36
37 output = "\nCAPTURE LIMITS:\n"
38 if max_hits is not None:
39 output += f"- Max Hits: {max_hits}\n"
40 if max_string_length is not None:
41 output += f"- Max String Length: {max_string_length}\n"
42 if max_collection_width is not None:
43 output += f"- Max Collection Width: {max_collection_width}\n"
44 if max_collection_depth is not None:
45 output += f"- Max Collection Depth: {max_collection_depth}\n"
46 if max_stack_frames is not None:
47 output += f"- Max Stack Frames: {max_stack_frames}\n"
48 if max_stack_trace_size is not None:
49 output += f"- Max Stack Trace Size: {max_stack_trace_size}\n"
50 if max_object_depth is not None:
51 output += f"- Max Object Depth: {max_object_depth}\n"
52 if max_fields_per_object is not None:
53 output += f"- Max Fields Per Object: {max_fields_per_object}\n"
54 return output
55
56
57def render_create_success_message(
58 response: Dict[str, Any],
59 normalized_type: str,
60 service: str,
61 environment: str,
62 location: Location,
63 ttl_hours: Optional[int],
64 capture_arguments: Optional[List[str]],
65 code_capture_locals: Optional[List[str]],
66 is_line_level: bool,
67 code_capture_return: Optional[bool],
68 code_capture_stack_trace: Optional[bool],
69 max_hits: Optional[int],
70 max_string_length: Optional[int],
71 max_collection_width: Optional[int],
72 max_collection_depth: Optional[int],
73 max_stack_frames: Optional[int],
74 max_stack_trace_size: Optional[int],
75 max_object_depth: Optional[int],
76 max_fields_per_object: Optional[int],
77 attribute_filters: Optional[List[Dict[str, str]]],
78) -> str:
79 """Render the success message for a created instrumentation configuration."""
80 location_hash = response.get("LocationHash", "N/A")
81 arn = response.get("ARN", "N/A")
82 created_at = format_timestamp(
83 response.get("CreatedAt"),
84 default=datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
85 )
86 actual_expires_at = format_timestamp(response.get("ExpiresAt"), default="")
87
88 success_message = f"""Successfully created {normalized_type} instrumentation
89
90INSTRUMENTATION CREATED:
91- Type: {normalized_type}
92- Service: {service}
93- Environment: {environment}
94- SignalType: {SNAPSHOT_SIGNAL_TYPE}
95- ARN: {arn}
96- CreatedAt: {created_at}
97"""
98
99 if actual_expires_at:
100 suffix = (
101 f' (requested {ttl_hours} hour{"s" if ttl_hours != 1 else ""})'
102 if ttl_hours is not None
103 else ""
104 )
105 success_message += f"- Expires: {actual_expires_at}{suffix}\n"
106 else:
107 success_message += "- Expires: Never (unless deleted)\n"
108
109 success_message += "\nLOCATION:\n"
110 success_message += render_location_block(location=location, location_hash=location_hash)
111 success_message += "\nCAPTURE CONFIGURATION:\n"
112
113 if not is_line_level:
114 if capture_arguments:
115 success_message += f'- Arguments: {", ".join(capture_arguments)}\n'
116 else:
117 success_message += "- Arguments: (none)\n"
118
119 if code_capture_locals:
120 success_message += f'- Local Variables: {", ".join(code_capture_locals)}\n'
121
122 if not is_line_level:
123 success_message += f'- Return Values: {"Enabled" if code_capture_return else "Disabled"}\n'
124 success_message += f'- Stack Traces: {"Enabled" if code_capture_stack_trace else "Disabled"}\n'
125 success_message += _render_create_capture_limits(
126 max_hits=max_hits,
127 max_string_length=max_string_length,
128 max_collection_width=max_collection_width,
129 max_collection_depth=max_collection_depth,
130 max_stack_frames=max_stack_frames,
131 max_stack_trace_size=max_stack_trace_size,
132 max_object_depth=max_object_depth,
133 max_fields_per_object=max_fields_per_object,
134 )
135
136 if attribute_filters:
137 success_message += (
138 f"\nATTRIBUTE FILTERS: {len(attribute_filters)} filter group(s) applied\n"
139 )
140
141 if normalized_type == "PROBE":
142 expected_ready = "~10-12 min"
143 else:
144 expected_ready = "~1-2 min"
145 success_message += (
146 f"\nNOTE: Allow {expected_ready} before this configuration reports READY. "
147 "Status checks immediately after creation may return no events yet — "
148 "wait and re-check rather than recreating.\n"
149 )
150
151 success_message += (
152 f"\nTIP: Use this LocationHash to delete: "
153 f'delete_instrumentation(location_hash="{location_hash}")'
154 )
155 return success_message
156
157
158def render_list_instrumentations_output(
159 data: Dict[str, Any],
160 normalized_type: str,
161 service: str,
162 environment: str,
163) -> str:
164 """Render the output for a list-instrumentations result."""
165 configs = data.get("LatestConfigurations", [])
166 next_token_response = data.get("NextToken")
167
168 if not configs:
169 return f"""No active {normalized_type} instrumentations found
170
171Service: {service}
172Environment: {environment}
173
174TIP: Use create_instrumentation to add instrumentations."""
175
176 output = f"""Active {normalized_type} Instrumentations ({len(configs)} found)
177
178Service: {service}
179Environment: {environment}
180Synced At: {format_timestamp(data.get('SyncedAt'))}
181
182"""
183
184 for index, config in enumerate(configs, 1):
185 cap = capture_from_response(config.get("CaptureConfiguration", {}))
186
187 output += f"""{'=' * 60}
188INSTRUMENTATION #{index}
189{'=' * 60}
190LOCATION:
191"""
192 output += render_location_block(
193 location=location_from_response(config.get("Location", {})),
194 location_hash=config.get("LocationHash"),
195 )
196
197 output += "\nCAPTURE SETTINGS:\n"
198 if isinstance(cap, CodeCapture):
199 output += f'- Return: {"Enabled" if cap.capture_return else "Disabled"}\n'
200 output += f'- Stack Traces: {"Enabled" if cap.capture_stack_trace else "Disabled"}\n'
201
202 if cap.capture_arguments is None:
203 output += "- Arguments: (not set)\n"
204 elif cap.capture_arguments:
205 output += f'- Arguments: {", ".join(cap.capture_arguments)}\n'
206 else:
207 output += "- Arguments: (empty list)\n"
208
209 if cap.capture_locals is None:
210 output += "- Locals: (not set)\n"
211 elif cap.capture_locals:
212 output += f'- Locals: {", ".join(cap.capture_locals)}\n'
213 else:
214 output += "- Locals: (empty list)\n"
215
216 limits = cap.limits
217 if not limits.is_empty():
218 limit_strs = []
219 if limits.max_hits is not None:
220 limit_strs.append(f"MaxHits={limits.max_hits}")
221 if limits.max_string_length is not None:
222 limit_strs.append(f"MaxStringLen={limits.max_string_length}")
223 if limits.max_collection_width is not None:
224 limit_strs.append(f"MaxCollWidth={limits.max_collection_width}")
225 if limit_strs:
226 output += f'- Limits: {", ".join(limit_strs)}\n'
227 else:
228 output += "- Capture payload could not be parsed.\n"
229
230 output += f"""
231TIMING:
232- Created: {format_timestamp(config.get('CreatedAt'))}
233- Expires: {format_timestamp(config.get('ExpiresAt'), default='Never')}
234
235Description: {config.get('Description', 'N/A')}
236ARN: {config.get('ARN', 'N/A')}
237
238"""
239
240 if next_token_response:
241 output += (
242 f'\nPAGINATION: More results available. Use next_token="{next_token_response}" '
243 "to retrieve next page."
244 )
245
246 return output
247
248
249def render_get_instrumentation_output(
250 config: Dict[str, Any],
251 service: str,
252 environment: str,
253) -> str:
254 """Render the output for a single get-instrumentation result."""
255 cap = capture_from_response(config.get("CaptureConfiguration", {}))
256
257 output = f"""INSTRUMENTATION CONFIGURATION
258
259TYPE: {config.get('InstrumentationType', 'N/A')}
260SERVICE: {service}
261ENVIRONMENT: {environment}
262SIGNAL TYPE: {config.get('SignalType', SNAPSHOT_SIGNAL_TYPE)}
263
264LOCATION:
265"""
266 output += render_location_block(
267 location=location_from_response(config.get("Location", {})),
268 location_hash=config.get("LocationHash"),
269 )
270
271 output += "\nCAPTURE CONFIGURATION:\n"
272 if isinstance(cap, CodeCapture):
273 output += f'- Return Values: {"Enabled" if cap.capture_return else "Disabled"}\n'
274 output += f'- Stack Traces: {"Enabled" if cap.capture_stack_trace else "Disabled"}\n'
275 if cap.capture_arguments is None:
276 output += "- Arguments: (not set)\n"
277 elif cap.capture_arguments:
278 output += f'- Arguments: {", ".join(cap.capture_arguments)}\n'
279 else:
280 output += "- Arguments: (empty list)\n"
281 if cap.capture_locals is None:
282 output += "- Local Variables: (not set)\n"
283 elif cap.capture_locals:
284 output += f'- Local Variables: {", ".join(cap.capture_locals)}\n'
285 else:
286 output += "- Local Variables: (empty list)\n"
287
288 limits = cap.limits
289 if not limits.is_empty():
290 output += "\nCAPTURE LIMITS:\n"
291 if limits.max_hits is not None:
292 output += f"- Max Hits: {limits.max_hits}\n"
293 if limits.max_string_length is not None:
294 output += f"- Max String Length: {limits.max_string_length}\n"
295 if limits.max_collection_width is not None:
296 output += f"- Max Collection Width: {limits.max_collection_width}\n"
297 if limits.max_collection_depth is not None:
298 output += f"- Max Collection Depth: {limits.max_collection_depth}\n"
299 if limits.max_stack_frames is not None:
300 output += f"- Max Stack Frames: {limits.max_stack_frames}\n"
301 if limits.max_stack_trace_size is not None:
302 output += f"- Max Stack Trace Size: {limits.max_stack_trace_size}\n"
303 if limits.max_object_depth is not None:
304 output += f"- Max Object Depth: {limits.max_object_depth}\n"
305 if limits.max_fields_per_object is not None:
306 output += f"- Max Fields Per Object: {limits.max_fields_per_object}\n"
307 else:
308 output += "- Capture payload could not be parsed.\n"
309
310 if config.get("AttributeFilters"):
311 output += f'\nATTRIBUTE FILTERS: {len(config["AttributeFilters"])} filter group(s)\n'
312 for index, filter_group in enumerate(config["AttributeFilters"], 1):
313 output += f" Group {index}: {filter_group}\n"
314
315 output += f"""
316METADATA:
317- Description: {config.get('Description', 'N/A')}
318- Created: {format_timestamp(config.get('CreatedAt'))}
319- Expires: {format_timestamp(config.get('ExpiresAt'), default='Never')}
320- ARN: {config.get('ARN', 'N/A')}
321"""
322 return output
323
324
325def _format_batch_delete_response(
326 mode: str,
327 data: Dict[str, Any],
328 instrumentation_type: str,
329 service: Optional[str] = None,
330 environment: Optional[str] = None,
331) -> str:
332 successful = data.get("SuccessfulDeletions", [])
333 errors = data.get("Errors", [])
334 deleted_count = data.get("DeletedCount", 0)
335
336 output = f"""BATCH DELETE COMPLETED
337
338Mode: {mode}
339InstrumentationType: {instrumentation_type}
340DeletedCount: {deleted_count}
341SuccessfulDeletions: {len(successful)}
342Errors: {len(errors)}
343"""
344 if service:
345 output += f"Service: {service}\n"
346 if environment:
347 output += f"Environment: {environment}\n"
348
349 if successful:
350 output += "\nSUCCESSFUL DELETIONS:\n"
351 for index, item in enumerate(successful, 1):
352 resource_arn = item.get("ResourceArn")
353 signal_type = item.get("SignalType")
354 location_hash = item.get("LocationHash")
355 if resource_arn:
356 output += f"- Item {index}: ResourceArn={resource_arn}\n"
357 else:
358 output += (
359 f'- Item {index}: SignalType={signal_type or "N/A"} | '
360 f'LocationHash={location_hash or "N/A"}\n'
361 )
362
363 if errors:
364 output += "\nDELETE ERRORS:\n"
365 for index, item in enumerate(errors, 1):
366 output += (
367 f'- Item {index}: ResourceArn={item.get("ResourceArn", "N/A")} | '
368 f'Code={item.get("Code", "N/A")} | '
369 f'Message={item.get("Message", "N/A")}\n'
370 )
371
372 return output