ashb commented on code in PR #53821:
URL: https://github.com/apache/airflow/pull/53821#discussion_r2245110371
##########
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:
I didn't think that once `REMOTE_TASK_LOG` was set that that Airflow3 will
use the task handler for anything anymore.
> Are you suggesting that both remote read & write should be handled in
ElasticsearchRemoteLogIO ?
Yes.
> And that we should get rid of ElasticsearchTaskHandler ?
No, this class needs to stay to support Airflow 2. Check out the GCS or S3
logging files
--
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]
