From 6ddeaa3477086d69f1155872757d5770d85dc928 Mon Sep 17 00:00:00 2001 From: Daniel Standish <15932138+dstandish@users.noreply.github.com> Date: Thu, 5 Oct 2023 14:02:11 -0700 Subject: [PATCH 1/3] Remove unused argument metadata from es_read --- airflow/providers/elasticsearch/log/es_task_handler.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/airflow/providers/elasticsearch/log/es_task_handler.py b/airflow/providers/elasticsearch/log/es_task_handler.py index 546dc579376ae..7c9e02faf0d63 100644 --- a/airflow/providers/elasticsearch/log/es_task_handler.py +++ b/airflow/providers/elasticsearch/log/es_task_handler.py @@ -269,7 +269,7 @@ def _read( offset = metadata["offset"] log_id = self._render_log_id(ti, try_number) - logs = self.es_read(log_id, offset, metadata) + logs = self.es_read(log_id, offset) logs_by_host = self._group_logs_by_host(logs) next_offset = offset if not logs else attrgetter(self.offset_field)(logs[-1]) # Ensure a string here. Large offset numbers will get JSON.parsed incorrectly @@ -330,13 +330,12 @@ def _format_msg(self, log_line): # Just a safe-guard to preserve backwards-compatibility return log_line.message - def es_read(self, log_id: str, offset: int | str, metadata: dict) -> list | ElasticSearchResponse: + def es_read(self, log_id: str, offset: int | str) -> list | ElasticSearchResponse: """ Return the logs matching log_id in Elasticsearch and next offset or ''. :param log_id: the log_id of the log to read. :param offset: the offset start to read log from. - :param metadata: log metadata, used for steaming log download. """ query: dict[Any, Any] = { "query": { From d512774a43c1e785619ad750cd44ac80572de17d Mon Sep 17 00:00:00 2001 From: Daniel Standish <15932138+dstandish@users.noreply.github.com> Date: Fri, 6 Oct 2023 09:41:51 -0700 Subject: [PATCH 2/3] renaem underscore; add notes --- airflow/providers/elasticsearch/CHANGELOG.rst | 11 +++++++++++ .../providers/elasticsearch/log/es_task_handler.py | 6 ++++-- 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/airflow/providers/elasticsearch/CHANGELOG.rst b/airflow/providers/elasticsearch/CHANGELOG.rst index 0f08f823ad231..8686d81348a7a 100644 --- a/airflow/providers/elasticsearch/CHANGELOG.rst +++ b/airflow/providers/elasticsearch/CHANGELOG.rst @@ -27,6 +27,17 @@ Changelog --------- +5.1.0 +..... + +Misc +~~~~ + +.. note:: + In PR #34790 we removed the unused argument ``metadata`` from method ``es_read``. We determined that ``es_read`` + is an internal method and therefore not subject to backcompat, so we did not bump major version for this provider. + In order to make clearer that this is an internal method we renamed it with an underscore prefix ``_es_read``. + .. note:: Upgrade to Elasticsearch 8. The ElasticsearchTaskHandler & ElasticsearchSQLHook will now use Elasticsearch 8 package. As explained https://elasticsearch-py.readthedocs.io/en/stable , Elasticsearch language clients are only backwards diff --git a/airflow/providers/elasticsearch/log/es_task_handler.py b/airflow/providers/elasticsearch/log/es_task_handler.py index 7c9e02faf0d63..ab07e12487a07 100644 --- a/airflow/providers/elasticsearch/log/es_task_handler.py +++ b/airflow/providers/elasticsearch/log/es_task_handler.py @@ -269,7 +269,7 @@ def _read( offset = metadata["offset"] log_id = self._render_log_id(ti, try_number) - logs = self.es_read(log_id, offset) + logs = self._es_read(log_id, offset) logs_by_host = self._group_logs_by_host(logs) next_offset = offset if not logs else attrgetter(self.offset_field)(logs[-1]) # Ensure a string here. Large offset numbers will get JSON.parsed incorrectly @@ -330,12 +330,14 @@ def _format_msg(self, log_line): # Just a safe-guard to preserve backwards-compatibility return log_line.message - def es_read(self, log_id: str, offset: int | str) -> list | ElasticSearchResponse: + def _es_read(self, log_id: str, offset: int | str) -> list | ElasticSearchResponse: """ Return the logs matching log_id in Elasticsearch and next offset or ''. :param log_id: the log_id of the log to read. :param offset: the offset start to read log from. + + :meta private: """ query: dict[Any, Any] = { "query": { From 849ca42c71f767950c9a7722233aef61c60bbd66 Mon Sep 17 00:00:00 2001 From: Daniel Standish <15932138+dstandish@users.noreply.github.com> Date: Fri, 6 Oct 2023 09:47:42 -0700 Subject: [PATCH 3/3] Update airflow/providers/elasticsearch/CHANGELOG.rst Co-authored-by: Elad Kalif <45845474+eladkal@users.noreply.github.com> --- airflow/providers/elasticsearch/CHANGELOG.rst | 5 ----- 1 file changed, 5 deletions(-) diff --git a/airflow/providers/elasticsearch/CHANGELOG.rst b/airflow/providers/elasticsearch/CHANGELOG.rst index 8686d81348a7a..ec37d47d44b47 100644 --- a/airflow/providers/elasticsearch/CHANGELOG.rst +++ b/airflow/providers/elasticsearch/CHANGELOG.rst @@ -27,11 +27,6 @@ Changelog --------- -5.1.0 -..... - -Misc -~~~~ .. note:: In PR #34790 we removed the unused argument ``metadata`` from method ``es_read``. We determined that ``es_read``