ashb commented on code in PR #53821:
URL: https://github.com/apache/airflow/pull/53821#discussion_r2241901966
##########
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:
By introducing this class andhaving an empty read method, this means that
log messages won't viewable from ElasticSearch anymore -- so this method will
need implementing.
The way we've done it for other logging providers is to change the
ElasticsearchTaskHandler class to have an `self.io` property to have all/most
of the read/write logic, and then do pretty much `return self.io.read(...)` etc.
--
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]
