diff --git a/packages/aws-durable-execution-sdk-python-insight/README.md b/packages/aws-durable-execution-sdk-python-insight/README.md index be1027df..2260a017 100644 --- a/packages/aws-durable-execution-sdk-python-insight/README.md +++ b/packages/aws-durable-execution-sdk-python-insight/README.md @@ -48,6 +48,160 @@ 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). +## Exporters + +All exporters live in `aws_durable_execution_sdk_python_insight.exporters` and +are re-exported from the package root. Each serializes the record as compact +JSON. Exporters that call AWS accept an injected `client=` for tests and use +the Lambda runtime's boto3 otherwise; none adds a required dependency. + +| Exporter | Destination | Upsert | Operations shape | Default size limit | +| --- | --- | --- | --- | --- | +| `LambdaLogExporter` | Function's own log group | No | `operationsByName` | 256 KB | +| `CloudWatchLogsExporter` | Any log group, one stream per day | No | `operationsByName` | 256 KB | +| `S3Exporter` | S3 object per execution | Yes (key) | `operations` | 5 MB | +| `DynamoDBExporter` | DynamoDB item | Configurable | `operationsByName` | 400 KB | +| `AuroraExporter` | Aurora MySQL/PostgreSQL row (Data API) | Yes (upsert) | full record as JSON column | 1 MB | +| `RedshiftExporter` | Redshift row (Data API) | Yes (MERGE) | full record as SUPER column | 1 MB | +| `OpenSearchExporter` | OpenSearch document | Yes (`_id`) | `operations` | 10 MB | +| `FirehoseExporter` | Firehose delivery stream | N/A | `operations_format` | 1 MB | +| `EventBridgeExporter` | EventBridge event | N/A | `operations_format` | 256 KB | +| `SQSExporter` | SQS message | N/A | `operations_format` | 256 KB | +| `OTelExporter` | OTLP/HTTP logs endpoint | N/A | `operations_format` | 1 MB | +| `HttpExporter` | Any HTTP endpoint | N/A | `operations_format` | none | +| `FileExporter` | Directory (EFS, mount, `/tmp`) | Configurable | `operations_format` | none | + +`operations_format` is `"array"` (default), `"by-name"`, or `"both"` +(`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. + +### CloudWatchLogsExporter + +Writes one `PutLogEvents` event per record to `log_group_name`, in a stream +named `{log_stream_prefix}{YYYY}/{MM}/{DD}` (default prefix `workflow-insight/`). +IAM: `logs:CreateLogStream`, `logs:PutLogEvents` on the log group. + +```python +CloudWatchLogsExporter(log_group_name="/custom/workflow-insight") +``` + +### DynamoDBExporter + +`PutItem` keyed by `partition_key` (default `pk`) = `executionArn`. With the +default `sort_key="sk"` (= `emittedAt`) every export adds an item; pass +`sort_key=None` for a key-only table that upserts. IAM: `dynamodb:PutItem`. + +```python +DynamoDBExporter(table_name="workflow-insight") +``` + +### AuroraExporter + +Upserts a row by `execution_arn` through the RDS Data API; `engine` is +`"postgresql"` or `"mysql"` and selects the dialect. Columns: `execution_arn, +execution_name, function_name, status, start_time, end_time, duration_ms, +record_json, emitted_at`. IAM: `rds-data:ExecuteStatement`, +`secretsmanager:GetSecretValue`. + +```python +AuroraExporter( + resource_arn="arn:aws:rds:us-east-1:123456789012:cluster:my-cluster", + secret_arn="arn:aws:secretsmanager:us-east-1:123456789012:secret:my-db-creds", + database="insight", + engine="postgresql", +) +``` + +### RedshiftExporter + +`MERGE` by `execution_arn` through the Redshift Data API into +`{schema}.{table}` (default `public.workflow_insight`, same columns as Aurora, +`record_json` as `SUPER`). Provide `workgroup_name` (Serverless) or +`cluster_identifier` (provisioned, with `db_user` or `secret_arn`). IAM: +`redshift-data:ExecuteStatement` plus `redshift-serverless:GetCredentials` or +`secretsmanager:GetSecretValue`. The statement is submitted and not awaited, so +statement failures are not reported and `on-change` records may land out of +order; prefer the default `on-complete` emit mode with this exporter. + +```python +RedshiftExporter(database="insight", workgroup_name="insight-wg") +``` + +### OpenSearchExporter + +`PUT {endpoint}/{index_name}/_doc/{executionArn}` (default index +`workflow-insight`). `auth="sigv4"` (default) signs with the runtime's +credentials via botocore; `auth="basic"` uses `username`/`password`. IAM: +`es:ESHttpPut` on the domain. + +```python +OpenSearchExporter(endpoint="https://my-domain.us-east-1.es.amazonaws.com", region="us-east-1") +``` + +### FirehoseExporter + +`PutRecord` of one JSON line (trailing newline) per record. IAM: +`firehose:PutRecord`. + +```python +FirehoseExporter(delivery_stream_name="workflow-insight-stream") +``` + +### EventBridgeExporter + +`PutEvents` with `Source` (default `aws.durable-execution.insight`), +`DetailType` = record status, `Detail` = record. All arguments optional. IAM: +`events:PutEvents`. + +```python +EventBridgeExporter(event_bus_name="default") +``` + +### SQSExporter + +`SendMessage` with the record as body and `status`/`functionName` message +attributes. A `.fifo` queue URL enables `MessageGroupId` (default +`executionArn`, or `message_group_id`) and a deduplication id of +`executionArn:emittedAt`. IAM: `sqs:SendMessage`. + +```python +SQSExporter(queue_url="https://sqs.us-east-1.amazonaws.com/123456789012/insight") +``` + +### OTelExporter + +POSTs one OTLP `ExportLogsServiceRequest` (`http/json` only) per record to +`endpoint`; identity fields become attributes and the record is the log body. +Pass vendor auth in `headers`. No IAM. + +```python +OTelExporter(endpoint="https://otlp.vendor.com/v1/logs", headers={"x-api-key": "..."}) +``` + +### HttpExporter + +`POST` (or `method="PUT"`) the record as JSON to `url` with +`Content-Type: application/json` plus `headers`; a non-2xx status raises. +`timeout_ms` defaults to 10000. No IAM. + +```python +HttpExporter(url="https://hooks.example.com/insight", headers={"Authorization": "Bearer ..."}) +``` + +### FileExporter + +`mode="ndjson"` (default) appends to `{directory}/{YYYY-MM-DD}.ndjson`; +`mode="json"` writes `{directory}/{executionName}.json`, overwriting on update. +For Lambda use an EFS mount (IAM: `elasticfilesystem:ClientMount`, +`elasticfilesystem:ClientWrite`) or `/tmp` for testing. Appends from many +concurrent environments to one NDJSON file on a network file system can +interleave; use `mode="json"` when several environments share a directory. + +```python +FileExporter(directory="/mnt/efs/workflow-insight") +``` + Emission behavior, record schema (`recordType: WorkflowInsight`, `schemaVersion: "1.0"`), sampling, content configuration (input/output omission, `include_errors`, per-operation result opt-in), truncation phases, diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/__init__.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/__init__.py index 0ea38d47..488c2ada 100644 --- a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/__init__.py +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/__init__.py @@ -5,11 +5,29 @@ from aws_durable_execution_sdk_python_insight.__about__ import __version__ from aws_durable_execution_sdk_python_insight.exporters import ( + AuroraEngine, + AuroraExporter, + CloudWatchLogsExporter, + DynamoDBExporter, + EventBridgeExporter, + FileExporter, + FileMode, + FirehoseExporter, + HttpExporter, + HttpMethod, LambdaLogExporter, + OpenSearchAuth, + OpenSearchExporter, + OTelExporter, + OTelProtocol, + RedshiftExporter, S3Exporter, S3Partitioning, + SQSExporter, ) from aws_durable_execution_sdk_python_insight.operations_index import ( + OperationsFormat, + apply_operations_format, build_operations_by_name, with_operations_by_name, ) @@ -31,17 +49,35 @@ __all__ = [ "__version__", + "AuroraEngine", + "AuroraExporter", + "CloudWatchLogsExporter", "ContentConfig", "ContentOperations", + "DynamoDBExporter", "EmitMode", + "EventBridgeExporter", + "FileExporter", + "FileMode", + "FirehoseExporter", + "HttpExporter", + "HttpMethod", "InsightExporter", "LambdaLogExporter", + "OTelExporter", + "OTelProtocol", + "OpenSearchAuth", + "OpenSearchExporter", "OperationDetail", "OperationOverride", + "OperationsFormat", + "RedshiftExporter", "S3Exporter", "S3Partitioning", + "SQSExporter", "WorkflowInsightConfig", "WorkflowInsightPlugin", + "apply_operations_format", "build_operations_by_name", "truncate_record", "with_operations_by_name", diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/__init__.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/__init__.py index 3d432de2..91ef44ad 100644 --- a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/__init__.py +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/__init__.py @@ -3,36 +3,84 @@ # 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. +One module per destination, so no single file accretes 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``. +Shared serialization and transport helpers live in the private ``_common`` +module. -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. +Every exporter serializes the curated record as compact JSON (no whitespace, +non-ASCII preserved). Records are written verbatim -- no synthetic emission. """ from __future__ import annotations +from aws_durable_execution_sdk_python_insight.exporters.aurora_exporter import ( + AuroraEngine, + AuroraExporter, +) +from aws_durable_execution_sdk_python_insight.exporters.cloudwatch_logs_exporter import ( + CloudWatchLogsExporter, +) +from aws_durable_execution_sdk_python_insight.exporters.dynamodb_exporter import ( + DynamoDBExporter, +) +from aws_durable_execution_sdk_python_insight.exporters.eventbridge_exporter import ( + EventBridgeExporter, +) +from aws_durable_execution_sdk_python_insight.exporters.file_exporter import ( + FileExporter, + FileMode, +) +from aws_durable_execution_sdk_python_insight.exporters.firehose_exporter import ( + FirehoseExporter, +) +from aws_durable_execution_sdk_python_insight.exporters.http_exporter import ( + HttpExporter, + HttpMethod, +) from aws_durable_execution_sdk_python_insight.exporters.lambda_log_exporter import ( LambdaLogExporter, ) +from aws_durable_execution_sdk_python_insight.exporters.opensearch_exporter import ( + OpenSearchAuth, + OpenSearchExporter, +) +from aws_durable_execution_sdk_python_insight.exporters.otel_exporter import ( + OTelExporter, + OTelProtocol, +) +from aws_durable_execution_sdk_python_insight.exporters.redshift_exporter import ( + RedshiftExporter, +) from aws_durable_execution_sdk_python_insight.exporters.s3_exporter import ( S3Exporter, S3Partitioning, ) +from aws_durable_execution_sdk_python_insight.exporters.sqs_exporter import ( + SQSExporter, +) __all__ = [ + "AuroraEngine", + "AuroraExporter", + "CloudWatchLogsExporter", + "DynamoDBExporter", + "EventBridgeExporter", + "FileExporter", + "FileMode", + "FirehoseExporter", + "HttpExporter", + "HttpMethod", "LambdaLogExporter", + "OTelExporter", + "OTelProtocol", + "OpenSearchAuth", + "OpenSearchExporter", + "RedshiftExporter", "S3Exporter", "S3Partitioning", + "SQSExporter", ] diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/_common.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/_common.py index 4401fce1..28ef3626 100644 --- a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/_common.py +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/_common.py @@ -10,11 +10,35 @@ from __future__ import annotations +import datetime import json import re +import urllib.error +import urllib.request from typing import Any +_SQL_IDENTIFIER = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$") +# Upper bound on how much of a non-2xx response body is read for diagnostics. +_MAX_ERROR_BODY_BYTES = 64 * 1024 + + +class _NoRedirect(urllib.request.HTTPRedirectHandler): + """Refuse every redirect so a 3xx surfaces as a failed status. + + Following a redirect would re-send a POST as a body-less GET and forward + configured credential headers to the new location. + """ + + def redirect_request( # type: ignore[override] # stdlib signature has no hints + self, req: Any, fp: Any, code: int, msg: str, headers: Any, newurl: str + ) -> None: + return None + + +_OPENER = urllib.request.build_opener(_NoRedirect()) + + def compact_dumps(value: Any) -> str: """Serialize ``value`` as compact JSON (no whitespace, non-ASCII preserved). @@ -27,3 +51,55 @@ def compact_dumps(value: Any) -> str: def sanitize(value: str) -> str: """Replace characters unsafe for object keys / file names with ``_``.""" return re.sub(r"[^a-zA-Z0-9._-]", "_", value) + + +def sql_identifier(name: str) -> str: + """Return ``name`` if it is a plain SQL identifier, else raise ``ValueError``. + + Table and schema names are interpolated into SQL text, so only letters, + digits, and underscores are accepted. + """ + if not _SQL_IDENTIFIER.match(name): + msg = ( + f'Invalid SQL identifier: "{name}". ' + "Only letters, digits, and underscores are allowed." + ) + raise ValueError(msg) + return name + + +def parse_iso_datetime(value: str) -> datetime.datetime: + """Parse an ISO-8601 timestamp (``Z`` or offset) into an aware UTC datetime.""" + parsed = datetime.datetime.fromisoformat(value.replace("Z", "+00:00")) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=datetime.UTC) + return parsed.astimezone(datetime.UTC) + + +def http_send( + method: str, + url: str, + headers: dict[str, str], + body: bytes, + timeout: float | None = None, +) -> tuple[int, str, str]: + """Send one HTTP request and return ``(status, reason, error_text)``. + + A non-2xx status is returned, not raised, so callers build their own error + message. Redirects are not followed: a 3xx is returned like any other + failure. ``error_text`` is the first ``_MAX_ERROR_BODY_BYTES`` of a non-2xx + response body and empty on success; a success body is never read. Network + errors and timeouts propagate. + """ + request = urllib.request.Request(url, data=body, method=method) + for key, value in headers.items(): + request.add_header(key, value) + try: + with _OPENER.open(request, timeout=timeout) as response: # noqa: S310 + return int(response.status), str(response.reason or ""), "" + except urllib.error.HTTPError as exc: + try: + detail = exc.read(_MAX_ERROR_BODY_BYTES).decode("utf-8", errors="replace") + except Exception: # noqa: BLE001 - the body is best-effort detail only + detail = "" + return int(exc.code), str(exc.reason or ""), detail diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/aurora_exporter.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/aurora_exporter.py new file mode 100644 index 00000000..8f393502 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/aurora_exporter.py @@ -0,0 +1,152 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Aurora (RDS Data API) Workflow Insight exporter.""" + +from __future__ import annotations + +from enum import StrEnum +from typing import Any, Literal + +from aws_durable_execution_sdk_python_insight.exporters._common import ( + compact_dumps, + sql_identifier, +) + + +class AuroraEngine(StrEnum): + """Database engine; selects the upsert dialect.""" + + POSTGRESQL = "postgresql" + MYSQL = "mysql" + + +# Accepted string inputs, kept in lockstep with the enum values above. +AuroraEngineInput = Literal["postgresql", "mysql"] + +_COLUMNS = ( + "execution_arn, execution_name, function_name, status, start_time, " + "end_time, duration_ms, record_json, emitted_at" +) + + +def _string_or_null(value: Any) -> dict[str, Any]: + return {"stringValue": value} if value else {"isNull": True} + + +class AuroraExporter: + """Upserts one row per execution through the RDS Data API. + + Rows are keyed by ``execution_arn``; a later export for the same execution + overwrites the row. The full record is stored as JSON in ``record_json``. + The cluster must have the Data API enabled. + """ + + def __init__( + self, + resource_arn: str, + secret_arn: str, + database: str, + engine: AuroraEngine | AuroraEngineInput, + table: str = "workflow_insight", + region: str | None = None, + max_record_size_bytes: int | None = None, + client: Any = None, + ) -> None: + self.resource_arn = resource_arn + self.secret_arn = secret_arn + self.database = database + self.table = sql_identifier(table) + self.engine = AuroraEngine(engine) + self.max_record_size_bytes: int | None = ( + 1_000_000 if max_record_size_bytes is None else max_record_size_bytes + ) + if client is not None: + self._client = client + else: + import boto3 # deferred: boto3 is provided by the Lambda runtime + + self._client = ( + boto3.client("rds-data", region_name=region) + if region + else boto3.client("rds-data") + ) + + def render(self, record: dict[str, Any]) -> dict[str, Any]: + return record + + def export(self, record: dict[str, Any]) -> None: + sql = ( + self._postgres_upsert() + if self.engine == AuroraEngine.POSTGRESQL + else self._mysql_upsert() + ) + duration = record.get("durationMs") + self._client.execute_statement( + resourceArn=self.resource_arn, + secretArn=self.secret_arn, + database=self.database, + sql=sql, + parameters=[ + { + "name": "execution_arn", + "value": {"stringValue": record["executionArn"]}, + }, + { + "name": "execution_name", + "value": _string_or_null(record.get("executionName")), + }, + { + "name": "function_name", + "value": {"stringValue": record["functionName"]}, + }, + {"name": "status", "value": {"stringValue": record["status"]}}, + {"name": "start_time", "value": {"stringValue": record["startTime"]}}, + {"name": "end_time", "value": _string_or_null(record.get("endTime"))}, + { + "name": "duration_ms", + "value": {"longValue": duration} + if duration is not None + else {"isNull": True}, + }, + { + "name": "record_json", + "value": {"stringValue": compact_dumps(record)}, + }, + {"name": "emitted_at", "value": {"stringValue": record["emittedAt"]}}, + ], + ) + + def flush(self) -> None: + return None + + def _postgres_upsert(self) -> str: + return ( + f"INSERT INTO {self.table}\n" + f" ({_COLUMNS})\n" + " VALUES\n" + " (:execution_arn, :execution_name, :function_name, :status, " + ":start_time::timestamptz, :end_time::timestamptz, :duration_ms, " + ":record_json::jsonb, :emitted_at::timestamptz)\n" + " ON CONFLICT (execution_arn) DO UPDATE SET\n" + " status = EXCLUDED.status,\n" + " end_time = EXCLUDED.end_time,\n" + " duration_ms = EXCLUDED.duration_ms,\n" + " record_json = EXCLUDED.record_json,\n" + " emitted_at = EXCLUDED.emitted_at" + ) + + def _mysql_upsert(self) -> str: + return ( + f"INSERT INTO {self.table}\n" + f" ({_COLUMNS})\n" + " VALUES\n" + " (:execution_arn, :execution_name, :function_name, :status, " + ":start_time, :end_time, :duration_ms, :record_json, :emitted_at)\n" + " ON DUPLICATE KEY UPDATE\n" + " status = VALUES(status),\n" + " end_time = VALUES(end_time),\n" + " duration_ms = VALUES(duration_ms),\n" + " record_json = VALUES(record_json),\n" + " emitted_at = VALUES(emitted_at)" + ) diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/cloudwatch_logs_exporter.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/cloudwatch_logs_exporter.py new file mode 100644 index 00000000..e2f8cab9 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/cloudwatch_logs_exporter.py @@ -0,0 +1,94 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""CloudWatch Logs (PutLogEvents) Workflow Insight exporter.""" + +from __future__ import annotations + +import datetime +import time +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, +) + + +def _is_already_exists(exc: BaseException) -> bool: + if type(exc).__name__ == "ResourceAlreadyExistsException": + return True + response = getattr(exc, "response", None) + if isinstance(response, dict): + code = response.get("Error", {}).get("Code") + return bool(code == "ResourceAlreadyExistsException") + return False + + +class CloudWatchLogsExporter: + """Writes ``operationsByName`` records to a chosen log group with PutLogEvents. + + Unlike ``LambdaLogExporter`` this targets any log group. One log stream is + created per UTC day, named ``{log_stream_prefix}{YYYY}/{MM}/{DD}``. + """ + + def __init__( + self, + log_group_name: str, + log_stream_prefix: str = "workflow-insight/", + region: str | None = None, + max_record_size_bytes: int | None = None, + client: Any = None, + ) -> None: + self.log_group_name = log_group_name + self.log_stream_prefix = log_stream_prefix + self.max_record_size_bytes: int | None = ( + 256_000 if max_record_size_bytes is None else max_record_size_bytes + ) + self._created_streams: set[str] = set() + if client is not None: + self._client = client + else: + import boto3 # deferred: boto3 is provided by the Lambda runtime + + self._client = ( + boto3.client("logs", region_name=region) + if region + else boto3.client("logs") + ) + + def render(self, record: dict[str, Any]) -> dict[str, Any]: + return with_operations_by_name(record) + + def export(self, record: dict[str, Any]) -> None: + stream_name = self._build_stream_name() + self._ensure_stream(stream_name) + self._client.put_log_events( + logGroupName=self.log_group_name, + logStreamName=stream_name, + logEvents=[ + { + "timestamp": int(time.time() * 1000), + "message": compact_dumps(self.render(record)), + } + ], + ) + + def flush(self) -> None: + return None + + def _build_stream_name(self) -> str: + now = datetime.datetime.now(datetime.UTC) + return f"{self.log_stream_prefix}{now:%Y/%m/%d}" + + def _ensure_stream(self, stream_name: str) -> None: + if stream_name in self._created_streams: + return + try: + self._client.create_log_stream( + logGroupName=self.log_group_name, logStreamName=stream_name + ) + except Exception as exc: + if not _is_already_exists(exc): + raise + self._created_streams.add(stream_name) diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/dynamodb_exporter.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/dynamodb_exporter.py new file mode 100644 index 00000000..9b139c10 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/dynamodb_exporter.py @@ -0,0 +1,73 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""DynamoDB Workflow Insight exporter.""" + +from __future__ import annotations + +import json +from decimal import Decimal +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 DynamoDBExporter: + """Writes ``operationsByName`` records to a DynamoDB table with PutItem. + + The partition key holds ``executionArn``. With the default sort key + (``sk`` = ``emittedAt``) every export adds a new item, keeping the full + history. With ``sort_key=None`` later exports overwrite the item. + """ + + def __init__( + self, + table_name: str, + partition_key: str = "pk", + sort_key: str | None = "sk", + region: str | None = None, + max_record_size_bytes: int | None = None, + client: Any = None, + ) -> None: + self.table_name = table_name + self.partition_key = partition_key + # ``None`` and ``""`` both disable the sort key. + self.sort_key = sort_key or None + self.max_record_size_bytes: int | None = ( + 400_000 if max_record_size_bytes is None else max_record_size_bytes + ) + if client is not None: + self._client = client + else: + import boto3 # deferred: boto3 is provided by the Lambda runtime + + self._client = ( + boto3.client("dynamodb", region_name=region) + if region + else boto3.client("dynamodb") + ) + from boto3.dynamodb.types import TypeSerializer # deferred, same reason + + self._serializer = TypeSerializer() + + def render(self, record: dict[str, Any]) -> dict[str, Any]: + return with_operations_by_name(record) + + def export(self, record: dict[str, Any]) -> None: + item = self.render(record) + item[self.partition_key] = record["executionArn"] + if self.sort_key: + item[self.sort_key] = record["emittedAt"] + self._client.put_item(TableName=self.table_name, Item=self._marshal(item)) + + def flush(self) -> None: + return None + + def _marshal(self, item: dict[str, Any]) -> dict[str, Any]: + # DynamoDB numbers must be ``Decimal``; a JSON round trip converts every + # float and leaves the rest of the record untouched. + plain = json.loads(compact_dumps(item), parse_float=Decimal) + return {key: self._serializer.serialize(value) for key, value in plain.items()} diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/eventbridge_exporter.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/eventbridge_exporter.py new file mode 100644 index 00000000..7b6e99da --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/eventbridge_exporter.py @@ -0,0 +1,82 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""EventBridge Workflow Insight exporter.""" + +from __future__ import annotations + +from typing import Any + +from aws_durable_execution_sdk_python_insight.exporters._common import ( + compact_dumps, + parse_iso_datetime, +) +from aws_durable_execution_sdk_python_insight.operations_index import ( + OperationsFormat, + OperationsFormatInput, + apply_operations_format, +) + + +class EventBridgeExporter: + """Publishes each record as one EventBridge event with PutEvents. + + ``Source`` is configurable, ``DetailType`` is the record status + (``SUCCEEDED``, ``RUNNING``, ``FAILED``), ``Detail`` is the rendered record + and ``Time`` is the record's ``emittedAt``. + """ + + def __init__( + self, + event_bus_name: str = "default", + source: str = "aws.durable-execution.insight", + region: str | None = None, + operations_format: OperationsFormat | OperationsFormatInput = ( + OperationsFormat.ARRAY + ), + max_record_size_bytes: int | None = None, + client: Any = None, + ) -> None: + self.event_bus_name = event_bus_name + self.source = source + self.operations_format = OperationsFormat(operations_format) + self.max_record_size_bytes: int | None = ( + 256_000 if max_record_size_bytes is None else max_record_size_bytes + ) + if client is not None: + self._client = client + else: + import boto3 # deferred: boto3 is provided by the Lambda runtime + + self._client = ( + boto3.client("events", region_name=region) + if region + else boto3.client("events") + ) + + def render(self, record: dict[str, Any]) -> dict[str, Any]: + return apply_operations_format(record, self.operations_format) + + def export(self, record: dict[str, Any]) -> None: + result = self._client.put_events( + Entries=[ + { + "EventBusName": self.event_bus_name, + "Source": self.source, + "DetailType": record["status"], + "Detail": compact_dumps(self.render(record)), + "Time": parse_iso_datetime(record["emittedAt"]), + } + ] + ) + if (result or {}).get("FailedEntryCount", 0) > 0: + entries = result.get("Entries") or [{}] + entry = entries[0] + msg = ( + "EventBridge PutEvents failed: " + f"{entry.get('ErrorCode')} — {entry.get('ErrorMessage')}" + ) + raise RuntimeError(msg) + + def flush(self) -> None: + return None diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/file_exporter.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/file_exporter.py new file mode 100644 index 00000000..30c4d026 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/file_exporter.py @@ -0,0 +1,88 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Filesystem Workflow Insight exporter.""" + +from __future__ import annotations + +import json +from enum import StrEnum +from pathlib import Path +from typing import Any, Literal + +from aws_durable_execution_sdk_python_insight.exporters._common import ( + compact_dumps, + sanitize, +) +from aws_durable_execution_sdk_python_insight.operations_index import ( + OperationsFormat, + OperationsFormatInput, + apply_operations_format, +) + + +class FileMode(StrEnum): + """File layout.""" + + # append every record to one ``{YYYY-MM-DD}.ndjson`` file per day (default) + NDJSON = "ndjson" + # one pretty-printed ``{executionName}.json`` per execution, overwritten + JSON = "json" + + +# Accepted string inputs, kept in lockstep with the enum values above. +FileModeInput = Literal["ndjson", "json"] + + +class FileExporter: + """Writes records under a directory (EFS mount, a mounted share, or ``/tmp``). + + ``ndjson`` appends one compact line per record to ``{directory}/{date}.ndjson`` + where the date comes from ``emittedAt``. ``json`` writes + ``{directory}/{executionName}.json`` and overwrites it on every export. + ``max_record_size_bytes`` has no default. + """ + + def __init__( + self, + directory: str | Path, + mode: FileMode | FileModeInput = FileMode.NDJSON, + operations_format: OperationsFormat | OperationsFormatInput = ( + OperationsFormat.ARRAY + ), + max_record_size_bytes: int | None = None, + ) -> None: + self.directory = Path(directory) + self.mode = FileMode(mode) + self.operations_format = OperationsFormat(operations_format) + self.max_record_size_bytes = max_record_size_bytes + self._dir_created = False + + def render(self, record: dict[str, Any]) -> dict[str, Any]: + return apply_operations_format(record, self.operations_format) + + def export(self, record: dict[str, Any]) -> None: + self._ensure_dir() + formatted = self.render(record) + if self.mode == FileMode.NDJSON: + date = str(record["emittedAt"])[:10] # YYYY-MM-DD + path = self.directory / f"{date}.ndjson" + with path.open("a", encoding="utf-8") as handle: + handle.write(compact_dumps(formatted) + "\n") + else: + file_name = ( + sanitize(record.get("executionName") or record["executionArn"]) + + ".json" + ) + (self.directory / file_name).write_text( + json.dumps(formatted, indent=2, ensure_ascii=False), encoding="utf-8" + ) + + def flush(self) -> None: + return None + + def _ensure_dir(self) -> None: + if self._dir_created: + return + self.directory.mkdir(parents=True, exist_ok=True) + self._dir_created = True diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/firehose_exporter.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/firehose_exporter.py new file mode 100644 index 00000000..0eca07dc --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/firehose_exporter.py @@ -0,0 +1,61 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Kinesis Data Firehose 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 ( + OperationsFormat, + OperationsFormatInput, + apply_operations_format, +) + + +class FirehoseExporter: + """Sends each record to a Firehose delivery stream with PutRecord. + + The data is one JSON line with a trailing newline, so records that Firehose + concatenates into a single object stay parseable as NDJSON. + """ + + def __init__( + self, + delivery_stream_name: str, + region: str | None = None, + operations_format: OperationsFormat | OperationsFormatInput = ( + OperationsFormat.ARRAY + ), + max_record_size_bytes: int | None = None, + client: Any = None, + ) -> None: + self.delivery_stream_name = delivery_stream_name + self.operations_format = OperationsFormat(operations_format) + self.max_record_size_bytes: int | None = ( + 1_000_000 if max_record_size_bytes is None else max_record_size_bytes + ) + if client is not None: + self._client = client + else: + import boto3 # deferred: boto3 is provided by the Lambda runtime + + self._client = ( + boto3.client("firehose", region_name=region) + if region + else boto3.client("firehose") + ) + + def render(self, record: dict[str, Any]) -> dict[str, Any]: + return apply_operations_format(record, self.operations_format) + + def export(self, record: dict[str, Any]) -> None: + data = (compact_dumps(self.render(record)) + "\n").encode("utf-8") + self._client.put_record( + DeliveryStreamName=self.delivery_stream_name, Record={"Data": data} + ) + + def flush(self) -> None: + return None diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/http_exporter.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/http_exporter.py new file mode 100644 index 00000000..fec78281 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/http_exporter.py @@ -0,0 +1,73 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""HTTP / webhook Workflow Insight exporter.""" + +from __future__ import annotations + +from enum import StrEnum +from typing import Any, Literal + +from aws_durable_execution_sdk_python_insight.exporters._common import ( + compact_dumps, + http_send, +) +from aws_durable_execution_sdk_python_insight.operations_index import ( + OperationsFormat, + OperationsFormatInput, + apply_operations_format, +) + + +class HttpMethod(StrEnum): + """Request method. ``PUT`` suits endpoints that upsert by URL path.""" + + POST = "POST" + PUT = "PUT" + + +# Accepted string inputs, kept in lockstep with the enum values above. +HttpMethodInput = Literal["POST", "PUT"] + + +class HttpExporter: + """Sends each record as a JSON body to any HTTP endpoint. + + The endpoint must answer 2xx; any other status raises. ``timeout_ms`` + bounds the whole request (default 10 seconds). ``max_record_size_bytes`` + has no default because a generic endpoint has no known limit. + """ + + def __init__( + self, + url: str, + headers: dict[str, str] | None = None, + method: HttpMethod | HttpMethodInput = HttpMethod.POST, + timeout_ms: int = 10_000, + operations_format: OperationsFormat | OperationsFormatInput = ( + OperationsFormat.ARRAY + ), + max_record_size_bytes: int | None = None, + ) -> None: + self.url = url + self.method = HttpMethod(method) + self.headers = dict(headers or {}) + self.timeout_ms = timeout_ms + self.operations_format = OperationsFormat(operations_format) + self.max_record_size_bytes = max_record_size_bytes + + def render(self, record: dict[str, Any]) -> dict[str, Any]: + return apply_operations_format(record, self.operations_format) + + def export(self, record: dict[str, Any]) -> None: + body = compact_dumps(self.render(record)).encode("utf-8") + headers = {"Content-Type": "application/json", **self.headers} + status, reason, _ = http_send( + self.method.value, self.url, headers, body, timeout=self.timeout_ms / 1000 + ) + if not 200 <= status < 300: + msg = f"HttpExporter: endpoint returned {status} {reason}" + raise RuntimeError(msg) + + def flush(self) -> None: + return None diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/opensearch_exporter.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/opensearch_exporter.py new file mode 100644 index 00000000..3636332b --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/opensearch_exporter.py @@ -0,0 +1,109 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""OpenSearch (Index API) Workflow Insight exporter.""" + +from __future__ import annotations + +import base64 +from enum import StrEnum +from typing import Any, Literal +from urllib.parse import quote + +from aws_durable_execution_sdk_python_insight.exporters._common import ( + compact_dumps, + http_send, +) + + +class OpenSearchAuth(StrEnum): + """Authentication method for the index request.""" + + # IAM credentials, SigV4-signed (Amazon OpenSearch Service) + SIGV4 = "sigv4" + # username / password (self-managed, or a domain with basic auth) + BASIC = "basic" + + +# Accepted string inputs, kept in lockstep with the enum values above. +OpenSearchAuthInput = Literal["sigv4", "basic"] + +# Same character set that browsers leave unescaped in URI components. +_DOC_ID_SAFE = "-_.!~*'()" + + +class OpenSearchExporter: + """Indexes each record as one document, keyed by ``executionArn``. + + A later export for the same execution overwrites the document. Supports + SigV4 (IAM) and basic authentication; no OpenSearch client library needed. + """ + + def __init__( + self, + endpoint: str, + region: str, + index_name: str = "workflow-insight", + auth: OpenSearchAuth | OpenSearchAuthInput = OpenSearchAuth.SIGV4, + username: str | None = None, + password: str | None = None, + max_record_size_bytes: int | None = None, + ) -> None: + self.endpoint = endpoint[:-1] if endpoint.endswith("/") else endpoint + self.index_name = index_name + self.region = region + self.auth = OpenSearchAuth(auth) + if self.auth == OpenSearchAuth.BASIC and (username is None or password is None): + msg = "OpenSearchExporter: auth='basic' requires username and password." + raise ValueError(msg) + self.username = username + self.password = password + self.max_record_size_bytes: int | None = ( + 10_000_000 if max_record_size_bytes is None else max_record_size_bytes + ) + self._credentials: Any = None + + def render(self, record: dict[str, Any]) -> dict[str, Any]: + return record + + def export(self, record: dict[str, Any]) -> None: + doc_id = quote(record["executionArn"], safe=_DOC_ID_SAFE) + url = f"{self.endpoint}/{self.index_name}/_doc/{doc_id}" + body = compact_dumps(record).encode("utf-8") + headers: dict[str, str] = {"Content-Type": "application/json"} + + if self.auth == OpenSearchAuth.BASIC: + token = base64.b64encode( + f"{self.username}:{self.password}".encode() + ).decode("ascii") + headers["Authorization"] = f"Basic {token}" + else: + # The signed header set is sent as-is: it already carries + # content-type, host, date, token and authorization. + headers = self._sign(url, body, headers) + + status, reason, detail = http_send("PUT", url, headers, body) + if not 200 <= status < 300: + msg = f"OpenSearch index failed: {status} {reason}" + if detail: + msg += f" — {detail[:500]}" + raise RuntimeError(msg) + + def flush(self) -> None: + return None + + def _sign(self, url: str, body: bytes, headers: dict[str, str]) -> dict[str, str]: + # deferred: botocore is provided by the Lambda runtime + from botocore.auth import SigV4Auth + from botocore.awsrequest import AWSRequest + + if self._credentials is None: + import botocore.session + + self._credentials = botocore.session.get_session().get_credentials() + if self._credentials is None: + msg = "OpenSearchExporter: no AWS credentials found for SigV4 signing." + raise RuntimeError(msg) + request = AWSRequest(method="PUT", url=url, data=body, headers=headers) + SigV4Auth(self._credentials, "es", self.region).add_auth(request) + return dict(request.headers.items()) diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/otel_exporter.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/otel_exporter.py new file mode 100644 index 00000000..001f105a --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/otel_exporter.py @@ -0,0 +1,151 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""OpenTelemetry (OTLP/HTTP logs) Workflow Insight exporter.""" + +from __future__ import annotations + +import datetime +from enum import StrEnum +from typing import Any, Literal + +from aws_durable_execution_sdk_python_insight.exporters._common import ( + compact_dumps, + http_send, + parse_iso_datetime, +) +from aws_durable_execution_sdk_python_insight.operations_index import ( + OperationsFormat, + OperationsFormatInput, + apply_operations_format, +) + + +class OTelProtocol(StrEnum): + """OTLP transport encoding.""" + + HTTP_JSON = "http/json" + HTTP_PROTOBUF = "http/protobuf" + + +# Accepted string inputs, kept in lockstep with the enum values above. +OTelProtocolInput = Literal["http/json", "http/protobuf"] + +_SCOPE_NAME = "aws-durable-execution-sdk-python-insight" +_SEVERITY = {"FAILED": 17, "RUNNING": 9, "SUCCEEDED": 9} # ERROR / INFO / INFO +_EPOCH = datetime.datetime(1970, 1, 1, tzinfo=datetime.UTC) + + +def _kv(key: str, value: str | int) -> dict[str, Any]: + if isinstance(value, int): + return {"key": key, "value": {"intValue": str(value)}} + return {"key": key, "value": {"stringValue": value}} + + +def _to_nano(iso: str) -> str: + # Keep the record's full microsecond precision (integer math, no float). + delta = parse_iso_datetime(iso) - _EPOCH + seconds = delta.days * 86_400 + delta.seconds + return str(seconds * 1_000_000_000 + delta.microseconds * 1_000) + + +class OTelExporter: + """Posts each record as one OTLP log record to an OTLP/HTTP endpoint. + + Record identity fields become resource and log attributes; the record + itself, rendered per ``operations_format``, is the log body. Only + ``http/json`` is supported; ``http/protobuf`` raises at construction. + """ + + def __init__( + self, + endpoint: str, + headers: dict[str, str] | None = None, + protocol: OTelProtocol | OTelProtocolInput = OTelProtocol.HTTP_JSON, + operations_format: OperationsFormat | OperationsFormatInput = ( + OperationsFormat.ARRAY + ), + max_record_size_bytes: int | None = None, + ) -> None: + if OTelProtocol(protocol) == OTelProtocol.HTTP_PROTOBUF: + msg = "OTelExporter: http/protobuf is not yet supported. Use http/json." + raise ValueError(msg) + self.endpoint = endpoint + self.headers = dict(headers or {}) + self.operations_format = OperationsFormat(operations_format) + self.max_record_size_bytes: int | None = ( + 1_000_000 if max_record_size_bytes is None else max_record_size_bytes + ) + + def render(self, record: dict[str, Any]) -> dict[str, Any]: + # The whole request is measured: the record sits inside body.stringValue, + # so the size limit covers both the record and the OTLP envelope. + return self._build_payload(record) + + 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) + if not 200 <= status < 300: + msg = f"OTelExporter: OTLP endpoint returned {status} {reason}" + raise RuntimeError(msg) + + def flush(self) -> None: + return None + + def _build_payload(self, record: dict[str, Any]) -> dict[str, Any]: + """Build an OTLP ``ExportLogsServiceRequest``.""" + function_name = record.get("functionName", "") + status = record.get("status", "") + return { + "resourceLogs": [ + { + "resource": { + "attributes": [ + _kv("service.name", function_name), + _kv("cloud.region", record.get("region", "")), + _kv("cloud.account.id", record.get("accountId", "")), + _kv("faas.name", function_name), + _kv("faas.version", record.get("functionQualifier", "")), + ] + }, + "scopeLogs": [ + { + "scope": { + "name": _SCOPE_NAME, + "version": record.get("schemaVersion", ""), + }, + "logRecords": [ + { + "timeUnixNano": _to_nano(record["emittedAt"]), + "severityNumber": _SEVERITY.get(status, 0), + "severityText": status, + "body": { + "stringValue": compact_dumps( + apply_operations_format( + record, self.operations_format + ) + ) + }, + "attributes": [ + _kv( + "workflow.execution_arn", + record["executionArn"], + ), + _kv( + "workflow.execution_name", + record.get("executionName") or "", + ), + _kv("workflow.status", status), + _kv( + "workflow.duration_ms", + record.get("durationMs") or 0, + ), + ], + } + ], + } + ], + } + ] + } diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/redshift_exporter.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/redshift_exporter.py new file mode 100644 index 00000000..d7154a1d --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/redshift_exporter.py @@ -0,0 +1,155 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Redshift (Data API) Workflow Insight exporter.""" + +from __future__ import annotations + +from typing import Any + +from aws_durable_execution_sdk_python_insight.exporters._common import ( + compact_dumps, + sql_identifier, +) + + +_COLUMNS = ( + "execution_arn, execution_name, function_name, status, start_time, " + "end_time, duration_ms, record_json, emitted_at" +) + + +class RedshiftExporter: + """Upserts one row per execution through the Redshift Data API. + + Works with Redshift Serverless (``workgroup_name``) and provisioned + clusters (``cluster_identifier``). Rows are merged on ``execution_arn`` and + the full record lands in the ``record_json`` SUPER column. The statement is + submitted and not awaited. + """ + + def __init__( + self, + database: str, + workgroup_name: str | None = None, + cluster_identifier: str | None = None, + db_user: str | None = None, + secret_arn: str | None = None, + table: str = "workflow_insight", + schema: str = "public", + region: str | None = None, + max_record_size_bytes: int | None = None, + client: Any = None, + ) -> None: + if bool(workgroup_name) == bool(cluster_identifier): + msg = ( + "RedshiftExporter: provide exactly one of workgroup_name or " + "cluster_identifier." + ) + raise ValueError(msg) + self.database = database + self.fq_table = f"{sql_identifier(schema)}.{sql_identifier(table)}" + self.workgroup_name = workgroup_name + self.cluster_identifier = cluster_identifier + self.db_user = db_user + self.secret_arn = secret_arn + self.max_record_size_bytes: int | None = ( + 1_000_000 if max_record_size_bytes is None else max_record_size_bytes + ) + if client is not None: + self._client = client + else: + import boto3 # deferred: boto3 is provided by the Lambda runtime + + self._client = ( + boto3.client("redshift-data", region_name=region) + if region + else boto3.client("redshift-data") + ) + + def render(self, record: dict[str, Any]) -> dict[str, Any]: + return record + + def export(self, record: dict[str, Any]) -> None: + parameters: list[dict[str, str]] = [ + {"name": "execution_arn", "value": record["executionArn"]}, + {"name": "function_name", "value": record["functionName"]}, + {"name": "status", "value": record["status"]}, + {"name": "start_time", "value": record["startTime"]}, + {"name": "record_json", "value": compact_dumps(record)}, + {"name": "emitted_at", "value": record["emittedAt"]}, + ] + + # The Data API takes neither NULL nor empty-string parameter values, so + # an absent nullable field becomes a typed NULL literal in the source + # projection instead of a bound parameter. + if record.get("endTime"): + parameters.append({"name": "end_time", "value": record["endTime"]}) + end_time_sel = ":end_time::timestamptz" + else: + end_time_sel = "NULL::timestamptz" + if record.get("durationMs") is not None: + parameters.append( + {"name": "duration_ms", "value": str(record["durationMs"])} + ) + duration_sel = ":duration_ms::bigint" + else: + duration_sel = "NULL::bigint" + if record.get("executionName"): + parameters.append( + {"name": "execution_name", "value": record["executionName"]} + ) + exec_name_sel = ":execution_name::varchar" + else: + exec_name_sel = "NULL::varchar" + + # The source row is a subquery and MERGE joins on a source column: + # Redshift rejects a MERGE whose join is on a parameter or constant + # (NestedLoop). Time columns are cast to timestamptz and record_json is + # JSON_PARSEd into the SUPER column. + sql = ( + f"MERGE INTO {self.fq_table} USING (\n" + " SELECT\n" + " :execution_arn::varchar AS execution_arn,\n" + f" {exec_name_sel} AS execution_name,\n" + " :function_name::varchar AS function_name,\n" + " :status::varchar AS status,\n" + " :start_time::timestamptz AS start_time,\n" + f" {end_time_sel} AS end_time,\n" + f" {duration_sel} AS duration_ms,\n" + " JSON_PARSE(:record_json) AS record_json,\n" + " :emitted_at::timestamptz AS emitted_at\n" + " ) AS src\n" + f" ON {self.fq_table}.execution_arn = src.execution_arn\n" + " WHEN MATCHED THEN UPDATE SET\n" + " status = src.status,\n" + " end_time = src.end_time,\n" + " duration_ms = src.duration_ms,\n" + " record_json = src.record_json,\n" + " emitted_at = src.emitted_at\n" + " WHEN NOT MATCHED THEN INSERT\n" + f" ({_COLUMNS})\n" + " VALUES\n" + " (src.execution_arn, src.execution_name, src.function_name, " + "src.status, src.start_time, src.end_time, src.duration_ms, " + "src.record_json, src.emitted_at)" + ) + + kwargs: dict[str, Any] = { + "Database": self.database, + "Sql": sql, + "Parameters": parameters, + } + # Unset optional targets are omitted rather than passed as None. + if self.workgroup_name: + kwargs["WorkgroupName"] = self.workgroup_name + if self.cluster_identifier: + kwargs["ClusterIdentifier"] = self.cluster_identifier + if self.db_user: + kwargs["DbUser"] = self.db_user + if self.secret_arn: + kwargs["SecretArn"] = self.secret_arn + self._client.execute_statement(**kwargs) + + def flush(self) -> None: + return None diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/s3_exporter.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/s3_exporter.py index 2aca47ae..3fc69e09 100644 --- a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/s3_exporter.py +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/s3_exporter.py @@ -57,7 +57,7 @@ def __init__( # string so an invalid scheme fails at construction rather than silently # falling through to no partitioning. self.partitioning = S3Partitioning(partitioning) - self.max_record_size_bytes = ( + self.max_record_size_bytes: int | None = ( 5_000_000 if max_record_size_bytes is None else max_record_size_bytes ) if client is not None: diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/sqs_exporter.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/sqs_exporter.py new file mode 100644 index 00000000..f9e11fea --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/sqs_exporter.py @@ -0,0 +1,94 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""SQS Workflow Insight exporter.""" + +from __future__ import annotations + +import hashlib +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 ( + OperationsFormat, + OperationsFormatInput, + apply_operations_format, +) + + +# SQS caps MessageGroupId and MessageDeduplicationId at 128 characters. +_MAX_ID_LENGTH = 128 + + +class SQSExporter: + """Sends each record as one SQS message with SendMessage. + + The rendered record is the message body; ``status`` and ``functionName`` + are message attributes. On a FIFO queue (URL ending in ``.fifo``) the group + id defaults to ``executionArn`` and the deduplication id is + ``executionArn:emittedAt``; an id longer than 128 characters is replaced by + its SHA-256 hex digest so the message is never rejected. + """ + + def __init__( + self, + queue_url: str, + message_group_id: str | None = None, + region: str | None = None, + operations_format: OperationsFormat | OperationsFormatInput = ( + OperationsFormat.ARRAY + ), + max_record_size_bytes: int | None = None, + client: Any = None, + ) -> None: + self.queue_url = queue_url + self.message_group_id = message_group_id + self.is_fifo = queue_url.endswith(".fifo") + self.operations_format = OperationsFormat(operations_format) + self.max_record_size_bytes: int | None = ( + 256_000 if max_record_size_bytes is None else max_record_size_bytes + ) + if client is not None: + self._client = client + else: + import boto3 # deferred: boto3 is provided by the Lambda runtime + + self._client = ( + boto3.client("sqs", region_name=region) + if region + else boto3.client("sqs") + ) + + def render(self, record: dict[str, Any]) -> dict[str, Any]: + return apply_operations_format(record, self.operations_format) + + def export(self, record: dict[str, Any]) -> None: + kwargs: dict[str, Any] = { + "QueueUrl": self.queue_url, + "MessageBody": compact_dumps(self.render(record)), + "MessageAttributes": { + "status": {"DataType": "String", "StringValue": record["status"]}, + "functionName": { + "DataType": "String", + "StringValue": record["functionName"], + }, + }, + } + if self.is_fifo: + kwargs["MessageGroupId"] = _bounded_id( + self.message_group_id or record["executionArn"] + ) + kwargs["MessageDeduplicationId"] = _bounded_id( + f"{record['executionArn']}:{record['emittedAt']}" + ) + self._client.send_message(**kwargs) + + def flush(self) -> None: + return None + + +def _bounded_id(value: str) -> str: + """Return ``value`` if it fits the SQS 128-character id limit, else its SHA-256.""" + if len(value) <= _MAX_ID_LENGTH: + return value + return hashlib.sha256(value.encode("utf-8")).hexdigest() diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/operations_index.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/operations_index.py index e348b4cd..a83bbaa4 100644 --- a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/operations_index.py +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/operations_index.py @@ -14,7 +14,23 @@ from __future__ import annotations -from typing import Any +from enum import StrEnum +from typing import Any, Literal + + +class OperationsFormat(StrEnum): + """How an exporter renders operations in the emitted record.""" + + # the canonical ``operations`` array (lossless; default) + ARRAY = "array" + # replace it with the name-keyed ``operationsByName`` summary map + BY_NAME = "by-name" + # include both the array and the map + BOTH = "both" + + +# Accepted string inputs, kept in lockstep with the enum values above. +OperationsFormatInput = Literal["array", "by-name", "both"] def build_operations_by_name( @@ -96,3 +112,21 @@ def with_operations_by_name(record: dict[str, Any]) -> dict[str, Any]: out = {key: value for key, value in record.items() if key != "operations"} out["operationsByName"] = build_operations_by_name(record.get("operations", [])) return out + + +def apply_operations_format( + record: dict[str, Any], operations_format: OperationsFormat | OperationsFormatInput +) -> dict[str, Any]: + """Render ``operations`` as an array, a name-keyed map, or both. + + ``"array"`` returns the record unchanged. ``"by-name"`` replaces the array + with ``operationsByName``. ``"both"`` keeps the array and adds the map. + """ + fmt = OperationsFormat(operations_format) + if fmt == OperationsFormat.BY_NAME: + return with_operations_by_name(record) + if fmt == OperationsFormat.BOTH: + out = dict(record) + out["operationsByName"] = build_operations_by_name(record.get("operations", [])) + return out + return record diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/types.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/types.py index 82426609..f813931c 100644 --- a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/types.py +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/types.py @@ -49,7 +49,9 @@ class InsightExporter(Protocol): """A destination that receives one curated Workflow Insight record. ``max_record_size_bytes`` bounds the serialized record body (the plugin's - size limiter measures ``render(record)``); ``None`` disables truncation. + size limiter measures ``render(record)``). First-party exporters default it + to their destination's practical limit; an exporter whose value is ``None`` + is never truncated. ``render`` maps the canonical record dict to the exact shape the exporter serializes (identity for array exporters, the ``operationsByName`` expansion for point-access exporters). diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/e2e/exporters_lifecycle_int_test.py b/packages/aws-durable-execution-sdk-python-insight/tests/e2e/exporters_lifecycle_int_test.py new file mode 100644 index 00000000..9f60ff0a --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/e2e/exporters_lifecycle_int_test.py @@ -0,0 +1,130 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""End-to-end test: first-party exporters receive records through the real plugin. + +Drives the plugin through the repository's LOCAL durable runner +(``DurableFunctionTestRunner``) and the real ``@durable_execution`` lifecycle +with three shipped exporters attached at once: + +* ``FileExporter`` in ``ndjson`` mode with ``operations_format="by-name"`` and + ``on-change`` emission, so each delivered snapshot and the terminal record + land as separate lines in one dated file (the export worker may coalesce + RUNNING snapshots, so their count is not asserted); +* ``FileExporter`` in ``json`` mode, so the per-execution file is overwritten + and ends holding only the terminal record; +* ``SQSExporter`` with an injected recording client and a small + ``max_record_size_bytes``, so the plugin's size limiter truncates the copy + that exporter receives without touching the copies the others receive. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Callable + +from aws_durable_execution_sdk_python.context import DurableContext +from aws_durable_execution_sdk_python.execution import ( + InvocationStatus, + durable_execution, +) + +from aws_durable_execution_sdk_python_insight import ( + FileExporter, + SQSExporter, + WorkflowInsightConfig, + workflow_insight, +) +from aws_durable_execution_sdk_python_testing.runner import ( + DurableFunctionTestResult, + DurableFunctionTestRunner, +) + + +_STEP_NAMES = ("first", "second", "third") + + +class _RecordingSqsClient: + def __init__(self) -> None: + self.sends: list[dict[str, Any]] = [] + + def send_message(self, **kwargs: Any) -> dict[str, Any]: + self.sends.append(kwargs) + return {"MessageId": "m"} + + +def _three_steps_handler(event: Any, context: DurableContext) -> str: # noqa: ARG001 + for name in _STEP_NAMES: + context.step(_step_returning(name), name=name) + return "done" + + +def _step_returning(value: str) -> Callable[[Any], str]: + return lambda _step_ctx: value + + +def test_exporters_receive_records_through_the_plugin_lifecycle( + tmp_path: Path, +) -> None: + ndjson_dir = tmp_path / "ndjson" + json_dir = tmp_path / "json" + sqs_client = _RecordingSqsClient() + sqs = SQSExporter( + queue_url="https://sqs.us-east-1.amazonaws.com/123456789012/insight", + max_record_size_bytes=800, + client=sqs_client, + ) + plugin = workflow_insight( + WorkflowInsightConfig( + exporters=[ + FileExporter(directory=ndjson_dir, operations_format="by-name"), + FileExporter(directory=json_dir, mode="json"), + sqs, + ], + emit_mode="on-change", + ) + ) + handler = durable_execution(_three_steps_handler, plugins=[plugin]) + + with DurableFunctionTestRunner(handler=handler, execution_timeout=15) as runner: + result: DurableFunctionTestResult = runner.run(input="{}") + assert result.status is InvocationStatus.SUCCEEDED + + # ndjson + by-name: one line per delivered emission, name-keyed operations, + # terminal last. The export worker coalesces pending on-change snapshots, so + # the number of RUNNING lines is not fixed; every line must still be well + # formed and the terminal record must be the last one written. + ndjson_files = list(ndjson_dir.iterdir()) + 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) >= 1 + assert all("operations" not in r and "operationsByName" in r for r in records) + assert {r["status"] for r in records[:-1]} <= {"RUNNING"} + terminal = records[-1] + assert terminal["status"] == "SUCCEEDED" + assert set(terminal["operationsByName"]) == set(_STEP_NAMES) + assert all( + terminal["operationsByName"][n]["status"] == "SUCCEEDED" for n in _STEP_NAMES + ) + + # json mode: overwritten per emission, so the file holds the terminal record. + json_files = list(json_dir.iterdir()) + assert len(json_files) == 1 + final = json.loads(json_files[0].read_text(encoding="utf-8")) + assert final["status"] == "SUCCEEDED" + assert [op["name"] for op in final["operations"]] == list(_STEP_NAMES) + assert final["executionArn"] == terminal["executionArn"] + + # SQS with a small limit: the same emissions arrive, and the terminal copy is + # truncated to fit while the file copies above stayed intact. + assert len(sqs_client.sends) == len(records) + sqs_terminal = json.loads(sqs_client.sends[-1]["MessageBody"]) + assert sqs_terminal["status"] == "SUCCEEDED" + assert sqs_terminal.get("truncated") is True + assert len(sqs_client.sends[-1]["MessageBody"].encode("utf-8")) <= 800 + assert "truncated" not in final + assert sqs_client.sends[-1]["MessageAttributes"]["status"]["StringValue"] == ( + "SUCCEEDED" + ) diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_aurora_exporter.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_aurora_exporter.py new file mode 100644 index 00000000..728892bf --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_aurora_exporter.py @@ -0,0 +1,158 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Unit tests for ``AuroraExporter`` (fake RDS Data API client, no AWS).""" + +from __future__ import annotations + +import json +from typing import Any + +import pytest + +from aws_durable_execution_sdk_python_insight import AuroraEngine, AuroraExporter +from aws_durable_execution_sdk_python_insight.exporters.aurora_exporter import ( + AuroraExporter as AuroraExporterFromModule, +) + + +def _record(**overrides: Any) -> dict[str, Any]: + """A complete SUCCEEDED record; keyword arguments override fields.""" + record: dict[str, Any] = { + "recordType": "WorkflowInsight", + "schemaVersion": "1.0", + "emittedAt": "2026-07-15T12:00:00.000Z", + "executionArn": ( + "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST" + "/durable-execution/my-exec/inv-1" + ), + "executionName": "my-exec", + "functionName": "fn", + "functionQualifier": "$LATEST", + "region": "us-east-1", + "accountId": "123456789012", + "status": "SUCCEEDED", + "startTime": "2026-07-15T11:59:58.000Z", + "endTime": "2026-07-15T12:00:00.000Z", + "durationMs": 2000, + "operations": [ + { + "id": "op-1", + "name": "fetch-user", + "type": "STEP", + "subType": "Step", + "status": "SUCCEEDED", + "durationMs": 12, + } + ], + } + record.update(overrides) + return record + + +class FakeRdsDataClient: + def __init__(self) -> None: + self.calls: list[dict[str, Any]] = [] + + def execute_statement(self, **kwargs: Any) -> dict[str, Any]: + self.calls.append(kwargs) + return {} + + +def _params(call: dict[str, Any]) -> dict[str, dict[str, Any]]: + return {p["name"]: p["value"] for p in call["parameters"]} + + +def _exporter(client: FakeRdsDataClient, **kw: Any) -> AuroraExporter: + base: dict[str, Any] = { + "resource_arn": "arn:aws:rds:us-east-1:123456789012:cluster:c", + "secret_arn": "arn:aws:secretsmanager:us-east-1:123456789012:secret:s", + "database": "insight", + "engine": "postgresql", + "client": client, + } + base.update(kw) + return AuroraExporter(**base) + + +def test_public_import_path_and_defaults() -> None: + assert AuroraExporter is AuroraExporterFromModule + exporter = _exporter(FakeRdsDataClient()) + assert exporter.max_record_size_bytes == 1_000_000 + assert exporter.table == "workflow_insight" + assert exporter.engine is AuroraEngine.POSTGRESQL + assert _exporter(FakeRdsDataClient(), engine=AuroraEngine.MYSQL).engine == "mysql" + exporter.flush() # no buffering: a no-op + + +def test_postgres_upsert_with_typed_casts_and_bound_parameters() -> None: + client = FakeRdsDataClient() + record = _record() + exporter = _exporter(client) + assert exporter.render(record) is record + exporter.export(record) + + assert len(client.calls) == 1 + call = client.calls[0] + assert call["resourceArn"] == "arn:aws:rds:us-east-1:123456789012:cluster:c" + assert call["secretArn"] == "arn:aws:secretsmanager:us-east-1:123456789012:secret:s" + assert call["database"] == "insight" + sql = call["sql"] + assert "INSERT INTO workflow_insight" in sql + assert "ON CONFLICT (execution_arn) DO UPDATE" in sql + assert ":start_time::timestamptz" in sql + assert ":record_json::jsonb" in sql + + params = _params(call) + assert [p["name"] for p in call["parameters"]] == [ + "execution_arn", + "execution_name", + "function_name", + "status", + "start_time", + "end_time", + "duration_ms", + "record_json", + "emitted_at", + ] + assert params["execution_arn"] == {"stringValue": record["executionArn"]} + assert params["execution_name"] == {"stringValue": "my-exec"} + assert params["function_name"] == {"stringValue": "fn"} + assert params["status"] == {"stringValue": "SUCCEEDED"} + assert params["start_time"] == {"stringValue": "2026-07-15T11:59:58.000Z"} + assert params["end_time"] == {"stringValue": "2026-07-15T12:00:00.000Z"} + assert params["duration_ms"] == {"longValue": 2000} + assert params["emitted_at"] == {"stringValue": "2026-07-15T12:00:00.000Z"} + record_json = params["record_json"]["stringValue"] + assert ", " not in record_json and '": ' not in record_json + assert json.loads(record_json) == record + + +def test_mysql_dialect_has_no_casts_and_custom_table() -> None: + client = FakeRdsDataClient() + _exporter(client, engine="mysql", table="custom_table").export(_record()) + sql = client.calls[0]["sql"] + assert "INSERT INTO custom_table" in sql + assert "ON DUPLICATE KEY UPDATE" in sql + assert "::timestamptz" not in sql + assert "ON CONFLICT" not in sql + + +def test_absent_nullable_fields_are_is_null() -> None: + client = FakeRdsDataClient() + record = _record(status="RUNNING") + del record["executionName"] + del record["endTime"] + del record["durationMs"] + _exporter(client).export(record) + params = _params(client.calls[0]) + assert params["execution_name"] == {"isNull": True} + assert params["end_time"] == {"isNull": True} + assert params["duration_ms"] == {"isNull": True} + + +def test_invalid_identifier_or_engine_fails_at_construction() -> None: + with pytest.raises(ValueError, match="Invalid SQL identifier"): + _exporter(FakeRdsDataClient(), table="bad-table; drop") + with pytest.raises(ValueError): + _exporter(FakeRdsDataClient(), engine="oracle") diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_cloudwatch_logs_exporter.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_cloudwatch_logs_exporter.py new file mode 100644 index 00000000..f8cd092a --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_cloudwatch_logs_exporter.py @@ -0,0 +1,155 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Unit tests for ``CloudWatchLogsExporter`` (fake client, no AWS).""" + +from __future__ import annotations + +import datetime +import json +import re +from typing import Any + +import pytest + +from aws_durable_execution_sdk_python_insight import CloudWatchLogsExporter +from aws_durable_execution_sdk_python_insight.exporters.cloudwatch_logs_exporter import ( + CloudWatchLogsExporter as CloudWatchLogsExporterFromModule, +) + + +def _record(**overrides: Any) -> dict[str, Any]: + """A complete SUCCEEDED record; keyword arguments override fields.""" + record: dict[str, Any] = { + "recordType": "WorkflowInsight", + "schemaVersion": "1.0", + "emittedAt": "2026-07-15T12:00:00.000Z", + "executionArn": ( + "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST" + "/durable-execution/my-exec/inv-1" + ), + "executionName": "my-exec", + "functionName": "fn", + "functionQualifier": "$LATEST", + "region": "us-east-1", + "accountId": "123456789012", + "status": "SUCCEEDED", + "startTime": "2026-07-15T11:59:58.000Z", + "endTime": "2026-07-15T12:00:00.000Z", + "durationMs": 2000, + "operations": [ + { + "id": "op-1", + "name": "fetch-user", + "type": "STEP", + "subType": "Step", + "status": "SUCCEEDED", + "durationMs": 12, + } + ], + } + record.update(overrides) + return record + + +class ResourceAlreadyExistsException(Exception): + """Stand-in for the modeled botocore exception (matched by class name).""" + + +class GenericClientError(Exception): + def __init__(self, code: str) -> None: + super().__init__(code) + self.response = {"Error": {"Code": code, "Message": code}} + + +class FakeLogsClient: + def __init__(self, create_error: Exception | None = None) -> None: + self.creates: list[dict[str, Any]] = [] + self.puts: list[dict[str, Any]] = [] + self._create_error = create_error + + def create_log_stream(self, **kwargs: Any) -> None: + self.creates.append(kwargs) + if self._create_error is not None: + raise self._create_error + + def put_log_events(self, **kwargs: Any) -> dict[str, Any]: + self.puts.append(kwargs) + return {} + + +def test_public_import_path_and_defaults() -> None: + assert CloudWatchLogsExporter is CloudWatchLogsExporterFromModule + exporter = CloudWatchLogsExporter(log_group_name="/g", client=FakeLogsClient()) + assert exporter.max_record_size_bytes == 256_000 + assert exporter.log_stream_prefix == "workflow-insight/" + exporter.flush() # no buffering: a no-op + + +def test_creates_dated_stream_then_puts_operations_by_name_event() -> None: + client = FakeLogsClient() + exporter = CloudWatchLogsExporter(log_group_name="/insight/records", client=client) + before_ms = int(datetime.datetime.now(datetime.UTC).timestamp() * 1000) + exporter.export(_record()) + + assert len(client.creates) == 1 + assert len(client.puts) == 1 + create = client.creates[0] + assert create["logGroupName"] == "/insight/records" + assert re.fullmatch(r"workflow-insight/\d{4}/\d{2}/\d{2}", create["logStreamName"]) + today = datetime.datetime.now(datetime.UTC).strftime("%Y/%m/%d") + assert create["logStreamName"] == f"workflow-insight/{today}" + + put = client.puts[0] + assert put["logGroupName"] == "/insight/records" + assert put["logStreamName"] == create["logStreamName"] + assert len(put["logEvents"]) == 1 + event = put["logEvents"][0] + assert isinstance(event["timestamp"], int) + assert event["timestamp"] >= before_ms + message = event["message"] + assert ", " not in message and '": ' not in message + parsed = json.loads(message) + assert parsed["operationsByName"]["fetch-user"]["count"] == 1 + assert "operations" not in parsed + + +def test_custom_prefix_is_used_in_stream_name() -> None: + client = FakeLogsClient() + CloudWatchLogsExporter( + log_group_name="/g", log_stream_prefix="wi-", client=client + ).export(_record()) + assert client.creates[0]["logStreamName"].startswith("wi-") + + +def test_stream_is_created_once_across_exports() -> None: + client = FakeLogsClient() + exporter = CloudWatchLogsExporter(log_group_name="/g", client=client) + exporter.export(_record()) + exporter.export(_record()) + assert len(client.creates) == 1 + assert len(client.puts) == 2 + + +@pytest.mark.parametrize( + "error", + [ + ResourceAlreadyExistsException("exists"), + GenericClientError("ResourceAlreadyExistsException"), + ], +) +def test_already_exists_from_create_is_swallowed(error: Exception) -> None: + client = FakeLogsClient(create_error=error) + exporter = CloudWatchLogsExporter(log_group_name="/g", client=client) + exporter.export(_record()) + exporter.export(_record()) + assert len(client.creates) == 1 + assert len(client.puts) == 2 + + +def test_other_create_errors_propagate() -> None: + client = FakeLogsClient(create_error=GenericClientError("AccessDeniedException")) + exporter = CloudWatchLogsExporter(log_group_name="/g", client=client) + with pytest.raises(GenericClientError): + exporter.export(_record()) + assert client.puts == [] diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_dynamodb_exporter.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_dynamodb_exporter.py new file mode 100644 index 00000000..7931b73b --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_dynamodb_exporter.py @@ -0,0 +1,118 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Unit tests for ``DynamoDBExporter`` (fake client, no AWS).""" + +from __future__ import annotations + +from decimal import Decimal +from typing import Any + +from aws_durable_execution_sdk_python_insight import DynamoDBExporter +from aws_durable_execution_sdk_python_insight.exporters.dynamodb_exporter import ( + DynamoDBExporter as DynamoDBExporterFromModule, +) + + +def _record(**overrides: Any) -> dict[str, Any]: + """A complete SUCCEEDED record; keyword arguments override fields.""" + record: dict[str, Any] = { + "recordType": "WorkflowInsight", + "schemaVersion": "1.0", + "emittedAt": "2026-07-15T12:00:00.000Z", + "executionArn": ( + "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST" + "/durable-execution/my-exec/inv-1" + ), + "executionName": "my-exec", + "functionName": "fn", + "functionQualifier": "$LATEST", + "region": "us-east-1", + "accountId": "123456789012", + "status": "SUCCEEDED", + "startTime": "2026-07-15T11:59:58.000Z", + "endTime": "2026-07-15T12:00:00.000Z", + "durationMs": 2000, + "operations": [ + { + "id": "op-1", + "name": "fetch-user", + "type": "STEP", + "subType": "Step", + "status": "SUCCEEDED", + "durationMs": 12, + } + ], + } + record.update(overrides) + return record + + +class FakeDynamoDBClient: + def __init__(self) -> None: + self.puts: list[dict[str, Any]] = [] + + def put_item(self, **kwargs: Any) -> None: + self.puts.append(kwargs) + + +def test_public_import_path() -> None: + assert DynamoDBExporter is DynamoDBExporterFromModule + + +def test_defaults_and_render_is_operations_by_name() -> None: + exporter = DynamoDBExporter(table_name="insight", client=FakeDynamoDBClient()) + assert exporter.max_record_size_bytes == 400_000 + assert exporter.partition_key == "pk" + assert exporter.sort_key == "sk" + shaped = exporter.render(_record()) + assert "operations" not in shaped + assert shaped["operationsByName"]["fetch-user"]["count"] == 1 + exporter.flush() # no buffering: a no-op + + +def test_export_writes_history_item_keyed_by_arn_and_emitted_at() -> None: + client = FakeDynamoDBClient() + record = _record() + DynamoDBExporter(table_name="insight", client=client).export(record) + + assert len(client.puts) == 1 + put = client.puts[0] + assert put["TableName"] == "insight" + item = put["Item"] + assert item["pk"] == {"S": record["executionArn"]} + assert item["sk"] == {"S": "2026-07-15T12:00:00.000Z"} + assert "operations" not in item + assert item["durationMs"] == {"N": "2000"} + summary = item["operationsByName"]["M"]["fetch-user"]["M"] + assert summary["count"] == {"N": "1"} + assert summary["type"] == {"S": "STEP"} + + +def test_export_upserts_without_sort_key_and_custom_partition_key() -> None: + client = FakeDynamoDBClient() + exporter = DynamoDBExporter( + table_name="insight", + partition_key="executionArnKey", + sort_key="", + client=client, + ) + assert exporter.sort_key is None + record = _record() + exporter.export(record) + item = client.puts[0]["Item"] + assert item["executionArnKey"] == {"S": record["executionArn"]} + assert "sk" not in item + + +def test_export_marshals_floats_as_numbers() -> None: + client = FakeDynamoDBClient() + DynamoDBExporter(table_name="insight", client=client).export( + _record(output={"ratio": 0.25, "n": 3, "ok": True, "none": None}) + ) + output = client.puts[0]["Item"]["output"]["M"] + assert output["ratio"] == {"N": "0.25"} + assert Decimal(output["ratio"]["N"]) == Decimal("0.25") + assert output["n"] == {"N": "3"} + assert output["ok"] == {"BOOL": True} + assert output["none"] == {"NULL": True} diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_eventbridge_exporter.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_eventbridge_exporter.py new file mode 100644 index 00000000..40ca19b9 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_eventbridge_exporter.py @@ -0,0 +1,126 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Unit tests for ``EventBridgeExporter`` (fake client, no AWS).""" + +from __future__ import annotations + +import datetime +import json +from typing import Any + +import pytest + +from aws_durable_execution_sdk_python_insight import EventBridgeExporter +from aws_durable_execution_sdk_python_insight.exporters.eventbridge_exporter import ( + EventBridgeExporter as EventBridgeExporterFromModule, +) + + +def _record(**overrides: Any) -> dict[str, Any]: + """A complete SUCCEEDED record; keyword arguments override fields.""" + record: dict[str, Any] = { + "recordType": "WorkflowInsight", + "schemaVersion": "1.0", + "emittedAt": "2026-07-15T12:00:00.000Z", + "executionArn": ( + "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST" + "/durable-execution/my-exec/inv-1" + ), + "executionName": "my-exec", + "functionName": "fn", + "functionQualifier": "$LATEST", + "region": "us-east-1", + "accountId": "123456789012", + "status": "SUCCEEDED", + "startTime": "2026-07-15T11:59:58.000Z", + "endTime": "2026-07-15T12:00:00.000Z", + "durationMs": 2000, + "operations": [ + { + "id": "op-1", + "name": "fetch-user", + "type": "STEP", + "subType": "Step", + "status": "SUCCEEDED", + "durationMs": 12, + } + ], + } + record.update(overrides) + return record + + +class FakeEventsClient: + def __init__(self, response: dict[str, Any] | None = None) -> None: + self.calls: list[dict[str, Any]] = [] + self._response = response or {"FailedEntryCount": 0, "Entries": [{}]} + + def put_events(self, **kwargs: Any) -> dict[str, Any]: + self.calls.append(kwargs) + return self._response + + +def test_public_import_path_and_defaults() -> None: + assert EventBridgeExporter is EventBridgeExporterFromModule + exporter = EventBridgeExporter(client=FakeEventsClient()) + assert exporter.event_bus_name == "default" + assert exporter.source == "aws.durable-execution.insight" + assert exporter.operations_format == "array" + assert exporter.max_record_size_bytes == 256_000 + exporter.flush() # no buffering: a no-op + + +def test_publishes_single_event_with_status_detail_type() -> None: + client = FakeEventsClient() + record = _record() + exporter = EventBridgeExporter(client=client) + assert exporter.render(record) is record + exporter.export(record) + + assert len(client.calls) == 1 + entries = client.calls[0]["Entries"] + assert len(entries) == 1 + entry = entries[0] + assert entry["EventBusName"] == "default" + assert entry["Source"] == "aws.durable-execution.insight" + assert entry["DetailType"] == "SUCCEEDED" + assert entry["Time"] == datetime.datetime( + 2026, 7, 15, 12, 0, 0, tzinfo=datetime.UTC + ) + detail = entry["Detail"] + assert ", " not in detail and '": ' not in detail + assert json.loads(detail) == record + assert isinstance(json.loads(detail)["operations"], list) + + +def test_by_name_format_with_custom_bus_and_source() -> None: + client = FakeEventsClient() + EventBridgeExporter( + event_bus_name="insight-bus", + source="my.source", + operations_format="by-name", + client=client, + ).export(_record(status="FAILED")) + entry = client.calls[0]["Entries"][0] + assert entry["EventBusName"] == "insight-bus" + assert entry["Source"] == "my.source" + assert entry["DetailType"] == "FAILED" + detail = json.loads(entry["Detail"]) + assert "operations" not in detail + assert detail["operationsByName"]["fetch-user"]["count"] == 1 + + +def test_failed_entry_raises() -> None: + client = FakeEventsClient( + response={ + "FailedEntryCount": 1, + "Entries": [{"ErrorCode": "ThrottlingException", "ErrorMessage": "slow"}], + } + ) + exporter = EventBridgeExporter(client=client) + with pytest.raises( + RuntimeError, + match=r"EventBridge PutEvents failed: ThrottlingException — slow", + ): + exporter.export(_record()) diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_exporters.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_exporters.py index 84bada35..4b078c1a 100644 --- a/packages/aws-durable-execution-sdk-python-insight/tests/test_exporters.py +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_exporters.py @@ -75,6 +75,16 @@ def test_public_import_paths_resolve_same_classes(): assert S3Partitioning is S3PartitioningFromModule +def test_every_exporter_is_re_exported_from_the_package_root(): + import aws_durable_execution_sdk_python_insight as pkg + from aws_durable_execution_sdk_python_insight import exporters + + assert set(exporters.__all__) <= set(pkg.__all__) + for name in exporters.__all__: + assert getattr(pkg, name) is getattr(exporters, name) + assert exporters.__all__ == sorted(exporters.__all__) + + # -- LambdaLogExporter -------------------------------------------------------- @@ -216,3 +226,46 @@ def test_s3_partitioning_invalid_string_raises_value_error(): S3Exporter(bucket="b", partitioning="function_name", client=FakeS3Client()) with pytest.raises(ValueError): S3Exporter(bucket="b", partitioning="bogus", client=FakeS3Client()) + + +def test_every_exporter_satisfies_the_insight_exporter_protocol() -> None: + # Annotated on purpose so mypy checks this body: each first-party exporter + # must be assignable to the protocol the plugin config is typed against. + from aws_durable_execution_sdk_python_insight import ( + AuroraExporter, + CloudWatchLogsExporter, + DynamoDBExporter, + EventBridgeExporter, + FileExporter, + FirehoseExporter, + HttpExporter, + InsightExporter, + OpenSearchExporter, + OTelExporter, + RedshiftExporter, + SQSExporter, + ) + + fake = FakeS3Client() + exporters: list[InsightExporter] = [ + LambdaLogExporter(), + S3Exporter(bucket="b", client=fake), + DynamoDBExporter(table_name="t", client=fake), + AuroraExporter( + resource_arn="r", secret_arn="s", database="d", engine="mysql", client=fake + ), + CloudWatchLogsExporter(log_group_name="/g", client=fake), + OTelExporter(endpoint="http://127.0.0.1:1/"), + FirehoseExporter(delivery_stream_name="s", client=fake), + EventBridgeExporter(client=fake), + RedshiftExporter(database="d", workgroup_name="w", client=fake), + OpenSearchExporter(endpoint="http://127.0.0.1:1", region="us-east-1"), + SQSExporter(queue_url="https://sqs.us-east-1.amazonaws.com/1/q", client=fake), + HttpExporter(url="http://127.0.0.1:1/"), + FileExporter(directory="/tmp/insight-protocol-check"), + ] + assert len(exporters) == 13 + for exporter in exporters: + assert exporter.max_record_size_bytes is None or isinstance( + exporter.max_record_size_bytes, int + ) diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_file_exporter.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_file_exporter.py new file mode 100644 index 00000000..96571de4 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_file_exporter.py @@ -0,0 +1,149 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Unit tests for ``FileExporter`` (``tmp_path``, no AWS).""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest + +from aws_durable_execution_sdk_python_insight import FileExporter, FileMode +from aws_durable_execution_sdk_python_insight.exporters.file_exporter import ( + FileExporter as FileExporterFromModule, +) + + +def _record(**overrides: Any) -> dict[str, Any]: + """A complete SUCCEEDED record; keyword arguments override fields.""" + record: dict[str, Any] = { + "recordType": "WorkflowInsight", + "schemaVersion": "1.0", + "emittedAt": "2026-07-15T12:00:00.000Z", + "executionArn": ( + "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST" + "/durable-execution/my-exec/inv-1" + ), + "executionName": "my-exec", + "functionName": "fn", + "functionQualifier": "$LATEST", + "region": "us-east-1", + "accountId": "123456789012", + "status": "SUCCEEDED", + "startTime": "2026-07-15T11:59:58.000Z", + "endTime": "2026-07-15T12:00:00.000Z", + "durationMs": 2000, + "operations": [ + { + "id": "op-1", + "name": "fetch-user", + "type": "STEP", + "subType": "Step", + "status": "SUCCEEDED", + "durationMs": 12, + } + ], + } + record.update(overrides) + return record + + +def test_public_import_path_and_defaults(tmp_path: Path) -> None: + assert FileExporter is FileExporterFromModule + exporter = FileExporter(directory=tmp_path) + assert exporter.mode is FileMode.NDJSON + assert exporter.operations_format == "array" + assert exporter.max_record_size_bytes is None + exporter.flush() # no buffering: a no-op + with pytest.raises(ValueError): + FileExporter( + directory=tmp_path, + mode="csv", # type: ignore[arg-type] # dynamic invalid value + ) + + +def test_ndjson_appends_date_partitioned_compact_lines(tmp_path: Path) -> None: + directory = tmp_path / "nested" / "insight" + exporter = FileExporter(directory=str(directory)) + first = _record() + second = _record(executionName="other", status="FAILED") + assert exporter.render(first) is first + exporter.export(first) + exporter.export(second) + + files = sorted(p.name for p in directory.iterdir()) + assert files == ["2026-07-15.ndjson"] + content = (directory / "2026-07-15.ndjson").read_text(encoding="utf-8") + assert content.endswith("\n") + lines = content.split("\n") + assert lines[-1] == "" + assert len(lines) == 3 + assert lines[0] == json.dumps(first, separators=(",", ":"), ensure_ascii=False) + assert json.loads(lines[1])["executionName"] == "other" + assert isinstance(json.loads(lines[0])["operations"], list) + + +def test_ndjson_uses_emitted_at_date_per_file(tmp_path: Path) -> None: + exporter = FileExporter(directory=tmp_path) + exporter.export(_record(emittedAt="2026-07-16T00:00:00.000Z")) + exporter.export(_record()) + assert sorted(p.name for p in tmp_path.iterdir()) == [ + "2026-07-15.ndjson", + "2026-07-16.ndjson", + ] + + +def test_json_mode_writes_pretty_file_per_execution_and_overwrites( + tmp_path: Path, +) -> None: + exporter = FileExporter(directory=tmp_path, mode="json") + exporter.export(_record(status="RUNNING")) + exporter.export(_record(status="SUCCEEDED")) + + assert sorted(p.name for p in tmp_path.iterdir()) == ["my-exec.json"] + content = (tmp_path / "my-exec.json").read_text(encoding="utf-8") + assert "\n " in content # 2-space pretty print + assert content == json.dumps( + _record(status="SUCCEEDED"), indent=2, ensure_ascii=False + ) + assert json.loads(content)["status"] == "SUCCEEDED" + + +def test_json_mode_sanitizes_name_and_falls_back_to_arn(tmp_path: Path) -> None: + exporter = FileExporter(directory=tmp_path, mode=FileMode.JSON) + exporter.export(_record(executionName="exec/with space")) + record = _record() + del record["executionName"] + exporter.export(record) + names = sorted(p.name for p in tmp_path.iterdir()) + assert "exec_with_space.json" in names + assert any(n.startswith("arn_aws_lambda") and n.endswith(".json") for n in names) + + +def test_by_name_format_applies_to_written_record(tmp_path: Path) -> None: + FileExporter(directory=tmp_path, operations_format="by-name").export(_record()) + line = (tmp_path / "2026-07-15.ndjson").read_text(encoding="utf-8").splitlines()[0] + parsed = json.loads(line) + assert "operations" not in parsed + assert parsed["operationsByName"]["fetch-user"]["count"] == 1 + + +def test_directory_is_created_once( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[Path] = [] + real_mkdir = Path.mkdir + + def counting_mkdir(self: Path, *args: Any, **kwargs: Any) -> None: + calls.append(self) + real_mkdir(self, *args, **kwargs) + + monkeypatch.setattr(Path, "mkdir", counting_mkdir) + exporter = FileExporter(directory=tmp_path / "d") + exporter.export(_record()) + exporter.export(_record()) + assert calls == [tmp_path / "d"] diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_firehose_exporter.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_firehose_exporter.py new file mode 100644 index 00000000..fb2dee0f --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_firehose_exporter.py @@ -0,0 +1,115 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Unit tests for ``FirehoseExporter`` (fake client, no AWS).""" + +from __future__ import annotations + +import json +from typing import Any + +import pytest + +from aws_durable_execution_sdk_python_insight import FirehoseExporter, OperationsFormat +from aws_durable_execution_sdk_python_insight.exporters.firehose_exporter import ( + FirehoseExporter as FirehoseExporterFromModule, +) + + +def _record(**overrides: Any) -> dict[str, Any]: + """A complete SUCCEEDED record; keyword arguments override fields.""" + record: dict[str, Any] = { + "recordType": "WorkflowInsight", + "schemaVersion": "1.0", + "emittedAt": "2026-07-15T12:00:00.000Z", + "executionArn": ( + "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST" + "/durable-execution/my-exec/inv-1" + ), + "executionName": "my-exec", + "functionName": "fn", + "functionQualifier": "$LATEST", + "region": "us-east-1", + "accountId": "123456789012", + "status": "SUCCEEDED", + "startTime": "2026-07-15T11:59:58.000Z", + "endTime": "2026-07-15T12:00:00.000Z", + "durationMs": 2000, + "operations": [ + { + "id": "op-1", + "name": "fetch-user", + "type": "STEP", + "subType": "Step", + "status": "SUCCEEDED", + "durationMs": 12, + } + ], + } + record.update(overrides) + return record + + +class FakeFirehoseClient: + def __init__(self) -> None: + self.puts: list[dict[str, Any]] = [] + + def put_record(self, **kwargs: Any) -> dict[str, Any]: + self.puts.append(kwargs) + return {} + + +def test_public_import_path_and_defaults() -> None: + assert FirehoseExporter is FirehoseExporterFromModule + exporter = FirehoseExporter(delivery_stream_name="s", client=FakeFirehoseClient()) + assert exporter.max_record_size_bytes == 1_000_000 + assert exporter.operations_format is OperationsFormat.ARRAY + exporter.flush() # no buffering: a no-op + + +def test_puts_single_ndjson_record_with_trailing_newline() -> None: + client = FakeFirehoseClient() + record = _record() + exporter = FirehoseExporter(delivery_stream_name="insight-stream", client=client) + assert exporter.render(record) is record + exporter.export(record) + + assert len(client.puts) == 1 + put = client.puts[0] + assert put["DeliveryStreamName"] == "insight-stream" + data = put["Record"]["Data"] + assert isinstance(data, bytes) + expected = json.dumps(record, separators=(",", ":"), ensure_ascii=False) + "\n" + assert data == expected.encode("utf-8") + assert data.endswith(b"\n") + assert len(data.rstrip(b"\n").split(b"\n")) == 1 + assert isinstance(json.loads(data)["operations"], list) + + +def test_by_name_format_renders_operations_by_name() -> None: + client = FakeFirehoseClient() + FirehoseExporter( + delivery_stream_name="s", operations_format="by-name", client=client + ).export(_record()) + parsed = json.loads(client.puts[0]["Record"]["Data"]) + assert "operations" not in parsed + assert parsed["operationsByName"]["fetch-user"]["count"] == 1 + + +def test_both_format_keeps_array_and_adds_map() -> None: + client = FakeFirehoseClient() + FirehoseExporter( + delivery_stream_name="s", operations_format=OperationsFormat.BOTH, client=client + ).export(_record()) + parsed = json.loads(client.puts[0]["Record"]["Data"]) + assert isinstance(parsed["operations"], list) + assert parsed["operationsByName"]["fetch-user"]["count"] == 1 + + +def test_invalid_operations_format_raises() -> None: + with pytest.raises(ValueError): + FirehoseExporter( + delivery_stream_name="s", + operations_format="list", # type: ignore[arg-type] # dynamic invalid value + client=FakeFirehoseClient(), + ) diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_http_exporter.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_http_exporter.py new file mode 100644 index 00000000..6fd64d6b --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_http_exporter.py @@ -0,0 +1,224 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Unit tests for ``HttpExporter`` against a local HTTP server.""" + +from __future__ import annotations + +import json +import threading +import time +from dataclasses import dataclass, field +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any, Iterator + +import pytest + +from aws_durable_execution_sdk_python_insight import HttpExporter, HttpMethod +from aws_durable_execution_sdk_python_insight.exporters.http_exporter import ( + HttpExporter as HttpExporterFromModule, +) + + +def _record(**overrides: Any) -> dict[str, Any]: + """A complete SUCCEEDED record; keyword arguments override fields.""" + record: dict[str, Any] = { + "recordType": "WorkflowInsight", + "schemaVersion": "1.0", + "emittedAt": "2026-07-15T12:00:00.000Z", + "executionArn": ( + "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST" + "/durable-execution/my-exec/inv-1" + ), + "executionName": "my-exec", + "functionName": "fn", + "functionQualifier": "$LATEST", + "region": "us-east-1", + "accountId": "123456789012", + "status": "SUCCEEDED", + "startTime": "2026-07-15T11:59:58.000Z", + "endTime": "2026-07-15T12:00:00.000Z", + "durationMs": 2000, + "operations": [ + { + "id": "op-1", + "name": "fetch-user", + "type": "STEP", + "subType": "Step", + "status": "SUCCEEDED", + "durationMs": 12, + } + ], + } + record.update(overrides) + return record + + +@dataclass +class CapturedRequest: + method: str + path: str + headers: dict[str, str] + body: bytes + + +@dataclass +class HttpCapture: + """A local HTTP server that records requests and replies with ``status``.""" + + url: str + requests: list[CapturedRequest] = field(default_factory=list) + status: int = 200 + response_body: bytes = b"" + delay_seconds: float = 0.0 + location: str | None = None + + +@pytest.fixture +def http_capture() -> Iterator[HttpCapture]: + capture = HttpCapture(url="") + + class Handler(BaseHTTPRequestHandler): + def _handle(self) -> None: + length = int(self.headers.get("Content-Length") or 0) + body = self.rfile.read(length) if length else b"" + capture.requests.append( + CapturedRequest( + method=self.command, + path=self.path, + headers={k.lower(): v for k, v in self.headers.items()}, + body=body, + ) + ) + if capture.delay_seconds: + time.sleep(capture.delay_seconds) + self.send_response(capture.status) + if capture.location is not None: + self.send_header("Location", capture.location) + self.send_header("Content-Length", str(len(capture.response_body))) + self.end_headers() + self.wfile.write(capture.response_body) + + do_POST = _handle + do_PUT = _handle + + def log_message(self, format: str, *args: Any) -> None: # noqa: A002 + return None + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + port = server.server_address[1] + capture.url = f"http://127.0.0.1:{port}" + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield capture + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +def test_public_import_path_and_defaults() -> None: + assert HttpExporter is HttpExporterFromModule + exporter = HttpExporter(url="http://127.0.0.1:1/insight") + assert exporter.method is HttpMethod.POST + assert exporter.timeout_ms == 10_000 + assert exporter.headers == {} + assert exporter.operations_format == "array" + assert exporter.max_record_size_bytes is None + exporter.flush() # no buffering: a no-op + with pytest.raises(ValueError): + HttpExporter( + url="http://127.0.0.1:1/", + method="PATCH", # type: ignore[arg-type] # dynamic invalid value + ) + + +def test_posts_compact_json_with_content_type(http_capture: HttpCapture) -> None: + record = _record() + exporter = HttpExporter(url=f"{http_capture.url}/insight") + assert exporter.render(record) is record + exporter.export(record) + + assert len(http_capture.requests) == 1 + req = http_capture.requests[0] + assert req.method == "POST" + assert req.path == "/insight" + assert req.headers["content-type"] == "application/json" + assert req.body == json.dumps( + record, separators=(",", ":"), ensure_ascii=False + ).encode("utf-8") + assert isinstance(json.loads(req.body)["operations"], list) + + +def test_put_with_custom_headers_and_by_name_format(http_capture: HttpCapture) -> None: + HttpExporter( + url=f"{http_capture.url}/upsert", + method="PUT", + headers={"Authorization": "Bearer token123"}, + operations_format="by-name", + ).export(_record()) + req = http_capture.requests[0] + assert req.method == "PUT" + assert req.headers["authorization"] == "Bearer token123" + assert req.headers["content-type"] == "application/json" + body = json.loads(req.body) + assert "operations" not in body + assert body["operationsByName"]["fetch-user"]["count"] == 1 + + +def test_custom_header_can_override_content_type(http_capture: HttpCapture) -> None: + HttpExporter( + url=http_capture.url, headers={"Content-Type": "application/x-ndjson"} + ).export(_record()) + assert http_capture.requests[0].headers["content-type"] == "application/x-ndjson" + + +def test_non_2xx_response_raises(http_capture: HttpCapture) -> None: + http_capture.status = 500 + exporter = HttpExporter(url=http_capture.url) + with pytest.raises(RuntimeError, match=r"HttpExporter: endpoint returned 500"): + exporter.export(_record()) + + +def test_timeout_is_enforced(http_capture: HttpCapture) -> None: + http_capture.delay_seconds = 1.0 + exporter = HttpExporter(url=http_capture.url, timeout_ms=100) + with pytest.raises(TimeoutError): + exporter.export(_record()) + + +def test_oversized_error_body_is_capped_in_the_message( + http_capture: HttpCapture, +) -> None: + http_capture.status = 502 + http_capture.response_body = b"x" * (200 * 1024) + exporter = HttpExporter(url=http_capture.url) + with pytest.raises(RuntimeError) as info: + exporter.export(_record()) + # the message carries status and reason only; the body is not echoed + assert str(info.value) == "HttpExporter: endpoint returned 502 Bad Gateway" + + +def test_large_success_body_is_not_read(http_capture: HttpCapture) -> None: + http_capture.response_body = b"y" * (4 * 1024 * 1024) + HttpExporter(url=http_capture.url).export(_record()) + assert len(http_capture.requests) == 1 + + +@pytest.mark.parametrize("status", [301, 302, 303, 307, 308]) +def test_redirects_are_not_followed(http_capture: HttpCapture, status: int) -> None: + http_capture.status = status + http_capture.location = f"{http_capture.url}/elsewhere" + exporter = HttpExporter( + url=f"{http_capture.url}/insight", headers={"Authorization": "Bearer secret"} + ) + with pytest.raises( + RuntimeError, match=rf"HttpExporter: endpoint returned {status}" + ): + exporter.export(_record()) + # exactly one request, to the configured URL, with the body; nothing was + # re-sent to the redirect target + assert [r.path for r in http_capture.requests] == ["/insight"] + assert http_capture.requests[0].method == "POST" + assert http_capture.requests[0].body diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_opensearch_exporter.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_opensearch_exporter.py new file mode 100644 index 00000000..7c6b4339 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_opensearch_exporter.py @@ -0,0 +1,251 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Unit tests for ``OpenSearchExporter`` against a local HTTP server.""" + +from __future__ import annotations + +import base64 +import json +import threading +import time +from dataclasses import dataclass, field +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any, Iterator +from urllib.parse import quote + +import pytest + +from aws_durable_execution_sdk_python_insight import ( + OpenSearchAuth, + OpenSearchExporter, +) +from aws_durable_execution_sdk_python_insight.exporters.opensearch_exporter import ( + OpenSearchExporter as OpenSearchExporterFromModule, +) + + +def _record(**overrides: Any) -> dict[str, Any]: + """A complete SUCCEEDED record; keyword arguments override fields.""" + record: dict[str, Any] = { + "recordType": "WorkflowInsight", + "schemaVersion": "1.0", + "emittedAt": "2026-07-15T12:00:00.000Z", + "executionArn": ( + "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST" + "/durable-execution/my-exec/inv-1" + ), + "executionName": "my-exec", + "functionName": "fn", + "functionQualifier": "$LATEST", + "region": "us-east-1", + "accountId": "123456789012", + "status": "SUCCEEDED", + "startTime": "2026-07-15T11:59:58.000Z", + "endTime": "2026-07-15T12:00:00.000Z", + "durationMs": 2000, + "operations": [ + { + "id": "op-1", + "name": "fetch-user", + "type": "STEP", + "subType": "Step", + "status": "SUCCEEDED", + "durationMs": 12, + } + ], + } + record.update(overrides) + return record + + +@dataclass +class CapturedRequest: + method: str + path: str + headers: dict[str, str] + body: bytes + + +@dataclass +class HttpCapture: + """A local HTTP server that records requests and replies with ``status``.""" + + url: str + requests: list[CapturedRequest] = field(default_factory=list) + status: int = 200 + response_body: bytes = b"" + delay_seconds: float = 0.0 + + +@pytest.fixture +def http_capture() -> Iterator[HttpCapture]: + capture = HttpCapture(url="") + + class Handler(BaseHTTPRequestHandler): + def _handle(self) -> None: + length = int(self.headers.get("Content-Length") or 0) + body = self.rfile.read(length) if length else b"" + capture.requests.append( + CapturedRequest( + method=self.command, + path=self.path, + headers={k.lower(): v for k, v in self.headers.items()}, + body=body, + ) + ) + if capture.delay_seconds: + time.sleep(capture.delay_seconds) + self.send_response(capture.status) + self.send_header("Content-Length", str(len(capture.response_body))) + self.end_headers() + self.wfile.write(capture.response_body) + + do_POST = _handle + do_PUT = _handle + + def log_message(self, format: str, *args: Any) -> None: # noqa: A002 + return None + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + port = server.server_address[1] + capture.url = f"http://127.0.0.1:{port}" + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield capture + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +def test_public_import_path_and_defaults() -> None: + assert OpenSearchExporter is OpenSearchExporterFromModule + exporter = OpenSearchExporter( + endpoint="https://d.us-east-1.es.amazonaws.com/", region="us-east-1" + ) + assert exporter.endpoint == "https://d.us-east-1.es.amazonaws.com" + assert exporter.index_name == "workflow-insight" + assert exporter.auth is OpenSearchAuth.SIGV4 + assert exporter.max_record_size_bytes == 10_000_000 + exporter.flush() # no buffering: a no-op + with pytest.raises(ValueError): + OpenSearchExporter( + endpoint="https://d", + region="us-east-1", + auth="token", # type: ignore[arg-type] # dynamic invalid value + ) + + +def test_sigv4_puts_signed_document_keyed_by_encoded_arn( + http_capture: HttpCapture, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("AWS_ACCESS_KEY_ID", "AKIAEXAMPLE") + monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "secret") + monkeypatch.setenv("AWS_SESSION_TOKEN", "token") + record = _record() + exporter = OpenSearchExporter(endpoint=f"{http_capture.url}/", region="us-east-1") + assert exporter.render(record) is record + exporter.export(record) + + assert len(http_capture.requests) == 1 + req = http_capture.requests[0] + assert req.method == "PUT" + encoded = quote(record["executionArn"], safe="-_.!~*'()") + assert "$" not in encoded and "/" not in encoded and ":" not in encoded + assert req.path == f"/workflow-insight/_doc/{encoded}" + assert req.headers["content-type"] == "application/json" + assert req.headers["authorization"].startswith("AWS4-HMAC-SHA256 ") + assert "Credential=AKIAEXAMPLE/" in req.headers["authorization"] + assert "/us-east-1/es/aws4_request" in req.headers["authorization"] + assert "x-amz-date" in req.headers + assert req.headers["x-amz-security-token"] == "token" + assert req.body == json.dumps( + record, separators=(",", ":"), ensure_ascii=False + ).encode("utf-8") + + +def test_basic_auth_sends_encoded_credentials_without_signing( + http_capture: HttpCapture, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("AWS_ACCESS_KEY_ID", raising=False) + monkeypatch.delenv("AWS_SECRET_ACCESS_KEY", raising=False) + OpenSearchExporter( + endpoint=http_capture.url, + region="us-east-1", + index_name="custom-index", + auth="basic", + username="admin", + password="secret", # noqa: S106 - test fixture value + ).export(_record()) + + req = http_capture.requests[0] + assert req.path.startswith("/custom-index/_doc/") + expected = "Basic " + base64.b64encode(b"admin:secret").decode("ascii") + assert req.headers["authorization"] == expected + assert "x-amz-date" not in req.headers + + +def test_non_2xx_raises_with_status_and_detail(http_capture: HttpCapture) -> None: + http_capture.status = 403 + http_capture.response_body = b'{"message":"User is not authorized"}' + exporter = OpenSearchExporter( + endpoint=http_capture.url, + region="us-east-1", + auth=OpenSearchAuth.BASIC, + username="u", + password="p", # noqa: S106 - test fixture value + ) + with pytest.raises(RuntimeError, match=r"OpenSearch index failed: 403") as info: + exporter.export(_record()) + assert "User is not authorized" in str(info.value) + + +def test_error_detail_is_truncated_to_500_chars(http_capture: HttpCapture) -> None: + http_capture.status = 500 + http_capture.response_body = b"e" * (100 * 1024) + exporter = OpenSearchExporter( + endpoint=http_capture.url, + region="us-east-1", + auth="basic", + username="u", + password="p", # noqa: S106 - test fixture value + ) + with pytest.raises(RuntimeError) as info: + exporter.export(_record()) + message = str(info.value) + assert message.startswith("OpenSearch index failed: 500 Internal Server Error — ") + assert message.endswith("e" * 500) + assert len(message) < 600 + + +@pytest.mark.parametrize( + ("username", "password"), + [(None, None), ("admin", None), (None, "secret")], +) +def test_basic_auth_requires_username_and_password( + username: str | None, password: str | None +) -> None: + with pytest.raises(ValueError, match="requires username and password"): + OpenSearchExporter( + endpoint="https://d", + region="us-east-1", + auth="basic", + username=username, + password=password, + ) + + +def test_basic_auth_accepts_an_empty_password(http_capture: HttpCapture) -> None: + OpenSearchExporter( + endpoint=http_capture.url, + region="us-east-1", + auth="basic", + username="admin", + password="", + ).export(_record()) + expected = "Basic " + base64.b64encode(b"admin:").decode("ascii") + assert http_capture.requests[0].headers["authorization"] == expected diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_otel_exporter.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_otel_exporter.py new file mode 100644 index 00000000..e42f509e --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_otel_exporter.py @@ -0,0 +1,252 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Unit tests for ``OTelExporter`` against a local HTTP server.""" + +from __future__ import annotations + +import json +import threading +import time +from dataclasses import dataclass, field +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any, Iterator + +import pytest + +from aws_durable_execution_sdk_python_insight import OTelExporter, OTelProtocol +from aws_durable_execution_sdk_python_insight.exporters.otel_exporter import ( + OTelExporter as OTelExporterFromModule, +) + + +def _record(**overrides: Any) -> dict[str, Any]: + """A complete SUCCEEDED record; keyword arguments override fields.""" + record: dict[str, Any] = { + "recordType": "WorkflowInsight", + "schemaVersion": "1.0", + "emittedAt": "2026-07-15T12:00:00.000Z", + "executionArn": ( + "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST" + "/durable-execution/my-exec/inv-1" + ), + "executionName": "my-exec", + "functionName": "fn", + "functionQualifier": "$LATEST", + "region": "us-east-1", + "accountId": "123456789012", + "status": "SUCCEEDED", + "startTime": "2026-07-15T11:59:58.000Z", + "endTime": "2026-07-15T12:00:00.000Z", + "durationMs": 2000, + "operations": [ + { + "id": "op-1", + "name": "fetch-user", + "type": "STEP", + "subType": "Step", + "status": "SUCCEEDED", + "durationMs": 12, + } + ], + } + record.update(overrides) + return record + + +@dataclass +class CapturedRequest: + method: str + path: str + headers: dict[str, str] + body: bytes + + +@dataclass +class HttpCapture: + """A local HTTP server that records requests and replies with ``status``.""" + + url: str + requests: list[CapturedRequest] = field(default_factory=list) + status: int = 200 + response_body: bytes = b"" + delay_seconds: float = 0.0 + location: str | None = None + + +@pytest.fixture +def http_capture() -> Iterator[HttpCapture]: + capture = HttpCapture(url="") + + class Handler(BaseHTTPRequestHandler): + def _handle(self) -> None: + length = int(self.headers.get("Content-Length") or 0) + body = self.rfile.read(length) if length else b"" + capture.requests.append( + CapturedRequest( + method=self.command, + path=self.path, + headers={k.lower(): v for k, v in self.headers.items()}, + body=body, + ) + ) + if capture.delay_seconds: + time.sleep(capture.delay_seconds) + self.send_response(capture.status) + if capture.location is not None: + self.send_header("Location", capture.location) + self.send_header("Content-Length", str(len(capture.response_body))) + self.end_headers() + self.wfile.write(capture.response_body) + + do_POST = _handle + do_PUT = _handle + + def log_message(self, format: str, *args: Any) -> None: # noqa: A002 + return None + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + port = server.server_address[1] + capture.url = f"http://127.0.0.1:{port}" + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield capture + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +def _log_record(payload: dict[str, Any]) -> dict[str, Any]: + resource_log = payload["resourceLogs"][0] + return dict(resource_log["scopeLogs"][0]["logRecords"][0]) + + +def _attrs(items: list[dict[str, Any]]) -> dict[str, Any]: + return {item["key"]: item["value"] for item in items} + + +def test_public_import_path_and_defaults() -> None: + assert OTelExporter is OTelExporterFromModule + exporter = OTelExporter(endpoint="http://127.0.0.1:1/v1/logs") + assert exporter.max_record_size_bytes == 1_000_000 + assert exporter.operations_format == "array" + assert exporter.headers == {} + exporter.flush() # no buffering: a no-op + + +def test_http_protobuf_is_rejected_at_construction() -> None: + with pytest.raises(ValueError, match="http/protobuf is not yet supported"): + OTelExporter(endpoint="http://127.0.0.1:1/v1/logs", protocol="http/protobuf") + with pytest.raises(ValueError): + OTelExporter( + endpoint="http://127.0.0.1:1/v1/logs", protocol=OTelProtocol.HTTP_PROTOBUF + ) + with pytest.raises(ValueError): + OTelExporter( + endpoint="http://127.0.0.1:1/v1/logs", + protocol="grpc", # type: ignore[arg-type] # dynamic invalid value + ) + + +def test_posts_otlp_request_with_record_in_log_body(http_capture: HttpCapture) -> None: + record = _record() + exporter = OTelExporter( + endpoint=f"{http_capture.url}/v1/logs", headers={"x-api-key": "k"} + ) + exporter.export(record) + + assert len(http_capture.requests) == 1 + req = http_capture.requests[0] + assert req.method == "POST" + assert req.path == "/v1/logs" + assert req.headers["content-type"] == "application/json" + assert req.headers["x-api-key"] == "k" + + payload = json.loads(req.body) + assert req.body == json.dumps( + exporter.render(record), separators=(",", ":"), ensure_ascii=False + ).encode("utf-8") + resource_attrs = _attrs(payload["resourceLogs"][0]["resource"]["attributes"]) + assert resource_attrs["service.name"] == {"stringValue": "fn"} + assert resource_attrs["cloud.region"] == {"stringValue": "us-east-1"} + assert resource_attrs["cloud.account.id"] == {"stringValue": "123456789012"} + assert resource_attrs["faas.name"] == {"stringValue": "fn"} + assert resource_attrs["faas.version"] == {"stringValue": "$LATEST"} + scope = payload["resourceLogs"][0]["scopeLogs"][0]["scope"] + assert scope == { + "name": "aws-durable-execution-sdk-python-insight", + "version": "1.0", + } + + log_record = _log_record(payload) + # 2026-07-15T12:00:00.000Z in nanoseconds since the epoch, as a string + assert log_record["timeUnixNano"] == "1784116800000000000" + assert log_record["severityNumber"] == 9 + assert log_record["severityText"] == "SUCCEEDED" + body = json.loads(log_record["body"]["stringValue"]) + assert body == record + attrs = _attrs(log_record["attributes"]) + assert attrs["workflow.execution_arn"] == {"stringValue": record["executionArn"]} + assert attrs["workflow.execution_name"] == {"stringValue": "my-exec"} + assert attrs["workflow.status"] == {"stringValue": "SUCCEEDED"} + assert attrs["workflow.duration_ms"] == {"intValue": "2000"} + + +def test_failed_maps_to_error_severity_and_missing_fields_default( + http_capture: HttpCapture, +) -> None: + record = _record(status="FAILED") + del record["executionName"] + del record["durationMs"] + OTelExporter(endpoint=f"{http_capture.url}/v1/logs").export(record) + log_record = _log_record(json.loads(http_capture.requests[0].body)) + assert log_record["severityNumber"] == 17 + assert log_record["severityText"] == "FAILED" + attrs = _attrs(log_record["attributes"]) + assert attrs["workflow.execution_name"] == {"stringValue": ""} + assert attrs["workflow.duration_ms"] == {"intValue": "0"} + + +def test_unknown_status_maps_to_unspecified_severity() -> None: + rendered = OTelExporter(endpoint="http://127.0.0.1:1/").render( + _record(status="STOPPED") + ) + assert _log_record(rendered)["severityNumber"] == 0 + + +def test_by_name_format_reshapes_log_body_only(http_capture: HttpCapture) -> None: + OTelExporter( + endpoint=f"{http_capture.url}/v1/logs", operations_format="by-name" + ).export(_record()) + body = json.loads( + _log_record(json.loads(http_capture.requests[0].body))["body"]["stringValue"] + ) + assert "operations" not in body + assert body["operationsByName"]["fetch-user"]["count"] == 1 + + +def test_non_2xx_response_raises(http_capture: HttpCapture) -> None: + http_capture.status = 503 + exporter = OTelExporter(endpoint=f"{http_capture.url}/v1/logs") + with pytest.raises(RuntimeError, match=r"OTelExporter: OTLP endpoint returned 503"): + exporter.export(_record()) + + +def test_time_unix_nano_keeps_microseconds() -> None: + rendered = OTelExporter(endpoint="http://127.0.0.1:1/").render( + _record(emittedAt="2026-07-15T12:00:00.123456Z") + ) + assert _log_record(rendered)["timeUnixNano"] == "1784116800123456000" + + +def test_redirect_is_reported_as_failure(http_capture: HttpCapture) -> None: + http_capture.status = 302 + http_capture.location = f"{http_capture.url}/other" + exporter = OTelExporter( + endpoint=f"{http_capture.url}/v1/logs", headers={"x-api-key": "k"} + ) + with pytest.raises(RuntimeError, match=r"OTLP endpoint returned 302"): + exporter.export(_record()) + assert [r.path for r in http_capture.requests] == ["/v1/logs"] diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_redshift_exporter.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_redshift_exporter.py new file mode 100644 index 00000000..724af05d --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_redshift_exporter.py @@ -0,0 +1,181 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Unit tests for ``RedshiftExporter`` (fake Redshift Data API client, no AWS).""" + +from __future__ import annotations + +import json +from typing import Any + +import pytest + +from aws_durable_execution_sdk_python_insight import RedshiftExporter +from aws_durable_execution_sdk_python_insight.exporters.redshift_exporter import ( + RedshiftExporter as RedshiftExporterFromModule, +) + + +def _record(**overrides: Any) -> dict[str, Any]: + """A complete SUCCEEDED record; keyword arguments override fields.""" + record: dict[str, Any] = { + "recordType": "WorkflowInsight", + "schemaVersion": "1.0", + "emittedAt": "2026-07-15T12:00:00.000Z", + "executionArn": ( + "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST" + "/durable-execution/my-exec/inv-1" + ), + "executionName": "my-exec", + "functionName": "fn", + "functionQualifier": "$LATEST", + "region": "us-east-1", + "accountId": "123456789012", + "status": "SUCCEEDED", + "startTime": "2026-07-15T11:59:58.000Z", + "endTime": "2026-07-15T12:00:00.000Z", + "durationMs": 2000, + "operations": [ + { + "id": "op-1", + "name": "fetch-user", + "type": "STEP", + "subType": "Step", + "status": "SUCCEEDED", + "durationMs": 12, + } + ], + } + record.update(overrides) + return record + + +class FakeRedshiftDataClient: + def __init__(self) -> None: + self.calls: list[dict[str, Any]] = [] + + def execute_statement(self, **kwargs: Any) -> dict[str, Any]: + self.calls.append(kwargs) + return {"Id": "stmt-1"} + + +def _params(call: dict[str, Any]) -> dict[str, str]: + return {p["name"]: p["value"] for p in call["Parameters"]} + + +def test_public_import_path_and_defaults() -> None: + assert RedshiftExporter is RedshiftExporterFromModule + exporter = RedshiftExporter( + database="insight", workgroup_name="wg", client=FakeRedshiftDataClient() + ) + assert exporter.max_record_size_bytes == 1_000_000 + assert exporter.fq_table == "public.workflow_insight" + exporter.flush() # no buffering: a no-op + + +def test_merge_joins_on_source_column_subquery_with_typed_casts() -> None: + client = FakeRedshiftDataClient() + record = _record() + exporter = RedshiftExporter( + database="insight", workgroup_name="insight-wg", client=client + ) + assert exporter.render(record) is record + exporter.export(record) + + assert len(client.calls) == 1 + call = client.calls[0] + assert call["WorkgroupName"] == "insight-wg" + assert call["Database"] == "insight" + assert "ClusterIdentifier" not in call + assert "DbUser" not in call + assert "SecretArn" not in call + + sql = call["Sql"] + assert "MERGE INTO public.workflow_insight USING (" in sql + assert "ON public.workflow_insight.execution_arn = src.execution_arn" in sql + assert ":execution_arn::varchar AS execution_arn" in sql + assert "USING (SELECT 1)" not in sql + assert ":start_time::timestamptz AS start_time" in sql + assert ":end_time::timestamptz AS end_time" in sql + assert ":duration_ms::bigint AS duration_ms" in sql + assert ":execution_name::varchar AS execution_name" in sql + assert ":emitted_at::timestamptz AS emitted_at" in sql + assert "JSON_PARSE(:record_json) AS record_json" in sql + assert "WHEN MATCHED THEN UPDATE SET" in sql + assert "WHEN NOT MATCHED THEN INSERT" in sql + + params = _params(call) + assert params["execution_arn"] == record["executionArn"] + assert params["function_name"] == "fn" + assert params["status"] == "SUCCEEDED" + assert params["start_time"] == "2026-07-15T11:59:58.000Z" + assert params["end_time"] == "2026-07-15T12:00:00.000Z" + assert params["duration_ms"] == "2000" + assert params["execution_name"] == "my-exec" + assert params["emitted_at"] == "2026-07-15T12:00:00.000Z" + assert all(isinstance(v, str) for v in params.values()) + assert ", " not in params["record_json"] + assert json.loads(params["record_json"]) == record + + +def test_absent_nullable_fields_become_typed_null_literals() -> None: + client = FakeRedshiftDataClient() + record = _record(status="RUNNING") + del record["executionName"] + del record["endTime"] + del record["durationMs"] + RedshiftExporter( + database="insight", + cluster_identifier="insight-cluster", + db_user="admin", + secret_arn="arn:aws:secretsmanager:us-east-1:123456789012:secret:s", + client=client, + ).export(record) + + call = client.calls[0] + assert call["ClusterIdentifier"] == "insight-cluster" + assert call["DbUser"] == "admin" + assert call["SecretArn"] == "arn:aws:secretsmanager:us-east-1:123456789012:secret:s" + assert "WorkgroupName" not in call + assert "NULL::timestamptz AS end_time" in call["Sql"] + assert "NULL::bigint AS duration_ms" in call["Sql"] + assert "NULL::varchar AS execution_name" in call["Sql"] + params = _params(call) + assert "end_time" not in params + assert "duration_ms" not in params + assert "execution_name" not in params + + +def test_custom_schema_and_table_are_validated() -> None: + exporter = RedshiftExporter( + database="d", + workgroup_name="wg", + schema="analytics", + table="wi_records", + client=FakeRedshiftDataClient(), + ) + assert exporter.fq_table == "analytics.wi_records" + with pytest.raises(ValueError, match="Invalid SQL identifier"): + RedshiftExporter( + database="d", + workgroup_name="wg", + schema="a.b", + client=FakeRedshiftDataClient(), + ) + + +def test_requires_workgroup_or_cluster() -> None: + with pytest.raises( + ValueError, match="exactly one of workgroup_name or cluster_identifier" + ): + RedshiftExporter(database="insight", client=FakeRedshiftDataClient()) + + +def test_rejects_both_workgroup_and_cluster() -> None: + with pytest.raises(ValueError, match="exactly one of"): + RedshiftExporter( + database="insight", + workgroup_name="wg", + cluster_identifier="cluster", + client=FakeRedshiftDataClient(), + ) diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_shaping.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_shaping.py index 57b4e27e..f5ac4288 100644 --- a/packages/aws-durable-execution-sdk-python-insight/tests/test_shaping.py +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_shaping.py @@ -6,6 +6,8 @@ from __future__ import annotations from aws_durable_execution_sdk_python_insight.operations_index import ( + OperationsFormat, + apply_operations_format, build_operations_by_name, with_operations_by_name, ) @@ -130,3 +132,29 @@ def test_truncation_noop_when_within_limit(): record = _record_with_results([5]) out = truncate_record(record, 5_000_000, render=lambda r: r) assert out is record + + +def test_apply_operations_format_array_returns_record_unchanged(): + record = {"executionArn": "arn", "operations": [_op("greet")]} + assert apply_operations_format(record, "array") is record + assert apply_operations_format(record, OperationsFormat.ARRAY) is record + + +def test_apply_operations_format_by_name_replaces_array(): + record = {"executionArn": "arn", "operations": [_op("greet")]} + out = apply_operations_format(record, "by-name") + assert "operations" not in out + assert out["operationsByName"]["greet"]["count"] == 1 + assert "operations" in record # input never mutated + + +def test_apply_operations_format_both_keeps_array_and_adds_map(): + record = {"executionArn": "arn", "operations": [_op("greet")]} + out = apply_operations_format(record, OperationsFormat.BOTH) + assert out["operations"] == record["operations"] + assert out["operationsByName"]["greet"]["count"] == 1 + assert "operationsByName" not in record + + +def test_operations_format_enum_values(): + assert {f.value for f in OperationsFormat} == {"array", "by-name", "both"} diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_sqs_exporter.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_sqs_exporter.py new file mode 100644 index 00000000..fcf9bc77 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_sqs_exporter.py @@ -0,0 +1,150 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Unit tests for ``SQSExporter`` (fake client, no AWS).""" + +from __future__ import annotations + +import hashlib +import json +from typing import Any + +from aws_durable_execution_sdk_python_insight import SQSExporter +from aws_durable_execution_sdk_python_insight.exporters.sqs_exporter import ( + SQSExporter as SQSExporterFromModule, +) + + +def _record(**overrides: Any) -> dict[str, Any]: + """A complete SUCCEEDED record; keyword arguments override fields.""" + record: dict[str, Any] = { + "recordType": "WorkflowInsight", + "schemaVersion": "1.0", + "emittedAt": "2026-07-15T12:00:00.000Z", + "executionArn": ( + "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST" + "/durable-execution/my-exec/inv-1" + ), + "executionName": "my-exec", + "functionName": "fn", + "functionQualifier": "$LATEST", + "region": "us-east-1", + "accountId": "123456789012", + "status": "SUCCEEDED", + "startTime": "2026-07-15T11:59:58.000Z", + "endTime": "2026-07-15T12:00:00.000Z", + "durationMs": 2000, + "operations": [ + { + "id": "op-1", + "name": "fetch-user", + "type": "STEP", + "subType": "Step", + "status": "SUCCEEDED", + "durationMs": 12, + } + ], + } + record.update(overrides) + return record + + +STANDARD_URL = "https://sqs.us-east-1.amazonaws.com/123456789012/insight" +FIFO_URL = "https://sqs.us-east-1.amazonaws.com/123456789012/insight.fifo" + + +class FakeSqsClient: + def __init__(self) -> None: + self.sends: list[dict[str, Any]] = [] + + def send_message(self, **kwargs: Any) -> dict[str, Any]: + self.sends.append(kwargs) + return {"MessageId": "m-1"} + + +def test_public_import_path_and_defaults() -> None: + assert SQSExporter is SQSExporterFromModule + exporter = SQSExporter(queue_url=STANDARD_URL, client=FakeSqsClient()) + assert exporter.max_record_size_bytes == 256_000 + assert exporter.operations_format == "array" + assert exporter.is_fifo is False + exporter.flush() # no buffering: a no-op + + +def test_standard_queue_message_has_attributes_and_no_fifo_fields() -> None: + client = FakeSqsClient() + record = _record() + exporter = SQSExporter(queue_url=STANDARD_URL, client=client) + assert exporter.render(record) is record + exporter.export(record) + + assert len(client.sends) == 1 + send = client.sends[0] + assert send["QueueUrl"] == STANDARD_URL + assert "MessageGroupId" not in send + assert "MessageDeduplicationId" not in send + assert send["MessageAttributes"] == { + "status": {"DataType": "String", "StringValue": "SUCCEEDED"}, + "functionName": {"DataType": "String", "StringValue": "fn"}, + } + body = send["MessageBody"] + assert ", " not in body and '": ' not in body + assert json.loads(body) == record + + +def test_fifo_queue_sets_group_and_dedup_ids() -> None: + client = FakeSqsClient() + record = _record() + exporter = SQSExporter(queue_url=FIFO_URL, client=client) + assert exporter.is_fifo is True + exporter.export(record) + send = client.sends[0] + assert send["MessageGroupId"] == record["executionArn"] + assert send["MessageDeduplicationId"] == ( + f"{record['executionArn']}:2026-07-15T12:00:00.000Z" + ) + + +def test_fifo_honors_explicit_message_group_id() -> None: + client = FakeSqsClient() + SQSExporter(queue_url=FIFO_URL, message_group_id="grp", client=client).export( + _record() + ) + assert client.sends[0]["MessageGroupId"] == "grp" + + +def test_by_name_format_renders_operations_by_name() -> None: + client = FakeSqsClient() + SQSExporter( + queue_url=STANDARD_URL, operations_format="by-name", client=client + ).export(_record()) + body = json.loads(client.sends[0]["MessageBody"]) + assert "operations" not in body + assert body["operationsByName"]["fetch-user"]["count"] == 1 + + +def test_fifo_ids_over_128_chars_are_hashed() -> None: + long_arn = ( + "arn:aws:lambda:us-east-1:123456789012:function:" + + "a-very-long-function-name-" * 3 + + ":$LATEST/durable-execution/" + + "e" * 64 + + "/inv-1" + ) + assert len(long_arn) > 128 + client = FakeSqsClient() + record = _record(executionArn=long_arn) + SQSExporter(queue_url=FIFO_URL, client=client).export(record) + send = client.sends[0] + expected_group = hashlib.sha256(long_arn.encode("utf-8")).hexdigest() + expected_dedup = hashlib.sha256( + f"{long_arn}:{record['emittedAt']}".encode() + ).hexdigest() + assert send["MessageGroupId"] == expected_group + assert send["MessageDeduplicationId"] == expected_dedup + assert len(send["MessageGroupId"]) <= 128 + assert len(send["MessageDeduplicationId"]) <= 128 + # short ids are still passed through untouched + short = _record() + SQSExporter(queue_url=FIFO_URL, client=client).export(short) + assert client.sends[1]["MessageGroupId"] == short["executionArn"]