feat(insight): add the remaining Workflow Insight exporters - #720
feat(insight): add the remaining Workflow Insight exporters#720wangyb-A wants to merge 1 commit into
Conversation
|
/ai review |
- DynamoDBExporter: PutItem, pk=executionArn, optional sk=emittedAt - AuroraExporter: RDS Data API upsert, postgresql or mysql dialect - CloudWatchLogsExporter: PutLogEvents into a per-day stream of any log group - OTelExporter: OTLP/HTTP log record, http/json only - FirehoseExporter: PutRecord, one JSON line per record - EventBridgeExporter: PutEvents, DetailType = record status - RedshiftExporter: Redshift Data API MERGE by execution_arn - OpenSearchExporter: Index API PUT, SigV4 or basic auth - SQSExporter: SendMessage, FIFO group and dedup ids - HttpExporter: POST or PUT JSON with a timeout - FileExporter: ndjson append or one json file per execution - OperationsFormat / apply_operations_format shared by the flexible exporters - README: exporter table and one setup block per exporter
ec3c719 to
32c68f1
Compare
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| url: str, | ||
| headers: dict[str, str], | ||
| body: bytes, | ||
| timeout: float | None = None, |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_vqw7dmpi77j3kupfbkxyltsain
[P1] Bound HTTP requests made from plugin hooks
OTelExporter and OpenSearchExporter call this helper without a timeout, permitting an indefinitely blocking connect or read. Exporters run synchronously during operation-change and invocation-end hooks, so a stalled endpoint can block checkpoint completion or the handler response until Lambda times out. Give these exporters a finite configurable timeout (or make this default finite) and test a stalled endpoint.
| kwargs["DbUser"] = self.db_user | ||
| if self.secret_arn: | ||
| kwargs["SecretArn"] = self.secret_arn | ||
| self._client.execute_statement(**kwargs) |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_f37mrckuri2f3eds4liaxgyw3y
[P1] Wait for the Redshift statement to complete
Redshift Data API ExecuteStatement only submits asynchronous work. In on-change mode, multiple MERGEs can therefore finish out of order, allowing an older RUNNING record to overwrite a terminal record; later SQL failures are also never surfaced. Poll the returned statement ID to a terminal state with a finite timeout, raising for FAILED/ABORTED, or otherwise serialize updates and reject stale emittedAt values. Add ordering and failure tests.
| with path.open("a", encoding="utf-8") as handle: | ||
| handle.write(compact_dumps(formatted) + "\n") |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_5r7yljuanxrd44ar6oy2to5gpf
[P1] Protect shared EFS appends from concurrent writers
The default mode sends every invocation for a day to one file using an unlocked append. EFS is NFS, where append positioning is not atomic across clients, so concurrently scaled Lambda environments can overwrite or interleave records and corrupt the NDJSON file. Use per-execution files or cross-process locking with an atomic record write, and add concurrent-writer coverage.
| file_name = ( | ||
| sanitize(record.get("executionName") or record["executionArn"]) | ||
| + ".json" |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_kslgwubds4itjgdcu3ikbfbsge
[P2] Key JSON files by the full execution identity
executionName is not unique across functions or execution-name reuse, and sanitize() is many-to-one. Distinct executions sharing a directory can consequently overwrite the same file. Derive the filename from executionArn, preferably using a stable digest plus a readable prefix, and test same-name and sanitization collisions.
| def export(self, record: dict[str, Any]) -> None: | ||
| body = compact_dumps(self._build_payload(record)).encode("utf-8") | ||
| headers = {"Content-Type": "application/json", **self.headers} | ||
| status, reason, _ = http_send("POST", self.endpoint, headers, body) |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_k23hi45mamy6lcwsmntfxhc4rt
[P2] Surface OTLP partial-success rejections
OTLP/HTTP may return status 200 with partialSuccess.rejectedLogRecords indicating that the log was rejected. Discarding the response body and accepting every 2xx makes this data loss silent. Parse successful JSON responses and raise or report when records were rejected, including errorMessage; add a partial-success response test.
| (`OperationsFormat`). `max_record_size_bytes` overrides the size limit on every | ||
| exporter; `None` disables truncation. |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_iog3uxk7xlfqe7iiawopnh7q6k
[P2] Make the documented None override achievable
Every new default-limited exporter interprets max_record_size_bytes=None as “use the default,” so callers cannot use None to disable truncation as documented. Large records may unexpectedly lose operations, input, or output. Distinguish an omitted argument with a sentinel and preserve explicit None, or document and test another disable value consistently.
| AuroraEngine, | ||
| AuroraExporter, | ||
| CloudWatchLogsExporter, |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_pwssfvb537ct7wulvz6g7segez
[P3] Add the required e2e coverage for the public exporter surface
These additions expose new public APIs and cross-component plugin behavior, but the PR adds only isolated unit tests. Repository policy requires e2e tests for public API changes. Add coverage under tests/e2e/ that drives a new exporter through the real plugin lifecycle, including replay or on-change emission and render/truncation interaction.
Codex AI reviewFound seven actionable issues. Highest risk: unbounded HTTP calls can stall durable execution, and asynchronous Redshift MERGEs can silently fail or overwrite newer state. File delivery also risks corruption/data loss. Required e2e coverage is absent. Reviewed commit |
1 participant
Summary
Adds the eleven Workflow Insight exporters that were missing from the insight package: DynamoDB, Aurora, CloudWatch Logs, OTel, Firehose, EventBridge, Redshift, OpenSearch, SQS, HTTP, and File. Each is one module under
exporters/, re-exported from the package root, with the same config fields, defaults, size limits, and failure behavior as the other SDKs' insight exporters. AddsOperationsFormat/apply_operations_format(array,by-name,both) used by the flexible-destination exporters.boto3/botocore stay runtime-provided; no new dependency or extra.
plugin.py,types.py, and the core SDK are untouched.Testing
packages/aws-durable-execution-sdk-python-insight/tests/, using injected recording clients that assert exact request kwargs and body bytes.http.server; OpenSearch asserts the SigV4Authorizationheader and the basic-auth header. File usestmp_pathin both modes.hatch fmt --check,hatch run types:check, and the insight test suite (128 tests) pass;hatch buildsucceeds.Known follow-ups
OTelExporterandOpenSearchExportersend their request with no timeout, same as the reference exporters; a hung endpoint blocks the invocation-end hook until the function times out.SQSExporterreplaces a FIFO group/deduplication id longer than SQS's 128-character cap with its SHA-256 digest; the other SDKs' exporters still send the raw id.OTelExportersupportshttp/jsononly;http/protobufraises at construction.