diff --git a/tools/bulk_executor/ai_lint/rules/dynamodb_failures_reported.md b/tools/bulk_executor/ai_lint/rules/dynamodb_failures_reported.md index c236deea..e3507a77 100644 --- a/tools/bulk_executor/ai_lint/rules/dynamodb_failures_reported.md +++ b/tools/bulk_executor/ai_lint/rules/dynamodb_failures_reported.md @@ -148,15 +148,32 @@ handler exists or is broad enough. ## Invariant 2 — driver-side work fails politely too A verb whose DynamoDB access happens on the **driver** (`find`, `count`, `sql` read -through `shared/glue_connector`'s `.load()`) has no worker code to check, but it is -not therefore exempt. A denied `.load()` must still reach the user as one sentence. - -The channel is `BulkExecutorError`: `root.py` catches it and calls -`sys.exit(str(e))`, Glue records that as the job's `ErrorMessage`, and the client -prints it as its closing line. The PITR guard already rides this path and produces -exactly one clean sentence. Anything else — a bare Spark/Py4J exception, or a plain -`Exception` — leaves the user a traceback prefixed with Glue's error category, even -when the text underneath is perfectly good. +through `shared/glue_connector`'s `.load()`; `load` writes through it) has no worker code +to check, but it is not therefore exempt. A denied `.load()` must still reach the user as +one sentence. + +Since #332 there is a net: `root.py` routes everything out of a verb through +`shared/driver_errors.py`, which classifies it the same way workers are classified and +**exits rather than re-raising**. Re-raising is what summons Glue's exception-analysis +blob and Py4J's restatement of the same failure. So a verb no longer *has* to convert its +own exception for the user to get a sentence — but converting is still better where the +verb knows something the net cannot, which is why `sql`, `load` and the transform loader +raise `BulkExecutorError` with their own wording. + +Two things to check when reading driver-side code: + +- **A marker must reach the console.** An understood failure logs behind + `driver_errors.EXPLAINED_FAILURE_PREFIX`, an unexpected one behind + `driver_errors.UNEXPECTED_FAILURE_BANNER`, and the client suppresses Glue's blob when it + sees either (or `BulkExecutorError`). Glue emits that blob after a clean `sys.exit` only + *sometimes* — measured: a denied `find` produced 183 blob lines while a denied `count` + in the same batch produced none — so a missing marker is a bug you will only see half + the time. +- **The message has to survive the Java stack.** A Py4J error's `str()` is the whole + stack. `get_error_message` unwraps it to the innermost cause, because the outer layer is + usually Spark boilerplate ("Job aborted due to stage failure") while the cause is AWS's + sentence. If you see a closing line reading `An error occurred while calling o304.load`, + that unwrapping did not happen. Two things that look like compliance and are not: @@ -280,13 +297,11 @@ State what you verified even when clean, so a pass is trustworthy. Outputs are kept in `~/Documents/bulk-331-runs/` with `before/` counterparts on `main` (597-668 lines each). -- **Does not conform to invariant 2, tracked as #332:** `find`, `count` and `sql`. - A table the role cannot `Scan` gives 314-324 lines closing on - `Error Category: UNCLASSIFIED_ERROR; Failed Line Number: 1362; An error occurred - while calling o304.load. User: ... is not authorized ...`. AWS's sentence is in - there, but it arrives as an unhandled Py4J exception. Compare the worker-side - verbs, which now close on `Error during delete: User ... is not authorized to - perform: dynamodb:BatchWriteItem` in 26-82 lines. +- Invariant 2 conforms as of #332, via the `root.py` net plus per-site wording in `sql`, + `load` and `transform_loader`. Before it, a table the role could not `Scan` gave + 314-324 lines closing on `Error Category: UNCLASSIFIED_ERROR; ...; An error occurred + while calling o304.load. User: ... is not authorized ...`; the sentence was in there, + but it arrived as an unhandled Py4J exception. - Partly audited: `load`, `load_export`, `revert_export`. Their shared write path (`shared/export/pipeline/writer.py`) is now correct, but whether every worker entry point in those paths records rather than escapes has not been checked — treat the diff --git a/tools/bulk_executor/client/src/runner.py b/tools/bulk_executor/client/src/runner.py index 188b96fa..ad751c39 100644 --- a/tools/bulk_executor/client/src/runner.py +++ b/tools/bulk_executor/client/src/runner.py @@ -55,12 +55,16 @@ ]) # Markers the job prints when it has already told the user what went wrong: a -# BulkExecutorError sentence, or a worker traceback that names the offending line. -# Either way Glue's exception analysis adds nothing but volume afterwards. -# The second must match worker_errors.UNEXPECTED_FAILURE_BANNER. +# BulkExecutorError sentence, or a traceback that names the offending line -- from a +# worker or from the driver. Either way Glue's exception analysis adds nothing but volume +# afterwards, and for a driver-side failure it also restates the same error through Py4J. +# The last two must match worker_errors.UNEXPECTED_FAILURE_BANNER and +# driver_errors.UNEXPECTED_FAILURE_BANNER (a guard test checks both). JOB_EXPLAINED_THE_FAILURE = ( 'BulkExecutorError', 'A worker failed in a way we did not expect. Traceback from the worker:', + 'The job failed in a way we did not expect. Traceback:', + 'Bulk Executor failure: ', ) # Timing constants diff --git a/tools/bulk_executor/client/src/utils/__init__.py b/tools/bulk_executor/client/src/utils/__init__.py index 80c34d92..a4b92a8a 100644 --- a/tools/bulk_executor/client/src/utils/__init__.py +++ b/tools/bulk_executor/client/src/utils/__init__.py @@ -38,6 +38,13 @@ # aws-glue-di-package.jar or metrics-core. Remove once AWS fixes the image. # Issue #334. r"Exception thrown from AWSDILyraMetricsReporter#report", + # Spark reports a query-analysis failure itself, at ERROR, before our handler sees it: + # one ~10 KB JSON event carrying the message we go on to print cleanly, plus ~90 Java + # frames and the unresolved query plan. Measured on `sql` with a mistyped column: 90 of + # the run's 148 lines. Anchored on the logger name inside that JSON, so it drops the + # whole event rather than leaving orphaned frames. The message still reaches the user + # through the verb's own "SQL query error: ..." line. Issue #332. + r'"logger": "SQLQueryContextLogger"', ] # Intentional nuanced configs: diff --git a/tools/bulk_executor/server/src/python_modules/load/__init__.py b/tools/bulk_executor/server/src/python_modules/load/__init__.py index fb9761ef..e82d42c7 100644 --- a/tools/bulk_executor/server/src/python_modules/load/__init__.py +++ b/tools/bulk_executor/server/src/python_modules/load/__init__.py @@ -81,20 +81,36 @@ def run(job, spark_context, glue_context, parsed_args): if not check_s3_file_exists(s3_path): raise BulkExecutorError(f"The S3 path '{s3_path}' doesn't exist or is not accessible") - dynamicFrame = read_data(glue_context, s3_path, parsed_args) - + # Inside the same handler as the count below: some formats fail here instead. Parquet + # reads its footer while the frame is being created, so `--format parquet` at a CSV + # file raises from read_data, while a JSON or CSV mismatch only surfaces at count(). + # Measured before this was wrapped: the Parquet case reported as an unexpected failure + # ("Py4JJavaError: ... is not a Parquet file") rather than as the user's own mistake. count = 0 try: + dynamicFrame = read_data(glue_context, s3_path, parsed_args) count = dynamicFrame.count() if count == 0: - log.error("No data found, please check your data source") # Should perhaps check that the path exists + # Logged as a warning, not an error: this run goes on to succeed, and an ERROR + # line above "Job completed successfully" is a contradiction the reader has to + # resolve. An empty drop is also a legitimate input -- the export pipeline + # treats a 0-item export the same way. Whether a source that holds bytes but + # yields no rows should fail instead is a separate question (#340). + log.warning( + f"Read 0 items from '{s3_path}' as {parsed_args.get('format')!r} -- " + f"nothing was loaded. If that is unexpected, check --format and the path.") return log.info(f"\nPreparing to load {count} items") log.info("Schema is:") dynamicFrame.printSchema() except Exception as e: - raise Exception(f"Failed to create DynamicFrame {e}") + # This is where Spark actually reads the source, so it fires on the everyday + # mistakes: --format json pointed at CSV, malformed JSON, an unreadable Parquet + # file. (A path that does not exist is caught earlier, before the job starts.) + raise BulkExecutorError( + f"Could not read the source at '{s3_path}' as {parsed_args.get('format')!r}: " + f"{get_error_message(e)}") from None if parsed_args.get('removeEmptyStringAttributes') is not None: log.debug(f"removeEmptyStringAttributes parameter was provided") @@ -114,7 +130,9 @@ def run(job, spark_context, glue_context, parsed_args): glue_context, df, table_name, parsed_args, write_rate=write_rate) log.info(f"Wrote {count} items to '{table_name}'") except Exception as e: - raise Exception(f"Error in writing to table: {get_error_message(e)}") from None + # The connector write: a read-only role or persistent throttling lands here, and + # get_error_message unwraps the Py4J stack to AWS's own sentence. + raise BulkExecutorError(f"Error in writing to table: {get_error_message(e)}") from None def check_s3_file_exists(s3_uri): """ diff --git a/tools/bulk_executor/server/src/python_modules/shared/driver_errors.py b/tools/bulk_executor/server/src/python_modules/shared/driver_errors.py new file mode 100644 index 00000000..200c78a1 --- /dev/null +++ b/tools/bulk_executor/server/src/python_modules/shared/driver_errors.py @@ -0,0 +1,83 @@ +"""Surfacing a failure that happened on the driver. + +`worker_errors.py` handles the failures a worker records; this handles the ones the +driver hits directly -- a denied `.load()` through the Glue connector, a source file +that isn't the format the user claimed, a typo in `--transform`. The classification is +the same one, imported from there, because the question is the same: did we understand +this? + +- understood -- a denial, expired credentials, throttling, or something a verb already + phrased as a BulkExecutorError. One sentence, no traceback. +- unexpected -- anything else. The traceback is printed once, on the console, and the + job still exits with a one-line reason. + +Either way `root.py` exits rather than re-raising, so Glue's exception-analysis blob and +its Py4J restatement of the same error never reach the user. That matters most for the +connector: a denied read arrives as a Py4JJavaError whose str() is a 300-line Java stack +with AWS's sentence buried on line two, and re-raising it means the last thing the user +sees is `An error occurred while calling o304.load`. +""" + +import sys +import traceback + +from python_modules.shared.bulk_executor_error import BulkExecutorError +from python_modules.shared.logger import log +from python_modules.shared.worker_errors import classify_failure + +UNEXPECTED_FAILURE_BANNER = "The job failed in a way we did not expect. Traceback:" + +# Printed in front of an understood failure's sentence. Two jobs: it tells the reader this +# is the explanation rather than one more log line, and the client watches for it to +# suppress Glue's exception-analysis blob. Glue emits that blob for a clean sys.exit only +# sometimes -- observed on a denied `find` but not a denied `count` in the same batch -- so +# the marker has to be there every time, not only when a BulkExecutorError was involved. +# +# Named rather than generic because the client matches it as a substring anywhere in a log +# line. "Failure: " alone appeared only in our own output across every run captured for +# #332, but Spark has shapes like ExecutorLostFailure and FetchFailure that could put +# "Failure: " in a line of its own -- and a false match would suppress Glue's diagnostics +# for a failure we had not explained. +EXPLAINED_FAILURE_PREFIX = "Bulk Executor failure: " + +# Cap on what we hand to sys.exit(): Glue records it as the job's ErrorMessage and the +# client prints it as its closing line. AWS's authorization sentences run ~400 chars, +# which is worth keeping whole; a Java stack that slipped through is not. +MAX_REASON_CHARS = 800 + + +def _one_line(message): + """Collapse to a single line and bound the length. The reason is a closing line, not + a report -- anything longer has already been printed above it.""" + collapsed = " ".join(str(message).split()) + if len(collapsed) > MAX_REASON_CHARS: + collapsed = collapsed[:MAX_REASON_CHARS - 3] + "..." + return collapsed + + +def surface(exception): + """Report a driver-side failure and exit. Never returns. + + Called by root.py for anything that escapes a verb. + """ + understood, message = classify_failure(exception) + if not understood: + # Ours to debug, or a user's generator/transform: the frames are the report, and + # printing them ourselves means the client suppresses Glue's blob (it watches for + # this banner and for BulkExecutorError) and the closing line stays short. + print(UNEXPECTED_FAILURE_BANNER) + print(traceback.format_exc()) + # classify_failure already extracted the readable message; the type goes in front + # because an unexpected message alone can be as bare as "'pk'". + reason = f"{type(exception).__name__}: {message}" + else: + reason = message + + reason = _one_line(reason) + if isinstance(exception, BulkExecutorError): + # Keep the name users have seen since before any of this, and the client's + # long-standing suppression marker. + log.error(f"BulkExecutorError: {reason}") + else: + log.error(f"{EXPLAINED_FAILURE_PREFIX}{reason}") + sys.exit(reason) diff --git a/tools/bulk_executor/server/src/python_modules/shared/errors.py b/tools/bulk_executor/server/src/python_modules/shared/errors.py index 6f58fb2a..56598cba 100644 --- a/tools/bulk_executor/server/src/python_modules/shared/errors.py +++ b/tools/bulk_executor/server/src/python_modules/shared/errors.py @@ -47,7 +47,7 @@ def get_error_message(e): except json.JSONDecodeError: pass # fallback - # Look for DynamoDB exception message + # Look for DynamoDB exception message (AWS SDK v1, which Glue 4 and earlier used) dynamo_match = re.search( r'com\.amazonaws\.services\.dynamodbv2\.model\.AmazonDynamoDBException:\s*(.*?)\s*\(Service:', msg @@ -55,6 +55,28 @@ def get_error_message(e): if dynamo_match: return dynamo_match.group(1).strip() + # A Java exception relayed through Py4J. str() on one of these is the entire Java + # stack -- hundreds of frames -- with the sentence that matters on the second line: + # + # py4j.protocol.Py4JJavaError: An error occurred while calling o304.load. + # : software.amazon.awssdk.services.dynamodb.model.DynamoDbException: User: ... + # is not authorized to perform: dynamodb:Scan on resource: ... + # at software.amazon.awssdk...(DynamoDbException.java:113) + # + # Take the innermost cause's message: for a wrapped failure the outer layer is + # usually Spark boilerplate ("Job aborted due to stage failure") and the cause is + # the AWS sentence the user needs. The message can wrap onto continuation lines, so + # keep going until a stack frame or a new exception header. + java_causes = re.findall( + r'^(?:: |Caused by: )(?:[\w$]+\.)+([\w$]*(?:Exception|Error)): ' + r'(.*(?:\n(?!\s*(?:at |\.\.\. )|: |Caused by: ).*)*)', + msg, re.MULTILINE) + if java_causes: + _cls, detail = java_causes[-1] + detail = ' '.join(detail.split()) + if detail: + return detail + # ParseException handling if hasattr(e, 'desc'): # ParseException msg = e.desc diff --git a/tools/bulk_executor/server/src/python_modules/shared/export/pipeline/transform_loader.py b/tools/bulk_executor/server/src/python_modules/shared/export/pipeline/transform_loader.py index b5093885..9c033c0c 100644 --- a/tools/bulk_executor/server/src/python_modules/shared/export/pipeline/transform_loader.py +++ b/tools/bulk_executor/server/src/python_modules/shared/export/pipeline/transform_loader.py @@ -1,4 +1,5 @@ import importlib +from ...bulk_executor_error import BulkExecutorError def load_transform_module(module_name, transform_package): @@ -18,4 +19,7 @@ def load_transform_module(module_name, transform_package): try: return importlib.import_module(f"{transform_package}.{module_name}") except ImportError as e: - raise ImportError(f"Cannot import transform module '{module_name}' from '{transform_package}': {e}") + # --transform is user input, so a typo is a sentence rather than a traceback. + raise BulkExecutorError( + f"Cannot import transform module '{module_name}' from '{transform_package}': {e}" + ) from None diff --git a/tools/bulk_executor/server/src/python_modules/sql.py b/tools/bulk_executor/server/src/python_modules/sql.py index dbdf4c14..bb735134 100644 --- a/tools/bulk_executor/server/src/python_modules/sql.py +++ b/tools/bulk_executor/server/src/python_modules/sql.py @@ -92,7 +92,8 @@ def run(job, spark_context, glue_context, parsed_args): # without wrapping them into an opaque "SQL query error". raise except Exception as e: - raise Exception("SQL query error: " + get_error_message(e)) from None + # The query is the user's to fix, so this is a sentence, not a stack trace. + raise BulkExecutorError("SQL query error: " + get_error_message(e)) from None finally: # Ensure Spark session cleanup try: diff --git a/tools/bulk_executor/server/src/root.py b/tools/bulk_executor/server/src/root.py index 43d1c6c2..af2ce9e6 100644 --- a/tools/bulk_executor/server/src/root.py +++ b/tools/bulk_executor/server/src/root.py @@ -7,6 +7,7 @@ from awsglue.job import Job from awsglue.transforms import * from pyspark.context import SparkContext +from python_modules.shared import driver_errors from python_modules.shared.bulk_executor_error import BulkExecutorError @@ -103,11 +104,21 @@ def _get_parsed_glue_job_args(argv): action_script_function = getattr(module, action_script_function_name) try: action_script_function(job, spark_context, glue_context, parsed_args) # Run the function - except BulkExecutorError as e: - log.error(f"BulkExecutorError: {e}") - sys.exit(str(e)) - except Exception as e: - raise # Just let it propagate + except BaseException as e: + # Everything a verb can fail with lands here, and shared/driver_errors.py + # decides what the user sees: a denial or other understood failure exits with + # one sentence, anything else prints its traceback first and then exits with a + # one-line reason. Exiting rather than re-raising is the point -- a re-raise + # hands the user Glue's exception-analysis blob plus Py4J's restatement of the + # same error, with AWS's actual sentence buried inside a Java stack (#332). + # + # BaseException rather than Exception so nothing escapes by inheriting from + # the wrong base -- but SystemExit passes straight through: a helper that + # already called exit() has said its piece, and re-reporting it would relabel + # a deliberate exit as a surprise. + if isinstance(e, SystemExit): + raise + driver_errors.surface(e) else: raise Exception(f"Could not find the function '{action_script_function_name}' within the module '{module_name}'.") diff --git a/tools/bulk_executor/tests/client/test_runner.py b/tools/bulk_executor/tests/client/test_runner.py index 7b93559b..9e47edeb 100644 --- a/tools/bulk_executor/tests/client/test_runner.py +++ b/tools/bulk_executor/tests/client/test_runner.py @@ -345,6 +345,77 @@ def test_suppresses_glue_metrics_reporter_stack(self, bulk_runner, capsys): assert captured.out == '' assert captured.err == '' + def test_driver_failure_marker_suppresses_the_glue_blob(self, bulk_runner, capsys, monkeypatch): + """#332: an understood driver-side failure prints one sentence, and Glue's + exception-analysis blob that sometimes follows a clean sys.exit is dropped. + + Glue emits that blob unpredictably -- observed after a denied `find` but not after a + denied `count` in the same batch -- so this is asserted here rather than trusted to + show up in a live run.""" + monkeypatch.setattr(runner_module.utils, 'CONFIG_LOG_MESSAGE_KEYS', []) + monkeypatch.setattr(runner_module.utils, 'STD_ERROR_MESSAGE_KEYS', []) + + denial = _make_event(message=( + "Bulk Executor failure: User: arn:aws:sts::1:assumed-role/R/GlueJobRunnerSession is not " + "authorized to perform: dynamodb:Scan on resource: table/t" + )) + bulk_runner._pretty_print_log_event(denial) + first = capsys.readouterr().out + assert 'not authorized to perform' in first, "the sentence itself must print" + + blob = _make_event(message=( + "2026-09-01 08:49:19 ERROR GlueExceptionAnalysisListener:9 - " + "[Glue Exception Analysis] {\"Failure Reason\": \"Traceback (most recent call last)...\"}" + )) + bulk_runner._pretty_print_log_event(blob) + assert capsys.readouterr().out == '', "Glue's restatement adds nothing after it" + + def test_suppresses_sparks_own_query_analysis_dump(self, bulk_runner, capsys): + """#332: Spark logs an analysis failure itself, at ERROR, as one JSON event holding + the message plus ~90 Java frames and the query plan -- before our handler turns it + into "SQL query error: ...". Measured on a mistyped column: 90 of 148 lines. + + The message below is the head of the verbatim CloudWatch event (10,850 chars, a + single event, so one anchor drops all of it). + """ + ev = _make_event(message="{\"ts\": \"2026-09-01 09:31:40.589\", \"level\": \"ERROR\", \"logger\": \"SQLQueryContextLogger\", \"msg\": \"[UNRESOLVED_COLUMN.WITH_SUGGESTION] A column, variable, or function parameter with name `nosuchcolumn` cannot be resolved. Did you mean one of the following? [`payload`, `sk`, `pk`]. SQLSTATE: 42703\", \"context\": {\"errorClass\": \"UNRESOLVED_COLUMN.WITH_SUGGESTION\"}, \"exception\": {\"class\": \"Py4JJavaError\", ") + bulk_runner._pretty_print_log_event(ev) + captured = capsys.readouterr() + assert captured.out == '' and captured.err == '' + + def test_our_own_sql_query_error_still_prints(self, bulk_runner, capsys): + """Guard: the anchor is Spark's logger name, not the error text, so the sentence the + verb produces is unaffected.""" + ev = _make_event(message=( + "2026-09-01 09:31:41,173 ERROR - BulkExecutorError: SQL query error: " + "[UNRESOLVED_COLUMN.WITH_SUGGESTION] A column with name `nosuchcolumn` cannot be resolved" + )) + bulk_runner._pretty_print_log_event(ev) + captured = capsys.readouterr() + assert 'UNRESOLVED_COLUMN' in captured.out + captured.err + + def test_sparks_own_failure_wording_does_not_trip_the_marker(self, bulk_runner, monkeypatch): + """The client matches markers as substrings, so a generic word would misfire. Spark + has plenty of "...Failure" shapes; none of them may be mistaken for the job saying + it has explained itself, or Glue's diagnostics get suppressed for a failure nobody + described.""" + monkeypatch.setattr(runner_module.utils, 'LOG_PATTERN_IGNORE_LIST', []) + monkeypatch.setattr(runner_module.utils, 'CONFIG_LOG_MESSAGE_KEYS', []) + monkeypatch.setattr(runner_module.utils, 'STD_ERROR_MESSAGE_KEYS', []) + + for spark_line in ( + "ERROR TaskSetManager: Lost task 3.0 in stage 2.0: ExecutorLostFailure: " + "executor 7 exited unrelated to the running tasks", + "WARN TaskSetManager: Lost task 1.0: FetchFailure: shuffle block missing", + 'ERROR GlueExceptionAnalysisListener:9 - {"Failure Reason": "boom"}', + "ERROR DAGScheduler: Job aborted due to stage failure: Task 0 failed 4 times", + ): + bulk_runner._suppress_glue_noise = False + bulk_runner._pretty_print_log_event(_make_event(message=spark_line)) + assert bulk_runner._suppress_glue_noise is False, ( + f"{spark_line[:60]!r} must not read as the job explaining itself" + ) + def test_real_worker_traceback_still_prints(self, bulk_runner, capsys): """Guard for #334: the anchor is Glue's reporter, not the frames. An unexpected worker failure prints its traceback through the same path and must survive.""" diff --git a/tools/bulk_executor/tests/server/shared/export/pipeline/test_transform_loader.py b/tools/bulk_executor/tests/server/shared/export/pipeline/test_transform_loader.py index 363ae5ed..cad942e1 100644 --- a/tools/bulk_executor/tests/server/shared/export/pipeline/test_transform_loader.py +++ b/tools/bulk_executor/tests/server/shared/export/pipeline/test_transform_loader.py @@ -24,5 +24,7 @@ def test_default_incremental_is_callable(self): assert callable(module.transform_incremental_record) def test_load_nonexistent_module(self): - with pytest.raises(ImportError): + """--transform names the module, so a typo is a sentence, not a traceback (#332).""" + from python_modules.shared.bulk_executor_error import BulkExecutorError + with pytest.raises(BulkExecutorError, match="Cannot import transform module"): load_transform_module('nonexistent_module', 'python_modules.load_export.transform') diff --git a/tools/bulk_executor/tests/server/shared/test_driver_errors.py b/tools/bulk_executor/tests/server/shared/test_driver_errors.py new file mode 100644 index 00000000..483eaa01 --- /dev/null +++ b/tools/bulk_executor/tests/server/shared/test_driver_errors.py @@ -0,0 +1,137 @@ +"""Unit tests for server/src/python_modules/shared/driver_errors.py. + +Covers surface(): an understood failure exits with one sentence and prints nothing else; +an unexpected one prints a banner plus its traceback and still exits with a one-line +reason; the reason is collapsed to a single line and bounded. + +The contract these protect is what #332 is about. A driver-side failure used to be +re-raised, which handed the user three copies of the same problem -- Glue's +exception-analysis blob, Py4J's restatement, and the Python traceback -- with AWS's +actual sentence buried in a Java stack. Now the driver decides, the same way it does for +failures a worker recorded. +""" + +import pytest + +from python_modules.shared import driver_errors, worker_errors +from python_modules.shared.bulk_executor_error import BulkExecutorError +from python_modules.shared.driver_errors import UNEXPECTED_FAILURE_BANNER, surface + +PY4J_DENIAL = """An error occurred while calling o304.load. +: software.amazon.awssdk.services.dynamodb.model.DynamoDbException: User: \ +arn:aws:sts::1:assumed-role/Role/GlueJobRunnerSession is not authorized to perform: \ +dynamodb:Scan on resource: arn:aws:dynamodb:us-east-1:1:table/t because no \ +identity-based policy allows the dynamodb:Scan action (Service: DynamoDb, Status Code: 400) +\tat software.amazon.awssdk.services.dynamodb.model.DynamoDbException$BuilderImpl.build(DynamoDbException.java:113) +\tat software.amazon.awssdk.core.internal.http.pipeline.stages.RetryableStage.execute(RetryableStage.java:86) +""" + + +def _real_errors_module(): + """Load shared/errors.py from disk. conftest replaces that module with a Mock for the + whole server suite, and str(exception) is not a substitute: the entire point here is + the extraction get_error_message does on a Py4J stack.""" + import importlib.util + from pathlib import Path + + path = Path(__file__).resolve().parents[3] / "server/src/python_modules/shared/errors.py" + spec = importlib.util.spec_from_file_location("_real_errors_for_test", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +@pytest.fixture(autouse=True) +def real_message_extraction(monkeypatch): + """The classifier binds these at import time, against conftest's Mock.""" + real = _real_errors_module() + monkeypatch.setattr(worker_errors, 'get_error_code', real.get_error_code) + monkeypatch.setattr(worker_errors, 'get_error_message', real.get_error_message) + + +class TestSurfaceUnderstood: + + def test_bulk_executor_error_exits_with_its_sentence(self, capsys): + with pytest.raises(SystemExit) as raised: + surface(BulkExecutorError("Invalid 'where': no such column")) + + assert str(raised.value) == "Invalid 'where': no such column" + assert capsys.readouterr().out == '', "nothing to add; the sentence is the report" + + def test_py4j_denial_exits_with_aws_own_sentence(self, capsys): + """The row that made #332 worth filing: 300 lines of Java, one useful sentence.""" + with pytest.raises(SystemExit) as raised: + surface(Exception(PY4J_DENIAL)) + + reason = str(raised.value) + assert 'is not authorized to perform: dynamodb:Scan' in reason + assert 'at software.amazon' not in reason, "no Java frames in the closing line" + assert 'o304.load' not in reason, "Py4J's own wrapper text is not the problem" + assert capsys.readouterr().out == '', "a denial needs no traceback" + + def test_logs_behind_the_marker_the_client_watches(self, caplog): + """The sentence must carry EXPLAINED_FAILURE_PREFIX. Without it the client cannot + tell the job has explained itself, and Glue's exception-analysis blob -- which + follows a clean exit only sometimes -- reaches the user anyway.""" + import logging + + with caplog.at_level(logging.ERROR): + with pytest.raises(SystemExit): + surface(Exception(PY4J_DENIAL)) + + logged = ' '.join(r.message for r in caplog.records) + assert driver_errors.EXPLAINED_FAILURE_PREFIX in logged + assert 'is not authorized' in logged + + def test_bulk_executor_error_keeps_its_own_marker(self, caplog): + """The name users have seen for years, and the client's original marker.""" + import logging + + with caplog.at_level(logging.ERROR): + with pytest.raises(SystemExit): + surface(BulkExecutorError("PITR must be enabled")) + + logged = ' '.join(r.message for r in caplog.records) + assert 'BulkExecutorError: PITR must be enabled' in logged + + def test_reason_is_one_line(self): + with pytest.raises(SystemExit) as raised: + surface(BulkExecutorError("first line\nsecond line\n\tindented third")) + + assert str(raised.value) == "first line second line indented third" + + def test_long_reason_is_bounded(self): + with pytest.raises(SystemExit) as raised: + surface(BulkExecutorError("x" * 5000)) + + reason = str(raised.value) + assert len(reason) == driver_errors.MAX_REASON_CHARS + assert reason.endswith('...') + + +class TestSurfaceUnexpected: + + def test_prints_banner_and_traceback_then_exits_one_line(self, capsys): + try: + raise KeyError('pk') + except KeyError as e: + with pytest.raises(SystemExit) as raised: + surface(e) + + out = capsys.readouterr().out + assert UNEXPECTED_FAILURE_BANNER in out + assert 'Traceback' in out and "raise KeyError('pk')" in out, \ + "the frames are the report for something we did not expect" + + reason = str(raised.value) + assert reason == "KeyError: 'pk'", "the type, since a KeyError message is bare" + assert 'Traceback' not in reason, \ + "the reason becomes Glue's ErrorMessage and the client's closing line" + + def test_exits_rather_than_re_raising(self): + """Re-raising is what produced the Glue blob and the Py4J restatement.""" + try: + raise RuntimeError('boom') + except RuntimeError as e: + with pytest.raises(SystemExit): + surface(e) diff --git a/tools/bulk_executor/tests/server/shared/test_errors_message_extraction.py b/tools/bulk_executor/tests/server/shared/test_errors_message_extraction.py new file mode 100644 index 00000000..92e5e25c --- /dev/null +++ b/tools/bulk_executor/tests/server/shared/test_errors_message_extraction.py @@ -0,0 +1,126 @@ +"""Unit tests for get_error_message's Java/Py4J unwrapping in shared/errors.py. + +A denied connector read reaches the driver as a Py4JJavaError whose str() is the whole +Java stack -- hundreds of frames -- with the sentence the user needs on the second line. +Before #332 nothing extracted it: the SDK v1 pattern in this function stopped matching +when Glue moved to the v2 SDK (`software.amazon.awssdk...DynamoDbException` rather than +`com.amazonaws...AmazonDynamoDBException`), so the "message" was the stack itself. + +shared/errors.py is replaced by a Mock for the whole server suite, so these tests load it +from disk. That is deliberate: the extraction is the behaviour under test. +""" + +import importlib + +import pytest + + +@pytest.fixture +def errors(): + """Import the real module under its own name, with conftest's Mock put back after. + + Loading it from a file path under a different module name also works, but coverage + does not credit the execution, so the extraction below would read as untested and + invite someone to delete it.""" + import sys + + saved = {name: sys.modules[name] for name in + ('python_modules.shared.errors', 'shared.errors') if name in sys.modules} + for name in saved: + del sys.modules[name] + try: + yield importlib.import_module('python_modules.shared.errors') + finally: + for name in list(sys.modules): + if name in ('python_modules.shared.errors', 'shared.errors'): + del sys.modules[name] + sys.modules.update(saved) + + +DENIAL_V2 = """An error occurred while calling o304.load. +: software.amazon.awssdk.services.dynamodb.model.DynamoDbException: User: \ +arn:aws:sts::1:assumed-role/R/GlueJobRunnerSession is not authorized to perform: \ +dynamodb:Scan on resource: arn:aws:dynamodb:us-east-1:1:table/t (Service: DynamoDb, Status Code: 400) +\tat software.amazon.awssdk.services.dynamodb.model.DynamoDbException$BuilderImpl.build(DynamoDbException.java:113) +\tat org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils$.loadV2Source(DataSourceV2Utils.scala:157) +""" + +WRAPPED_IN_SPARK = """An error occurred while calling o92.save. +: org.apache.spark.SparkException: Job aborted due to stage failure: Task 3 in stage 2.0 failed 4 times +\tat org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2905) +Caused by: software.amazon.awssdk.services.dynamodb.model.DynamoDbException: User: R is not \ +authorized to perform: dynamodb:BatchWriteItem on resource: table/t +\tat software.amazon.awssdk.core.internal.http.pipeline.stages.RetryableStage.execute(RetryableStage.java:86) +""" + +MULTILINE_DETAIL = """An error occurred while calling o1.load. +: java.lang.IllegalArgumentException: Unsupported option 'dynamodb.throughput.read.percent' + supplied for this connector version; use dynamodb.throughput.read instead +\tat com.amazonaws.services.glue.connectors.DynamoDbOptions.validate(DynamoDbOptions.scala:88) +""" + + +class TestJavaErrorUnwrapping: + + def test_sdk_v2_denial_yields_aws_sentence(self, errors): + message = errors.get_error_message(Exception(DENIAL_V2)) + assert message.startswith('User: arn:aws:sts::1:assumed-role/R/GlueJobRunnerSession') + assert 'dynamodb:Scan' in message + assert 'at software.amazon' not in message and 'at org.apache.spark' not in message + assert 'o304.load' not in message, "Py4J's wrapper line is not the error" + + def test_innermost_cause_wins_over_spark_boilerplate(self, errors): + """The outer layer says "Job aborted due to stage failure"; the cause says why.""" + message = errors.get_error_message(Exception(WRAPPED_IN_SPARK)) + assert 'not authorized to perform: dynamodb:BatchWriteItem' in message + assert 'Job aborted due to stage failure' not in message + + def test_message_continuation_lines_are_joined(self, errors): + message = errors.get_error_message(Exception(MULTILINE_DETAIL)) + assert message == ( + "Unsupported option 'dynamodb.throughput.read.percent' supplied for this " + "connector version; use dynamodb.throughput.read instead" + ) + + def test_sdk_v1_pattern_still_works(self, errors): + """Glue 4 and earlier; kept so an older runtime does not regress.""" + v1 = ("com.amazonaws.services.dynamodbv2.model.AmazonDynamoDBException: " + "Requested resource not found (Service: AmazonDynamoDBv2; Status Code: 400)") + assert errors.get_error_message(Exception(v1)) == "Requested resource not found" + + def test_spark_parse_exception_splits_off_the_sql(self, errors): + """Spark's ParseException carries the offending SQL after a marker line; the + function pulls it onto one line. Pre-existing behaviour, untested until this + file had a way to import the module.""" + class ParseError(Exception): + desc = "Syntax error at or near 'FRM'\n== SQL ==\nSELECT * FRM t\n ^^^" + + assert errors.get_error_message(ParseError()) == ( + "Syntax error at or near 'FRM' | SQL: SELECT * FRM t | ^^^" + ) + + def test_parse_exception_without_a_sql_section(self, errors): + class ParseError(Exception): + desc = " Syntax error, nothing more " + + assert errors.get_error_message(ParseError()) == "Syntax error, nothing more" + + def test_exception_with_a_message_attribute(self, errors): + """Some Py4J/Java wrappers expose .message rather than a useful str().""" + class WithMessage(Exception): + message = " the useful part " + + def __str__(self): + return "an unhelpful repr" + + assert errors.get_error_message(WithMessage()) == "the useful part" + + def test_a_plain_exception_is_unchanged(self, errors): + assert errors.get_error_message(ValueError("just a message")) == "just a message" + + def test_boto_error_response_still_preferred(self, errors): + """An AWS SDK error carries its own message; do not go looking in the string.""" + class Fake(Exception): + response = {'Error': {'Code': 'AccessDeniedException', 'Message': 'denied by policy'}} + + assert errors.get_error_message(Fake()) == 'denied by policy' diff --git a/tools/bulk_executor/tests/server/test_load.py b/tools/bulk_executor/tests/server/test_load.py index 96107c54..a08ba256 100644 --- a/tools/bulk_executor/tests/server/test_load.py +++ b/tools/bulk_executor/tests/server/test_load.py @@ -273,8 +273,50 @@ def test_returns_early_when_count_is_zero(self, monkeypatch): {'table': 't', 's3_path': 's3://b/k', 'format': 'csv'}) assert result is None + def test_zero_rows_warns_and_succeeds(self, monkeypatch, caplog): + """A run that goes on to succeed must not print an ERROR line. Measured before this + change: "ERROR - No data found, please check your data source" immediately above + "Job completed successfully", with exit 0.""" + import logging + + monkeypatch.setattr(load_module, 'check_s3_file_exists', lambda uri: True) + df = MagicMock() + df.count.return_value = 0 + monkeypatch.setattr(load_module, 'read_data', lambda *a: df) + write = MagicMock() + monkeypatch.setattr(load_module, 'write_dynamodb_dataframe', write) + + with caplog.at_level(logging.WARNING): + load_module.run(MagicMock(), MagicMock(), MagicMock(), + {'table': 't', 's3_path': 's3://b/k', 'format': 'json'}) + + levels = {r.levelname for r in caplog.records} + assert 'ERROR' not in levels, "a successful run must not log an error" + message = ' '.join(r.message for r in caplog.records) + assert 'Read 0 items' in message and "'json'" in message and '--format' in message, \ + "say what happened and what to check" + write.assert_not_called() + + def test_raises_when_the_frame_cannot_even_be_created(self, monkeypatch): + """Parquet reads its footer while the frame is created, so `--format parquet` at a + CSV file raises from read_data rather than from count(). Both are the same mistake + and must read the same way; measured live before this was wrapped, the Parquet case + came out as "Py4JJavaError: ... is not a Parquet file" with a traceback.""" + monkeypatch.setattr(load_module, 'check_s3_file_exists', lambda uri: True) + + def boom(*_args): + raise RuntimeError("s3://b/k/data.csv is not a Parquet file") + + monkeypatch.setattr(load_module, 'read_data', boom) + + with pytest.raises(load_module.BulkExecutorError, match="Could not read the source") as exc: + load_module.run(MagicMock(), MagicMock(), MagicMock(), + {'table': 't', 's3_path': 's3://b/k/data.csv', 'format': 'parquet'}) + assert "'parquet'" in str(exc.value), "name the format the user claimed" + def test_raises_on_count_exception(self, monkeypatch): - """Line 99-100: exception during count() re-raises wrapped.""" + """A failure reading the source is the user's to fix -- wrong --format, malformed + data -- so it reports as a sentence naming the path and format (#332).""" monkeypatch.setattr(load_module, 'check_s3_file_exists', lambda uri: True) monkeypatch.setattr(load_module, 'get_dynamodb_throughput_configs', lambda *a, **kw: {}) @@ -282,9 +324,12 @@ def test_raises_on_count_exception(self, monkeypatch): df.count.side_effect = RuntimeError('spark error') monkeypatch.setattr(load_module, 'read_data', lambda *a: df) - with pytest.raises(Exception, match="Failed to create DynamicFrame"): + with pytest.raises(load_module.BulkExecutorError, match="Could not read the source") as exc: load_module.run(MagicMock(), MagicMock(), MagicMock(), {'table': 't', 's3_path': 's3://b/k', 'format': 'csv'}) + assert "s3://b/k" in str(exc.value) and "'csv'" in str(exc.value), ( + "name the path and the format the user claimed it was" + ) class TestRunRemoveEmptyStrings: diff --git a/tools/bulk_executor/tests/server/test_root.py b/tools/bulk_executor/tests/server/test_root.py index cc04f7c9..3a102b06 100644 --- a/tools/bulk_executor/tests/server/test_root.py +++ b/tools/bulk_executor/tests/server/test_root.py @@ -113,12 +113,18 @@ def _install_logger_stub(modules_to_install): def _install_bulk_executor_error_stub(modules_to_install): - """Provide a real exception class so `except BulkExecutorError` actually matches.""" - be_module = types.ModuleType("python_modules.shared.bulk_executor_error") - - class BulkExecutorError(Exception): - pass + """Expose the class the classifier itself holds, under the stubbed module path. + + Not a fresh subclass. root.py now reports through shared/driver_errors.py, which asks + shared/worker_errors.py whether the failure is understood -- and that check is an + isinstance against the class *it* bound at import time. conftest registers both + `shared.` and `python_modules.shared.` over the same files, so importing the module + again can produce a second, distinct class; the isinstance would then miss and the + test would assert nothing about the branch it names. + """ + from python_modules.shared.worker_errors import BulkExecutorError + be_module = types.ModuleType("python_modules.shared.bulk_executor_error") be_module.BulkExecutorError = BulkExecutorError modules_to_install["python_modules.shared.bulk_executor_error"] = be_module return BulkExecutorError @@ -144,6 +150,23 @@ def _load_root(monkeypatch, argv, verb_module=None, verb_name=None, BulkExecutorError = _install_bulk_executor_error_stub(install) + # tests/server/conftest.py replaces shared.errors with a Mock, so get_error_message + # would hand back a Mock repr and any assertion on the message text would be vacuous. + # driver_errors needs a working one to build the closing line. + errors_module = types.ModuleType("python_modules.shared.errors") + errors_module.get_error_message = lambda e: str(e) + errors_module.get_error_code = lambda e: None + errors_module.ListAccumulator = object + install["python_modules.shared.errors"] = errors_module + + # shared/worker_errors.py holds the classifier root.py's reporting path uses, and it + # binds these two names at import time -- which already happened, against conftest's + # Mock. Installing a module above does not rebind them, so patch them where they are + # read or every assertion on message text is vacuous. + import python_modules.shared.worker_errors as _worker_errors + monkeypatch.setattr(_worker_errors, "get_error_message", lambda e: str(e)) + monkeypatch.setattr(_worker_errors, "get_error_code", lambda e: None) + # The verb module is what root.py imports via importlib.import_module. # If `import_should_fail` is set, we leave python_modules. # uninstalled and patch importlib.import_module to raise ImportError so @@ -581,23 +604,51 @@ def test_bulk_executor_error_exits_cleanly(self, monkeypatch): class TestRootGenericExceptionPropagation: - """Generic Exception from a verb is re-raised (lines 88-89).""" + """A non-BulkExecutorError from a verb is reported, then exits (#332). + + It used to be re-raised, which handed the user Glue's exception-analysis blob and + Py4J's restatement of the same failure on top of the traceback. root.py now routes + everything through shared/driver_errors.py: the traceback is printed once for the + user, and the job exits with a one-line reason that becomes Glue's ErrorMessage. + """ - def test_generic_exception_propagates(self, monkeypatch): - """Lines 88-89: a non-BulkExecutorError exception is re-raised verbatim.""" + def test_generic_exception_is_reported_then_exits(self, monkeypatch, capsys): run_mock = MagicMock(side_effect=RuntimeError("boom")) verb = _make_verb_module("copy", run_callable=run_mock) - with pytest.raises(RuntimeError, match="boom"): + with pytest.raises(SystemExit) as exc_info: _load_root(monkeypatch, ["root.py", "--XAction", "copy"], verb_module=verb, verb_name="copy") + reason = str(exc_info.value) + assert "boom" in reason, "the closing line names the failure" + assert "Traceback" not in reason, ( + "the reason is Glue's ErrorMessage and the client's last line -- keep it to one" + ) + out = capsys.readouterr().out + assert "did not expect" in out and "Traceback" in out, ( + "an unexpected failure prints its traceback where the user will see it" + ) + + def test_a_verbs_own_exit_passes_straight_through(self, monkeypatch, capsys): + """A helper that already called exit() has said its piece. Re-reporting it would + relabel a deliberate exit as a surprise and print a traceback for it.""" + run_mock = MagicMock(side_effect=SystemExit("PITR must be enabled first")) + verb = _make_verb_module("copy", run_callable=run_mock) + with pytest.raises(SystemExit) as exc_info: + _load_root(monkeypatch, + ["root.py", "--XAction", "copy"], + verb_module=verb, verb_name="copy") + + assert str(exc_info.value) == "PITR must be enabled first", "unchanged, not re-wrapped" + assert "did not expect" not in capsys.readouterr().out + def test_generic_exception_skips_commit_and_stop(self, monkeypatch): - """Lines 93-94: exception path bypasses job.commit() and spark.stop().""" + """A failed job must not be committed -- that is what makes Glue mark it FAILED.""" run_mock = MagicMock(side_effect=ValueError("nope")) verb = _make_verb_module("copy", run_callable=run_mock) awsglue = _build_awsglue_stubs() - with pytest.raises(ValueError): + with pytest.raises(SystemExit): _load_root(monkeypatch, ["root.py", "--XAction", "copy"], verb_module=verb, verb_name="copy", diff --git a/tools/bulk_executor/tests/server/test_sql.py b/tools/bulk_executor/tests/server/test_sql.py index 190cad97..cedd0382 100644 --- a/tools/bulk_executor/tests/server/test_sql.py +++ b/tools/bulk_executor/tests/server/test_sql.py @@ -424,9 +424,11 @@ def test_valid_limit_runtime_failure_wraps_as_sql_query_error(self, monkeypatch, mock_table_info, mock_spark_session, mock_get_error_message, glue_context, base_args): - """A valid integer limit that then fails in Spark is a runtime error, not a - bad-parameter error: it flows to the outer 'SQL query error' handler rather - than being mislabeled 'Invalid limit'.""" + """A valid integer limit that then fails in Spark still reports as a query error + rather than being mislabeled 'Invalid limit'. + + Since #332 both are BulkExecutorError -- a query is the user's to fix either way, + so neither deserves a stack trace. The distinction that matters is the message.""" base_args['limit'] = '5' result = MagicMock() result.limit.side_effect = RuntimeError("spark limit failure") @@ -434,10 +436,11 @@ def test_valid_limit_runtime_failure_wraps_as_sql_query_error(self, monkeypatch, df = MagicMock() glue_context.create_dynamic_frame.from_options.return_value.toDF.return_value = df - with pytest.raises(Exception, match="SQL query error") as exc: + with pytest.raises(sql_module.BulkExecutorError, match="SQL query error") as exc: sql_module.run(MagicMock(), MagicMock(), glue_context, base_args) - # It is NOT a clean user-parameter error. - assert not isinstance(exc.value, sql_module.BulkExecutorError) + assert "Invalid 'limit'" not in str(exc.value), ( + "a Spark-side failure must not be reported as a bad --limit" + ) mock_get_error_message.assert_called() diff --git a/tools/bulk_executor/tests/server/test_worker_failures_go_through_worker_errors.py b/tools/bulk_executor/tests/server/test_worker_failures_go_through_worker_errors.py index ac5d2433..44f8563e 100644 --- a/tools/bulk_executor/tests/server/test_worker_failures_go_through_worker_errors.py +++ b/tools/bulk_executor/tests/server/test_worker_failures_go_through_worker_errors.py @@ -18,6 +18,8 @@ """ import re + +import pytest from pathlib import Path _REPO = Path(__file__).resolve().parents[2] @@ -89,16 +91,23 @@ def test_the_seam_is_actually_used(): f"expected several verbs to surface worker failures, found {surfacing}") -def test_the_client_watches_for_the_banner_the_job_prints(): +@pytest.mark.parametrize("constant", [ + ("worker_errors.py", "UNEXPECTED_FAILURE_BANNER"), + ("driver_errors.py", "UNEXPECTED_FAILURE_BANNER"), + ("driver_errors.py", "EXPLAINED_FAILURE_PREFIX"), +]) +def test_the_client_watches_for_every_marker_the_job_prints(constant): """The client suppresses Glue's exception-analysis noise once the job has explained - itself. That handshake is two string literals in two trees: if they drift, an - unexpected failure gets the worker traceback *and* a Glue traceback of our plumbing - on top of it, which is exactly what the banner exists to prevent.""" - server = (_SERVER_SRC / "python_modules" / "shared" / "worker_errors.py").read_text() - banner = re.search(r'UNEXPECTED_FAILURE_BANNER = "([^"]+)"', server) - assert banner, "worker_errors.py no longer defines UNEXPECTED_FAILURE_BANNER" + itself. That handshake is string literals in two trees: if they drift, an unexpected + failure gets our traceback *and* Glue's blob on top of it, which is exactly what the + banner exists to prevent. Both sides print one -- a worker failure and a driver + failure -- so both are checked.""" + module, name = constant + server = (_SERVER_SRC / "python_modules" / "shared" / module).read_text() + marker = re.search(rf'{name} = "([^"]+)"', server) + assert marker, f"{module} no longer defines {name}" client = (_REPO / "client" / "src" / "runner.py").read_text() - assert banner.group(1) in client, ( - f"client/src/runner.py does not watch for {banner.group(1)!r}; " - "JOB_EXPLAINED_THE_FAILURE has drifted from worker_errors.py") + assert marker.group(1) in client, ( + f"client/src/runner.py does not watch for {marker.group(1)!r}; " + f"JOB_EXPLAINED_THE_FAILURE has drifted from {module}.{name}")