feat: add Workflow Insight plugin - #661
Conversation
5d83fc9 to
9cb8201
Compare
GitHub Actions
Inactive
GitHub Actions
Failure
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.
|
Use /ai review command to trigger AI review workflows for draft PRs |
d90f91f to
bdbd88f
Compare
|
/ai review |
This comment has been minimized.
This comment has been minimized.
bdbd88f to
24e28b8
Compare
|
/ai review |
| if (state == null || !state.sampledIn) { | ||
| return; | ||
| } | ||
| emit(buildRecord( |
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
| // across multiple emissions (ON_CHANGE), so a transform that mutates its argument in place must not | ||
| // corrupt that snapshot or any later emission's view of it. Omit on throw so a failing redactor never | ||
| // leaks the raw value. | ||
| return transform.apply(Json.deepCopyContent(value)); |
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
| if (!anyResultDropped[0] && droppedOperations[0] == 0 && !droppedInput[0] && !droppedOutput[0]) { | ||
| return record; | ||
| } | ||
| return candidate(record, ops, kept, droppedOperations[0], droppedInput[0], droppedOutput[0]); |
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
| static Object applyDataContent(Object value, boolean include, Function<Object, Object> transform) { | ||
| if (!include || value == null) { | ||
| return null; |
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
| if (value == null | ||
| || value instanceof String | ||
| || value instanceof Boolean | ||
| || value instanceof Character | ||
| || value instanceof Enum<?>) { | ||
| return value; | ||
| } | ||
| if (value instanceof Map<?, ?> map) { | ||
| Map<Object, Object> copy = new LinkedHashMap<>(); | ||
| for (Map.Entry<?, ?> e : map.entrySet()) { | ||
| copy.put(e.getKey(), deepCopyContent(e.getValue())); |
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
| public S3Exporter build() { | ||
| return new S3Exporter(this); |
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
| return false; | ||
| } | ||
| long unsigned = fnv1a32(executionArn) & 0xffffffffL; | ||
| return (double) unsigned / 0xffffffffL < rate; |
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.
24e28b8 to
3b6eb77
Compare
|
Tracking the following up items in #679 |
GitHub Actions
Inactive
GitHub Actions
Failure
GitHub Actions
Inactive
| if (state == null || !state.sampledIn) { | ||
| return; | ||
| } | ||
| emit(buildRecord( |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_xryomwjxrzqt3z55wozggpsrsl
[P2] Keep ON_CHANGE I/O off checkpoint callbacks
This synchronously exports and flushes while CheckpointManager still holds its polling lock, before pollers finish and checkpoint processing is cleared. Slow S3/Logs calls therefore delay durable progress. The completed operation can also wake the handler concurrently, allowing a terminal export to race this RUNNING export; the RUNNING S3 write may finish last and overwrite the terminal object. Queue immutable/coalesced snapshots, serialize exports outside this callback, and drain them from the awaited onInvocationEnd. Add a delayed-exporter race test.
| if (!anyResultDropped[0] && droppedOperations[0] == 0 && !droppedInput[0] && !droppedOutput[0]) { | ||
| return record; | ||
| } | ||
| return candidate(record, ops, kept, droppedOperations[0], droppedInput[0], droppedOutput[0]); |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_sulajh3ii6jyv6tb56c7lizmvp
[P2] Enforce the size limit after every truncation phase
The final candidate is returned without remeasurement. An oversized top-level execution error or another retained scalar can therefore remain above the exporter's limit after operations and payloads are removed, causing the exporter to reject the record and emit() to discard it. Remeasure the final shape and bound remaining optional fields such as error text; if no legal record fits, skip it with an explicit diagnostic. Test oversized errors in both render shapes.
| // across multiple emissions (ON_CHANGE), so a transform that mutates its argument in place must not | ||
| // corrupt that snapshot or any later emission's view of it. Omit on throw so a failing redactor never | ||
| // leaks the raw value. | ||
| return transform.apply(Json.deepCopyContent(value)); |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_j5amytymbyibhzrtph7noqfti6
[P2] Preserve runtime types passed to content transforms
deepCopyContent converts DTOs into maps and Java-time values into strings before invoking the transform; input has already undergone the same normalization when cached. A transform that casts the execution input/output to its Java type consequently throws, and the catch silently omits that field. Create type-preserving defensive snapshots for transform invocation, then normalize only the transform result. Add DTO, Java-time, and custom-SerDes transform tests.
| if (!include || value == null) { | ||
| return null; |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_oirdg3rnpdnwvpldafyiqrnncz
[P2] Preserve included JSON null values
An enabled value that is actually null is represented identically to excluded or unavailable content, and the wire map subsequently omits it. Thus a successful handler returning null has no output field despite output inclusion being enabled. Track field presence separately from value and emit JSON null when included; omit only excluded or unavailable fields. Add null input/output wire-shape tests.
| // content transform can mutate it. This raw, detached snapshot is the single source of truth for input on | ||
| // every emission (start / change / end); each build hands transforms a separate defensive copy so a | ||
| // mutating transform cannot corrupt it. | ||
| state.cachedInput = Json.deepCopyContent(info.executionInput()); |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_65fn2ndq4okrooanfbimyf6a6m
[P2] Do not snapshot input when input collection is disabled
This always runs the input through the plugin's Jackson mapper, even for content.input(false). With a custom-SerDes value that Jackson cannot copy, the start hook throws before the ON_CHANGE start record is emitted, although the input was explicitly excluded; large excluded inputs also incur unnecessary serialization cost. Check includeInput before snapshotting and cache explicit absence without inspecting the value. Add an ON_CHANGE test using an excluded, non-Jackson input.
| public S3Exporter build() { | ||
| return new S3Exporter(this); |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_5cs3isxmbcgkzlluwxepe4qxv4
[P3] Reject invalid exporter configuration at build time
The builder accepts a null/blank bucket and non-positive record limits. Missing destinations fail only during export, where the plugin catches the exception and silently produces no record; non-positive limits disable truncation. Validate required destinations and positive limits before creating the client, apply equivalent validation to CloudWatch Logs and Lambda log exporters, and add invalid-configuration tests.
| || value instanceof Enum<?>) { | ||
| return value; | ||
| } | ||
| if (value instanceof Map<?, ?> map) { | ||
| Map<Object, Object> copy = new LinkedHashMap<>(); | ||
| for (Map.Entry<?, ?> e : map.entrySet()) { | ||
| copy.put(e.getKey(), deepCopyContent(e.getValue())); |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_rzwdoyquezpr2qgby63imtmxcn
[P3] Detach enums and map keys between exporters
The copy still shares enum instances and every map key. Enums may expose mutable state or a mutable @JsonValue, while keys are reachable through toWireMap(); mutation by one exporter can therefore alter later exporters' records. Convert enums to detached JSON scalars and copy keys using the mapper's key-serialization semantics. Add mutable-enum and mutable-key isolation tests.
| return false; | ||
| } | ||
| long unsigned = fnv1a32(executionArn) & 0xffffffffL; | ||
| return (double) unsigned / 0xffffffffL < rate; |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_47kz274xvx4sow4mmbm7gzor7a
[P3] Normalize sampling into the documented half-open interval
Dividing by 0xffffffff maps the maximum unsigned hash to exactly 1.0, contradicting the documented [0,1) range and shifting deterministic boundary decisions. Divide by 2^32 (0x1_0000_0000L) and add shared fixed-vector tests, coordinating the boundary change with the other SDK implementations.
Codex AI reviewFound 8 actionable issues, primarily around ON_CHANGE lifecycle races and silent record loss during truncation/content handling. Additional issues affect excluded inputs, exporter validation/isolation, and sampling boundaries. Reviewed commit |
2 participants
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
Issue Link, if available
N/A. Workflow Insight conformance requirements are being reviewed in aws/aws-durable-execution-conformance-tests#73.
Description
Adds the preview Workflow Insight plugin as a new
insight-pluginMaven module.The plugin emits schema-versioned execution records from the existing Java plugin hooks and matches the JavaScript Workflow Insight contract. It includes:
operationsByNamerecord shapesThe implementation uses the invocation operation snapshots, execution payloads, errors, attempts, and serialized operation results already exposed by the Java plugin API. No core hook changes are required.
Demo/Screenshots
No UI change.
insight-plugin/README.mdcontains a usage example. The draft conformance PR linked above contains Java handlers for all 18 Workflow Insight requirements.Checklist
Testing
Unit Tests
Yes. The module adds 35 tests covering record construction, sampling, emission modes, replay/suspension state, operation filtering, result transforms, aggregation, truncation, exporter requests, failure isolation, and flush behavior.
mvn -q -pl insight-plugin -am test mvn -q -pl insight-plugin -am clean verifyBoth pass.
The full repository reactor also passes:
Integration Tests
Yes.
WorkflowInsightPluginTestexercises the plugin throughLocalDurableTestRunner.The draft Workflow Insight conformance suite was also run against deployed Java 21 Lambda functions in
us-west-2:operationsByNameleg: 18 passed, 0 failedSink-specific assertions were capability-gated and exercised by the complementary leg.
Examples
The module README includes configuration and exporter usage. Customer-facing conformance handlers are in aws/aws-durable-execution-conformance-tests#73 rather than this repository, following the repository's conformance-test ownership rule.