You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "pankajastro (via GitHub)" <gi...@apache.org> on 2024/02/09 14:48:15 UTC
[PR] Modify KPO to log container log periodically [airflow]
pankajastro opened a new pull request, #37279:
URL: https://github.com/apache/airflow/pull/37279
The current state of the KubernetesPodOperator (KPO)
only prints container logs at the end of task execution.
While this is sufficient for short-running tasks,
it becomes less user-friendly when the container runs for an extended period.
This PR enhances the KPO by modifying the trigger and operator
to fetch container logs periodically
making it possible to monitor the task's progress in the Airflow task UI.
a new parameter have been introduced to the operator:
- logging_interval: This parameter specifies the maximum time,
in seconds, that the task should remain deferred before resuming to fetch the latest logs.
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "pankajkoti (via GitHub)" <gi...@apache.org>.
pankajkoti commented on code in PR #37279:
URL: https://github.com/apache/airflow/pull/37279#discussion_r1486041543
##########
airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -659,10 +672,79 @@ def invoke_defer_method(self):
startup_check_interval=self.startup_check_interval_seconds,
base_container_name=self.base_container_name,
on_finish_action=self.on_finish_action.value,
+ last_log_time=last_log_time,
+ logging_interval=self.logging_interval,
),
- method_name="execute_complete",
+ method_name="trigger_reentry",
)
+ @staticmethod
+ def raise_for_trigger_status(event: dict[str, Any]) -> None:
+ """Raise exception if pod is not in expected state."""
+ if event["status"] == "error":
+ error_type = event["error_type"]
+ description = event["description"]
+ if error_type == "PodLaunchTimeoutException":
+ raise PodLaunchTimeoutException(description)
+ else:
+ raise AirflowException(description)
+
+ def trigger_reentry(self, context: Context, event: dict[str, Any]) -> Any:
+ """
+ Point of re-entry from trigger.
+
+ If ``logging_interval`` is None, then at this point the pod should be done and we'll just fetch
+ the logs and exit.
+
+ If ``logging_interval`` is not None, it could be that the pod is still running and we'll just
+ grab the latest logs and defer back to the trigger again.
+ """
+ remote_pod = None
+ try:
+ self.pod_request_obj = self.build_pod_request_obj(context)
+ self.pod = self.find_pod(
+ namespace=self.namespace or self.pod_request_obj.metadata.namespace,
+ context=context,
+ )
+
+ # we try to find pod before possibly raising so that on_kill will have `pod` attr
+ self.raise_for_trigger_status(event)
+
+ if not self.pod:
+ raise PodNotFoundException("Could not find pod after resuming from deferral")
+
+ if self.get_logs:
+ last_log_time = event and event.get("last_log_time")
+ if last_log_time:
+ self.log.info("Resuming logs read from time %r", last_log_time)
+ pod_log_status = self.pod_manager.fetch_container_logs(
+ pod=self.pod,
+ container_name=self.BASE_CONTAINER_NAME,
+ follow=self.logging_interval is None,
+ since_time=last_log_time,
+ )
+ if pod_log_status.running:
+ self.log.info("Container still running; deferring again.")
+ self.invoke_defer_method(pod_log_status.last_log_time)
+
+ if self.do_xcom_push:
+ result = self.extract_xcom(pod=self.pod)
+ remote_pod = self.pod_manager.await_pod_completion(self.pod)
+ except TaskDeferred:
+ raise
+ except Exception:
+ self.cleanup(
+ pod=self.pod or self.pod_request_obj,
+ remote_pod=remote_pod,
+ )
+ raise
+ self.cleanup(
+ pod=self.pod or self.pod_request_obj,
+ remote_pod=remote_pod,
+ )
+ if self.do_xcom_push:
+ return result
+
def execute_complete(self, context: Context, event: dict, **kwargs):
Review Comment:
Thanks @hussein-awala
Actually, would be nice for us to include this in the providers wave as we are depending on the change 🙂 .
We tested this on our internal tests and they worked fine. How about we test this with the RC? Would that time be sufficient if we include this? We can work on providing fixes with utmost priority if a bug is identified. or You see some other risks?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #37279:
URL: https://github.com/apache/airflow/pull/37279#discussion_r1485757767
##########
airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -203,6 +211,9 @@ class KubernetesPodOperator(BaseOperator):
of KubernetesPodOperator.
:param progress_callback: Callback function for receiving k8s container logs.
`progress_callback` is deprecated, please use :param `callbacks` instead.
+ :param logging_interval: max time in seconds that task should be in deferred state before
Review Comment:
consider adding the `logging_interval` param to the example DAG?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "pankajastro (via GitHub)" <gi...@apache.org>.
pankajastro commented on code in PR #37279:
URL: https://github.com/apache/airflow/pull/37279#discussion_r1485828910
##########
airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -203,6 +211,9 @@ class KubernetesPodOperator(BaseOperator):
of KubernetesPodOperator.
:param progress_callback: Callback function for receiving k8s container logs.
`progress_callback` is deprecated, please use :param `callbacks` instead.
+ :param logging_interval: max time in seconds that task should be in deferred state before
Review Comment:
Added example. thank you
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "pankajkoti (via GitHub)" <gi...@apache.org>.
pankajkoti commented on PR #37279:
URL: https://github.com/apache/airflow/pull/37279#issuecomment-1938193642
The only failed test in the previous CI run has passed now: https://github.com/apache/airflow/actions/runs/7868578876/job/21466117939#step:24:1.
I think we're good to merge @pankajastro
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "pankajkoti (via GitHub)" <gi...@apache.org>.
pankajkoti commented on code in PR #37279:
URL: https://github.com/apache/airflow/pull/37279#discussion_r1485981757
##########
airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -659,10 +672,79 @@ def invoke_defer_method(self):
startup_check_interval=self.startup_check_interval_seconds,
base_container_name=self.base_container_name,
on_finish_action=self.on_finish_action.value,
+ last_log_time=last_log_time,
+ logging_interval=self.logging_interval,
),
- method_name="execute_complete",
+ method_name="trigger_reentry",
)
+ @staticmethod
+ def raise_for_trigger_status(event: dict[str, Any]) -> None:
+ """Raise exception if pod is not in expected state."""
+ if event["status"] == "error":
+ error_type = event["error_type"]
+ description = event["description"]
+ if error_type == "PodLaunchTimeoutException":
+ raise PodLaunchTimeoutException(description)
+ else:
+ raise AirflowException(description)
+
+ def trigger_reentry(self, context: Context, event: dict[str, Any]) -> Any:
+ """
+ Point of re-entry from trigger.
+
+ If ``logging_interval`` is None, then at this point the pod should be done and we'll just fetch
+ the logs and exit.
+
+ If ``logging_interval`` is not None, it could be that the pod is still running and we'll just
+ grab the latest logs and defer back to the trigger again.
+ """
+ remote_pod = None
+ try:
+ self.pod_request_obj = self.build_pod_request_obj(context)
+ self.pod = self.find_pod(
+ namespace=self.namespace or self.pod_request_obj.metadata.namespace,
+ context=context,
+ )
+
+ # we try to find pod before possibly raising so that on_kill will have `pod` attr
+ self.raise_for_trigger_status(event)
+
+ if not self.pod:
+ raise PodNotFoundException("Could not find pod after resuming from deferral")
+
+ if self.get_logs:
+ last_log_time = event and event.get("last_log_time")
+ if last_log_time:
+ self.log.info("Resuming logs read from time %r", last_log_time)
+ pod_log_status = self.pod_manager.fetch_container_logs(
+ pod=self.pod,
+ container_name=self.BASE_CONTAINER_NAME,
+ follow=self.logging_interval is None,
+ since_time=last_log_time,
+ )
+ if pod_log_status.running:
+ self.log.info("Container still running; deferring again.")
+ self.invoke_defer_method(pod_log_status.last_log_time)
+
+ if self.do_xcom_push:
+ result = self.extract_xcom(pod=self.pod)
+ remote_pod = self.pod_manager.await_pod_completion(self.pod)
+ except TaskDeferred:
+ raise
+ except Exception:
+ self.cleanup(
+ pod=self.pod or self.pod_request_obj,
+ remote_pod=remote_pod,
+ )
+ raise
+ self.cleanup(
+ pod=self.pod or self.pod_request_obj,
+ remote_pod=remote_pod,
+ )
+ if self.do_xcom_push:
+ return result
+
def execute_complete(self, context: Context, event: dict, **kwargs):
Review Comment:
@pankajastro can you please take a look? With the use of the new method `trigger_reentry`, if `execute_complete` is no longer getting used anywhere, perhaps we could remove the method and clean this up?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "pankajastro (via GitHub)" <gi...@apache.org>.
pankajastro merged PR #37279:
URL: https://github.com/apache/airflow/pull/37279
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "pankajkoti (via GitHub)" <gi...@apache.org>.
pankajkoti commented on PR #37279:
URL: https://github.com/apache/airflow/pull/37279#issuecomment-1938781390
> Subclasses of KPO still call the execute_complete method - https://github.com/apache/airflow/blob/main/airflow/providers/google/cloud/operators/kubernetes_engine.py#L584-L593
>
> Is this expected?
Yes, I think that should not break. The existing public method names should continue to work as they were.
I think you've caught the cause for issue in https://github.com/apache/airflow/pull/37279#issuecomment-1938777428. Thanks for the help in testing and identifying the issue.
We will fix this
cc: @pankajastro
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "pankajkoti (via GitHub)" <gi...@apache.org>.
pankajkoti commented on PR #37279:
URL: https://github.com/apache/airflow/pull/37279#issuecomment-1938766241
@vchiapaikeo Would you please offer more help in understanding the issue and reproducibility?
If I am understanding correctly, is it that we are getting an issue with the `GKEStartPodOperator` around arguments `deferrable` and `do_xcom_push`?
Could you provide details in which case we're observing failures (with what values of this params)?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "jonathonbattista (via GitHub)" <gi...@apache.org>.
jonathonbattista commented on PR #37279:
URL: https://github.com/apache/airflow/pull/37279#issuecomment-2087353608
Should this already be compatible with the `GKEStartPodOperator` out of the box or does it also need updated?
I am on `apache-airflow-providers-cncf-kubernetes v8.1.1` and `apache-airflow-providers-google v10.17.0`
I have `logging_internval` set, but its not polling the logs when the task is deferred:
https://github.com/apache/airflow/blob/main/airflow/providers/google/cloud/operators/kubernetes_engine.py#L792
```
with DAG(
"gke_start_pod_test",
schedule_interval=None,
start_date=days_ago(1),
tags=["test"],
) as dag:
gke_start_pod_test = GKEStartPodOperator(
task_id="sleep",
project_id="xxxxx",
location="xxx",
cluster_name="xxxxx",
do_xcom_push=True,
namespace="default",
image="ubuntu:jammy",
cmds=["sh", "-c", "timeout 300 bash -c 'while true; do echo \"meow\"; sleep 3; done'"],
name="test-sleep",
in_cluster=False,
on_finish_action="delete_pod",
deferrable=True,
get_logs=True,
logging_interval=10
)
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "pankajkoti (via GitHub)" <gi...@apache.org>.
pankajkoti commented on PR #37279:
URL: https://github.com/apache/airflow/pull/37279#issuecomment-2088033059
I think it should work `out of the box`. @pankajastro could you please test this once?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "pankajastro (via GitHub)" <gi...@apache.org>.
pankajastro commented on PR #37279:
URL: https://github.com/apache/airflow/pull/37279#issuecomment-1944009155
Hey @vchiapaikeo would it be possible for you to test https://github.com/apache/airflow/pull/37363 with your use case?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on code in PR #37279:
URL: https://github.com/apache/airflow/pull/37279#discussion_r1486016644
##########
airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -659,10 +672,79 @@ def invoke_defer_method(self):
startup_check_interval=self.startup_check_interval_seconds,
base_container_name=self.base_container_name,
on_finish_action=self.on_finish_action.value,
+ last_log_time=last_log_time,
+ logging_interval=self.logging_interval,
),
- method_name="execute_complete",
+ method_name="trigger_reentry",
)
+ @staticmethod
+ def raise_for_trigger_status(event: dict[str, Any]) -> None:
+ """Raise exception if pod is not in expected state."""
+ if event["status"] == "error":
+ error_type = event["error_type"]
+ description = event["description"]
+ if error_type == "PodLaunchTimeoutException":
+ raise PodLaunchTimeoutException(description)
+ else:
+ raise AirflowException(description)
+
+ def trigger_reentry(self, context: Context, event: dict[str, Any]) -> Any:
+ """
+ Point of re-entry from trigger.
+
+ If ``logging_interval`` is None, then at this point the pod should be done and we'll just fetch
+ the logs and exit.
+
+ If ``logging_interval`` is not None, it could be that the pod is still running and we'll just
+ grab the latest logs and defer back to the trigger again.
+ """
+ remote_pod = None
+ try:
+ self.pod_request_obj = self.build_pod_request_obj(context)
+ self.pod = self.find_pod(
+ namespace=self.namespace or self.pod_request_obj.metadata.namespace,
+ context=context,
+ )
+
+ # we try to find pod before possibly raising so that on_kill will have `pod` attr
+ self.raise_for_trigger_status(event)
+
+ if not self.pod:
+ raise PodNotFoundException("Could not find pod after resuming from deferral")
+
+ if self.get_logs:
+ last_log_time = event and event.get("last_log_time")
+ if last_log_time:
+ self.log.info("Resuming logs read from time %r", last_log_time)
+ pod_log_status = self.pod_manager.fetch_container_logs(
+ pod=self.pod,
+ container_name=self.BASE_CONTAINER_NAME,
+ follow=self.logging_interval is None,
+ since_time=last_log_time,
+ )
+ if pod_log_status.running:
+ self.log.info("Container still running; deferring again.")
+ self.invoke_defer_method(pod_log_status.last_log_time)
+
+ if self.do_xcom_push:
+ result = self.extract_xcom(pod=self.pod)
+ remote_pod = self.pod_manager.await_pod_completion(self.pod)
+ except TaskDeferred:
+ raise
+ except Exception:
+ self.cleanup(
+ pod=self.pod or self.pod_request_obj,
+ remote_pod=remote_pod,
+ )
+ raise
+ self.cleanup(
+ pod=self.pod or self.pod_request_obj,
+ remote_pod=remote_pod,
+ )
+ if self.do_xcom_push:
+ return result
+
def execute_complete(self, context: Context, event: dict, **kwargs):
Review Comment:
I wonder if it was kept for b/c.
I prefer to not include this change in the current providers wave if possible to have the time to test it. WDYT?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "pankajastro (via GitHub)" <gi...@apache.org>.
pankajastro commented on code in PR #37279:
URL: https://github.com/apache/airflow/pull/37279#discussion_r1486042792
##########
airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -659,10 +672,79 @@ def invoke_defer_method(self):
startup_check_interval=self.startup_check_interval_seconds,
base_container_name=self.base_container_name,
on_finish_action=self.on_finish_action.value,
+ last_log_time=last_log_time,
+ logging_interval=self.logging_interval,
),
- method_name="execute_complete",
+ method_name="trigger_reentry",
)
+ @staticmethod
+ def raise_for_trigger_status(event: dict[str, Any]) -> None:
+ """Raise exception if pod is not in expected state."""
+ if event["status"] == "error":
+ error_type = event["error_type"]
+ description = event["description"]
+ if error_type == "PodLaunchTimeoutException":
+ raise PodLaunchTimeoutException(description)
+ else:
+ raise AirflowException(description)
+
+ def trigger_reentry(self, context: Context, event: dict[str, Any]) -> Any:
+ """
+ Point of re-entry from trigger.
+
+ If ``logging_interval`` is None, then at this point the pod should be done and we'll just fetch
+ the logs and exit.
+
+ If ``logging_interval`` is not None, it could be that the pod is still running and we'll just
+ grab the latest logs and defer back to the trigger again.
+ """
+ remote_pod = None
+ try:
+ self.pod_request_obj = self.build_pod_request_obj(context)
+ self.pod = self.find_pod(
+ namespace=self.namespace or self.pod_request_obj.metadata.namespace,
+ context=context,
+ )
+
+ # we try to find pod before possibly raising so that on_kill will have `pod` attr
+ self.raise_for_trigger_status(event)
+
+ if not self.pod:
+ raise PodNotFoundException("Could not find pod after resuming from deferral")
+
+ if self.get_logs:
+ last_log_time = event and event.get("last_log_time")
+ if last_log_time:
+ self.log.info("Resuming logs read from time %r", last_log_time)
+ pod_log_status = self.pod_manager.fetch_container_logs(
+ pod=self.pod,
+ container_name=self.BASE_CONTAINER_NAME,
+ follow=self.logging_interval is None,
+ since_time=last_log_time,
+ )
+ if pod_log_status.running:
+ self.log.info("Container still running; deferring again.")
+ self.invoke_defer_method(pod_log_status.last_log_time)
+
+ if self.do_xcom_push:
+ result = self.extract_xcom(pod=self.pod)
+ remote_pod = self.pod_manager.await_pod_completion(self.pod)
+ except TaskDeferred:
+ raise
+ except Exception:
+ self.cleanup(
+ pod=self.pod or self.pod_request_obj,
+ remote_pod=remote_pod,
+ )
+ raise
+ self.cleanup(
+ pod=self.pod or self.pod_request_obj,
+ remote_pod=remote_pod,
+ )
+ if self.do_xcom_push:
+ return result
+
def execute_complete(self, context: Context, event: dict, **kwargs):
Review Comment:
Thanks for the feedback. Yes, this method is public so we can't remove it. similarly, a couple of methods in the trigger are unused now I kept them as well because they might be used by someone.
We have conducted some testing on our end. Once RC is available, we will perform additional tests. Hopefully, some community members will also test RC so we should be able find and fix if something unusual happend before releasing it.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "vchiapaikeo (via GitHub)" <gi...@apache.org>.
vchiapaikeo commented on PR #37279:
URL: https://github.com/apache/airflow/pull/37279#issuecomment-1938777428
Also, you changed name to pod_name. Did you mean to do that?
<img width="1470" alt="image" src="https://github.com/apache/airflow/assets/9200263/26f3f66d-1059-4ae8-81f7-48eac9dcf09c">
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "vchiapaikeo (via GitHub)" <gi...@apache.org>.
vchiapaikeo commented on PR #37279:
URL: https://github.com/apache/airflow/pull/37279#issuecomment-1938773700
Subclasses of KPO still call the execute_complete method - https://github.com/apache/airflow/blob/main/airflow/providers/google/cloud/operators/kubernetes_engine.py#L584-L593
Is this expected?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on code in PR #37279:
URL: https://github.com/apache/airflow/pull/37279#discussion_r1485880148
##########
airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -659,10 +672,79 @@ def invoke_defer_method(self):
startup_check_interval=self.startup_check_interval_seconds,
base_container_name=self.base_container_name,
on_finish_action=self.on_finish_action.value,
+ last_log_time=last_log_time,
+ logging_interval=self.logging_interval,
),
- method_name="execute_complete",
+ method_name="trigger_reentry",
)
+ @staticmethod
+ def raise_for_trigger_status(event: dict[str, Any]) -> None:
+ """Raise exception if pod is not in expected state."""
+ if event["status"] == "error":
+ error_type = event["error_type"]
+ description = event["description"]
+ if error_type == "PodLaunchTimeoutException":
+ raise PodLaunchTimeoutException(description)
+ else:
+ raise AirflowException(description)
+
+ def trigger_reentry(self, context: Context, event: dict[str, Any]) -> Any:
+ """
+ Point of re-entry from trigger.
+
+ If ``logging_interval`` is None, then at this point the pod should be done and we'll just fetch
+ the logs and exit.
+
+ If ``logging_interval`` is not None, it could be that the pod is still running and we'll just
+ grab the latest logs and defer back to the trigger again.
+ """
+ remote_pod = None
+ try:
+ self.pod_request_obj = self.build_pod_request_obj(context)
+ self.pod = self.find_pod(
+ namespace=self.namespace or self.pod_request_obj.metadata.namespace,
+ context=context,
+ )
+
+ # we try to find pod before possibly raising so that on_kill will have `pod` attr
+ self.raise_for_trigger_status(event)
+
+ if not self.pod:
+ raise PodNotFoundException("Could not find pod after resuming from deferral")
+
+ if self.get_logs:
+ last_log_time = event and event.get("last_log_time")
+ if last_log_time:
+ self.log.info("Resuming logs read from time %r", last_log_time)
+ pod_log_status = self.pod_manager.fetch_container_logs(
+ pod=self.pod,
+ container_name=self.BASE_CONTAINER_NAME,
+ follow=self.logging_interval is None,
+ since_time=last_log_time,
+ )
+ if pod_log_status.running:
+ self.log.info("Container still running; deferring again.")
+ self.invoke_defer_method(pod_log_status.last_log_time)
+
+ if self.do_xcom_push:
+ result = self.extract_xcom(pod=self.pod)
+ remote_pod = self.pod_manager.await_pod_completion(self.pod)
+ except TaskDeferred:
+ raise
+ except Exception:
+ self.cleanup(
+ pod=self.pod or self.pod_request_obj,
+ remote_pod=remote_pod,
+ )
+ raise
+ self.cleanup(
+ pod=self.pod or self.pod_request_obj,
+ remote_pod=remote_pod,
+ )
+ if self.do_xcom_push:
+ return result
+
def execute_complete(self, context: Context, event: dict, **kwargs):
Review Comment:
how this method is used and where?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "pankajastro (via GitHub)" <gi...@apache.org>.
pankajastro commented on PR #37279:
URL: https://github.com/apache/airflow/pull/37279#issuecomment-1938782796
hmm, look then "name" => "pod_name" is issue
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "vchiapaikeo (via GitHub)" <gi...@apache.org>.
vchiapaikeo commented on PR #37279:
URL: https://github.com/apache/airflow/pull/37279#issuecomment-1938757838
@pankajastro , @pankajkoti - this is causing failures for deferrable tasks that should complete. I reverted this commit and things work normally after. Hoping you can help figure out the issue or revert this before the providers push.
Sample DAG:
```py
from airflow import DAG
from airflow.providers.google.cloud.operators.kubernetes_engine import (
GKEStartPodOperator,
)
DEFAULT_TASK_ARGS = {
"owner": "gcp-data-platform",
"start_date": "2021-04-20",
"retries": 0,
"retry_delay": 60,
}
with DAG(
dag_id="test_gke_op",
schedule_interval="@daily",
max_active_runs=1,
max_active_tasks=5,
catchup=False,
default_args=DEFAULT_TASK_ARGS,
) as dag:
_ = GKEStartPodOperator(
task_id="whoami",
name="whoami",
cmds=["gcloud"],
arguments=["auth", "list"],
image="gcr.io/google.com/cloudsdktool/cloud-sdk:slim",
project_id="redacted-project-id",
namespace="airflow-default",
location="us-central1",
cluster_name="airflow-gke-cluster",
service_account_name="default",
deferrable=True,
# do_xcom_push=True,
)
_ = GKEStartPodOperator(
task_id="fail",
name="fail",
cmds=["bash"],
arguments=["-xc", "sleep 2 && exit 1"],
image="gcr.io/google.com/cloudsdktool/cloud-sdk:slim",
project_id="redacted-project-id",
namespace="airflow-default",
location="us-central1",
cluster_name="airflow-gke-cluster",
service_account_name="default",
deferrable=True,
# do_xcom_push=True,
)
```
Expected result (obtained after reverting):
<img width="1158" alt="image" src="https://github.com/apache/airflow/assets/9200263/57e89f7a-abe3-412c-b5e9-4d005fa12186">
Unexpected result (after rebasing - whoami should succeed):
<img width="993" alt="image" src="https://github.com/apache/airflow/assets/9200263/4060dffd-6496-4e39-98da-5383a1db2407">
whoami logs:
```
fcd7bd221fe9
*** Found local files:
*** * /root/airflow/logs/dag_id=test_gke_op/run_id=scheduled__2024-02-11T00:00:00+00:00/task_id=whoami/attempt=11.log
*** * /root/airflow/logs/dag_id=test_gke_op/run_id=scheduled__2024-02-11T00:00:00+00:00/task_id=whoami/attempt=11.log.trigger.40.log
[2024-02-12, 09:12:23 EST] {taskinstance.py:1994} INFO - Dependencies all met for dep_context=non-requeueable deps ti=
[2024-02-12, 09:12:23 EST] {taskinstance.py:1994} INFO - Dependencies all met for dep_context=requeueable deps ti=
[2024-02-12, 09:12:23 EST] {taskinstance.py:2208} INFO - Starting attempt 11 of 11
[2024-02-12, 09:12:23 EST] {taskinstance.py:2229} INFO - Executing on 2024-02-11 00:00:00+00:00
[2024-02-12, 09:12:23 EST] {standard_task_runner.py:60} INFO - Started process 579 to run task
[2024-02-12, 09:12:23 EST] {standard_task_runner.py:87} INFO - Running: ['***', 'tasks', 'run', 'test_gke_op', 'whoami', 'scheduled__2024-02-11T00:00:00+00:00', '--job-id', '42', '--raw', '--subdir', 'DAGS_FOLDER/test_deferrable_xcom.py', '--cfg-path', '/tmp/tmpkupoc4wd']
[2024-02-12, 09:12:23 EST] {standard_task_runner.py:88} INFO - Job 42: Subtask whoami
[2024-02-12, 09:12:23 EST] {task_command.py:423} INFO - Running on host fcd7bd221fe9
[2024-02-12, 09:12:23 EST] {taskinstance.py:2529} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='gcp-data-platform' AIRFLOW_CTX_DAG_ID='test_gke_op' AIRFLOW_CTX_TASK_ID='whoami' AIRFLOW_CTX_EXECUTION_DATE='2024-02-11T00:00:00+00:00' AIRFLOW_CTX_TRY_NUMBER='11' AIRFLOW_CTX_DAG_RUN_ID='scheduled__2024-02-11T00:00:00+00:00'
[2024-02-12, 09:12:23 EST] {connection.py:269} WARNING - Connection schemes (type: google_cloud_platform) shall not contain '_' according to RFC3986.
[2024-02-12, 09:12:23 EST] {base.py:83} INFO - Using connection ID 'google_cloud_default' for task execution.
[2024-02-12, 09:12:23 EST] {kubernetes_engine.py:289} INFO - Fetching cluster (project_id=redacted-project-id, location=us-central1, cluster_name=***-gke-cluster)
[2024-02-12, 09:12:23 EST] {credentials_provider.py:353} INFO - Getting connection using `google.auth.default()` since no explicit credentials are provided.
[2024-02-12, 09:12:24 EST] {pod.py:1079} INFO - Building pod whoami-75olgaah with labels: {'dag_id': 'test_gke_op', 'task_id': 'whoami', 'run_id': 'scheduled__2024-02-11T0000000000-2e3e3ab6f', 'kubernetes_pod_operator': 'True', 'try_number': '11'}
[2024-02-12, 09:12:24 EST] {connection.py:269} WARNING - Connection schemes (type: google_cloud_platform) shall not contain '_' according to RFC3986.
[2024-02-12, 09:12:24 EST] {base.py:83} INFO - Using connection ID 'google_cloud_default' for task execution.
[2024-02-12, 09:12:24 EST] {credentials_provider.py:353} INFO - Getting connection using `google.auth.default()` since no explicit credentials are provided.
[2024-02-12, 09:12:25 EST] {taskinstance.py:2382} INFO - Pausing task as DEFERRED. dag_id=test_gke_op, task_id=whoami, execution_date=20240211T000000, start_date=20240212T141223
[2024-02-12, 09:12:25 EST] {local_task_job_runner.py:231} INFO - Task exited with return code 100 (task deferral)
[2024-02-12, 09:12:26 EST] {pod.py:160} INFO - Checking pod 'whoami-75olgaah' in namespace '***-default'.
[2024-02-12, 09:12:26 EST] {connection.py:269} WARNING - Connection schemes (type: google_cloud_platform) shall not contain '_' according to RFC3986.
[2024-02-12, 09:12:26 EST] {base.py:83} INFO - Using connection ID 'google_cloud_default' for task execution.
[2024-02-12, 09:12:29 EST] {triggerer_job_runner.py:604} INFO - Trigger test_gke_op/scheduled__2024-02-11T00:00:00+00:00/whoami/-1/11 (ID 15) fired: TriggerEvent<{'status': 'done', 'namespace': 'airflow-default', 'pod_name': 'whoami-75olgaah'}>
[2024-02-12, 09:12:31 EST] {taskinstance.py:1994} INFO - Dependencies all met for dep_context=non-requeueable deps ti=
[2024-02-12, 09:12:31 EST] {taskinstance.py:1994} INFO - Dependencies all met for dep_context=requeueable deps ti=
[2024-02-12, 09:12:31 EST] {taskinstance.py:2206} INFO - Resuming after deferral
[2024-02-12, 09:12:31 EST] {taskinstance.py:2229} INFO - Executing on 2024-02-11 00:00:00+00:00
[2024-02-12, 09:12:31 EST] {standard_task_runner.py:60} INFO - Started process 649 to run task
[2024-02-12, 09:12:31 EST] {standard_task_runner.py:87} INFO - Running: ['***', 'tasks', 'run', 'test_gke_op', 'whoami', 'scheduled__2024-02-11T00:00:00+00:00', '--job-id', '43', '--raw', '--subdir', 'DAGS_FOLDER/test_deferrable_xcom.py', '--cfg-path', '/tmp/tmpbojzr3x_']
[2024-02-12, 09:12:31 EST] {standard_task_runner.py:88} INFO - Job 43: Subtask whoami
[2024-02-12, 09:12:31 EST] {task_command.py:423} INFO - Running on host fcd7bd221fe9
[2024-02-12, 09:12:31 EST] {connection.py:269} WARNING - Connection schemes (type: google_cloud_platform) shall not contain '_' according to RFC3986.
[2024-02-12, 09:12:31 EST] {base.py:83} INFO - Using connection ID 'google_cloud_default' for task execution.
[2024-02-12, 09:12:31 EST] {credentials_provider.py:353} INFO - Getting connection using `google.auth.default()` since no explicit credentials are provided.
[2024-02-12, 09:12:34 EST] {taskinstance.py:2751} ERROR - Task failed with exception
Traceback (most recent call last):
File "/opt/airflow/airflow/providers/cncf/kubernetes/operators/pod.py", line 753, in execute_complete
event["name"],
KeyError: 'name'
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/opt/airflow/airflow/models/taskinstance.py", line 446, in _execute_task
result = _execute_callable(context=context, **execute_callable_kwargs)
File "/opt/airflow/airflow/models/taskinstance.py", line 416, in _execute_callable
return execute_callable(context=context, **execute_callable_kwargs)
File "/opt/airflow/airflow/models/baseoperator.py", line 1623, in resume_execution
return execute_callable(context)
File "/opt/airflow/airflow/providers/google/cloud/operators/kubernetes_engine.py", line 593, in execute_complete
return super().execute_complete(context, event, **kwargs)
File "/opt/airflow/airflow/providers/cncf/kubernetes/operators/pod.py", line 786, in execute_complete
pod = self.pod_manager.await_pod_completion(pod, istio_enabled, self.base_container_name)
File "/opt/airflow/airflow/providers/cncf/kubernetes/utils/pod_manager.py", line 611, in await_pod_completion
remote_pod = self.read_pod(pod)
File "/usr/local/lib/python3.10/site-packages/tenacity/__init__.py", line 289, in wrapped_f
return self(f, *args, **kw)
File "/usr/local/lib/python3.10/site-packages/tenacity/__init__.py", line 379, in __call__
do = self.iter(retry_state=retry_state)
File "/usr/local/lib/python3.10/site-packages/tenacity/__init__.py", line 325, in iter
raise retry_exc.reraise()
File "/usr/local/lib/python3.10/site-packages/tenacity/__init__.py", line 158, in reraise
raise self.last_attempt.result()
File "/usr/local/lib/python3.10/concurrent/futures/_base.py", line 451, in result
return self.__get_result()
File "/usr/local/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
raise self._exception
File "/usr/local/lib/python3.10/site-packages/tenacity/__init__.py", line 382, in __call__
result = fn(*args, **kwargs)
File "/opt/airflow/airflow/providers/cncf/kubernetes/utils/pod_manager.py", line 713, in read_pod
return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
AttributeError: 'NoneType' object has no attribute 'metadata'
[2024-02-12, 09:12:34 EST] {taskinstance.py:1166} INFO - Marking task as FAILED. dag_id=test_gke_op, task_id=whoami, execution_date=20240211T000000, start_date=20240212T141223, end_date=20240212T141234
[2024-02-12, 09:12:34 EST] {standard_task_runner.py:107} ERROR - Failed to execute job 43 for task whoami ('NoneType' object has no attribute 'metadata'; 649)
[2024-02-12, 09:12:34 EST] {local_task_job_runner.py:234} INFO - Task exited with return code 1
[2024-02-12, 09:12:34 EST] {taskinstance.py:3332} INFO - 0 downstream tasks scheduled from follow-on schedule check
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Modify KPO to log container log periodically [airflow]
Posted by "pankajastro (via GitHub)" <gi...@apache.org>.
pankajastro commented on PR #37279:
URL: https://github.com/apache/airflow/pull/37279#issuecomment-2088470139
hmm, this will require small changes PR https://github.com/apache/airflow/pull/39348
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org