feat(insight): add the remaining Workflow Insight exporters by wangyb-A · Pull Request #720 · aws/aws-durable-execution-sdk-python

wangyb-A · 2026-09-11T05:00:48Z

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. Adds OperationsFormat / 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. S3Exporter gets a one-line type annotation so it, like the new exporters, is assignable to InsightExporter under mypy.

Small, deliberate differences from the other SDKs' exporters, each rejecting a configuration or input that could never succeed there either: SQS FIFO ids over 128 characters are hashed; Redshift requires exactly one of workgroup_name / cluster_identifier; OpenSearch basic auth requires both credentials at construction; the OTLP timeUnixNano keeps the record's microseconds; non-2xx HTTP bodies are read only up to 64 KiB for diagnostics; HTTP redirects are not followed (a 3xx is a failed export), so a POST is never re-sent as a body-less GET and configured credential headers never reach another origin.

Testing

One unit test file per exporter under packages/aws-durable-execution-sdk-python-insight/tests/, using injected recording clients that assert exact request kwargs and body bytes.
HTTP, OTel, and OpenSearch run against a local http.server; OpenSearch asserts the SigV4 Authorization header and the basic-auth header. File uses tmp_path in both modes.
tests/e2e/exporters_lifecycle_int_test.py drives the real plugin through the local durable runner with three exporters attached (FileExporter ndjson/by-name, FileExporter json, SQSExporter with a small size limit) in on-change mode, checking emission count, rendering, overwrite semantics, and per-exporter truncation.
hatch fmt --check, hatch run types:check, and hatch run test:all over the core, insight, otel, and testing packages pass.

Known follow-ups

OTelExporter and OpenSearchExporter send their request with no timeout, same as the reference exporters; a hung endpoint blocks the invocation-end hook until the function times out.
OTelExporter accepts any 2xx and does not inspect an OTLP partialSuccess body, same as the reference exporter.
RedshiftExporter submits the statement and does not await it, same as the reference exporter; documented in the README with a recommendation to use on-complete.
OTelExporter supports http/json only; http/protobuf raises at construction.
Repo-wide: every package's test package is named tests; under --import-mode=importlib the first import claims the name, so a conftest.py inside any non-core tests/ breaks the core SDK's tests.* imports.

wangyb-A · 2026-09-11T16:59:23Z

/ai review

wangyb-A · 2026-09-11T22:34:28Z

Sign in to view

+    url: str,
+    headers: dict[str, str],
+    body: bytes,
+    timeout: float | None = None,


Repeat of a finding from the previous run; disposition unchanged (see the reply on the earlier thread). Intentional: no timeout, same as the other SDKs' exporters; listed under Known follow-ups.

wangyb-A · 2026-09-11T22:34:29Z

Sign in to view

+            kwargs["DbUser"] = self.db_user
+        if self.secret_arn:
+            kwargs["SecretArn"] = self.secret_arn
+        self._client.execute_statement(**kwargs)


Repeat of a finding from the previous run; disposition unchanged (see the reply on the earlier thread). Intentional: submit without awaiting, same as the other SDKs' Redshift exporter; README recommends on-complete.

wangyb-A · 2026-09-11T22:34:30Z

Sign in to view

+            with path.open("a", encoding="utf-8") as handle:
+                handle.write(compact_dumps(formatted) + "\n")


Repeat of a finding from the previous run; disposition unchanged (see the reply on the earlier thread). Per-execution files exist as mode="json"; README documents the shared-directory caveat.

wangyb-A · 2026-09-11T22:34:35Z

Sign in to view

+            return (
+                int(response.status),
+                str(response.reason or ""),
+                response.read().decode("utf-8", errors="replace"),


Done in 9b517c0: http_send no longer reads a success body and reads at most 64 KiB of a non-2xx body for the error message. Tests: test_http_exporter.py (200 KiB error body, 4 MiB success body) and test_opensearch_exporter.py (detail still truncated to 500 chars).

wangyb-A · 2026-09-11T22:34:36Z

Sign in to view

+        if not workgroup_name and not cluster_identifier:
+            msg = (
+                "RedshiftExporter: provide either workgroup_name or cluster_identifier."
+            )
+            raise ValueError(msg)


Done in 9b517c0 for the target: exactly one of workgroup_name / cluster_identifier is now required at construction, with a negative test. Left the db_user/secret_arn combination unvalidated: I could not confirm which pairings the service rejects, and a wrong rule here would block valid configs.

wangyb-A · 2026-09-11T22:34:31Z

Sign in to view

+            file_name = (
+                sanitize(record.get("executionName") or record["executionArn"])
+                + ".json"


Repeat of a finding from the previous run; disposition unchanged (see the reply on the earlier thread). Same file-name rule as S3Exporter on main and the other SDKs' file exporter.

wangyb-A · 2026-09-11T22:34:33Z

Sign in to view

+        status, reason, _ = http_send("POST", self.endpoint, headers, body)
+        if not 200 <= status < 300:
+            msg = f"OTelExporter: OTLP endpoint returned {status} {reason}"
+            raise RuntimeError(msg)


Repeat of a finding from the previous run; disposition unchanged (see the reply on the earlier thread). OTLP partialSuccess handling is listed under Known follow-ups.

wangyb-A · 2026-09-11T22:34:34Z

Sign in to view

+(`OperationsFormat`). `max_record_size_bytes` raises or lowers an exporter's
+size limit; omitting it keeps the default. `HttpExporter` and `FileExporter`
+have no default and do not truncate unless a limit is set.


Repeat of a finding from the previous run; disposition unchanged (see the reply on the earlier thread). Omitting the argument keeps the default, matching the existing S3Exporter/LambdaLogExporter; README reworded in the previous push.

wangyb-A · 2026-09-11T22:34:37Z

Sign in to view

+            (self.directory / file_name).write_text(
+                json.dumps(formatted, indent=2, ensure_ascii=False), encoding="utf-8"
+            )


Kept as is: the limiter bounds the record body measured as compact JSON, not the destination's formatting, which is how the other SDKs' file exporter behaves too. FileExporter has no default limit, so this only affects callers who opt in.

wangyb-A · 2026-09-11T22:34:39Z

Sign in to view

+        self.auth = OpenSearchAuth(auth)
+        self.username = username
+        self.password = password


Done in 9b517c0: auth="basic" raises ValueError at construction when username or password is None; an empty password is still accepted. Tests added for both.

wangyb-A · 2026-09-11T22:34:40Z

Sign in to view

+def _to_nano(iso: str) -> str:
+    millis = (parse_iso_datetime(iso) - _EPOCH) // datetime.timedelta(milliseconds=1)
+    return str(millis * 1_000_000)


Done in 9b517c0 with the suggested integer conversion: timeUnixNano now carries the record's microseconds. Test added for a fractional-millisecond emittedAt.

wangyb-A · 2026-09-11T22:34:41Z

Sign in to view

+    assert len(ndjson_files) == 1
+    lines = ndjson_files[0].read_text(encoding="utf-8").splitlines()
+    records = [json.loads(line) for line in lines]
+    assert len(records) >= 2, "on-change must emit RUNNING snapshots before the end"


Done in 9b517c0 after rebasing onto the asynchronous export change: the test no longer asserts a RUNNING-line count. It requires every line to be by-name shaped, non-terminal lines to be RUNNING, the last line to be the terminal SUCCEEDED record, and the SQS send count to match the file line count. Stable across repeated local runs.

wangyb-A · 2026-09-11T22:34:42Z

Sign in to view

+                            "scope": {
+                                "name": _SCOPE_NAME,
+                                "version": record.get("schemaVersion", ""),
+                            },


Kept as is: scope.version = schemaVersion is what the other SDKs' OTel exporter sends, and changing only this one would group Python telemetry differently. Worth changing in all SDKs together; noted as a follow-up.

- 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

github-actions · 2026-09-11T23:35:14Z

+        self.db_user = db_user
+        self.secret_arn = secret_arn


Codex AI review · Finding arf_v1_lmu5i6xt6xpfayumozke7vmhli

[P2] Validate Redshift authentication combinations

db_user is valid only with a provisioned cluster using temporary credentials; it cannot accompany a workgroup or secret_arn. These combinations are accepted here and every export subsequently fails service validation, which the scheduler reduces to a warning. Reject invalid combinations during construction and test each supported authentication mode.

github-actions · 2026-09-11T23:35:22Z

Codex AI review

Found eight actionable issues, including unbounded exporter waits, stale Redshift writes, and EFS corruption risks. Live AWS-service behavior remains the main residual test risk.

Reviewed commit 3a424e7bf6d3c768f38024007e42038eb43fcbde. Workflow run

wangyb-A force-pushed the feat/insight-exporters-parity branch from ec3c719 to 32c68f1 Compare September 11, 2026 17:28

wangyb-A marked this pull request as ready for review September 11, 2026 18:01

wangyb-A deployed to ai-pr-review-runtime September 11, 2026 18:01 — with GitHub Actions Active

wangyb-A deployed to ai-pr-review-runtime September 11, 2026 18:25 — with GitHub Actions Active