You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by el...@apache.org on 2026/04/16 03:36:31 UTC
(airflow) branch main updated: refactoring e2e elasticsearch test (#65316)
This is an automated email from the ASF dual-hosted git repository.
eladkal 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 6e57bf5ccf1 refactoring e2e elasticsearch test (#65316)
6e57bf5ccf1 is described below
commit 6e57bf5ccf14a07d20e87445b75f5234b5382ac1
Author: Owen Leung <ow...@gmail.com>
AuthorDate: Thu Apr 16 11:36:23 2026 +0800
refactoring e2e elasticsearch test (#65316)
---
.../base_remote_log_search_test.py} | 48 ++++----
.../test_remote_logging_elasticsearch.py | 130 +--------------------
2 files changed, 33 insertions(+), 145 deletions(-)
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/remote_log_elasticsearch_tests/test_remote_logging_elasticsearch.py b/airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/base_remote_log_search_test.py
similarity index 80%
copy from airflow-e2e-tests/tests/airflow_e2e_tests/remote_log_elasticsearch_tests/test_remote_logging_elasticsearch.py
copy to airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/base_remote_log_search_test.py
index c17f5109f80..fb03d8bb2ba 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/remote_log_elasticsearch_tests/test_remote_logging_elasticsearch.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/base_remote_log_search_test.py
@@ -21,10 +21,18 @@ from datetime import datetime, timezone
import pytest
-from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient, get_elasticsearch_session
+from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient
-class TestRemoteLoggingElasticsearch:
+class BaseRemoteLoggingSearchTest:
+ """
+ Base class for remote logging e2e tests against search backends (Elasticsearch, OpenSearch).
+
+ Subclasses must set:
+ - ``search_url``: base URL of the search backend, e.g. ``"http://localhost:9200"``
+ - ``_get_session``: zero-arg callable returning a requests.Session with retry logic
+ """
+
airflow_client = AirflowClient()
dag_id = "example_xcom_test"
task_id = "bash_pull"
@@ -32,9 +40,12 @@ class TestRemoteLoggingElasticsearch:
max_retries = 12
target_index = "airflow-e2e-logs"
expected_message = "finished"
- elasticsearch_url = "http://localhost:9200"
+ search_url: str # set by subclass, e.g. "http://localhost:9200"
expected_log_id_prefix = f"{dag_id}-{task_id}-"
+ def _get_session(self):
+ raise NotImplementedError
+
def _matches_expected_log(self, log_source: dict, run_id: str) -> bool:
log_id = log_source.get("log_id", "")
log_message = log_source.get("event", "")
@@ -46,30 +57,27 @@ class TestRemoteLoggingElasticsearch:
and self.expected_message in log_message
)
- def test_remote_logging_elasticsearch(self):
- """Test that a DAG using remote logging to Elasticsearch completes successfully."""
-
- self.airflow_client.un_pause_dag(TestRemoteLoggingElasticsearch.dag_id)
+ def test_remote_logging(self):
+ """Test that a DAG using remote logging to the search backend completes successfully."""
+ self.airflow_client.un_pause_dag(self.dag_id)
resp = self.airflow_client.trigger_dag(
- TestRemoteLoggingElasticsearch.dag_id,
+ self.dag_id,
json={"logical_date": datetime.now(timezone.utc).isoformat()},
)
run_id = resp["dag_run_id"]
state = self.airflow_client.wait_for_dag_run(
- dag_id=TestRemoteLoggingElasticsearch.dag_id,
+ dag_id=self.dag_id,
run_id=run_id,
)
- assert state == "success", (
- f"DAG {TestRemoteLoggingElasticsearch.dag_id} did not complete successfully. Final state: {state}"
- )
+ assert state == "success", f"DAG {self.dag_id} did not complete successfully. Final state: {state}"
- elasticsearch_session = get_elasticsearch_session()
+ session = self._get_session()
matching_logs = []
for _ in range(self.max_retries):
- response = elasticsearch_session.post(
- f"{self.elasticsearch_url}/{self.target_index}/_search",
+ response = session.post(
+ f"{self.search_url}/{self.target_index}/_search",
json={"size": 200, "query": {"match_all": {}}},
)
if response.status_code == 404:
@@ -86,13 +94,13 @@ class TestRemoteLoggingElasticsearch:
if not matching_logs:
pytest.fail(
- f"Expected Elasticsearch logs for run_id {run_id} in index {self.target_index}, "
+ f"Expected search backend logs for run_id {run_id} in index {self.target_index}, "
f"but none contained log_id starting with {self.expected_log_id_prefix!r} and event "
f"or message containing {self.expected_message!r}"
)
task_logs = self.airflow_client.get_task_logs(
- dag_id=TestRemoteLoggingElasticsearch.dag_id,
+ dag_id=self.dag_id,
task_id=self.task_id,
run_id=run_id,
)
@@ -102,8 +110,8 @@ class TestRemoteLoggingElasticsearch:
f"Expected task logs to contain {self.expected_message!r}, got events: {events}"
)
- def test_remote_logging_elasticsearch_error_detail(self):
- """Test that log error_detail is retrieved correctly from Elasticsearch."""
+ def test_remote_logging_error_detail(self):
+ """Test that log error_detail is retrieved correctly from the search backend."""
dag_id = "example_failed_dag"
task_id = "fail_task"
@@ -117,7 +125,6 @@ class TestRemoteLoggingElasticsearch:
assert state == "failed"
- # Logs might take some time to appear in ES
task_logs_content = []
for _ in range(self.max_retries):
task_logs_resp = self.airflow_client.get_task_logs(
@@ -126,7 +133,6 @@ class TestRemoteLoggingElasticsearch:
run_id=run_id,
)
task_logs_content = task_logs_resp.get("content", [])
- # Search for the log entry with error_detail
if any("error_detail" in item for item in task_logs_content if isinstance(item, dict)):
break
time.sleep(self.retry_interval_in_seconds)
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/remote_log_elasticsearch_tests/test_remote_logging_elasticsearch.py b/airflow-e2e-tests/tests/airflow_e2e_tests/remote_log_elasticsearch_tests/test_remote_logging_elasticsearch.py
index c17f5109f80..424fa921043 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/remote_log_elasticsearch_tests/test_remote_logging_elasticsearch.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/remote_log_elasticsearch_tests/test_remote_logging_elasticsearch.py
@@ -16,130 +16,12 @@
# under the License.
from __future__ import annotations
-import time
-from datetime import datetime, timezone
+from airflow_e2e_tests.e2e_test_utils.base_remote_log_search_test import BaseRemoteLoggingSearchTest
+from airflow_e2e_tests.e2e_test_utils.clients import get_elasticsearch_session
-import pytest
-from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient, get_elasticsearch_session
+class TestRemoteLoggingElasticsearch(BaseRemoteLoggingSearchTest):
+ search_url = "http://localhost:9200"
-
-class TestRemoteLoggingElasticsearch:
- airflow_client = AirflowClient()
- dag_id = "example_xcom_test"
- task_id = "bash_pull"
- retry_interval_in_seconds = 5
- max_retries = 12
- target_index = "airflow-e2e-logs"
- expected_message = "finished"
- elasticsearch_url = "http://localhost:9200"
- expected_log_id_prefix = f"{dag_id}-{task_id}-"
-
- def _matches_expected_log(self, log_source: dict, run_id: str) -> bool:
- log_id = log_source.get("log_id", "")
- log_message = log_source.get("event", "")
- if not isinstance(log_message, str):
- log_message = log_source.get("message", "")
- return (
- log_id.startswith(self.expected_log_id_prefix)
- and run_id in log_id
- and self.expected_message in log_message
- )
-
- def test_remote_logging_elasticsearch(self):
- """Test that a DAG using remote logging to Elasticsearch completes successfully."""
-
- self.airflow_client.un_pause_dag(TestRemoteLoggingElasticsearch.dag_id)
-
- resp = self.airflow_client.trigger_dag(
- TestRemoteLoggingElasticsearch.dag_id,
- json={"logical_date": datetime.now(timezone.utc).isoformat()},
- )
- run_id = resp["dag_run_id"]
- state = self.airflow_client.wait_for_dag_run(
- dag_id=TestRemoteLoggingElasticsearch.dag_id,
- run_id=run_id,
- )
-
- assert state == "success", (
- f"DAG {TestRemoteLoggingElasticsearch.dag_id} did not complete successfully. Final state: {state}"
- )
-
- elasticsearch_session = get_elasticsearch_session()
- matching_logs = []
- for _ in range(self.max_retries):
- response = elasticsearch_session.post(
- f"{self.elasticsearch_url}/{self.target_index}/_search",
- json={"size": 200, "query": {"match_all": {}}},
- )
- if response.status_code == 404:
- time.sleep(self.retry_interval_in_seconds)
- continue
- response.raise_for_status()
- hits = response.json()["hits"]["hits"]
- matching_logs = [
- hit["_source"] for hit in hits if self._matches_expected_log(hit["_source"], run_id)
- ]
- if matching_logs:
- break
- time.sleep(self.retry_interval_in_seconds)
-
- if not matching_logs:
- pytest.fail(
- f"Expected Elasticsearch logs for run_id {run_id} in index {self.target_index}, "
- f"but none contained log_id starting with {self.expected_log_id_prefix!r} and event "
- f"or message containing {self.expected_message!r}"
- )
-
- task_logs = self.airflow_client.get_task_logs(
- dag_id=TestRemoteLoggingElasticsearch.dag_id,
- task_id=self.task_id,
- run_id=run_id,
- )
-
- events = [item.get("event", "") for item in task_logs.get("content", []) if isinstance(item, dict)]
- assert any(self.expected_message in event for event in events), (
- f"Expected task logs to contain {self.expected_message!r}, got events: {events}"
- )
-
- def test_remote_logging_elasticsearch_error_detail(self):
- """Test that log error_detail is retrieved correctly from Elasticsearch."""
- dag_id = "example_failed_dag"
- task_id = "fail_task"
-
- self.airflow_client.un_pause_dag(dag_id)
- resp = self.airflow_client.trigger_dag(
- dag_id,
- json={"logical_date": datetime.now(timezone.utc).isoformat()},
- )
- run_id = resp["dag_run_id"]
- state = self.airflow_client.wait_for_dag_run(dag_id=dag_id, run_id=run_id)
-
- assert state == "failed"
-
- # Logs might take some time to appear in ES
- task_logs_content = []
- for _ in range(self.max_retries):
- task_logs_resp = self.airflow_client.get_task_logs(
- dag_id=dag_id,
- task_id=task_id,
- run_id=run_id,
- )
- task_logs_content = task_logs_resp.get("content", [])
- # Search for the log entry with error_detail
- if any("error_detail" in item for item in task_logs_content if isinstance(item, dict)):
- break
- time.sleep(self.retry_interval_in_seconds)
-
- error_entries = [
- item for item in task_logs_content if isinstance(item, dict) and "error_detail" in item
- ]
- assert len(error_entries) > 0, (
- f"Expected error_detail in logs, but none found. Logs: {task_logs_content}"
- )
-
- error_detail = error_entries[0]["error_detail"]
- assert isinstance(error_detail, list), f"Expected error_detail to be a list, got {type(error_detail)}"
- assert len(error_detail) > 0, "Expected error_detail to have at least one exception"
- assert error_detail[0]["exc_type"] == "RuntimeError"
- assert "This is a test exception for stacktrace rendering" in error_detail[0]["exc_value"]
+ def _get_session(self):
+ return get_elasticsearch_session()