You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "hussein-awala (via GitHub)" <gi...@apache.org> on 2023/11/17 21:08:20 UTC
[PR] [WIP] Create a generic callbacks class for KubernetesPodOperator [airflow]
hussein-awala opened a new pull request, #35714:
URL: https://github.com/apache/airflow/pull/35714
(no comment)
--
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] Create a generic callbacks class for KubernetesPodOperator [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #35714:
URL: https://github.com/apache/airflow/pull/35714#issuecomment-1885520418
Also doc paragraph about this new cool ability is good to have
--
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] Create a generic callbacks class for KubernetesPodOperator [airflow]
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on PR #35714:
URL: https://github.com/apache/airflow/pull/35714#issuecomment-1901447474
I just tested on a real cluster and found that the type I provide to the trigger is converted to a string. I even tried providing an object instance but had the same problem. That's because we don't serialize the trigger kwargs, but we keep sqlalchemy convert them to string (`ExtendedJSON`).
I need to revert the async part and keep it only for the sync part, and later we can add support for async callbacks once we find a clean solution for the issue; I will take this into account while working on the kwargs encryption.
--
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] Create a generic callbacks class for KubernetesPodOperator [airflow]
Posted by "sean-rose (via GitHub)" <gi...@apache.org>.
sean-rose commented on code in PR #35714:
URL: https://github.com/apache/airflow/pull/35714#discussion_r2051262279
##########
tests/providers/cncf/kubernetes/operators/test_pod.py:
##########
@@ -1446,6 +1446,119 @@ def test_get_logs_but_not_for_base_container(
# check that we wait for the xcom sidecar to start before extracting XCom
mock_await_xcom_sidecar.assert_called_once_with(pod=pod)
+ @patch(HOOK_CLASS, new=MagicMock)
+ @patch(KUB_OP_PATH.format("find_pod"))
+ def test_execute_sync_callbacks(self, find_pod_mock):
+ from airflow.providers.cncf.kubernetes.callbacks import ExecutionMode
+
+ from ..test_callbacks import MockKubernetesPodOperatorCallback, MockWrapper
+
+ MockWrapper.reset()
+ mock_callbacks = MockWrapper.mock_callbacks
+ found_pods = [MagicMock(), MagicMock(), MagicMock()]
+ find_pod_mock.side_effect = [None] + found_pods
+
+ remote_pod_mock = MagicMock()
+ remote_pod_mock.status.phase = "Succeeded"
+ self.await_pod_mock.return_value = remote_pod_mock
+ k = KubernetesPodOperator(
+ namespace="default",
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo 10"],
+ labels={"foo": "bar"},
+ name="test",
+ task_id="task",
+ do_xcom_push=False,
+ callbacks=MockKubernetesPodOperatorCallback,
+ )
+ self.run_pod(k)
+
+ # check on_sync_client_creation callback
+ mock_callbacks.on_sync_client_creation.assert_called_once()
+ assert mock_callbacks.on_sync_client_creation.call_args.kwargs == {"client": k.client}
+
+ # check on_pod_creation callback
+ mock_callbacks.on_pod_creation.assert_called_once()
+ assert mock_callbacks.on_pod_creation.call_args.kwargs == {
+ "client": k.client,
+ "mode": ExecutionMode.SYNC,
+ "pod": found_pods[0],
+ }
+
+ # check on_pod_starting callback
+ mock_callbacks.on_pod_starting.assert_called_once()
+ assert mock_callbacks.on_pod_starting.call_args.kwargs == {
+ "client": k.client,
+ "mode": ExecutionMode.SYNC,
+ "pod": found_pods[1],
+ }
+
+ # check on_pod_completion callback
+ mock_callbacks.on_pod_completion.assert_called_once()
+ assert mock_callbacks.on_pod_completion.call_args.kwargs == {
+ "client": k.client,
+ "mode": ExecutionMode.SYNC,
+ "pod": found_pods[2],
+ }
+
+ # check on_pod_cleanup callback
+ mock_callbacks.on_pod_cleanup.assert_called_once()
+ assert mock_callbacks.on_pod_cleanup.call_args.kwargs == {
+ "client": k.client,
+ "mode": ExecutionMode.SYNC,
+ "pod": k.pod,
+ }
+
+ @patch(HOOK_CLASS, new=MagicMock)
+ def test_execute_async_callbacks(self):
+ from airflow.providers.cncf.kubernetes.callbacks import ExecutionMode
+
+ from ..test_callbacks import MockKubernetesPodOperatorCallback, MockWrapper
+
+ MockWrapper.reset()
+ mock_callbacks = MockWrapper.mock_callbacks
+ remote_pod_mock = MagicMock()
+ remote_pod_mock.status.phase = "Succeeded"
+ self.await_pod_mock.return_value = remote_pod_mock
+
+ k = KubernetesPodOperator(
+ namespace="default",
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo 10"],
+ labels={"foo": "bar"},
+ name="test",
+ task_id="task",
+ do_xcom_push=False,
+ callbacks=MockKubernetesPodOperatorCallback,
+ )
+ k.execute_complete(
+ context=create_context(k),
+ event={
+ "status": "success",
+ "message": TEST_SUCCESS_MESSAGE,
+ "name": TEST_NAME,
+ "namespace": TEST_NAMESPACE,
+ },
+ )
+
+ # check on_operator_resuming callback
+ mock_callbacks.on_pod_cleanup.assert_called_once()
+ assert mock_callbacks.on_pod_cleanup.call_args.kwargs == {
+ "client": k.client,
+ "mode": ExecutionMode.SYNC,
+ "pod": remote_pod_mock,
+ }
Review Comment:
FYI, while working on #49441 I noticed that this code isn't testing `on_operator_resuming` like the comment says it does.
--
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] Create a generic callbacks class for KubernetesPodOperator [airflow]
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on PR #35714:
URL: https://github.com/apache/airflow/pull/35714#issuecomment-1885943230
@eladkal could you check the Google provider check? is it's complicated, I can bump the min `cncf-kubernetes` version to avoid this complexity.
--
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] Create a generic callbacks class for KubernetesPodOperator [airflow]
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on code in PR #35714:
URL: https://github.com/apache/airflow/pull/35714#discussion_r1451020561
##########
airflow/providers/google/provider.yaml:
##########
@@ -151,7 +151,7 @@ additional-extras:
- apache-beam[gcp]
- name: cncf.kubernetes
dependencies:
- - apache-airflow-providers-cncf-kubernetes>=7.2.0
+ - apache-airflow-providers-cncf-kubernetes>=7.14.0
Review Comment:
I think it's the cleanest way, otherwise the logic will be a bit complicated (check my previous commit)
--
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] Create a generic callbacks class for KubernetesPodOperator [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #35714:
URL: https://github.com/apache/airflow/pull/35714#issuecomment-1891050408
There is a test failure though :(
--
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] Create a generic callbacks class for KubernetesPodOperator [airflow]
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on code in PR #35714:
URL: https://github.com/apache/airflow/pull/35714#discussion_r1451020122
##########
airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -550,7 +561,17 @@ def execute_sync(self, context: Context):
# get remote pod for use in cleanup methods
self.remote_pod = self.find_pod(self.pod.metadata.namespace, context=context)
+ if self.callbacks:
+ self.callbacks.on_pod_creation(
+ pod=self.remote_pod, client=self.client, mode=ExecutionMode.SYNC
+ )
self.await_pod_start(pod=self.pod)
+ if self.callbacks:
+ self.callbacks.on_pod_starting(
+ pod=self.find_pod(self.pod.metadata.namespace, context=context),
+ client=self.client,
+ mode=ExecutionMode.SYNC,
+ )
Review Comment:
I changed the default value to None and added a check if it's different to None in order to avoid calling useless methods like `find_pod` when the user doesn't provide a callbacks class.
--
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] Create a generic callbacks class for KubernetesPodOperator [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #35714:
URL: https://github.com/apache/airflow/pull/35714#discussion_r1451719498
##########
airflow/providers/google/provider.yaml:
##########
@@ -151,7 +151,7 @@ additional-extras:
- apache-beam[gcp]
- name: cncf.kubernetes
dependencies:
- - apache-airflow-providers-cncf-kubernetes>=7.2.0
+ - apache-airflow-providers-cncf-kubernetes>=7.14.0
Review Comment:
I am OK with bumping the min version
--
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] Create a generic callbacks class for KubernetesPodOperator [airflow]
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on code in PR #35714:
URL: https://github.com/apache/airflow/pull/35714#discussion_r1447830148
##########
airflow/providers/google/cloud/triggers/kubernetes_engine.py:
##########
@@ -25,6 +25,7 @@
from google.cloud.container_v1.types import Operation
from airflow.exceptions import AirflowProviderDeprecationWarning
+from airflow.providers.cncf.kubernetes.callbacks import KubernetesPodOperatorCallback
Review Comment:
you're right; let me fix 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] Create a generic callbacks class for KubernetesPodOperator [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #35714:
URL: https://github.com/apache/airflow/pull/35714#discussion_r1451789690
##########
airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -550,7 +561,17 @@ def execute_sync(self, context: Context):
# get remote pod for use in cleanup methods
self.remote_pod = self.find_pod(self.pod.metadata.namespace, context=context)
+ if self.callbacks:
+ self.callbacks.on_pod_creation(
+ pod=self.remote_pod, client=self.client, mode=ExecutionMode.SYNC
+ )
self.await_pod_start(pod=self.pod)
+ if self.callbacks:
+ self.callbacks.on_pod_starting(
+ pod=self.find_pod(self.pod.metadata.namespace, context=context),
+ client=self.client,
+ mode=ExecutionMode.SYNC,
+ )
Review Comment:
Good idea/
--
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] Create a generic callbacks class for KubernetesPodOperator [airflow]
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala merged PR #35714:
URL: https://github.com/apache/airflow/pull/35714
--
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] Create a generic callbacks class for KubernetesPodOperator [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #35714:
URL: https://github.com/apache/airflow/pull/35714#discussion_r1447825928
##########
airflow/providers/google/cloud/triggers/kubernetes_engine.py:
##########
@@ -25,6 +25,7 @@
from google.cloud.container_v1.types import Operation
from airflow.exceptions import AirflowProviderDeprecationWarning
+from airflow.providers.cncf.kubernetes.callbacks import KubernetesPodOperatorCallback
Review Comment:
This will not work if user will have combination of newer google provider with older version of k8s provider.
I think we better wrap this with some mechanisem that detects this combo and warn the user that he needs to bump k8s provider to use this
--
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] Create a generic callbacks class for KubernetesPodOperator [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #35714:
URL: https://github.com/apache/airflow/pull/35714#discussion_r1451790580
##########
airflow/providers/google/provider.yaml:
##########
@@ -151,7 +151,7 @@ additional-extras:
- apache-beam[gcp]
- name: cncf.kubernetes
dependencies:
- - apache-airflow-providers-cncf-kubernetes>=7.2.0
+ - apache-airflow-providers-cncf-kubernetes>=7.14.0
Review Comment:
Yeah. No big issue here. Though I think we should review fhose checks and add runtime checks/warnings, because those dependencies are rarely actually used and they are mostly informative.
But I think (as a separate PR) in the future providers we could implement a runtime warning that would be logged at provider import in case we have two providers that are essentially incompatible.
It should be rather easy to add - just next to our "hard" failure for minimum airllow version.
Created an issue here: https://github.com/apache/airflow/issues/36773
--
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