Source code for airflow.providers.google.cloud.triggers.kubernetes_engine

# 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.

from __future__ import annotations

import asyncio
import warnings
from functools import cached_property
from typing import TYPE_CHECKING, Any, AsyncIterator, Sequence

from google.cloud.container_v1.types import Operation
from packaging.version import parse as parse_version

from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
from airflow.providers.cncf.kubernetes.triggers.pod import KubernetesPodTrigger
from airflow.providers.cncf.kubernetes.utils.pod_manager import OnFinishAction, PodManager
from airflow.providers.cncf.kubernetes.utils.xcom_sidecar import PodDefaults
from airflow.providers.google.cloud.hooks.kubernetes_engine import (
    GKEAsyncHook,
    GKEKubernetesAsyncHook,
    GKEKubernetesHook,
)
from airflow.providers_manager import ProvidersManager
from airflow.triggers.base import BaseTrigger, TriggerEvent

if TYPE_CHECKING:
    from datetime import datetime

    from kubernetes_asyncio.client import V1Job


[docs]class GKEStartPodTrigger(KubernetesPodTrigger): """ Trigger for checking pod status until it finishes its job. :param pod_name: The name of the pod. :param pod_namespace: The namespace of the pod. :param cluster_url: The URL pointed to the cluster. :param ssl_ca_cert: SSL certificate that is used for authentication to the pod. :param cluster_context: Context that points to kubernetes cluster. :param poll_interval: Polling period in seconds to check for the status. :param trigger_start_time: time in Datetime format when the trigger was started :param in_cluster: run kubernetes client with in_cluster configuration. :param get_logs: get the stdout of the container as logs of the tasks. :param startup_timeout: timeout in seconds to start up the pod. :param base_container_name: The name of the base container in the pod. This container's logs will appear as part of this task's logs if get_logs is True. Defaults to None. If None, will consult the class variable BASE_CONTAINER_NAME (which defaults to "base") for the base container name to use. :param on_finish_action: What to do when the pod reaches its final state, or the execution is interrupted. If "delete_pod", the pod will be deleted regardless its state; if "delete_succeeded_pod", only succeeded pod will be deleted. You can set to "keep_pod" to keep the pod. :param should_delete_pod: What to do when the pod reaches its final state, or the execution is interrupted. If True (default), delete the pod; if False, leave the pod. Deprecated - use `on_finish_action` instead. """ def __init__( self, pod_name: str, pod_namespace: str, cluster_url: str, ssl_ca_cert: str, base_container_name: str, trigger_start_time: datetime, cluster_context: str | None = None, poll_interval: float = 2, in_cluster: bool | None = None, get_logs: bool = True, startup_timeout: int = 120, on_finish_action: str = "delete_pod", should_delete_pod: bool | None = None, gcp_conn_id: str = "google_cloud_default", impersonation_chain: str | Sequence[str] | None = None, *args, **kwargs, ): super().__init__( pod_name, pod_namespace, trigger_start_time, base_container_name, *args, **kwargs, ) self.pod_name = pod_name self.pod_namespace = pod_namespace self.trigger_start_time = trigger_start_time self.base_container_name = base_container_name self.poll_interval = poll_interval self.cluster_context = cluster_context self.in_cluster = in_cluster self.get_logs = get_logs self.startup_timeout = startup_timeout self.gcp_conn_id = gcp_conn_id self.impersonation_chain = impersonation_chain if should_delete_pod is not None: warnings.warn( "`should_delete_pod` parameter is deprecated, please use `on_finish_action`", AirflowProviderDeprecationWarning, stacklevel=2, ) self.on_finish_action = ( OnFinishAction.DELETE_POD if should_delete_pod else OnFinishAction.KEEP_POD ) self.should_delete_pod = should_delete_pod else: self.on_finish_action = OnFinishAction(on_finish_action) self.should_delete_pod = self.on_finish_action == OnFinishAction.DELETE_POD self._cluster_url = cluster_url self._ssl_ca_cert = ssl_ca_cert
[docs] def serialize(self) -> tuple[str, dict[str, Any]]: return ( "airflow.providers.google.cloud.triggers.kubernetes_engine.GKEStartPodTrigger", { "pod_name": self.pod_name, "pod_namespace": self.pod_namespace, "cluster_url": self._cluster_url, "ssl_ca_cert": self._ssl_ca_cert, "poll_interval": self.poll_interval, "cluster_context": self.cluster_context, "in_cluster": self.in_cluster, "get_logs": self.get_logs, "startup_timeout": self.startup_timeout, "trigger_start_time": self.trigger_start_time, "base_container_name": self.base_container_name, "should_delete_pod": self.should_delete_pod, "on_finish_action": self.on_finish_action.value, "gcp_conn_id": self.gcp_conn_id, "impersonation_chain": self.impersonation_chain, "logging_interval": self.logging_interval, "last_log_time": self.last_log_time, }, )
@cached_property
[docs] def hook(self) -> GKEKubernetesAsyncHook: # type: ignore[override] return GKEKubernetesAsyncHook( cluster_url=self._cluster_url, ssl_ca_cert=self._ssl_ca_cert, gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain, enable_tcp_keepalive=True, )
[docs]class GKEOperationTrigger(BaseTrigger): """Trigger which checks status of the operation.""" def __init__( self, operation_name: str, project_id: str | None, location: str, gcp_conn_id: str = "google_cloud_default", impersonation_chain: str | Sequence[str] | None = None, poll_interval: int = 10, ): super().__init__() self.operation_name = operation_name self.project_id = project_id self.location = location self.gcp_conn_id = gcp_conn_id self.impersonation_chain = impersonation_chain self.poll_interval = poll_interval self._hook: GKEAsyncHook | None = None
[docs] def serialize(self) -> tuple[str, dict[str, Any]]: """Serialize GKEOperationTrigger arguments and classpath.""" return ( "airflow.providers.google.cloud.triggers.kubernetes_engine.GKEOperationTrigger", { "operation_name": self.operation_name, "project_id": self.project_id, "location": self.location, "gcp_conn_id": self.gcp_conn_id, "impersonation_chain": self.impersonation_chain, "poll_interval": self.poll_interval, }, )
[docs] async def run(self) -> AsyncIterator[TriggerEvent]: # type: ignore[override] """Get operation status and yields corresponding event.""" hook = self._get_hook() try: while True: operation = await hook.get_operation( operation_name=self.operation_name, project_id=self.project_id, ) status = operation.status if status == Operation.Status.DONE: yield TriggerEvent( { "status": "success", "message": "Operation is successfully ended.", "operation_name": operation.name, } ) return elif status in (Operation.Status.RUNNING, Operation.Status.PENDING): self.log.info("Operation is still running.") self.log.info("Sleeping for %ss...", self.poll_interval) await asyncio.sleep(self.poll_interval) else: yield TriggerEvent( { "status": "failed", "message": f"Operation has failed with status: {operation.status}", } ) return except Exception as e: self.log.exception("Exception occurred while checking operation status") yield TriggerEvent( { "status": "error", "message": str(e), } )
def _get_hook(self) -> GKEAsyncHook: if self._hook is None: self._hook = GKEAsyncHook( gcp_conn_id=self.gcp_conn_id, location=self.location, impersonation_chain=self.impersonation_chain, ) return self._hook
[docs]class GKEJobTrigger(BaseTrigger): """GKEJobTrigger run on the trigger worker to check the state of Job.""" def __init__( self, cluster_url: str, ssl_ca_cert: str, job_name: str, job_namespace: str, pod_name: str, pod_namespace: str, base_container_name: str, gcp_conn_id: str = "google_cloud_default", poll_interval: float = 2, impersonation_chain: str | Sequence[str] | None = None, get_logs: bool = True, do_xcom_push: bool = False, ) -> None: super().__init__() self.cluster_url = cluster_url self.ssl_ca_cert = ssl_ca_cert self.job_name = job_name self.job_namespace = job_namespace self.pod_name = pod_name self.pod_namespace = pod_namespace self.base_container_name = base_container_name self.gcp_conn_id = gcp_conn_id self.poll_interval = poll_interval self.impersonation_chain = impersonation_chain self.get_logs = get_logs self.do_xcom_push = do_xcom_push
[docs] def serialize(self) -> tuple[str, dict[str, Any]]: """Serialize KubernetesCreateJobTrigger arguments and classpath.""" return ( "airflow.providers.google.cloud.triggers.kubernetes_engine.GKEJobTrigger", { "cluster_url": self.cluster_url, "ssl_ca_cert": self.ssl_ca_cert, "job_name": self.job_name, "job_namespace": self.job_namespace, "pod_name": self.pod_name, "pod_namespace": self.pod_namespace, "base_container_name": self.base_container_name, "gcp_conn_id": self.gcp_conn_id, "poll_interval": self.poll_interval, "impersonation_chain": self.impersonation_chain, "get_logs": self.get_logs, "do_xcom_push": self.do_xcom_push, }, )
[docs] async def run(self) -> AsyncIterator[TriggerEvent]: # type: ignore[override] """Get current job status and yield a TriggerEvent.""" if self.get_logs or self.do_xcom_push: pod = await self.hook.get_pod(name=self.pod_name, namespace=self.pod_namespace) if self.do_xcom_push: kubernetes_provider = ProvidersManager().providers["apache-airflow-providers-cncf-kubernetes"] kubernetes_provider_name = kubernetes_provider.data["package-name"] kubernetes_provider_version = kubernetes_provider.version min_version = "8.4.1" if parse_version(kubernetes_provider_version) < parse_version(min_version): raise AirflowException( "You are trying to use do_xcom_push in `GKEStartJobOperator` with the provider " f"package {kubernetes_provider_name}=={kubernetes_provider_version} which doesn't " f"support this feature. Please upgrade it to version higher than or equal to {min_version}." ) await self.hook.wait_until_container_complete( name=self.pod_name, namespace=self.pod_namespace, container_name=self.base_container_name, poll_interval=self.poll_interval, ) self.log.info("Checking if xcom sidecar container is started.") await self.hook.wait_until_container_started( name=self.pod_name, namespace=self.pod_namespace, container_name=PodDefaults.SIDECAR_CONTAINER_NAME, poll_interval=self.poll_interval, ) self.log.info("Extracting result from xcom sidecar container.") loop = asyncio.get_running_loop() xcom_result = await loop.run_in_executor(None, self.pod_manager.extract_xcom, pod) job: V1Job = await self.hook.wait_until_job_complete( name=self.job_name, namespace=self.job_namespace, poll_interval=self.poll_interval ) job_dict = job.to_dict() error_message = self.hook.is_job_failed(job=job) status = "error" if error_message else "success" message = f"Job failed with error: {error_message}" if error_message else "Job completed successfully" yield TriggerEvent( { "name": job.metadata.name, "namespace": job.metadata.namespace, "pod_name": pod.metadata.name if self.get_logs else None, "pod_namespace": pod.metadata.namespace if self.get_logs else None, "status": status, "message": message, "job": job_dict, "xcom_result": xcom_result if self.do_xcom_push else None, } )
@cached_property
[docs] def hook(self) -> GKEKubernetesAsyncHook: return GKEKubernetesAsyncHook( cluster_url=self.cluster_url, ssl_ca_cert=self.ssl_ca_cert, gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain, )
@cached_property
[docs] def pod_manager(self) -> PodManager: sync_hook = GKEKubernetesHook( gcp_conn_id=self.gcp_conn_id, cluster_url=self.cluster_url, ssl_ca_cert=self.ssl_ca_cert, impersonation_chain=self.impersonation_chain, ) return PodManager(kube_client=sync_hook.core_v1_client)

Was this entry helpful?