[dagster-aws] Add EMR step failure diagnostics with automatic log retrieval#33303
[dagster-aws] Add EMR step failure diagnostics with automatic log retrieval#33303Vamsi-klu wants to merge 1 commit into
Conversation
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: eec9fff387
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| try: | ||
| _, end_idx = decoder.raw_decode(trimmed) | ||
| except JSONDecodeError: | ||
| buffer = trimmed | ||
| break |
There was a problem hiding this comment.
If any log line contains a { that isn’t valid JSON (e.g., normal log messages or structured logs that aren’t DagsterEvent JSON), raw_decode throws and the handler keeps buffer set to that invalid prefix and breaks. Because the next iteration starts from the same bad {, subsequent valid Dagster event JSON in later lines never gets parsed, so CLI event extraction can silently drop all remaining events for that run. Consider clearing or advancing the buffer on JSONDecodeError (e.g., searching for the next {) so non-JSON braces don’t permanently block parsing.
Useful? React with 👍 / 👎.
|
@Vamsi-klu it looks like there are a lot of unrelated docs changes in this PR -- would you be able to move those changes to a different branch to keep this PR focused on EMR step failure diagnostics? |
…rieval Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Vamsi-klu
force-pushed
the
emr-step-failure-diagnostics
branch
from
eec9fff to
cfb87c8
Compare
|
@neverett thanks for the feedback and i have addressed your concern. can you please review my PR and let me know your thoughts now. Thanks! |
Vamsi-klu
force-pushed
the
emr-step-failure-diagnostics
branch
from
cfb87c8 to
f9d927a
Compare
Greptile SummaryThis PR adds automatic log retrieval on EMR step failure — when a step fails,
Confidence Score: 3/5The changes are additive and the failure path was previously a no-op, but two correctness issues need fixing before merging. The python_modules/libraries/dagster-aws/dagster_aws/emr/emr.py (_retrieve_step_failure_diagnostics timeout budget) and python_modules/libraries/dagster-aws/dagster_aws/pipes/clients/emr.py (_get_step_failure_details pagination)
|
| Filename | Overview |
|---|---|
| python_modules/libraries/dagster-aws/dagster_aws/emr/emr.py | Adds FailureLogConfig, StepFailureDiagnostics, and _retrieve_step_failure_diagnostics; the wait_timeout is applied per-log-file so the actual max wait can double the configured value |
| python_modules/libraries/dagster-aws/dagster_aws/pipes/clients/emr.py | Adds _get_step_failure_details to surface per-step errors; list_steps call is not paginated so clusters with >10 steps can silently miss the failing step |
| python_modules/libraries/dagster-aws/dagster_aws/emr/init.py | Re-exports FailureLogConfig and StepFailureDiagnostics from the package; straightforward addition |
| python_modules/libraries/dagster-aws/dagster_aws_tests/emr_tests/test_emr.py | Adds four new tests for failure diagnostics; mocks cover the main paths but the timeout-doubling scenario and pagination gap are not exercised |
Sequence Diagram
%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
participant Caller
participant EmrJobRunner
participant EMR as AWS EMR
participant S3
Caller->>EmrJobRunner: is_emr_step_complete(cluster_id, step_id)
EmrJobRunner->>EMR: describe_step()
EMR-->>EmrJobRunner: "step state = FAILED"
EmrJobRunner->>EmrJobRunner: _retrieve_step_failure_diagnostics()
EmrJobRunner->>EMR: "describe_cluster() [log_location_for_cluster #1]"
EMR-->>EmrJobRunner: LogUri
EmrJobRunner->>EMR: "describe_cluster() [log_location_for_cluster #2 inside retrieve_logs_for_step_id]"
EMR-->>EmrJobRunner: LogUri (duplicate call)
EmrJobRunner->>S3: wait_for_log(stdout.gz) up to wait_timeout seconds
S3-->>EmrJobRunner: stdout log
EmrJobRunner->>S3: wait_for_log(stderr.gz) up to wait_timeout seconds again
S3-->>EmrJobRunner: stderr log
EmrJobRunner-->>Caller: raise EmrError(message + diagnostics)
%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
participant Caller
participant EmrJobRunner
participant EMR as AWS EMR
participant S3
Caller->>EmrJobRunner: is_emr_step_complete(cluster_id, step_id)
EmrJobRunner->>EMR: describe_step()
EMR-->>EmrJobRunner: "step state = FAILED"
EmrJobRunner->>EmrJobRunner: _retrieve_step_failure_diagnostics()
EmrJobRunner->>EMR: "describe_cluster() [log_location_for_cluster #1]"
EMR-->>EmrJobRunner: LogUri
EmrJobRunner->>EMR: "describe_cluster() [log_location_for_cluster #2 inside retrieve_logs_for_step_id]"
EMR-->>EmrJobRunner: LogUri (duplicate call)
EmrJobRunner->>S3: wait_for_log(stdout.gz) up to wait_timeout seconds
S3-->>EmrJobRunner: stdout log
EmrJobRunner->>S3: wait_for_log(stderr.gz) up to wait_timeout seconds again
S3-->>EmrJobRunner: stderr log
EmrJobRunner-->>Caller: raise EmrError(message + diagnostics)
Reviews (1): Last reviewed commit: "[dagster-aws] Add EMR step failure diagn..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| # Calculate waiter settings from timeout | ||
| waiter_delay = 30 | ||
| waiter_max_attempts = max(1, self.failure_log_config.wait_timeout // waiter_delay) | ||
|
|
||
| stdout_log, stderr_log = self.retrieve_logs_for_step_id( | ||
| log, | ||
| cluster_id, | ||
| step_id, | ||
| waiter_delay=waiter_delay, | ||
| waiter_max_attempts=waiter_max_attempts, | ||
| ) |
There was a problem hiding this comment.
wait_timeout is applied per log file, doubling the actual maximum wait
retrieve_logs_for_step_id calls wait_for_log twice — once for stdout.gz and once for stderr.gz — passing the same waiter_max_attempts to each. With the default wait_timeout=600, each file can wait up to 600 s, making the real ceiling 1200 s (20 min) rather than the configured 10 min. A user who sets wait_timeout=60 to keep CI responsive will actually wait up to 2 minutes.
Halve waiter_max_attempts before passing it in (e.g. max(1, self.failure_log_config.wait_timeout // (2 * waiter_delay))), or split the budget after confirming the first file exists.
| self, | ||
| context: Union[OpExecutionContext, AssetExecutionContext], | ||
| cluster_id: str, | ||
| ) -> str: | ||
| """Retrieve failure details for all failed steps in the cluster. | ||
|
|
||
| Args: | ||
| context: The execution context for logging | ||
| cluster_id: The EMR cluster ID |
There was a problem hiding this comment.
list_steps is not paginated — failed steps beyond the first page are silently dropped
AWS's list_steps returns at most 10 steps per call and includes a Marker token for subsequent pages. Any cluster with more than 10 steps will have later-page steps omitted from failed_steps, so the error message can report "No step-level failure information available" even when a step did fail. The fix is to loop on Marker until it is absent from the response.
| try: | ||
| log_bucket, log_key_prefix = self.log_location_for_cluster(cluster_id) | ||
| logs_s3_uri = f"s3://{log_bucket}/{log_key_prefix}{cluster_id}/steps/{step_id}/" | ||
|
|
There was a problem hiding this comment.
log_location_for_cluster is called twice — one redundant describe_cluster API call
_retrieve_step_failure_diagnostics calls self.log_location_for_cluster(cluster_id) to build logs_s3_uri, and then retrieve_logs_for_step_id calls the same method internally (line 544) for its own log_bucket / log_key_prefix. This results in two describe_cluster API calls for every failure. Consider building the URI from the result already returned by retrieve_logs_for_step_id, or refactoring to pass the bucket/prefix down.
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: f9d927af95
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| self.aws_secret_access_key = check.opt_str_param( | ||
| aws_secret_access_key, "aws_secret_access_key" | ||
| ) | ||
| self.failure_log_config = failure_log_config or FailureLogConfig() |
There was a problem hiding this comment.
When EmrPySparkStepLauncher constructs EmrJobRunner without a config, this new default enables the 600s S3 log waiter even though the launcher’s wait_for_logs resource option defaults to False and documents that disabling it avoids the several-minute EMR log wait. In a failed Spark step on a cluster with LogUri, users who intentionally left wait_for_logs off now still block while logs are mirrored before the original failure is raised, and the launcher exposes no way to pass FailureLogConfig(retrieve_logs=False).
Useful? React with 👍 / 👎.
| waiter_delay = 30 | ||
| waiter_max_attempts = max(1, self.failure_log_config.wait_timeout // waiter_delay) | ||
|
|
||
| stdout_log, stderr_log = self.retrieve_logs_for_step_id( |
There was a problem hiding this comment.
For failures where stdout.gz is absent or delayed but stderr.gz exists, this combined call waits for stdout first and lets the EmrError abort the whole diagnostics path before stderr is read. That leaves the new failure message without the stderr excerpt users most need, and can spend the full timeout on a missing stdout object; fetch the two streams independently or try stderr first so one missing stream does not discard the other.
Useful? React with 👍 / 👎.
| context.log.error(f"[pipes] EMR job {cluster_id} failed") | ||
| raise Exception(f"[pipes] EMR job {cluster_id} failed:\n{cluster}") | ||
| # Get step-level failure information | ||
| failure_details = self._get_step_failure_details(context, cluster_id) |
There was a problem hiding this comment.
In real PipesEMRClient runs, cluster["Cluster"]["Status"]["State"] is a boto string (ClusterStateType), but EMR_CLUSTER_TERMINATED_STATES contains EmrClusterState enum members. This means the branch enclosing this newly added diagnostics call is skipped for TERMINATED_WITH_ERRORS, so _get_step_failure_details() never runs and the failed cluster can still be returned as a successful invocation; normalize state to EmrClusterState or compare against enum values before adding these details.
Useful? React with 👍 / 👎.
| A formatted string with step failure details | ||
| """ | ||
| try: | ||
| steps_response = self._client.list_steps(ClusterId=cluster_id) |
There was a problem hiding this comment.
When an EMR cluster has more steps than fit in one list_steps response, this reads only the first page before deciding whether any failed steps exist. If the failed/cancelled step is on a later page, the raised error says no step-level failure information is available even though EMR has it; use the paginator/marker or filter by failed step states so all relevant steps are considered.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: f9d927af95
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| wait_timeout: Seconds to wait for logs to appear in S3. Defaults to 600. | ||
| """ | ||
|
|
||
| retrieve_logs: bool = True |
There was a problem hiding this comment.
With retrieve_logs enabled by default, existing emr_pyspark_step_launcher users cannot opt out of the new S3 wait: that launcher still constructs EmrJobRunner(region=self.region_name) (pyspark_step_launcher.py:257) while its wait_for_logs option defaults to false and warns that waiting adds minutes (pyspark_step_launcher.py:68-75). On any failed step, is_emr_step_complete() now retrieves logs before raising, so failures can block for the 10-minute default even when users left wait_for_logs=False; please thread this setting into FailureLogConfig or keep retrieval opt-in for that path.
Useful? React with 👍 / 👎.
| stderr_excerpt = "\n".join(stderr_lines[-self.failure_log_config.stderr_lines :]) | ||
| if stdout_log: | ||
| stdout_lines = stdout_log.strip().split("\n") | ||
| stdout_excerpt = "\n".join(stdout_lines[-self.failure_log_config.stdout_lines :]) |
There was a problem hiding this comment.
If a caller sets FailureLogConfig(stderr_lines=0) or stdout_lines=0 to suppress excerpts, these slices use lines[-0:], which Python treats as lines[0:], so the error message includes the entire log instead of no lines. This defeats the new size/privacy limit for the zero case; guard on > 0 or validate that the configured line counts must be positive.
Useful? React with 👍 / 👎.
| waiter_delay = 30 | ||
| waiter_max_attempts = max(1, self.failure_log_config.wait_timeout // waiter_delay) | ||
|
|
||
| stdout_log, stderr_log = self.retrieve_logs_for_step_id( |
There was a problem hiding this comment.
For failed EMR steps where only one stdio object exists or one object is delayed, this combined call loses the useful stream: retrieve_logs_for_step_id() waits for stdout.gz and stderr.gz serially and raises if either waiter times out, so _retrieve_step_failure_diagnostics() catches the EmrError and returns no excerpt even if the other log was available. To make the new diagnostics reliable for steps that only emit stderr or only emit stdout, fetch each stream independently and include whichever one succeeds.
Useful? React with 👍 / 👎.
| A formatted string with step failure details | ||
| """ | ||
| try: | ||
| steps_response = self._client.list_steps(ClusterId=cluster_id) |
There was a problem hiding this comment.
For clusters with more than 50 steps, list_steps only returns the first page unless the Marker is followed, so this can miss failed/cancelled steps beyond that page and report No step-level failure information available even though failed steps exist. Since this helper is meant to surface all failed steps on a terminated Pipes EMR cluster, use the paginator or loop over Marker before deciding there are no failures.
Useful? React with 👍 / 👎.
| context.log.error(f"[pipes] EMR job {cluster_id} failed") | ||
| raise Exception(f"[pipes] EMR job {cluster_id} failed:\n{cluster}") | ||
| # Get step-level failure information | ||
| failure_details = self._get_step_failure_details(context, cluster_id) |
There was a problem hiding this comment.
When a Pipes EMR cluster finishes in TERMINATED_WITH_ERRORS, botocore's cluster_terminated waiter treats that state as a failure and raises before execution reaches this new call, so the added step-level diagnostics are skipped for the exact failure case they are meant to explain. Catch the waiter failure or poll/describe the cluster yourself before raising so _get_step_failure_details() can run on failed clusters.
Useful? React with 👍 / 👎.
2 participants
Summary & Motivation
EMR step failures previously only showed
"EMR step s-XXX failed"with no context. Users had to manually find logs in S3 via the AWS console. This PR makes failures self-diagnosing by automatically retrieving stdout/stderr excerpts from S3 and including them in the error message.What changed:
FailureLogConfig— controls how many log lines to retrieve, timeouts, and whether to fetch logs at allStepFailureDiagnostics— structured container for step ID, cluster ID, failure reason, log excerpts, and S3 URIEmrError— now includes formatted diagnostics (cluster ID, reason, log excerpts) when availableEmrJobRunner.is_emr_step_complete()— retrieves diagnostics on failure via new_retrieve_step_failure_diagnostics()PipesEMRClient— new_get_step_failure_details()surfaces step-level errors when clusters terminateCloses #1954
How I Tested These Changes
All 12 EMR tests pass (
pytest dagster_aws_tests/emr_tests/test_emr.py -v):test_emr_step_failure_diagnostics_with_config— log truncationtest_emr_step_failure_logs_disabled— config disables retrievaltest_emr_error_message_formatting— error message structuretest_is_emr_step_complete— diagnostics on failureChangelog
FailureLogConfigandStepFailureDiagnosticsfor configurable, structured failure diagnostics.