Lee-W commented on code in PR #69300:
URL: https://github.com/apache/airflow/pull/69300#discussion_r3628922468
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -736,15 +739,31 @@ def _get_pod_namespace(self, ti: TaskInstance):
return namespace or self.conf.get("kubernetes_executor", "namespace")
def get_task_log(self, ti: TaskInstance, try_number: int) ->
tuple[list[str], list[str]]:
- messages = []
- log = []
+ messages: list[str] = []
+ log: list[str] = []
+ try:
+ messages, log_streams = self.get_streaming_task_log(ti, try_number)
+ log = ["\n".join(stream) for stream in log_streams]
+ except Exception as e:
+ messages.append(f"Reading from k8s pod logs failed: {e}")
+ return messages, log or [""]
+
+ def get_streaming_task_log(self, ti: TaskInstance, try_number: int) ->
StreamingLogResponse:
+ messages: list[str] = []
+ log_streams: list[RawLogStream] = []
+
+ def create_log_stream(logs: Iterable[bytes]) -> RawLogStream:
Review Comment:
do we need to make it an inner function? or would it be better to make it an
underscored static method?
--
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]
