You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2021/06/22 09:12:51 UTC
[GitHub] [airflow] ashb commented on a change in pull request #16571: Implemented Basic EKS Integration
ashb commented on a change in pull request #16571:
URL: https://github.com/apache/airflow/pull/16571#discussion_r656027992
##########
File path: airflow/providers/amazon/aws/hooks/eks.py
##########
@@ -0,0 +1,346 @@
+# 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.
+
+# pylint: disable=invalid-name
+"""Interact with Amazon EKS, using the boto3 library."""
+
+import json
+from typing import Dict, List, Optional
+
+from botocore.exceptions import ClientError
+
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+from airflow.utils.json import AirflowJsonEncoder
+
+DEFAULT_RESULTS_PER_PAGE = 100
+DEFAULT_PAGINATION_TOKEN = ''
+
+
+class EKSHook(AwsBaseHook):
+ """
+ Interact with Amazon EKS, using the boto3 library.
+
+ Additional arguments (such as ``aws_conn_id``) may be specified and
+ are passed down to the underlying AwsBaseHook.
+
+ .. seealso::
+ :class:`~airflow.providers.amazon.aws.hooks.base_aws.AwsBaseHook`
+ """
+
+ conn_type = 'eks'
+ conn_name = 'eks'
+ client_type = 'eks'
+ hook_name = 'EKS'
+
+ def __init__(self, *args, **kwargs) -> None:
+ kwargs["client_type"] = self.client_type
+ super().__init__(*args, **kwargs)
+
+ def create_cluster(self, name: str, roleArn: str, resourcesVpcConfig: Dict, **kwargs) -> str:
+ """
+ Creates an Amazon EKS control plane.
+
+ :param name: The unique name to give to your Amazon EKS Cluster.
+ :type name: str
+ :param roleArn: The Amazon Resource Name (ARN) of the IAM role that provides permissions
+ for the Kubernetes control plane to make calls to AWS API operations on your behalf.
+ :type roleArn: str
+ :param resourcesVpcConfig: The VPC configuration used by the cluster control plane.
+ :type resourcesVpcConfig: Dict
+
+ :return: A JSON serialized string of the API call results.
Review comment:
Why return it as a string, not the Python object/dict?
##########
File path: airflow/providers/amazon/aws/operators/eks.py
##########
@@ -0,0 +1,646 @@
+# 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.
+
+# pylint: disable=invalid-name
+"""This module contains Amazon EKS operators."""
+import json
+import os
+from datetime import datetime
+from time import sleep
+from typing import Dict, List, Optional
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.eks import DEFAULT_PAGINATION_TOKEN, DEFAULT_RESULTS_PER_PAGE, EKSHook
+from airflow.providers.amazon.aws.utils.eks_kube_config import (
+ DEFAULT_CONTEXT_NAME,
+ DEFAULT_KUBE_CONFIG_PATH,
+ DEFAULT_NAMESPACE_NAME,
+ DEFAULT_POD_USERNAME,
+ generate_config_file,
+)
+from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
+
+CHECK_INTERVAL_SECONDS = 15
+TIMEOUT_SECONDS = 25 * 60
+DEFAULT_COMPUTE_TYPE = 'nodegroup'
+DEFAULT_NODEGROUP_NAME_SUFFIX = '-nodegroup'
+DEFAULT_POD_NAME = 'pod'
+KUBE_CONFIG_ENV_VAR = 'KUBECONFIG'
+
+
+class EKSCreateClusterOperator(BaseOperator):
+ """
+ Creates an Amazon EKS Cluster control plane.
+
+ Optionally, can also create the supporting compute architecture:
+ If argument 'compute' is provided with a value of 'nodegroup', will also attempt to create an Amazon
+ EKS Managed Nodegroup for the cluster. See EKSCreateNodegroupOperator documentation for requirements.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:EKSCreateClusterOperator`
+
+ :param cluster_name: The unique name to give to your Amazon EKS Cluster.
+ :type cluster_name: str
+ :param cluster_role_arn: The Amazon Resource Name (ARN) of the IAM role that provides permissions for the
+ Kubernetes control plane to make calls to AWS API operations on your behalf.
+ :type cluster_role_arn: str
+ :param resources_vpc_config: The VPC configuration used by the cluster control plane.
+ :type resources_vpc_config: Dict
+ :param compute: The type of compute architecture to generate along with the cluster.
+ Defaults to 'nodegroup' to generate an EKS Managed Nodegroup.
+ :type compute: str
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then the default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :type aws_conn_id: str
+
+ If 'compute' is 'nodegroup', the following are required:
+ :param nodegroup_name: The unique name to give your EKS Managed Nodegroup.
+ :type nodegroup_name: str
+ :param nodegroup_role_arn: The Amazon Resource Name (ARN) of the IAM role to associate
+ with the EKS Managed Nodegroup.
+ :type nodegroup_role_arn: str
+
+ """
+
+ def __init__(
+ self,
+ cluster_name: str,
+ cluster_role_arn: str,
+ resources_vpc_config: Dict,
+ nodegroup_name: Optional[str] = None,
+ nodegroup_role_arn: Optional[str] = None,
+ compute: Optional[str] = DEFAULT_COMPUTE_TYPE,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.eks_hook = EKSHook(**kwargs)
+ self.clusterName = cluster_name
+ self.clusterRoleArn = cluster_role_arn
+ self.resourcesVpcConfig = resources_vpc_config
+ self.compute = compute
+
+ if self.compute == 'nodegroup':
+ self.nodegroupName = nodegroup_name or self.clusterName + DEFAULT_NODEGROUP_NAME_SUFFIX
+ if nodegroup_role_arn:
+ self.nodegroupRoleArn = nodegroup_role_arn
+ else:
+ message = "Creating an EKS Managed Nodegroup requires nodegroup_role_arn to be passed in."
+ self.log.error(message)
+ raise AttributeError(message)
+
+ def execute(self, context):
+ self.eks_hook.create_cluster(
+ name=self.clusterName,
+ roleArn=self.clusterRoleArn,
+ resourcesVpcConfig=self.resourcesVpcConfig,
+ )
+
+ if self.compute is not None:
+ self.log.info("Waiting for EKS Cluster to provision. This will take some time.")
+
+ countdown = TIMEOUT_SECONDS
+ while self.eks_hook.get_cluster_state(clusterName=self.clusterName) != "ACTIVE":
+ if countdown >= CHECK_INTERVAL_SECONDS:
+ countdown -= CHECK_INTERVAL_SECONDS
+ self.log.info(
+ "Waiting for cluster to start. Checking again in %d seconds", CHECK_INTERVAL_SECONDS
+ )
+ sleep(CHECK_INTERVAL_SECONDS)
+ else:
+ message = "Cluster is still inactive after the allocated time limit. Aborting."
+ self.log.error(message)
+ raise RuntimeError(message)
+
+ if self.compute == 'nodegroup':
+ self.eks_hook.create_nodegroup(
+ clusterName=self.clusterName,
+ nodegroupName=self.nodegroupName,
+ subnets=self.resourcesVpcConfig.get('subnetIds'),
+ nodeRole=self.nodegroupRoleArn,
+ )
+
+
+class EKSCreateNodegroupOperator(BaseOperator):
+ """
+ Creates am Amazon EKS Managed Nodegroup for an existing Amazon EKS Cluster.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:EKSCreateNodegroupOperator`
+
+ :param cluster_name: The name of the Amazon EKS Cluster to create the managed nodegroup in.
+ :type cluster_name: str
+ :param nodegroup_name: The unique name to give your managed nodegroup.
+ :type nodegroup_name: str
+ :param nodegroup_subnets:
+ The subnets to use for the Auto Scaling group that is created for the managed nodegroup.
+ :type nodegroup_subnets: List[str]
+ :param nodegroup_role_arn:
+ The Amazon Resource Name (ARN) of the IAM role to associate with the managed nodegroup.
+ :type nodegroup_role_arn: str
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then the default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :type aws_conn_id: str
+
+ """
+
+ def __init__(
+ self,
+ cluster_name: str,
+ nodegroup_subnets: List[str],
+ nodegroup_role_arn: str,
+ nodegroup_name: Optional[str],
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.eks_hook = EKSHook(**kwargs)
Review comment:
Plese don't create hooks in Operator constructors -- we generally try to avoid it as it _can_ end up creating network requests during DAG parse time which we want to avoid.
##########
File path: airflow/providers/amazon/aws/hooks/base_aws.py
##########
@@ -347,6 +347,7 @@ def __init__(
client_type: Optional[str] = None,
resource_type: Optional[str] = None,
config: Optional[Config] = None,
+ **kwargs,
Review comment:
If we don't _do_ anything with these kwargs why do we need to accept them here?
##########
File path: airflow/providers/amazon/aws/hooks/eks.py
##########
@@ -0,0 +1,346 @@
+# 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.
+
+# pylint: disable=invalid-name
+"""Interact with Amazon EKS, using the boto3 library."""
+
+import json
+from typing import Dict, List, Optional
+
+from botocore.exceptions import ClientError
+
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+from airflow.utils.json import AirflowJsonEncoder
+
+DEFAULT_RESULTS_PER_PAGE = 100
+DEFAULT_PAGINATION_TOKEN = ''
+
+
+class EKSHook(AwsBaseHook):
+ """
+ Interact with Amazon EKS, using the boto3 library.
+
+ Additional arguments (such as ``aws_conn_id``) may be specified and
+ are passed down to the underlying AwsBaseHook.
+
+ .. seealso::
+ :class:`~airflow.providers.amazon.aws.hooks.base_aws.AwsBaseHook`
+ """
+
+ conn_type = 'eks'
+ conn_name = 'eks'
+ client_type = 'eks'
+ hook_name = 'EKS'
+
+ def __init__(self, *args, **kwargs) -> None:
+ kwargs["client_type"] = self.client_type
+ super().__init__(*args, **kwargs)
+
+ def create_cluster(self, name: str, roleArn: str, resourcesVpcConfig: Dict, **kwargs) -> str:
Review comment:
Interesting conundrum here. The Python naming style says these args should be `role_arn` and `resources_vpc_config` -- but _maybe_ it makes sense to mirror the Boto/AWS API names here.
What do other think, and what do we do in other AWS hooks?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org