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_validation.py
Python·294 lines·11 KB
"Python"
,
"java"
:
"Java"
,
"javascript"
:
"Javascript"
}
11
12
13def canonical_language(language: Optional[str]) -> Optional[str]:
14 """Return the API's canonical ``ProgrammingLanguage`` casing, or None if unknown.
15
16 The API's ``ProgrammingLanguage`` enum is case-sensitive (``Java``, ``Python``,
17 ``Javascript``). Callers accept any casing (e.g. ``"javascript"``,
18 ``"JavaScript"``) and map to the canonical form before sending to the API,
19 so a validated language is not rejected by the backend on a casing mismatch.
20 """
21 return _CANONICAL_LANGUAGES.get((language or "").strip().lower())
22
23
24def normalize_instrumentation_type(
25 instrumentation_type: str,
26) -> Tuple[str, Optional[str]]:
27 """Normalize the type to upper-case; return ``(normalized, error)``.
28
29 The normalized value is always a ``str`` (the upper-cased received value
30 even on the error path) so callers get a non-optional type once they
31 return early on ``error``. Callers must check ``error`` before using
32 ``normalized``.
33 """
34 normalized = (instrumentation_type or "").strip().upper()
35 allowed = {"BREAKPOINT", "PROBE"}
36 if normalized not in allowed:
37 return normalized, (
38 "ERROR: instrumentation_type must be one of BREAKPOINT, PROBE "
39 f"(received: {instrumentation_type})"
40 )
41 return normalized, None
42
43
44def validate_capture_names(field_name: str, names: Optional[List[str]]) -> Optional[str]:
45 """Validate a capture-name list (``capture_arguments`` / ``capture_locals``).
46
47 Returns an error string if invalid, else ``None``. An omitted list
48 (``None``) is valid and means "capture nothing for that field". A provided
49 list must be non-empty and may not contain the ``*`` wildcard — both the
50 empty list and ``*`` are rejected so the ambiguous "capture all" shapes
51 never reach the API.
52 """
53 if names is None:
54 return None
55 if not names:
56 return (
57 f"ERROR: {field_name} must contain at least one name if provided. "
58 "Omit it to capture none."
59 )
60 if "*" in names:
61 return (
62 f'ERROR: {field_name} does not support the wildcard "*". '
63 "List explicit names, or omit it to capture none."
64 )
65 return None
66
67
68def validate_probe_constraints(
69 normalized_type: str,
70 language: Optional[str],
71 line_number: Optional[int],
72) -> Optional[str]:
73 """Validate PROBE-only constraints; return error text if invalid, else None.
74
75 PROBE differs from BREAKPOINT in two ways the SDKs enforce:
76
77 * PROBE is not supported for JavaScript.
78 * PROBE is method/function-level only — the SDKs ignore line_number, so a
79 PROBE with line_number set would silently not behave as written.
80 """
81 if normalized_type != "PROBE":
82 return None
83 lang = (language or "").strip().lower()
84 if lang == "javascript":
85 return (
86 "ERROR: PROBE is not supported for JavaScript. "
87 "Use instrumentation_type=BREAKPOINT for JavaScript targets."
88 )
89 if line_number is not None:
90 return (
91 "ERROR: PROBE does not support line_number (the SDKs ignore it). "
92 "Omit line_number for PROBE — it is method/function-level only."
93 )
94 return None
95
96
97def is_valid_location_hash(location_hash: Optional[str]) -> bool:
98 """Return True for a 16-character lowercase hexadecimal location hash.
99
100 Location hashes are 16 lowercase hex characters by API design. Validating
101 against this shape (rather than only checking length) lets snapshot/status
102 tools reject malformed input before it is interpolated into a CloudWatch
103 Logs Insights query — hex can never contain the double-quote that would
104 otherwise break out of a query string literal.
105 """
106 return bool(location_hash and _LOCATION_HASH_RE.fullmatch(location_hash))
107
108
109def validate_snapshot_signal(signal_type: str) -> Optional[str]:
110 """Return an error message unless ``signal_type`` is SNAPSHOT, else None."""
111 normalized = (signal_type or "").strip().upper()
112 if normalized != SNAPSHOT_SIGNAL_TYPE:
113 return f"ERROR: signal_type must be SNAPSHOT for this API (received: {signal_type})"
114 return None
115
116
117def _format_code_location_troubleshooting(
118 language: Optional[str],
119 file_path: Optional[str],
120 code_unit: Optional[str],
121 class_name: Optional[str],
122 method_name: Optional[str],
123 line_number: Optional[int],
124) -> str:
125 """Build troubleshooting guidance for code-location create failures.
126
127 ``language``/``file_path`` are ``Optional`` because callers pass raw,
128 unvalidated inputs (which may be ``None``); the body renders them
129 verbatim and guards with ``(language or '')`` where it matters.
130 """
131 lang = (language or "").strip().lower()
132
133 lines = [
134 "CODE LOCATION TROUBLESHOOTING:",
135 "- file_path: source file path for the target code.",
136 "- code_unit: Python runtime module path OR Java package name.",
137 "- class_name: use for class methods (Java: simple class name only).",
138 "- method_name: function/method name.",
139 "- line_number: set only for line-level breakpoints (1-based).",
140 ]
141
142 if line_number is None:
143 lines.append("- Breakpoint level: FUNCTION/METHOD-level (line_number omitted).")
144 else:
145 lines.append(f"- Breakpoint level: LINE-LEVEL (L{line_number}).")
146 if lang in ("python", "java"):
147 lines.append(
148 " * NOTE: target an executable statement. In Python/Java a non-executable "
149 "line (blank, comment, decorator, signature) is ignored and the breakpoint "
150 "never fires."
151 )
152 elif lang == "javascript":
153 lines.append(
154 " * NOTE: in JavaScript a breakpoint on a non-executable line slides to the "
155 "next parseable line and fires there — verify it lands where you intend."
156 )
157
158 if lang == "python":
159 lines.extend(
160 [
161 "- Python rules:",
162 " * Set code_unit to the dotted runtime import path for the module that defines the target code.",
163 " * Example: services.billing, not billing.py or /app/services/billing.py.",
164 ' * Use code_unit="__main__" only when the target file is executed',
165 " directly as the process entry script.",
166 " * If call site uses direct import aliasing, target importing module and alias name.",
167 " * If you cannot determine the runtime module path confidently, inspect first instead of guessing.",
168 ]
169 )
170 elif lang == "java":
171 lines.extend(
172 [
173 "- Java rules:",
174 " * Set code_unit to the Java package name (e.g., com.amazon.sampleapp).",
175 " * class_name must be simple name (e.g., OrderContext), not fully qualified.",
176 ]
177 )
178 elif lang == "javascript":
179 lines.extend(
180 [
181 "- JavaScript rules:",
182 " * JavaScript binds by file_path + line_number; line_number is required (>= 1).",
183 " * code_unit, class_name, and method_name are not used for JavaScript.",
184 " * Point line_number at the executable statement you want to observe.",
185 ]
186 )
187
188 lines.extend(
189 [
190 "LOCATION INPUTS RECEIVED:",
191 f"- language={language}",
192 f"- file_path={file_path}",
193 f"- code_unit={code_unit}",
194 f"- class_name={class_name}",
195 f"- method_name={method_name}",
196 f"- line_number={line_number}",
197 ]
198 )
199
200 return "\n".join(lines)
201
202
203def _validate_location_inputs(
204 language: str,
205 file_path: str,
206 code_unit: Optional[str],
207 class_name: Optional[str],
208 method_name: Optional[str],
209 line_number: Optional[int],
210) -> Optional[str]:
211 """Validate location fields and return actionable error text if invalid.
212
213 Enforces the per-language fields the SDK needs to bind the instrumentation;
214 without them the SDK silently drops the configuration and nothing fires:
215
216 * Java — requires code_unit, class_name, and method_name.
217 * Python — requires code_unit and method_name (class_name optional).
218 * JavaScript — requires line_number (>= 1); binds by file + line.
219 """
220 lang = (language or "").strip().lower()
221
222 errors: List[str] = []
223 suggestions: List[str] = []
224
225 if not file_path or not str(file_path).strip():
226 errors.append("file_path is required and must be non-empty.")
227
228 if line_number is not None and line_number < 1:
229 errors.append(f"line_number must be >= 1 (received: {line_number}).")
230
231 if lang not in {"python", "java", "javascript"}:
232 errors.append(f"language must be Python, Java, or JavaScript (received: {language}).")
233
234 if lang == "java":
235 if not code_unit:
236 errors.append("Java requires code_unit (the package name, e.g. com.amazon.sampleapp).")
237 if not class_name:
238 errors.append("Java requires class_name (the simple class name, e.g. OrderContext).")
239 if not method_name:
240 errors.append("Java requires method_name.")
241 if class_name and "." in class_name:
242 errors.append(
243 'For Java, class_name must be simple (e.g., "OrderContext"), '
244 'not fully qualified (e.g., "com.example.OrderContext").'
245 )
246 if not code_unit:
247 parts = class_name.split(".")
248 if len(parts) > 1:
249 suggestions.append(
250 f'Use code_unit="{".".join(parts[:-1])}" and class_name="{parts[-1]}".'
251 )
252 if code_unit and "/" in code_unit:
253 suggestions.append(
254 "Java code_unit should be a package name with dots, not a path with slashes."
255 )
256
257 elif lang == "python":
258 if not code_unit:
259 errors.append(
260 "Python requires code_unit (the dotted runtime module path, e.g. services.billing)."
261 )
262 if not method_name:
263 errors.append("Python requires method_name.")
264 if code_unit and code_unit.endswith(".py"):
265 suggestions.append(
266 "Python code_unit should be a module path (e.g., services.billing), not a .py filename."
267 )
268
269 elif lang == "javascript":
270 if line_number is None:
271 errors.append("JavaScript requires line_number (>= 1); it binds by file and line.")
272
273 if not errors:
274 return None
275
276 message = "Invalid breakpoint location inputs:\n"
277 for idx, err in enumerate(errors, 1):
278 message += f"{idx}. {err}\n"
279
280 if suggestions:
281 message += "\nSuggestions:\n"
282 for idx, item in enumerate(suggestions, 1):
283 message += f"{idx}. {item}\n"
284
285 message += "\n" + _format_code_location_troubleshooting(
286 language=language,
287 file_path=file_path,
288 code_unit=code_unit,
289 class_name=class_name,
290 method_name=method_name,
291 line_number=line_number,
292 )
293
294 return message