feat: add Workflow Insight instrumentation plugin - #632
Conversation
GitHub Actions
Failure
GitHub Actions
Error
GitHub Actions
Inactive
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
This comment has been minimized.
This comment has been minimized.
GitHub Actions
Inactive
GitHub Actions
Error
This comment has been minimized.
This comment has been minimized.
GitHub Actions
Error
GitHub Actions
Inactive
| # on-change mode exports an updated RUNNING record on each change so | ||
| # mid-invocation progress is observable, not only at start/end. | ||
| if self._emit_mode == EmitMode.ON_CHANGE: | ||
| self._emit( | ||
| arn, | ||
| state, | ||
| status="RUNNING", | ||
| end_time=None, | ||
| output_raw=None, | ||
| error=None, | ||
| ) |
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
| for exporter in self._exporters: | ||
| try: | ||
| shaped = truncate_record( | ||
| record, exporter.max_record_size_bytes, exporter.render | ||
| ) | ||
| exporter.export(shaped) |
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
| def _partition(self, record: dict[str, Any]) -> str: | ||
| if self.partitioning == "function-name": | ||
| return f"function={sanitize(record.get('functionName', ''))}/" | ||
| if self.partitioning == "date": | ||
| start = str(record.get("startTime", "")) | ||
| # YYYY-MM-DD... -> year=YYYY/month=MM/day=DD/ | ||
| if len(start) >= 10 and start[4] == "-" and start[7] == "-": | ||
| return f"year={start[0:4]}/month={start[5:7]}/day={start[8:10]}/" | ||
| return "" | ||
| return "" |
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
GitHub Actions
Inactive
GitHub Actions
Error
|
|
||
| [tool.hatch.build.targets.sdist.force-include] | ||
| "../../LICENSE" = "LICENSE" | ||
| "../../NOTICE" = "NOTICE" |
There was a problem hiding this comment.
add a plugin entry point here to allow this plugin to be auto loaded. See otel plugin for reference
| for exporter in self._exporters: | ||
| try: | ||
| shaped = truncate_record( | ||
| record, exporter.max_record_size_bytes, exporter.render | ||
| ) | ||
| exporter.export(shaped) |
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
| parsed_output: Any = None | ||
| if output_raw is not None and output_raw != "": | ||
| try: | ||
| parsed_output = json.loads(output_raw) | ||
| except (json.JSONDecodeError, TypeError): | ||
| parsed_output = output_raw | ||
| input_value = _apply_data_content( | ||
| state.cached_input, content.input if content else None | ||
| ) | ||
| output_value = _apply_data_content( | ||
| parsed_output, content.output if content else None | ||
| ) | ||
| if input_value is not None: | ||
| record["input"] = input_value | ||
| if output_value is not None: | ||
| record["output"] = output_value |
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
This comment has been minimized.
This comment has been minimized.
ParidelPooya
left a comment
ParidelPooya
left a comment
There was a problem hiding this comment.
Reviewed locally at 953e66f. Verified: 60 tests pass (~1s), hatch run types:check clean, hatch fmt --check clean for this package, wheel/sdist build with LICENSE+NOTICE correctly placed.
Solid, careful port. truncation.py and operations_index.py are near line-for-line faithful to the JS reference, the ARN parsing and FNV-1a math match, and sourcing operations from the SDK's authoritative operations snapshots (rather than accumulating on_operation_end events) is the right call — the e2e suspend/resume test proves it survives a cold resume. The comments are unusually good at explaining why.
Concerns are concentrated in one behavioral gap plus a few correctness edges and test holes. Three I'd want addressed before merge:
flush()is declared on the exporter protocol but never called anywhere.- There is no export scheduler / coalescing, contrary to the PR description, and exports run inline on the checkpoint thread in
on-changemode. sampling_rate=NaNsilently disables all instrumentation (JS fails open to 1.0 with a warning).
Details inline.
| record["error"] = {"name": error.type, "message": error.message} | ||
| record["operations"] = self._build_operations(operations) | ||
|
|
||
| for exporter in self._exporters: |
There was a problem hiding this comment.
flush() is never called.
InsightExporter declares flush(), both shipped exporters implement it, and the JS plugin calls flushAll(exporters) in wrapInvocation. Nothing in this package ever invokes it — grep -rn flush src/ returns only definitions.
Both first-party exporters are unbuffered, so this is latent today, but the protocol advertises a lifecycle contract the plugin doesn't honor: any customer exporter that batches will silently drop records. The Python SDK has no wrap_invocation hook, so the natural place is right after the terminal export here (or at the end of on_invocation_end).
Alternatively, if flushing is deliberately out of scope, make flush() optional in the protocol rather than required.
| error=None, | ||
| ) | ||
|
|
||
| def on_operation_change(self, info: OperationChangeInfo) -> None: |
There was a problem hiding this comment.
No export scheduler — and the PR description claims one.
The description says "emit model … with export coalescing — a newer record supersedes a pending one." There is no ExportScheduler equivalent here; _emit calls exporter.export() inline. In on-change mode that has two consequences:
OperationChangeInfois dispatched by the SDK withsync=True(PluginExecutor.execute_plugins,plugin.py:823), i.e. inline on the checkpoint-processing thread. A blockingput_objecthere stalls the SDK's checkpoint pump for every S3 round trip.- Every operation status change becomes one
PutObjectto the same key, with no coalescing. A 200-operation execution issues ~200 writes where JS would issue far fewer.
I checked for an out-of-order/overwrite hazard and there isn't one: on_invocation_start runs before the checkpoint thread starts, changes are serialized on the single checkpoint thread, and on_invocation_end fires from handle_durable_output after the ThreadPoolExecutor(max_workers=2) block has joined both workers. So the terminal record is always written last. The issues are latency and write amplification, not correctness.
Either implement coalescing or correct the description — right now the two disagree.
Separately: this hook calls _ensure_state and _adopt_operations unconditionally, whereas JS returns early unless emitMode === "on-change". In the default on-complete mode that work is pure waste, since on_invocation_end re-adopts a fresh snapshot anyway. Cheap (shallow copy, small next to the SDK's own eager _to_operation_info_map), but an early return is strictly better.
| return _fnv1a32(execution_arn) / 0xFFFFFFFF < rate | ||
|
|
||
|
|
||
| def _resolve_sampling_rate(rate: float | None) -> float: |
There was a problem hiding this comment.
sampling_rate=NaN silently disables all instrumentation.
NaN falls through every guard here (nan < 0 and nan > 1 are both False), and then _should_sample returns False for every ARN. Verified:
rate_in=nan resolved=nan sampled_in=False
rate_in='0.5' resolved=1.0 sampled_in=True
JS handles this explicitly — Number.isNaN(rate) → 1.0 plus a console.warn. Fail-open on misconfiguration is the safer default, and the current fail-closed behavior is completely invisible: no records, no warning.
Related: the string case silently coerces to 1.0 with no warning either. That sits oddly next to EmitMode / OperationDetail / S3Partitioning, which all raise ValueError on a bad dynamic value. Worth picking one philosophy for the config surface.
| return None | ||
|
|
||
|
|
||
| def _apply_data_content(value: Any, setting: Any) -> Any: |
There was a problem hiding this comment.
A null execution output is dropped.
The early return on value is None collapses JS's undefined-vs-null distinction. A handler returning None yields execution_result == "null" → json.loads → None → output omitted from the record, where JS emits output: null.
Cosmetic, but it is a wire divergence in exactly the schema the cross-SDK conformance suite compares.
|
|
||
| # -- sampling / state ----------------------------------------------------- | ||
|
|
||
| def _sampled_in(self, execution_arn: str) -> bool: |
There was a problem hiding this comment.
Sampling is recomputed on every hook — this re-hashes the ~120-char ARN each time. JS computes it once and caches it in ExecutionState. Harmless, but the cache is free and the state object already exists.
| record, exporter.max_record_size_bytes, exporter.render | ||
| ) | ||
| exporter.export(shaped) | ||
| except Exception as exc: # noqa: BLE001 - one exporter must not break others / the execution |
There was a problem hiding this comment.
A library should use logging rather than print to stderr; the SDK already has a module logger and customers can't filter or route this.
Also worth knowing: the SDK's _dispatch_plugin already catches and logs exceptions escaping any hook, so this broad except is belt-and-braces rather than the only line of defense.
| # YYYY-MM-DD... -> year=YYYY/month=MM/day=DD/ | ||
| if len(start) >= 10 and start[4] == "-" and start[7] == "-": | ||
| return f"year={start[0:4]}/month={start[5:7]}/day={start[8:10]}/" | ||
| return "" |
There was a problem hiding this comment.
When startTime is absent or malformed, date partitioning silently degrades to no partition and the object lands at the prefix root, mixing unpartitioned objects into an otherwise partitioned layout — awkward for Athena partition projection.
This is better than JS, which produces year=NaN/month=NaN/day=NaN/, but a year=unknown/ sentinel would keep the layout uniform.
| assert exporter.records == [] | ||
|
|
||
|
|
||
| def test_sampling_zero_emits_nothing(): |
There was a problem hiding this comment.
Sampling's deterministic hash is never executed. Coverage shows plugin.py:90 (_fnv1a32 body) unhit — this is the only sampling test and rate 0 short-circuits before the hash. That leaves the one piece that must agree with JS record-for-record completely untested.
Worth adding:
- a vector test (
_fnv1a32(known_arn) == <value computed by the JS implementation>) to lock in cross-SDK parity; - a fractional-rate test asserting the same ARN yields the same decision across repeated calls / a fresh plugin instance;
- the clamp and non-numeric paths in
_resolve_sampling_rate(plugin.py:109,111).
Three other gaps in this file, all cheap:
OperationOverride.excludeis untested (plugin.py:339unhit) — a documented config knob with zero coverage.- Per-operation
errorinclusion is untested (plugin.py:360unhit); only theinclude_errors=Falsepath is covered. - Exporter-failure isolation is untested (
plugin.py:436unhit). "One exporter must not break others / the execution" is a core safety claim and a throwing-exporter test is a few lines.
| assert "bulk-3" in names # newest retained | ||
|
|
||
|
|
||
| def test_truncation_noop_when_within_limit(): |
There was a problem hiding this comment.
Truncation phase 3 is untested. truncation.py:100–105 and the pop lines 68/71 are unhit, so droppedInput / droppedOutput — documented in both the README and the PR description — are never exercised. Phases 1 and 2 are covered; a case where dropping every operation still leaves the record over the limit would close this out.
| packages/aws-durable-execution-sdk-python-otel/tests | ||
|
|
||
| mypy --install-types --non-interactive \ | ||
| packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight \ |
There was a problem hiding this comment.
Two CI-wiring omissions while you're in here:
ci.yml's "Verify legal files in published distributions" step takes an explicit package list (core / otel / testing) and doesn't include insight, even though this package declaresforce-includefor LICENSE and NOTICE in both sdist and wheel targets. I rancheck_dist_legal_files.pyagainst the built package manually and it passes — just add it to the list so the contract stays enforced.- The repo convention of a per-package
[tool.hatch.envs.dev-*]env plus an entry in.github/scripts/ci-checks.sh(present for core, otel, testing, examples) wasn't followed, so the local dev script skips insight entirely.
ci.yml's fmt and build steps loop over packages/*/, so those are already covered.
Port of the JS SDK's workflowInsight() plugin as a new package, aws-durable-execution-sdk-python-insight: listens to the SDK's instrumentation hooks and emits one curated WorkflowInsight record (schemaVersion 1.0, JS-identical camelCase wire format) per execution through configurable exporters (LambdaLogExporter default with the operationsByName summary; S3Exporter with the per-occurrence operations array). Mirrors the JS emit model: on-complete/on-failure/on-change scheduling with coalescing, ARN-hash sampling, content configuration (input/output omission and transforms, include_errors, per-operation result opt-in), two-phase truncation, top-level vs full-tree operation detail, and unnamed-operation dropping. Uses the invocation-hook execution_input/execution_result fields introduced in #616.
Convert the single exporters.py module into an exporters/ package with one module per destination (lambda_log_exporter, s3_exporter) plus a private _common helper, mirroring the JS package's src/exporters/ layout so the set can grow to full parity (DynamoDB, Firehose, CloudWatch Logs, ...) without a single file accreting every backend's imports. Public import paths are unchanged: 'from ...insight import S3Exporter' and 'from ...insight.exporters import S3Exporter' both still resolve. Adds test_exporters.py covering both exporters (previously untested).
- Seed operation map from InvocationStart/End/OperationChange snapshots instead of reconstructing via per-operation hooks (cold-resume correctness) - on-change mode emits an updated RUNNING record on each change - Drop on_operation_end/_current_execution_arn heuristic; key strictly by execution_arn to prevent cross-execution contamination - Clear per-execution state after every invocation end (bounded, no leak on suspend/retry/sampled-out) - Default to LambdaLogExporter when exporters omitted or empty - Always adopt authoritative execution_start_time on resume - Correct hook enum imports (InvocationStatus/OperationType from plugin) - Register insight tests in root testpaths and mypy type-checks
- 1: wire aws-durable-execution-sdk-python-insight into both the build and publish matrices of pypi-publish.yml; the generic legal-file verifier runs through the build matrix unchanged (LICENSE+NOTICE confirmed in whl+sdist). - 3: in on-change mode, a PENDING/RETRY invocation end maps to RUNNING and now omits endTime/durationMs; only terminal SUCCEEDED/FAILED records carry an end time (plus output/error). - 4: fix the README usage example to import WorkflowInsightConfig and call workflow_insight(WorkflowInsightConfig(exporters=[...])); add a smoke test for the documented call shape. - 5: back EmitMode/OperationDetail with StrEnum (JS-style values); config fields use Literal input typing and __post_init__ normalizes accepted strings to enum members (invalid dynamic strings raise ValueError); export the enums. - 6: add a checked-in tests/e2e local-runner integration test that drives the real durable_execution/PluginExecutor lifecycle through a suspend/resume wait and asserts the terminal record includes the prior step and completed wait. Comment 2 (asynchronous export scheduling) is intentionally deferred; no async queue/worker/coalescing/drain was added.
Address the S3 partition-validation review comment on PR #632. Add a public S3Partitioning(StrEnum) (DATE=date, FUNCTION_NAME=function-name, NONE=none) in the s3_exporter module. The constructor is typed as an S3Partitioning | Literal[...] union (never bare str) and normalizes input with S3Partitioning(partitioning), so an invalid dynamic value (e.g. function_name) raises ValueError at construction instead of silently falling through to no partitioning. Key building now compares enum members. Re-export S3Partitioning from the exporters package and top-level package alongside S3Exporter. Existing API-compatible string inputs are preserved. Scheduler/flush/queueing/draining behavior is intentionally unchanged.
953e66f to
455a5bc
Compare
GitHub Actions
Failure
GitHub Actions
Inactive
github-project-automation
Bot
moved this from In review
to Pending merge
in aws-durable-execution
| shaped = truncate_record( | ||
| record, exporter.max_record_size_bytes, exporter.render | ||
| ) | ||
| exporter.export(shaped) |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_unr63pqhsmclfj3xlxboslj7hs
[P1] Move exporter I/O off the synchronous hook path. This runs from operation-change hooks before checkpoint waiters are released and from invocation-end hooks before Lambda returns. A slow or hung S3/custom exporter can therefore stall workflow progress until timeout; catching exceptions only helps after the call returns. Queue exports to a bounded worker and use a deadline-bounded invocation-end drain, dropping telemetry if necessary.
| if input_value is not None: | ||
| record["input"] = input_value | ||
| if output_value is not None: | ||
| record["output"] = output_value |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_6rjbhuxglgvdeo3oibktujajdz
[P2] Preserve JSON null instead of treating it as absence. A handler returning None produces "null", which parses to None and is skipped here; null input and opted-in null operation results are similarly lost. Use a private omission sentinel, insert fields unless that sentinel is returned, and update the operation-index/truncation presence checks. Add null input, output, transform, and result tests.
| output_raw=info.execution_result if is_terminal else None, | ||
| error=info.error if is_terminal else None, |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_bsp5xlklqalwvh3ifya2nh7ioe
[P2] Preserve terminal payloads before the core SDK elides them from the Lambda response. InvocationEndInfo.execution_result is "" when a large result was checkpointed out-of-band, so this plugin omits the output and never invokes its configured transform. Large checkpointed failures can similarly lack info.error. Extend the hook data with the original serialized result/error and test both large success and failure paths.
| file_name = ( | ||
| sanitize( | ||
| record.get("executionName") or record.get("executionArn") or "record" | ||
| ) | ||
| + ".json" |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_44xqupbb34o4bqveyo5x65sdh4
[P2] Key objects by full execution identity. With date or no partitioning, different functions or qualifiers sharing a bucket/prefix and execution name generate the same key; function-name partitioning still collides across qualifiers. Later exports silently overwrite unrelated records. Include all identity components or a stable hash of executionArn, while retaining the same key for re-emissions, and add collision tests.
| Matches the JS exporters' ``JSON.stringify`` output so the wire bytes are | ||
| identical across SDKs. | ||
| """ | ||
| return json.dumps(value, separators=(",", ":"), ensure_ascii=False) |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_zl6lshrvmes7zy4ztypdet7a43
[P2] Normalize non-finite floats before serialization. Python emits NaN and Infinity tokens by default, which are invalid JSON and differ from JavaScript's JSON.stringify behavior of emitting null. Durable results and transforms can contain these values, producing incompatible S3/log records. Use one shared JS-compatible normalizer for both export serialization and byte-size measurement, with tests for NaN and both infinities.
Codex AI reviewFound five correctness issues affecting execution liveness, payload fidelity, JSON compatibility, and S3 record identity. Static review only, as requested. Reviewed commit |
github-project-automation
Bot
moved this from Pending merge
to Done
in aws-durable-execution
3 participants
Summary
Adds a Workflow Insight instrumentation plugin as a new package,
packages/aws-durable-execution-sdk-python-insight/— a port of the JS SDK'sworkflowInsight()plugin (aws-durable-execution-sdk-js-insight), treated as thereference implementation throughout. Experimental, matching the JS plugin's status.
It listens to the SDK's instrumentation hooks and emits one curated
WorkflowInsightrecord (
schemaVersion: "1.0") per execution. The wire record keeps the JS camelCasefield names so records read identically across SDKs and land in the same stores/queries.
Behavior (mirrors the JS plugin)
LambdaLogExporterdefault (one JSON line to the function's log group,carrying the name-keyed
operationsByNamesummary) andS3Exporter(the losslessper-occurrence
operationsarray; upsert-by-execution-name;none/date/function-namepartitioning).boto3is an extra ([s3]) since Lambda provides it.on-complete/on-failure/on-change. Exporter calls runsynchronously on the SDK checkpoint path today — there is no async scheduling or
export coalescing yet (deferred/tracked in [Feature]: Add workflow insight plugin #687). Exports never propagate errors
into the execution.
include_errorsgating operation-level error detail only, per-operation result opt-in with optional
transform.
operations oldest-first, input/output last; per-exporter
max_record_size_bytesmeasured against the exact shape each exporter emits.
top-level(default; children withparentIdsuppressed) vsfull-tree; unnamed operations are dropped (JS parity).Depends on #616 (merged)
The plugin reads
InvocationInfo.execution_input/InvocationEndInfo.execution_resultintroduced by #616 — the dependency floor is set to
>=1.8.0accordingly (first releasethat will carry those hooks). Capability note kept in the module docstring: the operations
map is reconstructed by accumulating per-operation hooks into per-execution state (keyed
by execution ARN to isolate warm-container reuse), since Python hooks carry no
end-of-invocation operations snapshot.
Conformance validation (live, us-west-2)
Validated against the cross-SDK
insightconformance suite(aws/aws-durable-execution-conformance-tests#73, 18 requirements): 18/18 on the s3
sink and 18/18 on the cloudwatch sink. Two known cross-SDK divergences are documented
in that suite rather than patched over here: operation ids pass through the SDK's native
blake2b[:64]format (JS usesMD5[:16]; the suite asserts ids as opaque), and theper-operation
error.namesurfaces the customer error class while the record-level errorcarries the SDK wrapper name (the suite asserts non-empty).
The suite's Python example handlers land in the conformance repo as a follow-up to #73
once this package is available.
Testing
hatch run test:all packages/aws-durable-execution-sdk-python-insight/tests/)covering record shaping, operations indexing, truncation phases, sampling, emit modes,
and exporter rendering
hatch fmtclean; package registered in the rootknown-first-party