Setting the file. One moment.
Emit · Dd Orchestrator · datadog-labs/agent-skills · Skills Docs
Back to the top of the page def _ndjson_path
— line 221
This file
Number 42.1
Position 1 of 4
Type Python
Size 19 KB
Lines 447 scripts/ emit.py
Python · 447 lines · 19 KB
16 lands events in the same internal ``service:setup-cli`` pipeline the onboarding-growth
17 team already consumes. It deliberately does NOT use the customer's DD_API_KEY: that
18 would write onboarding telemetry into the customer's own org, not Datadog's.
19
20 CLI (the runbook calls this at each dispatch boundary — see SKILL.md):
21 python3 emit.py skill_run --action started --session-id <uuid> [--field k=v ...]
22 python3 emit.py skill_step --action started --session-id <uuid> --field skill_id=<id> ...
23 python3 emit.py skill_run --action finished --session-id <uuid> --field result=success ...
24
25 The session id is minted and printed by resolve.py ("SESSION ID: <uuid>"); every
26 event in one DAG run must reuse it.
27 """
28 from __future__ import annotations
29
30 import argparse
31 import datetime
32 import json
33 import os
34 import sys
35 import tempfile
36 import time
37 import urllib.error
38 import urllib.request
39 from urllib.parse import quote
40
41 SCHEMA_VERSION = 1
42
43 # --- envelope the runtime owns (a caller can NEVER override these) -----------
44 RUNTIME_SOURCE = "setup-cli" # reuse the existing pipeline -> service:setup-cli
45 FLOW = "agent"
46 PLATFORM = "cli"
47
48 EVENT_TYPES = { "skill_run" , "skill_step" }
49 ACTIONS = {
50 "skill_run" : { "started" , "plan_resolved" , "finished" },
51 "skill_step" : { "planned" , "started" , "finished" , "skipped" },
52 }
53
54 # Reconciled with catalog.json `kinds`: a `lifecycle` node CAN enter a plan, so it is
55 # a valid skill_kind (the proposal's enum omitted it). `internal` is filtered by
56 # resolve.py and never reaches a step, so it is intentionally absent here.
57 SKILL_KINDS = { "foundation" , "platform-install" , "cloud-connect" ,
58 "product-enable" , "verify-troubleshoot" , "lifecycle" }
59
60 # --- client-side allow-list (mirrors the server-side allow-list intent) ------
61 # Any key not here is stripped before send (privacy + bounded cardinality). Grouped
62 # only for readability; the union is what gates.
63 _FACETS = {
64 "event_type" , "event_action" , "result" , "invocation_mode" , "intent_mode" , "entry_skill_id" ,
65 "agent_name" , "headless" , "target_platform" , "target_cloud" , "step_kind" ,
66 "skill_id" , "skill_kind" , "product" , "source_repo" , "source_mode" , "error_code" , "result_reported" ,
67 }
68 _MEASURES = {
69 "duration_ms" , "plan_position" , "dependency_count" ,
70 "planned_skill_count" , "dead_end_count" , "choice_count" ,
71 "step_success_count" , "step_failed_count" , "step_skipped_count" ,
72 }
73 _ATTRIBUTES = {
74 "schema_version" , "session_id" , "source" , "flow" , "platform" ,
75 "agent_version" , "recommended_products" , "emitted_at" , "depends_on" , "event_seq" ,
76 "skill_invoked" , "instrumentation_invoked" , "result_reconciled" ,
77 "org_id" , # authenticated org's public_id, captured once at auth; attribute (not a facet)
78 }
79 ALLOWED_KEYS = _FACETS | _MEASURES | _ATTRIBUTES
80
81 _BOOL_KEYS = { "headless" , "skill_invoked" , "instrumentation_invoked" , "result_reconciled" }
82
83 # setup-cli's publishable logs-intake client tokens (`pub…` = write-only, meant to
84 # ship in clients). Kept in sync with setup-cli/src/constants/credentials.ts. An env
85 # override always wins.
86 _CLIENT_TOKENS = {
87 "datadoghq.com" : "pub5b34c05a3abc178e4d2abddf11bb4f31" ,
88 "us3.datadoghq.com" : "pub82dbd99bf6299230cee3685535d4660d" ,
89 "us5.datadoghq.com" : "pubf9a4a5b93cc00e7f4d5628f05ba1e7f0" ,
90 "datadoghq.eu" : "pub471980dfdfb12f9cf2ffdf83551c9cb7" ,
91 "ap1.datadoghq.com" : "pubbbc10e19ab9fa2f7eb54250610e9bbad" ,
92 "ap2.datadoghq.com" : "pube391858a603cb55a0e72a8cb9f9fb627" ,
93 "uk1.datadoghq.com" : "pub90fe05f76003a728ed554865d39a88d1" ,
94 }
95 _TAGS = "service:setup-cli"
96
97 # Per-process breaker. Non-critical emits are 1-strike (a down intake can't add N*3s to a
98 # run). CRITICAL emits (resolve.py's plan-shape core) BYPASS an open breaker so a single
99 # transient blip cannot drop run:started + plan_resolved + planned×N; a critical failure
100 # trips the breaker only after 3 consecutive strikes (bounds the worst-case hang).
101 _BREAKER = { "open" : False , "critical_fails" : 0 }
102
103
104 def _trip_breaker (critical):
105 """Record one terminal failure (transport OR terminal HTTP) against the breaker.
106
107 Non-critical: a single strike opens it. Critical (resolve-core): opens only after
108 3 consecutive failures, so one blip cannot drop the plan shape — but once that
109 bound is reached the breaker applies to critical events too (see ``emit``)."""
110 if critical:
111 _BREAKER [ "critical_fails" ] += 1
112 if _BREAKER [ "critical_fails" ] >= 3 : # bound the worst-case hang
113 _BREAKER [ "open" ] = True
114 else :
115 _BREAKER [ "open" ] = True # non-critical: 1-strike
116
117
118 def _truthy (value):
119 return str (value).strip().lower() in ( "1" , "true" , "yes" , "on" )
120
121
122 def _is_disabled ():
123 if _truthy(os.environ.get( "DD_ORCH_TELEMETRY_DISABLED" , "" )):
124 return True
125 # ponytail: never emit real telemetry from a test run — the pub token writes to the
126 # internal setup-cli index. A genuine CLI / orchestrator run never imports these.
127 if "unittest" in sys.modules or "pytest" in sys.modules:
128 return True
129 return False
130
131
132 def _debug_log_path ():
133 return os.environ.get(
134 "DD_ORCH_TELEMETRY_LOG" ,
135 os.path.join(tempfile.gettempdir(), "dd-orchestrator-telemetry.log" ),
136 )
137
138
139 def _debug (msg):
140 """Append one line to the local debug log. Never raises."""
141 try :
142 with open (_debug_log_path(), "a" ) as handle:
143 handle.write(msg.rstrip( " \n " ) + " \n " )
144 except Exception :
145 pass
146
147
148 def _coerce (key, value):
149 if key in _BOOL_KEYS :
150 return value if isinstance (value, bool ) else _truthy(value)
151 if key in _MEASURES :
152 try :
153 return int (value)
154 except ( TypeError , ValueError ):
155 return None # drop an unparseable measure rather than send a bad type
156 return value
157
158
159 # --- run-scoped envelope: persisted ONCE by resolve.py, re-applied to every event ------
160 # Fixes F4 — SKILL.md-runbook emits (separate emit.py processes) dropped the envelope on
161 # started/finished/skill_run:finished. Write-once; read-only here; best-effort (never raises).
162 _ENVELOPE_KEYS = ( "entry_skill_id" , "invocation_mode" , "intent_mode" , "agent_name" ,
163 "target_platform" , "target_cloud" , "org_id" )
164
165
166 def _session_state_path (session_id):
167 safe = "" .join(c for c in str (session_id) if c.isalnum() or c in "-_" )
168 return os.path.join(tempfile.gettempdir(), f "dd-orch- { safe } .json" )
169
170
171 def _load_state (session_id):
172 try :
173 with open (_session_state_path(session_id)) as fh:
174 state = json.load(fh)
175 return state if isinstance (state, dict ) else {}
176 except Exception :
177 return {}
178
179
180 def write_session_state (session_id, envelope, shape = None ):
181 """Persist the run's envelope (re-attached to every later emit) and, when given, the plan
182 shape (`dead_end_count`, ...), so the terminal skill_run:finished can reconcile `result`
183 against the real counts. Preserves the seq counter across re-writes. Never raises."""
184 try :
185 if not session_id:
186 return
187 state = _load_state(session_id)
188 state[ "envelope" ] = {k: (envelope or {})[k] for k in _ENVELOPE_KEYS
189 if (envelope or {}).get(k) not in ( None , "" )}
190 if shape is not None :
191 state[ "shape" ] = {k: int (v) for k, v in shape.items()}
192 state.setdefault( "seq" , 0 )
193 with open (_session_state_path(session_id), "w" ) as fh:
194 json.dump(state, fh)
195 except Exception :
196 pass
197
198
199 def _read_session_envelope (session_id):
200 env = _load_state(session_id).get( "envelope" , {})
201 return {k: v for k, v in env.items() if k in _ENVELOPE_KEYS and v not in ( None , "" )}
202
203
204 def _next_seq (session_id):
205 """Return the next per-session monotonic ordinal (1-based). A missing ordinal on the
206 export means an event was dropped, not that the step never ran (gap detection). Never
207 raises; returns None if state is unwritable (then event_seq is simply omitted).
208 ponytail: no lock — dispatch is sequential (one emit.py process at a time)."""
209 try :
210 state = _load_state(session_id)
211 seq = int (state.get( "seq" , 0 )) + 1
212 state[ "seq" ] = seq
213 state.setdefault( "envelope" , {})
214 with open (_session_state_path(session_id), "w" ) as fh:
215 json.dump(state, fh)
216 return seq
217 except Exception :
218 return None
219
220
221 def _ndjson_path ():
222 return os.environ.get(
223 "DD_ORCH_TELEMETRY_NDJSON" ,
224 os.path.join(tempfile.gettempdir(), "dd-orchestrator-telemetry.ndjson" ),
225 )
226
227
228 def _record_ndjson (event, status):
229 """Append one durable NDJSON line per ATTEMPTED event so drops are auditable/replayable
230 offline. Never raises."""
231 try :
232 with open (_ndjson_path(), "a" ) as fh:
233 fh.write(json.dumps({ "status" : status, "event" : event}) + " \n " )
234 except Exception :
235 pass
236
237
238 def _now_iso ():
239 return datetime.datetime.now(datetime.timezone.utc).isoformat( timespec = "milliseconds" )
240
241
242 def _read_run_shape (session_id):
243 shape = _load_state(session_id).get( "shape" , {})
244 return shape if isinstance (shape, dict ) else {}
245
246
247 # Verdicts we re-check against the counts. An agent-reported `blocked` / `cancelled` is trusted
248 # and never overridden; but a reported success/partial/failed that contradicts a pure opt-out
249 # (only skips) or all-dead-end count-shape is corrected TO `cancelled` / `blocked` — so a user
250 # who declined is not lumped into the failure funnel (PR #194 review). Both are SKILL.md verdicts.
251 _RECONCILABLE_RESULTS = ( "success" , "partial_success" , "failed" )
252
253
254 def _reconcile_run_result (out, session_id):
255 """Guard the terminal run `result` against the real counts, so an agent verdict that
256 contradicts them (e.g. `partial_success` with 5 success / 0 failed / 0 skipped / 0 dead
257 ends) cannot skew the funnel. On disagreement, keep the agent's value as `result_reported`
258 and set `result` to the count-consistent verdict (`result_reconciled: true`)."""
259 result = out.get( "result" )
260 if result not in _RECONCILABLE_RESULTS :
261 return
262 if not any (k in out for k in ( "step_success_count" , "step_failed_count" , "step_skipped_count" )):
263 return # no counts on this event -> nothing to reconcile against
264 s = int (out.get( "step_success_count" , 0 ) or 0 )
265 f = int (out.get( "step_failed_count" , 0 ) or 0 )
266 k = int (out.get( "step_skipped_count" , 0 ) or 0 )
267 d = int (_read_run_shape(session_id).get( "dead_end_count" , 0 ) or 0 )
268 if s >= 1 and f == 0 and k == 0 :
269 derived = "success" # everything dispatched succeeded; dead-ends are coverage
270 # gaps (a product with no skill yet), not partial failures
271 elif s >= 1 :
272 derived = "partial_success" # something succeeded, but a step failed or was skipped
273 elif f == 0 and d == 0 and k >= 1 :
274 derived = "cancelled" # nothing ran, nothing errored: user declined every step
275 elif f == 0 and d >= 1 :
276 derived = "blocked" # nothing errored, but dead-ends stopped all progress
277 else :
278 derived = "failed" # something actually failed
279 if derived != result:
280 out[ "result_reported" ] = result # preserve the agent's verdict
281 out[ "result" ] = derived # count-consistent truth for the funnel
282 out[ "result_reconciled" ] = True
283
284
285 def build_event (event_type, event_action, session_id, fields = None ):
286 """Return the validated event dict.
287
288 Caller `fields` are allow-list-stripped and type-coerced first; then the
289 runtime-owned envelope is applied ON TOP, so a caller can never spoof
290 session_id / source / event identity.
291 """
292 out = {}
293 for key, value in (fields or {}).items():
294 if key not in ALLOWED_KEYS or value is None or value == "" :
295 continue
296 coerced = _coerce(key, value)
297 if coerced is not None :
298 out[key] = coerced
299 # Re-attach the run-scoped envelope (persisted by resolve.py) so runbook emits that did
300 # not re-pass it still carry agent/platform/entry (F4). Runtime-owned -> un-spoofable.
301 out.update(_read_session_envelope(session_id))
302 if event_type == "skill_run" and event_action == "finished" :
303 _reconcile_run_result(out, session_id) # `result` must match the step counts
304 out.update({
305 "event_type" : event_type,
306 "event_action" : event_action,
307 "schema_version" : SCHEMA_VERSION ,
308 "session_id" : session_id,
309 "source" : RUNTIME_SOURCE ,
310 "flow" : FLOW ,
311 "platform" : PLATFORM ,
312 "emitted_at" : _now_iso(), # F1: client ms timestamp -> reliable order + duration
313 })
314 return out
315
316
317 def _log_body (event):
318 """The flat logs-intake log line (mirrors setup-cli's sendToIntake body)."""
319 return json.dumps({
320 "service" : RUNTIME_SOURCE ,
321 "message" : f " { event[ 'event_type' ] } : { event[ 'event_action' ] } " ,
322 ** event,
323 })
324
325
326 def _resolve_token (site):
327 override = os.environ.get( "DD_ONBOARDING_CLIENT_TOKEN" )
328 if override:
329 return override
330 return _CLIENT_TOKENS .get(site)
331
332
333 def _intake_url (site, token):
334 return ( f "https://http-intake.logs. { site } /v1/input/ { token } "
335 f "?ddsource=dd-orchestrator&ddtags= { quote( _TAGS ) } " )
336
337
338 def _default_post (url, body):
339 """POST the JSON body with a 3s timeout. Returns the HTTP status (int) or None on a
340 transport error / timeout. Never raises."""
341 request = urllib.request.Request(
342 url, data = body.encode( "utf-8" ), method = "POST" ,
343 headers = { "Content-Type" : "application/json" },
344 )
345 try :
346 with urllib.request.urlopen(request, timeout = 3 ) as response:
347 return response.getcode()
348 except urllib.error.HTTPError as err:
349 return err.code
350 except Exception :
351 return None
352
353
354 def emit (event_type, event_action, session_id, fields = None ,
355 * , site = None , disabled = None , _post = None , critical = False ):
356 """Best-effort emit of one event. Returns True iff intake accepted it. Never raises,
357 never blocks beyond the bounded timeout / retries.
358
359 ``critical`` marks resolve.py's plan-shape core (run:started, plan_resolved, planned×N):
360 it BYPASSES an open breaker — but only until 3 consecutive critical failures trip it —
361 so one transient blip cannot drop the whole core, yet a persistently down intake still
362 stops after the bound instead of adding a timeout per plan node.
363 """
364 try :
365 if disabled is None :
366 disabled = _is_disabled()
367 if disabled:
368 return False
369 if event_type not in EVENT_TYPES or event_action not in ACTIONS .get(event_type, ()):
370 _debug( f "drop: unknown event { event_type } : { event_action } " )
371 return False
372 if not session_id:
373 _debug( f "drop: missing session_id for { event_type } : { event_action } " )
374 return False
375
376 site = site or os.environ.get( "DD_SITE" ) or "datadoghq.com"
377 token = _resolve_token(site)
378 if not token:
379 _debug( f "drop: no client token for site { site !r} " )
380 return False
381
382 event = build_event(event_type, event_action, session_id, fields)
383 seq = _next_seq(session_id)
384 if seq is not None :
385 event[ "event_seq" ] = seq # gap-detection ordinal
386 body = _log_body(event)
387
388 # Softened breaker: a critical (resolve-core) emit bypasses an open breaker so a
389 # single earlier blip cannot drop the plan shape — but only until the critical
390 # failure bound is reached; past that the breaker applies to critical events too,
391 # so a down intake cannot keep adding a timeout per plan node.
392 if _BREAKER [ "open" ] and ( not critical or _BREAKER [ "critical_fails" ] >= 3 ):
393 _record_ndjson(event, "suppressed" )
394 return False
395
396 url = _intake_url(site, token)
397 post = _post or _default_post
398
399 status = None
400 for attempt in range ( 3 ):
401 status = post(url, body)
402 if status is not None and 500 <= status <= 599 : # retry transient 5xx only
403 _debug( f "retry { event_type } : { event_action } attempt { attempt + 1 } status { status } " )
404 time.sleep( 0.2 )
405 continue
406 break # 2xx, 429, other 4xx, or None: stop
407
408 _record_ndjson(event, status if status is not None else "transport_failed" )
409
410 if status is None : # transport failure
411 _trip_breaker(critical)
412 _debug( f "transport-failed { event_type } : { event_action } " )
413 return False
414 if status >= 400 : # terminal HTTP failure (4xx/5xx/429)
415 _trip_breaker(critical) # count it toward the same bound
416 _debug( f "dropped { event_type } : { event_action } status { status } " )
417 return False
418 _BREAKER [ "critical_fails" ] = 0 # a success clears the streak
419 return True
420 except Exception as err: # best-effort: NEVER propagate
421 _debug( f "emit-exception { event_type } : { event_action } : { err !r} " )
422 return False
423
424
425 def _cli (argv = None ):
426 parser = argparse.ArgumentParser(
427 description = "best-effort dd-orchestrator telemetry emitter (v1)" )
428 parser.add_argument( "event_type" , choices = sorted ( EVENT_TYPES ))
429 parser.add_argument( "--action" , required = True , help = "event lifecycle action" )
430 parser.add_argument( "--session-id" , required = True ,
431 help = "the run's shared session id (from resolve.py 'SESSION ID:')" )
432 parser.add_argument( "--field" , action = "append" , default = [], metavar = "KEY=VALUE" ,
433 help = "event field (repeatable); non-allow-listed keys are dropped" )
434 args = parser.parse_args(argv)
435 fields = {}
436 for item in args.field:
437 if "=" not in item:
438 _debug( f "cli: ignoring malformed --field { item !r} " )
439 continue
440 key, value = item.split( "=" , 1 )
441 fields[key.strip()] = value
442 emit(args.event_type, args.action, args.session_id, fields)
443 return 0 # ALWAYS succeed — telemetry must never fail the caller
444
445
446 if __name__ == "__main__" :
447 sys.exit(_cli())