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_snapshot_tools.py
scripts/cloudwatch/di_snapshot_tools.py
Python·265 lines·12 KB
11 render_search_snapshots_for_status_event_output,
12)
13from di_validation import is_valid_location_hash
14
15
16def _build_base_filters(location_hash: str, service_name: str, environment: str) -> str:
17 """Build the resource-matching Logs Insights filter shared by both snapshot tools.
18
19 All three values are escaped for double-quoted string-literal context so a
20 caller-supplied quote cannot break out of the literal and inject query
21 syntax (which, for ``service_name``/``environment``, could otherwise widen
22 the match across services). ``location_hash`` is additionally validated as
23 16-char hex by the callers before reaching here.
24
25 Tolerant resource matching:
26 - For Java snapshots ``resource.attributes.*`` is populated; require an exact match.
27 - For Python snapshots the SDK currently emits an empty resource block, so we accept
28 records where the field is absent (``not ispresent(...)``). location_hash by itself
29 uniquely identifies (service, environment, location), so this fallback does not
30 widen the match across services.
31 - Assumption: location_hash collisions across services are negligible. If a future
32 SDK bug ever produces records with both a colliding hash and missing resource
33 attributes, this filter could return cross-service results.
34 """
35 location_hash_esc = _escape_logs_insights_string(location_hash)
36 service_name_esc = _escape_logs_insights_string(service_name)
37 environment_esc = _escape_logs_insights_string(environment)
38 return (
39 f'attributes.aws.di.location_hash = "{location_hash_esc}"'
40 f' and (resource.attributes.service.name = "{service_name_esc}"'
41 f" or not ispresent(resource.attributes.service.name))"
42 f' and (resource.attributes.deployment.environment = "{environment_esc}"'
43 f' or resource.attributes.deployment.environment.name = "{environment_esc}"'
44 f" or not ispresent(resource.attributes.deployment.environment))"
45 )
46
47
48def search_snapshots_for_status_event(
49 service: str,
50 environment: str,
51 location_hash: str,
52 status_timestamp: str,
53 limit: int = 10,
54 max_timeout: int = 30,
55 custom_filters: Optional[List[str]] = None,
56 start_time: Optional[str] = None,
57 end_time: Optional[str] = None,
58) -> str:
59 """Search CloudWatch Logs snapshots near a known instrumentation status timestamp.
60
61 This helper builds a Logs Insights query around the supplied status event time,
62 searches for records containing the `location_hash`, and returns a JSON string
63 with query metadata, parsed snapshot summaries, and raw results.
64
65 Args:
66 service: Service label echoed back in the response for operator context.
67 environment: Environment label echoed back in the response for operator context.
68 location_hash: 16-character lowercase hex instrumentation location hash used to filter snapshot records.
69 status_timestamp: ISO 8601 status-event timestamp used as the default search anchor.
70 limit: Maximum number of matching log records to return.
71 max_timeout: Maximum polling time in seconds for the Logs Insights query.
72 custom_filters: Optional raw Logs Insights filter fragments appended with `and`.
73 Accepts a JSON array of strings, e.g. ["@message like /ORD-123/"]. A single
74 bare string is also accepted and treated as a one-element list.
75 start_time: Optional ISO 8601 lower bound for the search window. When provided
76 with `end_time`, overrides the default `status_timestamp`-anchored window so
77 the caller can sweep an arbitrary span (e.g. the full breakpoint lifetime) in
78 one query. Both must be supplied together.
79 end_time: Optional ISO 8601 upper bound for the search window. See `start_time`.
80
81 Notes:
82 - The default search window is `status_timestamp - 5 seconds` through
83 `status_timestamp + 1 minute`. Pass `start_time`/`end_time` to widen it.
84 - The response is JSON text, not a human-formatted prose summary.
85 - Custom filters should already be valid Logs Insights expressions.
86
87 Returns:
88 A JSON string containing query status, query metadata, parsed snapshot
89 summaries, duration hints, and raw CloudWatch query results.
90 """
91 if not is_valid_location_hash(location_hash):
92 return "ERROR: location_hash must be a 16-character hex string"
93
94 try:
95 limit = int(limit)
96 except (TypeError, ValueError):
97 return "ERROR: limit must be an integer"
98
99 # A single filter passed as a bare string is the natural shape; the op documents a
100 # list, so coerce string -> [string] rather than mis-iterating the string per character
101 # (which would validate the first quote char and emit a misleading 'unbalanced quotes').
102 if isinstance(custom_filters, str):
103 custom_filters = [custom_filters]
104
105 try:
106 event_time = datetime.fromisoformat(status_timestamp.replace("Z", "+00:00"))
107 if event_time.tzinfo is None:
108 event_time = event_time.replace(tzinfo=timezone.utc)
109 except ValueError:
110 return 'ERROR: status_timestamp must be ISO 8601 format like "2025-02-03T18:42:00Z"'
111
112 # Window resolution: explicit start_time/end_time override the anchored default. Both
113 # must be supplied together so the window is never half-specified.
114 if (start_time is None) != (end_time is None):
115 return (
116 "ERROR: start_time and end_time must be provided together "
117 "(both ISO 8601), or both omitted to use the status_timestamp-anchored window"
118 )
119 if start_time is not None and end_time is not None:
120 try:
121 window_start = datetime.fromisoformat(start_time.replace("Z", "+00:00"))
122 if window_start.tzinfo is None:
123 window_start = window_start.replace(tzinfo=timezone.utc)
124 window_end = datetime.fromisoformat(end_time.replace("Z", "+00:00"))
125 if window_end.tzinfo is None:
126 window_end = window_end.replace(tzinfo=timezone.utc)
127 except ValueError:
128 return (
129 'ERROR: start_time and end_time must be ISO 8601 format like "2025-02-03T18:42:00Z"'
130 )
131 if window_end <= window_start:
132 return "ERROR: end_time must be after start_time"
133 start_time_dt = window_start
134 end_time_dt = window_end
135 else:
136 start_time_dt = event_time - timedelta(seconds=5)
137 end_time_dt = event_time + timedelta(minutes=1)
138
139 start_time_utc = start_time_dt.astimezone(timezone.utc)
140 end_time_utc = end_time_dt.astimezone(timezone.utc)
141 start_epoch = int(start_time_utc.timestamp())
142 end_epoch = int(end_time_utc.timestamp())
143
144 base_filters = _build_base_filters(location_hash, service, environment)
145 if custom_filters:
146 for custom_filter in custom_filters:
147 custom_filter = custom_filter.strip()
148 if not custom_filter:
149 continue
150 # custom_filters are documented as raw Logs Insights fragments the
151 # caller appends on purpose, so they are passed through rather than
152 # escaped. Reject only the realistic corruption vector: an unbalanced
153 # double-quote that would leak into (or truncate) the rest of the query.
154 if (custom_filter.count('"') - custom_filter.count('\\"')) % 2 != 0:
155 return f"ERROR: custom_filters has unbalanced quotes: {custom_filter!r}"
156 base_filters += f" and {custom_filter}"
157
158 query_string = (
159 "fields @timestamp, @message\n"
160 f"| filter {base_filters}\n"
161 "| sort @timestamp asc\n"
162 f"| limit {limit}"
163 )
164 query_result = _execute_cloudwatch_query(
165 query_string=query_string,
166 start_epoch=start_epoch,
167 end_epoch=end_epoch,
168 log_group_name=resolve_snapshot_log_group(service),
169 max_timeout=max_timeout,
170 )
171
172 return render_search_snapshots_for_status_event_output(
173 service_name=service,
174 environment=environment,
175 location_hash=location_hash,
176 custom_filters=custom_filters,
177 start_time_utc=start_time_utc.isoformat().replace("+00:00", "Z"),
178 end_time_utc=end_time_utc.isoformat().replace("+00:00", "Z"),
179 start_epoch=start_epoch,
180 end_epoch=end_epoch,
181 query_string=query_string,
182 query_result=query_result,
183 )
184
185
186def get_sample_snapshot_for_breakpoint(
187 service: str,
188 environment: str,
189 location_hash: str,
190 status_timestamp: str,
191 max_timeout: int = 30,
192 include_raw: bool = False,
193) -> str:
194 """Fetch one nearby snapshot to inspect the structure of captured data.
195
196 This is a discovery helper intended to show the shape of one snapshot record
197 before building narrower CloudWatch queries or deciding which capture fields
198 matter.
199
200 Args:
201 service: Service label echoed back in the response for operator context.
202 environment: Environment label echoed back in the response for operator context.
203 location_hash: 16-character lowercase hex instrumentation location hash used to filter snapshot records.
204 status_timestamp: ISO 8601 status-event timestamp used as the search anchor.
205 max_timeout: Maximum polling time in seconds for the Logs Insights query.
206 include_raw: When True, always include the full raw snapshot in the response.
207 When False (default), raw snapshots larger than 10 KB are replaced with a
208 compact parsed summary produced by _parse_snapshot_fields(). Small snapshots
209 are returned in full regardless of this flag.
210
211 Notes:
212 - The search window is currently `status_timestamp - 30 seconds` through
213 `status_timestamp + 90 seconds` (wider than search to accommodate
214 CloudWatch Logs ingestion delay).
215 - This helper requests only one result, sorted by most recent timestamp first.
216 - The response is JSON text, not a human-formatted prose summary.
217
218 Returns:
219 A JSON string containing query metadata plus one parsed sample snapshot,
220 or a structured timeout/error response when the query fails.
221 """
222 if not is_valid_location_hash(location_hash):
223 return "ERROR: location_hash must be a 16-character hex string"
224
225 try:
226 event_time = datetime.fromisoformat(status_timestamp.replace("Z", "+00:00"))
227 if event_time.tzinfo is None:
228 event_time = event_time.replace(tzinfo=timezone.utc)
229 except ValueError:
230 return 'ERROR: status_timestamp must be ISO 8601 format like "2025-02-03T18:42:00Z"'
231
232 start_time = event_time - timedelta(seconds=30)
233 end_time = event_time + timedelta(seconds=90)
234
235 start_time_utc = start_time.astimezone(timezone.utc)
236 end_time_utc = end_time.astimezone(timezone.utc)
237 start_epoch = int(start_time_utc.timestamp())
238 end_epoch = int(end_time_utc.timestamp())
239
240 query_string = (
241 "fields @timestamp, @message\n"
242 f"| filter {_build_base_filters(location_hash, service, environment)}\n"
243 "| sort @timestamp desc\n"
244 "| limit 1"
245 )
246
247 query_result = _execute_cloudwatch_query(
248 query_string=query_string,
249 start_epoch=start_epoch,
250 end_epoch=end_epoch,
251 log_group_name=resolve_snapshot_log_group(service),
252 max_timeout=max_timeout,
253 )
254
255 return render_get_sample_snapshot_for_breakpoint_output(
256 service_name=service,
257 environment=environment,
258 location_hash=location_hash,
259 start_time_utc=start_time_utc.isoformat().replace("+00:00", "Z"),
260 end_time_utc=end_time_utc.isoformat().replace("+00:00", "Z"),
261 max_timeout=max_timeout,
262 query_string=query_string,
263 query_result=query_result,
264 include_raw=include_raw,
265 )