-
Notifications
You must be signed in to change notification settings - Fork 23
feat: add Workflow Insight instrumentation plugin #632
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
cad922e
1b7e04f
2dd923e
ec7b2ea
455a5bc
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| # AWS Durable Execution SDK for Python — Workflow Insight plugin | ||
|
|
||
| Workflow Insight instrumentation plugin for the AWS Durable Execution SDK for | ||
| Python. A port of the JavaScript SDK's `workflowInsight()` plugin: it listens to | ||
| the SDK's instrumentation hooks and emits one curated `WorkflowInsight` record | ||
| per execution to the configured exporters. The wire record keeps the JS | ||
| camelCase field names so records read identically across SDKs. | ||
|
|
||
| > **Experimental.** Like its JS counterpart, this plugin is experimental and may | ||
| > change or be removed in future releases. | ||
|
|
||
| ## Install | ||
|
|
||
| ```bash | ||
| pip install aws-durable-execution-sdk-python-insight | ||
| # with the S3 exporter's local-dev dependency: | ||
| pip install "aws-durable-execution-sdk-python-insight[s3]" | ||
| ``` | ||
|
|
||
| ## Usage | ||
|
|
||
| ```python | ||
| from aws_durable_execution_sdk_python import durable_execution | ||
| from aws_durable_execution_sdk_python_insight import ( | ||
| WorkflowInsightConfig, | ||
| workflow_insight, | ||
| ) | ||
| from aws_durable_execution_sdk_python_insight.exporters import S3Exporter | ||
|
|
||
| @durable_execution( | ||
| plugins=[ | ||
| workflow_insight( | ||
| WorkflowInsightConfig( | ||
| exporters=[ | ||
| S3Exporter(bucket="my-bucket", prefix="workflow-insight/") | ||
| ], | ||
| ) | ||
| ) | ||
|
wangyb-A marked this conversation as resolved.
|
||
| ] | ||
| ) | ||
| def handler(event, context): | ||
| ... | ||
| ``` | ||
|
|
||
| With no exporter configured, records are written to the function's own | ||
| CloudWatch log group as single JSON lines (the `LambdaLogExporter` default), | ||
| carrying the name-keyed `operationsByName` summary. The `S3Exporter` writes the | ||
| lossless per-occurrence `operations` array, one object per execution | ||
| (upsert-by-execution-name, so re-emission overwrites rather than appends). | ||
|
|
||
| Emission behavior, record schema (`recordType: WorkflowInsight`, | ||
| `schemaVersion: "1.0"`), sampling, content configuration (input/output | ||
| omission, `include_errors`, per-operation result opt-in), truncation phases, | ||
| and `top-level` vs `full-tree` operation detail all mirror the JS plugin. | ||
| Behavior is validated cross-SDK by the `insight` conformance suite | ||
| (`aws-durable-execution-conformance-tests-insight`). | ||
|
|
||
| ## Requirements | ||
|
|
||
| - `aws-durable-execution-sdk-python` with the plugin invocation hooks that | ||
| surface `execution_input` / `execution_result` (included since the version | ||
| this package declares as its minimum). | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,79 @@ | ||
| [build-system] | ||
| requires = ["hatchling"] | ||
| build-backend = "hatchling.build" | ||
|
|
||
| [project] | ||
| name = "aws-durable-execution-sdk-python-insight" | ||
|
wangyb-A marked this conversation as resolved.
wangyb-A marked this conversation as resolved.
|
||
| dynamic = ["version"] | ||
| description = 'Workflow Insight instrumentation plugin for the AWS Durable Execution SDK for Python' | ||
| readme = "README.md" | ||
| requires-python = ">=3.11" | ||
| license = "Apache-2.0" | ||
| keywords = ["observability", "workflow-insight", "durable-execution"] | ||
| authors = [{ name = "AWS durable-execution-dev", email = "durable-execution-dev@amazon.com" }] | ||
| classifiers = [ | ||
| "Development Status :: 4 - Beta", | ||
| "Programming Language :: Python", | ||
| "Programming Language :: Python :: 3.11", | ||
| "Programming Language :: Python :: 3.12", | ||
| "Programming Language :: Python :: 3.13", | ||
| "Programming Language :: Python :: 3.14", | ||
| "Programming Language :: Python :: Implementation :: CPython", | ||
| ] | ||
| dependencies = [ | ||
| # >=1.8.0: first release carrying the plugin invocation-hook fields | ||
| # (InvocationInfo.execution_input / InvocationEndInfo.execution_result). | ||
| "aws-durable-execution-sdk-python>=1.8.0", | ||
| ] | ||
|
|
||
| [project.optional-dependencies] | ||
| # boto3 is provided by the Lambda runtime; declared as an extra for local dev | ||
| # (e.g. the S3Exporter) without vendoring it into deployments. | ||
| s3 = ["boto3>=1.26.0"] | ||
|
|
||
| [project.urls] | ||
| Documentation = "https://github.com/aws/aws-durable-execution-sdk-python#readme" | ||
| Issues = "https://github.com/aws/aws-durable-execution-sdk-python/issues" | ||
| Source = "https://github.com/aws/aws-durable-execution-sdk-python" | ||
|
|
||
| [tool.hatch.build.targets.sdist.force-include] | ||
| "../../LICENSE" = "LICENSE" | ||
| "../../NOTICE" = "NOTICE" | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. add a plugin entry point here to allow this plugin to be auto loaded. See otel plugin for reference |
||
|
|
||
| [tool.hatch.build.targets.wheel] | ||
| packages = ["src/aws_durable_execution_sdk_python_insight"] | ||
|
|
||
| [tool.hatch.build.targets.wheel.force-include] | ||
| "../../LICENSE" = "aws_durable_execution_sdk_python_insight/LICENSE" | ||
| "../../NOTICE" = "aws_durable_execution_sdk_python_insight/NOTICE" | ||
|
|
||
| [tool.hatch.version] | ||
| path = "src/aws_durable_execution_sdk_python_insight/__about__.py" | ||
|
|
||
| [tool.hatch.publish.index] | ||
| disable = true | ||
|
|
||
| [tool.coverage.run] | ||
| source_pkgs = ["aws_durable_execution_sdk_python_insight"] | ||
| branch = true | ||
| parallel = true | ||
| omit = ["src/aws_durable_execution_sdk_python_insight/__about__.py"] | ||
|
|
||
| [tool.coverage.report] | ||
| exclude_lines = ["no cov", "if __name__ == .__main__.:", "if TYPE_CHECKING:"] | ||
|
|
||
| [tool.ruff] | ||
| line-length = 88 | ||
| target-version = "py311" | ||
|
|
||
| [tool.ruff.lint] | ||
| preview = true | ||
| select = ["E4", "E7", "E9", "F", "TID252"] | ||
|
|
||
| [tool.ruff.lint.isort] | ||
| known-first-party = ["aws_durable_execution_sdk_python_insight"] | ||
| force-single-line = false | ||
| lines-after-imports = 2 | ||
|
|
||
| [tool.ruff.lint.per-file-ignores] | ||
| "tests/**" = ["ARG001", "ARG002", "ARG005", "S101", "PLR2004", "PLR6301", "SIM117", "TRY301"] | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,4 @@ | ||
| # SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. | ||
| # | ||
| # SPDX-License-Identifier: Apache-2.0 | ||
| __version__ = "0.0.1" |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| # SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. | ||
| # | ||
| # SPDX-License-Identifier: Apache-2.0 | ||
| """Workflow Insight instrumentation plugin for the AWS Durable Execution Python SDK.""" | ||
|
|
||
| from aws_durable_execution_sdk_python_insight.__about__ import __version__ | ||
| from aws_durable_execution_sdk_python_insight.exporters import ( | ||
| LambdaLogExporter, | ||
| S3Exporter, | ||
| S3Partitioning, | ||
| ) | ||
| from aws_durable_execution_sdk_python_insight.operations_index import ( | ||
| build_operations_by_name, | ||
| with_operations_by_name, | ||
| ) | ||
| from aws_durable_execution_sdk_python_insight.plugin import ( | ||
| WorkflowInsightPlugin, | ||
| workflow_insight, | ||
| ) | ||
| from aws_durable_execution_sdk_python_insight.truncation import truncate_record | ||
| from aws_durable_execution_sdk_python_insight.types import ( | ||
| ContentConfig, | ||
| ContentOperations, | ||
| EmitMode, | ||
| InsightExporter, | ||
| OperationDetail, | ||
| OperationOverride, | ||
| WorkflowInsightConfig, | ||
| ) | ||
|
|
||
|
|
||
| __all__ = [ | ||
| "__version__", | ||
| "ContentConfig", | ||
| "ContentOperations", | ||
| "EmitMode", | ||
| "InsightExporter", | ||
| "LambdaLogExporter", | ||
| "OperationDetail", | ||
| "OperationOverride", | ||
| "S3Exporter", | ||
| "S3Partitioning", | ||
| "WorkflowInsightConfig", | ||
| "WorkflowInsightPlugin", | ||
| "build_operations_by_name", | ||
| "truncate_record", | ||
| "with_operations_by_name", | ||
| "workflow_insight", | ||
| ] |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,38 @@ | ||
| # SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. | ||
| # | ||
| # SPDX-License-Identifier: Apache-2.0 | ||
| """First-party Workflow Insight exporters. | ||
|
|
||
| One module per exporter, mirroring the JS package's ``src/exporters/`` layout | ||
| (``aws-durable-execution-sdk-js-insight``). Each destination lives in its own | ||
| module so the set can grow to the full JS parity surface (S3, CloudWatch Logs, | ||
| DynamoDB, Firehose, EventBridge, SQS, OpenSearch, Redshift, Aurora, HTTP, OTel, | ||
| file, ...) without any single file accreting every backend's imports and | ||
| optional dependencies. | ||
|
|
||
| Concrete exporters are re-exported here so the public import path is stable: | ||
| ``from aws_durable_execution_sdk_python_insight.exporters import S3Exporter`` | ||
| keeps working exactly as before this package was split out of a single module. | ||
| Shared serialization helpers live in the private ``_common`` module. | ||
|
|
||
| Both shipped exporters serialize the curated record with JS-compatible compact | ||
| JSON (no whitespace) so the wire bytes match across SDKs. Records are written | ||
| verbatim -- no synthetic emission. | ||
| """ | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| from aws_durable_execution_sdk_python_insight.exporters.lambda_log_exporter import ( | ||
| LambdaLogExporter, | ||
| ) | ||
| from aws_durable_execution_sdk_python_insight.exporters.s3_exporter import ( | ||
| S3Exporter, | ||
| S3Partitioning, | ||
| ) | ||
|
|
||
|
|
||
| __all__ = [ | ||
| "LambdaLogExporter", | ||
| "S3Exporter", | ||
| "S3Partitioning", | ||
| ] |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,29 @@ | ||
| # SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. | ||
| # | ||
| # SPDX-License-Identifier: Apache-2.0 | ||
| """Shared serialization helpers for the Workflow Insight exporters. | ||
|
|
||
| Kept private to the ``exporters`` package: every backend needs the same | ||
| JS-compatible compact JSON encoding and the same key/file-name sanitizer, so | ||
| they live here rather than being duplicated per exporter module. | ||
| """ | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| import json | ||
| import re | ||
| from typing import Any | ||
|
|
||
|
|
||
| def compact_dumps(value: Any) -> str: | ||
| """Serialize ``value`` as compact JSON (no whitespace, non-ASCII preserved). | ||
|
|
||
| Matches the JS exporters' ``JSON.stringify`` output so the wire bytes are | ||
| identical across SDKs. | ||
| """ | ||
| return json.dumps(value, separators=(",", ":"), ensure_ascii=False) | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Codex AI review · Finding [P2] Normalize non-finite floats before serialization. Python emits |
||
|
|
||
|
|
||
| def sanitize(value: str) -> str: | ||
| """Replace characters unsafe for object keys / file names with ``_``.""" | ||
| return re.sub(r"[^a-zA-Z0-9._-]", "_", value) | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,39 @@ | ||
| # SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. | ||
| # | ||
| # SPDX-License-Identifier: Apache-2.0 | ||
| """Lambda log (CloudWatch) Workflow Insight exporter.""" | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| from typing import Any | ||
|
|
||
| from aws_durable_execution_sdk_python_insight.exporters._common import compact_dumps | ||
| from aws_durable_execution_sdk_python_insight.operations_index import ( | ||
| with_operations_by_name, | ||
| ) | ||
|
|
||
|
|
||
| class LambdaLogExporter: | ||
| """Writes ``operationsByName`` records to the function's own log group via ``print``. | ||
|
|
||
| Port of the JS ``LambdaLogExporter``: ``console.log(JSON.stringify( | ||
| withOperationsByName(record)))``. Requires no extra IAM. Emits the name-keyed | ||
| summary map (``OPERATIONS_BY_NAME``). | ||
| """ | ||
|
|
||
| def __init__(self, max_record_size_bytes: int | None = None) -> None: | ||
| self.max_record_size_bytes: int | None = ( | ||
| 256_000 if max_record_size_bytes is None else max_record_size_bytes | ||
| ) | ||
|
|
||
| def render(self, record: dict[str, Any]) -> dict[str, Any]: | ||
| return with_operations_by_name(record) | ||
|
|
||
| def export(self, record: dict[str, Any]) -> None: | ||
| # Raw JSON line to stdout -> the function's CloudWatch log group. The | ||
| # conformance CloudWatch sink json.loads each line (and unwraps the | ||
| # Lambda structured-log envelope when present). | ||
| print(compact_dumps(self.render(record)), flush=True) # noqa: T201 | ||
|
|
||
| def flush(self) -> None: | ||
| return None |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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.[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 overpackages/*/, so those are already covered.