This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 863440537b7 Keep service logs and OpenLineage e2e noise out of the job
output (#75811)
863440537b7 is described below
commit 863440537b73c8773392a1a2a57d93d78a9c6cbf
Author: Kacper Muda <[email protected]>
AuthorDate: Wed Oct 28 15:58:06 2026 +0100
Keep service logs and OpenLineage e2e noise out of the job output (#75811)
* Keep service logs and OpenLineage e2e noise out of the job output
The e2e conftest dumped the full stdout of every compose service into the
job output on failure, and the OpenLineage task log highlights printed every
warning and error line. The result is output too long to read, while the
service logs were not part of the uploaded logs.zip.
Write the service logs to files next to the other uploaded logs, and print
only the lines tagged as OpenLineage validation errors.
Co-Authored-By: Claude <[email protected]>
* Tell where the e2e logs went instead of printing them
Co-Authored-By: Claude <[email protected]>
---------
Co-authored-by: Claude <[email protected]>
---
.../tests/airflow_e2e_tests/conftest.py | 20 ++++----
.../airflow_e2e_tests/openlineage_tests/README.md | 13 ++---
.../openlineage_tests/task_logs.py | 59 +++++++++-------------
.../tests/system/openlineage/operator.py | 32 ++++++++----
4 files changed, 64 insertions(+), 60 deletions(-)
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
index 09331b1645d..ca6477878bc 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
@@ -993,20 +993,22 @@ def spin_up_airflow_environment(tmp_path_factory:
pytest.TempPathFactory):
except Exception:
console.print("[red]Failed to start docker compose")
if _E2ETestState.compose_instance:
- _print_logs(_E2ETestState.compose_instance)
+ _save_service_logs(_E2ETestState.compose_instance)
_E2ETestState.compose_instance.stop()
raise
-def _print_logs(compose_instance: DockerCompose):
- containers = compose_instance.get_containers()
- for container in containers:
+def _save_service_logs(compose_instance: DockerCompose):
+ LOGS_FOLDER.mkdir(parents=True, exist_ok=True)
+ for container in compose_instance.get_containers():
service = container.Service
if service:
stdout, _ = compose_instance.get_logs(service)
- console.print(f"::group:: {service} Logs")
- console.print(stdout, style="red", soft_wrap=True, markup=False)
- console.print("::endgroup::")
+ (LOGS_FOLDER / f"{service}.log").write_text(stdout)
+ console.print(
+ f"[yellow]Service and task logs are not printed here: find them in
{LOGS_FOLDER} "
+ f"(in CI, the `e2e-test-logs-<mode>` artifact, `logs.zip`)"
+ )
def pytest_sessionstart(session: pytest.Session):
@@ -1044,9 +1046,9 @@ def pytest_sessionfinish(session: pytest.Session,
exitstatus: int | pytest.ExitC
copytree(_E2ETestState.airflow_logs_path, LOGS_FOLDER,
dirs_exist_ok=True)
if _E2ETestState.compose_instance:
- # If any test failures lets print the services logs
+ # If any test failures lets save the services logs next to the
uploaded task logs
if any(r["status"] == "failed" for r in test_results):
- _print_logs(_E2ETestState.compose_instance)
+ _save_service_logs(_E2ETestState.compose_instance)
if not os.environ.get("SKIP_DOCKER_COMPOSE_DELETION"):
_E2ETestState.compose_instance.stop()
diff --git
a/airflow-e2e-tests/tests/airflow_e2e_tests/openlineage_tests/README.md
b/airflow-e2e-tests/tests/airflow_e2e_tests/openlineage_tests/README.md
index feb1c169cfb..5ef8c621f48 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/openlineage_tests/README.md
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/openlineage_tests/README.md
@@ -54,12 +54,13 @@ breeze testing airflow-e2e-tests --e2e-test-mode openlineage
- **pytest summary:** `1 passed` means every expected DAG is green. On failure
it prints the
offenders, e.g. `... not successful: {'openlineage_docs_file_dag':
'failed'}`, and logs
`⚠ DAGs that passed only on retry (flaky first run): [...]`.
-- **On failure, the relevant task logs are printed right in the pytest
output** (`check_events` plus
- any task that didn't succeed, filtered to warning/error lines — see
`task_logs.py`) — no need to
- open the log artifact for the common case, e.g. the exact mismatch
- (``Path `job > facets > ...`: expected X but got Y``).
+- **On failure, the OpenLineage validation errors are printed right in the
pytest output**: only the
+ lines tagged `[OL_VALIDATION_ERROR]` (by `OpenLineageTestOperator`) in the
logs of `check_events` and of any
+ task that didn't succeed, without the tag — see `task_logs.py`. For example
+ ``Path `job > facets > ...`: expected X but got Y``. Keep the tag in
`task_logs.py` in sync with
+ `VALIDATION_ERROR_TAG` in the provider's `system/openlineage/operator.py`.
- **Per-DAG logs** are still written in full under `airflow-e2e-tests/logs/`
for deeper digging
- beyond what gets printed.
+ beyond what gets printed, together with one `<service>.log` per compose
service.
- With `--skip-docker-compose-deletion`, the stack stays up: get a token from
`curl -s http://localhost:8080/auth/token` and check `GET
/api/v2/importErrors` for DAG parse
errors, or browse the runs in the UI at <http://localhost:8080> (login
`airflow` / `airflow`).
@@ -90,7 +91,7 @@ when a new DAG is version-gated.
```text
prepare_dags.py # sources DAGs from the provider system tests into
dags/ at runtime
harness.py # triggers every DAG via the REST API and collects
final run states
-task_logs.py # locates and prints the relevant task logs on
failure
+task_logs.py # prints the tagged validation errors of task logs
on failure
test_openlineage_e2e.py # the pytest entrypoint
dags_extra/ # harness-only DAGs/modules (warmup DAG, versioned
bundle) copied into dags/
```
diff --git
a/airflow-e2e-tests/tests/airflow_e2e_tests/openlineage_tests/task_logs.py
b/airflow-e2e-tests/tests/airflow_e2e_tests/openlineage_tests/task_logs.py
index f5d3c19170b..53615033e00 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/openlineage_tests/task_logs.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/openlineage_tests/task_logs.py
@@ -15,12 +15,13 @@
# specific language governing permissions and limitations
# under the License.
"""
-Read and format Airflow task logs for display in pytest output on failure.
+Print the OpenLineage validation errors of Airflow task logs in pytest output
on failure.
Task logs are structlog JSON-lines written by the deployed Airflow stack; they
land on the host via
-the ``./logs:/opt/airflow/logs`` bind mount in ``docker-compose.yaml``. Rather
than requiring someone
-to unzip the CI log artifact and open a specific ``attempt=N.log`` file, the
warning/error highlights
-are printed straight into the failing test's captured output.
+the ``./logs:/opt/airflow/logs`` bind mount in ``docker-compose.yaml``. Full
logs are too long for the CI
+output, so only the lines tagged with ``VALIDATION_ERROR_TAG`` are printed:
``OpenLineageTestOperator``
+tags every reason its event validation failed, as do the other helpers of the
system test Dags. The full
+logs stay in the CI log artifact.
"""
from __future__ import annotations
@@ -34,31 +35,17 @@ from typing import TYPE_CHECKING
if TYPE_CHECKING:
from rich.console import Console
-LOG_LEVELS_OF_INTEREST = frozenset({"warning", "error", "critical"})
-MAX_HIGHLIGHT_LINES = 100
+# Prefix of the log lines describing OpenLineage validation errors, keep in
sync with VALIDATION_ERROR_TAG in
+# providers/openlineage/tests/system/openlineage/operator.py
+VALIDATION_ERROR_TAG = "[OL_VALIDATION_ERROR]"
+MAX_ERROR_LINES = 100
_ATTEMPT_NUMBER_PATTERN = re.compile(r"attempt=(\d+)\.log$")
-def format_log_line(record: dict) -> str:
- """Render one parsed structlog record as a compact, human-readable line."""
- level = str(record.get("level", "info")).upper()
- event = record.get("event", "")
- lines = [f"[{level}] {event}"]
- for exc in record.get("error_detail") or []:
- exc_type = exc.get("exc_type", "Exception")
- exc_value = exc.get("exc_value", "")
- lines.append(f" {exc_type}: {exc_value}")
- frames = exc.get("frames") or []
- if frames:
- frame = frames[-1]
- lines.append(f" at
{Path(frame['filename']).name}:{frame['lineno']} in {frame['name']}")
- return "\n".join(lines)
-
-
-def filter_log_highlights(raw_lines: Iterable[str], max_lines: int =
MAX_HIGHLIGHT_LINES) -> list[str]:
- """Parse structlog JSON lines and keep only warning/error/critical ones,
formatted for display."""
- highlights = []
+def extract_validation_errors(raw_lines: Iterable[str], max_lines: int =
MAX_ERROR_LINES) -> list[str]:
+ """Parse structlog JSON lines and return the tagged validation errors,
without the tag."""
+ errors = []
for raw_line in raw_lines:
stripped_line = raw_line.strip()
if not stripped_line:
@@ -68,14 +55,13 @@ def filter_log_highlights(raw_lines: Iterable[str],
max_lines: int = MAX_HIGHLIG
except json.JSONDecodeError:
# Task logs aren't guaranteed to be 100% JSON (e.g. stray print
output) — skip those lines.
continue
- if record.get("level") in LOG_LEVELS_OF_INTEREST:
- highlights.append(format_log_line(record))
- if not highlights:
- return ["(no warning/error lines in this task's log)"]
- if len(highlights) > max_lines:
- dropped = len(highlights) - max_lines
- highlights = highlights[:max_lines] + [f"... ({dropped} more lines
suppressed)"]
- return highlights
+ for line in str(record.get("event", "")).splitlines():
+ if VALIDATION_ERROR_TAG in line:
+ errors.append(line.split(VALIDATION_ERROR_TAG, 1)[1].strip())
+ if len(errors) > max_lines:
+ dropped = len(errors) - max_lines
+ errors = errors[:max_lines] + [f"... ({dropped} more lines
suppressed)"]
+ return errors
def _attempt_number(path: Path) -> int:
@@ -95,7 +81,7 @@ def find_log_paths(logs_folder: Path, dag_id: str, run_id:
str, task_id: str) ->
def print_task_log(console: Console, logs_folder: Path, dag_id: str, run_id:
str, task_id: str) -> None:
- """Print the warning/error highlights of a task's log(s), or a note if
none are found."""
+ """Print the OpenLineage validation errors of a task's log(s), or a note
if there are none."""
console.print(f"[bold cyan]--- {dag_id} / {task_id} / START---[/]")
log_paths = find_log_paths(logs_folder, dag_id, run_id, task_id)
if not log_paths:
@@ -106,6 +92,7 @@ def print_task_log(console: Console, logs_folder: Path,
dag_id: str, run_id: str
for log_path in log_paths:
if len(log_paths) > 1:
console.print(f"[cyan]{log_path.relative_to(logs_folder)}[/]")
- for line in filter_log_highlights(log_path.read_text().splitlines()):
- console.print(line)
+ errors = extract_validation_errors(log_path.read_text().splitlines())
+ for line in errors or ["(no OpenLineage validation errors in this
task's log)"]:
+ console.print(line, markup=False, highlight=False)
console.print(f"[bold cyan]--- {dag_id} / {task_id} / END---[/]")
diff --git a/providers/openlineage/tests/system/openlineage/operator.py
b/providers/openlineage/tests/system/openlineage/operator.py
index bcdd66f5d4b..d3b3180f127 100644
--- a/providers/openlineage/tests/system/openlineage/operator.py
+++ b/providers/openlineage/tests/system/openlineage/operator.py
@@ -43,6 +43,14 @@ if TYPE_CHECKING:
log = logging.getLogger(__name__)
+# Marks every log line describing why event validation failed. The OpenLineage
e2e tests (task_logs.py) print
+# only these lines of a failed check, keep the tag in sync with them.
+VALIDATION_ERROR_TAG = "[OL_VALIDATION_ERROR]"
+
+
+def log_validation_error(msg: str, *args: Any) -> None:
+ log.error("%s %s", VALIDATION_ERROR_TAG, msg % args if args else msg)
+
def any(result: Any) -> Any:
return result
@@ -155,13 +163,15 @@ def match(expected, result, env: Environment, path: list
| None = None) -> bool:
if isinstance(expected, dict):
# Only keys present in expected are checked — extra keys in the actual
event are ignored.
if not isinstance(result, dict):
- log.error("Path `%s`: expected a dict but got `%s` (%s)",
path_str, result, type(result).__name__)
+ log_validation_error(
+ "Path `%s`: expected a dict but got `%s` (%s)", path_str,
result, type(result).__name__
+ )
return False
for k, v in expected.items():
# null sentinel: assert the key must NOT appear in the actual
event.
if v is None:
if k in result:
- log.error(
+ log_validation_error(
"Path `%s`: key `%s` must be absent but got `%s` (%s)",
path_str,
k,
@@ -190,7 +200,7 @@ def match(expected, result, env: Environment, path: list |
None = None) -> bool:
# At this point v is a plain template (dict, list, or scalar) with
no sentinels.
if k not in result:
- log.error("Path `%s`: key `%s` not found in received event",
path_str, k)
+ log_validation_error("Path `%s`: key `%s` not found in
received event", path_str, k)
return False
if not match(v, result[k], env, path + [k]):
return False
@@ -198,10 +208,14 @@ def match(expected, result, env: Environment, path: list
| None = None) -> bool:
# Lists must match exactly in length and order. Each element is
compared recursively,
# so nested sentinels (null, $optional) work inside list items too.
if not isinstance(result, list):
- log.error("Path `%s`: expected a list but got `%s` (%s)",
path_str, result, type(result).__name__)
+ log_validation_error(
+ "Path `%s`: expected a list but got `%s` (%s)", path_str,
result, type(result).__name__
+ )
return False
if len(expected) != len(result):
- log.error("Path `%s`: expected %d item(s) but got %d", path_str,
len(expected), len(result))
+ log_validation_error(
+ "Path `%s`: expected %d item(s) but got %d", path_str,
len(expected), len(result)
+ )
return False
for i, x in enumerate(expected):
if not match(x, result[i], env, path + [i]):
@@ -215,11 +229,11 @@ def match(expected, result, env: Environment, path: list
| None = None) -> bool:
try:
rendered = env.from_string(expected).render(result=result)
except (ValueError, TemplateError) as e:
- log.error("Path `%s`: failed to render template `%s`: %s",
path_str, expected, e)
+ log_validation_error("Path `%s`: failed to render template
`%s`: %s", path_str, expected, e)
return False
if str(rendered).lower() == "true" or rendered == result:
return True
- log.error(
+ log_validation_error(
"Path `%s`: template `%s` rendered to `%s` but expected `true`
or actual value `%s` (%s)",
path_str,
expected,
@@ -230,7 +244,7 @@ def match(expected, result, env: Environment, path: list |
None = None) -> bool:
return False
# Plain string: exact equality check.
if expected != result:
- log.error(
+ log_validation_error(
"Path `%s`: expected `%s` (str) but got `%s` (%s)",
path_str,
expected,
@@ -240,7 +254,7 @@ def match(expected, result, env: Environment, path: list |
None = None) -> bool:
return False
elif expected != result:
# Scalar (int, bool, float, …): exact equality check.
- log.error(
+ log_validation_error(
"Path `%s`: expected `%s` (%s) but got `%s` (%s)",
path_str,
expected,
