ashb commented on code in PR #53821:
URL: https://github.com/apache/airflow/pull/53821#discussion_r2243585689
##########
providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py:
##########
@@ -677,26 +716,17 @@ def _write_to_es(self, log_lines: list[dict[str, Any]])
-> bool:
:param log_lines: the log_lines to write to the ElasticSearch.
"""
+ es_kwargs = get_es_kwargs_from_config()
+
+ client = elasticsearch.Elasticsearch(self.host, **es_kwargs)
# Prepare the bulk request for Elasticsearch
bulk_actions = [{"_index": self.target_index, "_source": log} for log
in log_lines]
try:
- _ = helpers.bulk(self.client, bulk_actions)
+ _ = helpers.bulk(client, bulk_actions)
return True
except Exception as e:
self.log.exception("Unable to insert logs into Elasticsearch.
Reason: %s", str(e))
return False
-
-def getattr_nested(obj, item, default):
- """
- Get item from obj but return default if not found.
-
- E.g. calling ``getattr_nested(a, 'b.c', "NA")`` will return
- ``a.b.c`` if such a value exists, and "NA" otherwise.
-
- :meta private:
- """
- try:
- return attrgetter(item)(obj)
- except AttributeError:
- return default
+ def read(self, relative_path: str, ti: RuntimeTI) -> tuple[LogSourceInfo,
LogMessages]: # type: ignore[empty-body]
+ pass
Review Comment:
With the new config you have set here
https://github.com/apache/airflow/pull/53821/files#diff-48f5a59b2f88ffccd5a091698f2f8856854804418c6fc7ae77837397d2483b80R321
it shouldn't be using ElasticsearchTaskHandler for anything.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
