Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

KPO to tolerate missing xcom config functions on hook #31258

Closed
wants to merge 2 commits into from
Closed

KPO to tolerate missing xcom config functions on hook #31258

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0f5c17b
Make KPO tolerant of hooks that don't have xcom methods
dstandish May 12, 2023
cf1ba27
extract logic to method
dstandish May 12, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 17 additions & 6 deletions airflow/providers/cncf/kubernetes/operators/pod.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -861,12 +861,7 @@ def build_pod_request_obj(self, context: Context | None = None) -> k8s.V1Pod:
pod = secret.attach_to_pod(pod)
if self.do_xcom_push:
self.log.debug("Adding xcom sidecar to task %s", self.task_id)
pod = xcom_sidecar.add_xcom_sidecar(
pod,
sidecar_container_image=self.hook.get_xcom_sidecar_container_image(),
sidecar_container_resources=self.hook.get_xcom_sidecar_container_resources(),
)

pod = self._add_xcom_sidecar(pod)
labels = self._get_ti_pod_labels(context)
self.log.info("Building pod %s with labels: %s", pod.metadata.name, labels)

Expand All @@ -883,6 +878,22 @@ def build_pod_request_obj(self, context: Context | None = None) -> k8s.V1Pod:
pod_mutation_hook(pod)
return pod

def _add_xcom_sidecar(self, pod):
"""Add xcom sidecar to pod."""
sidecar_container_image = None
sidecar_container_resources = None
# self.hook may not be subclass of KubernetesHook (see GKEStartPodOperator) so we must
# check to make sure these methods exist before calling
if hasattr(self.hook, "get_xcom_sidecar_container_image"):
sidecar_container_image = self.hook.get_xcom_sidecar_container_image()
if hasattr(self.hook, "get_xcom_sidecar_container_resources"):
sidecar_container_resources = self.hook.get_xcom_sidecar_container_resources()
Comment on lines +887 to +890
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

  1. I think that it would be better to replace these two if statements with one if isinstance(self.hook, GKEPodHook). For the long term, maybe we should consider refactoring the abstraction.
  2. Is it possible to unit test it? If so, it would be nice to implement.

return xcom_sidecar.add_xcom_sidecar(
pod,
sidecar_container_image=sidecar_container_image,
sidecar_container_resources=sidecar_container_resources,
)

def dry_run(self) -> None:
"""
Prints out the pod definition that would be created by this operator.
Expand Down