Source code for toil.provisioners.abstractProvisioner

# Copyright (C) 2015-2021 Regents of the University of California
# Licensed 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
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.
import configparser
import json
import logging
import os.path
import subprocess
import tempfile
import textwrap
from abc import ABC, abstractmethod
from functools import total_ordering
from typing import Any, Dict, List, Optional, Set, Tuple, Union
from urllib.parse import quote
from uuid import uuid4

from toil import applianceSelf, customDockerInitCmd, customInitCmd
from toil.provisioners import ClusterTypeNotSupportedException
from toil.provisioners.node import Node

a_short_time = 5
logger = logging.getLogger(__name__)

[docs] class ManagedNodesNotSupportedException(RuntimeError): """ Raised when attempting to add managed nodes (which autoscale up and down by themselves, without the provisioner doing the work) to a provisioner that does not support them. Polling with this and try/except is the Right Way to check if managed nodes are available from a provisioner. """
[docs] @total_ordering class Shape: """ Represents a job or a node's "shape", in terms of the dimensions of memory, cores, disk and wall-time allocation. The wallTime attribute stores the number of seconds of a node allocation, e.g. 3600 for AWS. FIXME: and for jobs? The memory and disk attributes store the number of bytes required by a job (or provided by a node) in RAM or on disk (SSD or HDD), respectively. """ def __init__( self, wallTime: Union[int, float], memory: int, cores: Union[int, float], disk: int, preemptible: bool, ) -> None: self.wallTime = wallTime self.memory = memory self.cores = cores self.disk = disk self.preemptible = preemptible
[docs] def __eq__(self, other: Any) -> bool: return (self.wallTime == other.wallTime and self.memory == other.memory and self.cores == other.cores and self.disk == other.disk and self.preemptible == other.preemptible)
[docs] def greater_than(self, other: Any) -> bool: if self.preemptible < other.preemptible: return True elif self.preemptible > other.preemptible: return False elif self.memory > other.memory: return True elif self.memory < other.memory: return False elif self.cores > other.cores: return True elif self.cores < other.cores: return False elif self.disk > other.disk: return True elif self.disk < other.disk: return False elif self.wallTime > other.wallTime: return True elif self.wallTime < other.wallTime: return False else: return False
[docs] def __gt__(self, other: Any) -> bool: return self.greater_than(other)
[docs] def __repr__(self) -> str: return "Shape(wallTime=%s, memory=%s, cores=%s, disk=%s, preemptible=%s)" % \ (self.wallTime, self.memory, self.cores, self.disk, self.preemptible)
[docs] def __str__(self) -> str: return self.__repr__()
[docs] def __hash__(self) -> int: # Since we replaced __eq__ we need to replace __hash__ as well. return hash( (self.wallTime, self.memory, self.cores, self.disk, self.preemptible))
[docs] class AbstractProvisioner(ABC): """Interface for provisioning worker nodes to use in a Toil cluster.""" LEADER_HOME_DIR = '/root/' # home directory in the Toil appliance on an instance cloud: str = None def __init__( self, clusterName: Optional[str] = None, clusterType: Optional[str] = "mesos", zone: Optional[str] = None, nodeStorage: int = 50, nodeStorageOverrides: Optional[List[str]] = None, ) -> None: """ Initialize provisioner. Implementations should raise ClusterTypeNotSupportedException if presented with an unimplemented clusterType. :param clusterName: The cluster identifier. :param clusterType: The kind of cluster to make; 'mesos' or 'kubernetes'. :param zone: The zone the cluster runs in. :param nodeStorage: The amount of storage on the worker instances, in gigabytes. """ self.clusterName = clusterName self.clusterType = clusterType if self.clusterType not in self.supportedClusterTypes(): # This isn't actually a cluster type we can do raise ClusterTypeNotSupportedException(type(self), clusterType) self._zone = zone self._nodeStorage = nodeStorage self._nodeStorageOverrides = {} for override in nodeStorageOverrides or []: nodeShape, storageOverride = override.split(':') self._nodeStorageOverrides[nodeShape] = int(storageOverride) self._leaderPrivateIP = None # This will hold an SSH public key for Mesos clusters, or the # Kubernetes joining information as a dict for Kubernetes clusters. self._leaderWorkerAuthentication = None if clusterName: # Making a new cluster self.createClusterSettings() else: # Starting up on an existing cluster self.readClusterSettings()
[docs] @abstractmethod def supportedClusterTypes(self) -> Set[str]: """ Get all the cluster types that this provisioner implementation supports. """ raise NotImplementedError
[docs] @abstractmethod def createClusterSettings(self): """ Initialize class for a new cluster, to be deployed, when running outside the cloud. """ raise NotImplementedError
[docs] @abstractmethod def readClusterSettings(self): """ Initialize class from an existing cluster. This method assumes that the instance we are running on is the leader. Implementations must call _setLeaderWorkerAuthentication(). """ raise NotImplementedError
def _write_file_to_cloud(self, key: str, contents: bytes) -> str: """ Write a file to a physical storage system that is accessible to the leader and all nodes during the life of the cluster. Additional resources should be cleaned up in `self.destroyCluster()`. :return: A public URL that can be used to retrieve the file. """ raise NotImplementedError def _read_file_from_cloud(self, key: str) -> bytes: """ Return the contents of the file written by `self._write_file_to_cloud()`. """ raise NotImplementedError def _get_user_data_limit(self) -> int: """ Get the maximum number of bytes that can be passed as the user data during node creation. """ raise NotImplementedError def _setLeaderWorkerAuthentication(self, leader: Node = None): """ Configure authentication between the leader and the workers. Assumes that we are running on the leader, unless a Node is given, in which case credentials will be pulled from or created there. Configures the backing cluster scheduler so that the leader and workers will be able to communicate securely. Authentication may be one-way or mutual. Until this is called, new nodes may not be able to communicate with the leader. Afterward, the provisioner will include the necessary authentication information when provisioning nodes. :param leader: Node to pull credentials from, if not the current machine. """ if self.clusterType == 'mesos': # We're using a Mesos cluster, so set up SSH from leader to workers. self._leaderWorkerAuthentication = self._setSSH(leader=leader) elif self.clusterType == 'kubernetes': # We're using a Kubernetes cluster. self._leaderWorkerAuthentication = self._getKubernetesJoiningInfo(leader=leader) def _clearLeaderWorkerAuthentication(self): """ Forget any authentication information populated by _setLeaderWorkerAuthentication(). It will need to be called again to provision more workers. """ self._leaderWorkerAuthentication = None def _setSSH(self, leader: Node = None) -> str: """ Generate a key pair, save it in /root/.ssh/ on the leader, and return the public key. The file /root/.sshSuccess is used to prevent this operation from running twice. Also starts the ssh agent on the local node, if operating on the local node. :param leader: Node to operate on, if not the current machine. :return: Public key, without the "ssh-rsa" part. """ # To work locally or remotely we need to do all our setup work as one # big bash -c command = ['bash', '-c', ('set -e; if [ ! -e /root/.sshSuccess ] ; ' 'then ssh-keygen -f /root/.ssh/id_rsa -t rsa -N ""; ' 'touch /root/.sshSuccess; fi; chmod 700 /root/.ssh;')] if leader is None: # Run locally subprocess.check_call(command) # Grab from local file with open('/root/.ssh/') as f: leaderPublicKey = else: # Run remotely leader.sshInstance(*command, appliance=True) # Grab from remote file with tempfile.TemporaryDirectory() as tmpdir: localFile = os.path.join(tmpdir, '') leader.extractFile('/root/.ssh/', localFile, 'toil_leader') with open(localFile) as f: leaderPublicKey = # Drop the key type and keep just the key data leaderPublicKey = leaderPublicKey.split(' ')[1] # confirm it really is an RSA public key assert leaderPublicKey.startswith('AAAAB3NzaC1yc2E'), leaderPublicKey return leaderPublicKey def _getKubernetesJoiningInfo(self, leader: Node = None) -> Dict[str, str]: """ Get the Kubernetes joining info created when Kubernetes was set up on this node, which is the leader, or on a different specified Node. Returns a dict of JOIN_TOKEN, JOIN_CERT_HASH, and JOIN_ENDPOINT, which can be inserted into our Kubernetes worker setup script and config. :param leader: Node to operate on, if not the current machine. """ # Make a parser for the config config = configparser.ConfigParser(interpolation=None) # Leave case alone config.optionxform = str if leader is None: # This info is always supposed to be set up before the Toil appliance # starts, and mounted in at the same path as on the host. So we just go # read it. with open('/etc/kubernetes/worker.ini') as f: config.read_file(f) else: # Grab from remote file with tempfile.TemporaryDirectory() as tmpdir: localFile = os.path.join(tmpdir, 'worker.ini') leader.extractFile('/etc/kubernetes/worker.ini', localFile, 'toil_leader') with open(localFile) as f: config.read_file(f) # Grab everything out of the default section where our setup script put # it. return dict(config['DEFAULT'])
[docs] def setAutoscaledNodeTypes(self, nodeTypes: List[Tuple[Set[str], Optional[float]]]): """ Set node types, shapes and spot bids for Toil-managed autoscaling. :param nodeTypes: A list of node types, as parsed with parse_node_types. """ # This maps from an equivalence class of instance names to a spot bid. self._spotBidsMap = {} # This maps from a node Shape object to the instance type that has that # shape. TODO: what if multiple instance types in a cloud provider have # the same shape (e.g. AMD and Intel instances)??? self._shape_to_instance_type = {} for node_type in nodeTypes: preemptible = node_type[1] is not None if preemptible: # Record the spot bid for the whole equivalence class self._spotBidsMap[frozenset(node_type[0])] = node_type[1] for instance_type_name in node_type[0]: # Record the instance shape and associated type. shape = self.getNodeShape(instance_type_name, preemptible) self._shape_to_instance_type[shape] = instance_type_name
[docs] def hasAutoscaledNodeTypes(self) -> bool: """ Check if node types have been configured on the provisioner (via setAutoscaledNodeTypes). :returns: True if node types are configured for autoscaling, and false otherwise. """ return len(self.getAutoscaledInstanceShapes()) > 0
[docs] def getAutoscaledInstanceShapes(self) -> Dict[Shape, str]: """ Get all the node shapes and their named instance types that the Toil autoscaler should manage. """ if hasattr(self, '_shape_to_instance_type'): # We have had Toil-managed autoscaling set up return dict(self._shape_to_instance_type) else: # Nobody has called setAutoscaledNodeTypes yet, so nothing is to be autoscaled. return {}
[docs] @staticmethod def retryPredicate(e): """ Return true if the exception e should be retried by the cluster scaler. For example, should return true if the exception was due to exceeding an API rate limit. The error will be retried with exponential backoff. :param e: exception raised during execution of setNodeCount :return: boolean indicating whether the exception e should be retried """ return False
[docs] @abstractmethod def launchCluster(self, *args, **kwargs): """ Initialize a cluster and create a leader node. Implementations must call _setLeaderWorkerAuthentication() with the leader so that workers can be launched. :param leaderNodeType: The leader instance. :param leaderStorage: The amount of disk to allocate to the leader in gigabytes. :param owner: Tag identifying the owner of the instances. """ raise NotImplementedError
[docs] @abstractmethod def addNodes( self, nodeTypes: Set[str], numNodes: int, preemptible: bool, spotBid: Optional[float] = None, ) -> int: """ Used to add worker nodes to the cluster :param numNodes: The number of nodes to add :param preemptible: whether or not the nodes will be preemptible :param spotBid: The bid for preemptible nodes if applicable (this can be set in config, also). :return: number of nodes successfully added """ raise NotImplementedError
[docs] def addManagedNodes(self, nodeTypes: Set[str], minNodes, maxNodes, preemptible, spotBid=None) -> None: """ Add a group of managed nodes of the given type, up to the given maximum. The nodes will automatically be launched and terminated depending on cluster load. Raises ManagedNodesNotSupportedException if the provisioner implementation or cluster configuration can't have managed nodes. :param minNodes: The minimum number of nodes to scale to :param maxNodes: The maximum number of nodes to scale to :param preemptible: whether or not the nodes will be preemptible :param spotBid: The bid for preemptible nodes if applicable (this can be set in config, also). """ # Not available by default raise ManagedNodesNotSupportedException("Managed nodes not supported by this provisioner")
[docs] @abstractmethod def terminateNodes(self, nodes: List[Node]) -> None: """ Terminate the nodes represented by given Node objects :param nodes: list of Node objects """ raise NotImplementedError
[docs] @abstractmethod def getLeader(self): """ :return: The leader node. """ raise NotImplementedError
[docs] @abstractmethod def getProvisionedWorkers(self, instance_type: Optional[str] = None, preemptible: Optional[bool] = None) -> List[Node]: """ Gets all nodes, optionally of the given instance type or preemptability, from the provisioner. Includes both static and autoscaled nodes. :param preemptible: Boolean value to restrict to preemptible nodes or non-preemptible nodes :return: list of Node objects """ raise NotImplementedError
[docs] @abstractmethod def getNodeShape(self, instance_type: str, preemptible=False): """ The shape of a preemptible or non-preemptible node managed by this provisioner. The node shape defines key properties of a machine, such as its number of cores or the time between billing intervals. :param str instance_type: Instance type name to return the shape of. :rtype: Shape """ raise NotImplementedError
[docs] @abstractmethod def destroyCluster(self) -> None: """ Terminates all nodes in the specified cluster and cleans up all resources associated with the cluster. :param clusterName: identifier of the cluster to terminate. """ raise NotImplementedError
[docs] class InstanceConfiguration: """ Allows defining the initial setup for an instance and then turning it into an Ignition configuration for instance user data. """ def __init__(self): # Holds dicts with keys 'path', 'owner', 'permissions', and 'content' for files to create. # Permissions is a string octal number with leading 0. self.files = [] # Holds dicts with keys 'name', 'command', and 'content' defining Systemd units to create self.units = [] # Holds strings like "ssh-rsa actualKeyData" for keys to authorize (independently of cloud provider's system) self.sshPublicKeys = []
[docs] def addFile(self, path: str, filesystem: str = 'root', mode: Union[str, int] = '0755', contents: str = '', append: bool = False): """ Make a file on the instance with the given filesystem, mode, and contents. See the storage.files section: """ if isinstance(mode, str): # Convert mode from octal to decimal mode = int(mode, 8) assert isinstance(mode, int) contents = 'data:,' + quote(contents.encode('utf-8')) ignition_file = {'path': path, 'filesystem': filesystem, 'mode': mode, 'contents': {'source': contents}} if append: ignition_file["append"] = append self.files.append(ignition_file)
[docs] def addUnit(self, name: str, enabled: bool = True, contents: str = ''): """ Make a systemd unit on the instance with the given name (including .service), and content. Units will be enabled by default. Unit logs can be investigated with: systemctl status whatever.service or: journalctl -xe """ self.units.append({'name': name, 'enabled': enabled, 'contents': contents})
[docs] def addSSHRSAKey(self, keyData: str): """ Authorize the given bare, encoded RSA key (without "ssh-rsa"). """ self.sshPublicKeys.append("ssh-rsa " + keyData)
[docs] def toIgnitionConfig(self) -> str: """ Return an Ignition configuration describing the desired config. """ # Define the base config. We're using Flatcar's v2.2.0 fork # See: config = { 'ignition': { 'version': '2.2.0' }, 'storage': { 'files': self.files }, 'systemd': { 'units': self.units } } if len(self.sshPublicKeys) > 0: # Add SSH keys if needed config['passwd'] = { 'users': [ { 'name': 'core', 'sshAuthorizedKeys': self.sshPublicKeys } ] } # Serialize as JSON return json.dumps(config, separators=(',', ':'))
[docs] def getBaseInstanceConfiguration(self) -> InstanceConfiguration: """ Get the base configuration for both leader and worker instances for all cluster types. """ config = self.InstanceConfiguration() # We set Flatcar's update reboot strategy to off config.addFile("/etc/coreos/update.conf", mode='0644', contents=textwrap.dedent("""\ GROUP=stable REBOOT_STRATEGY=off """)) # Then we have volume mounting. That always happens. self.addVolumesService(config) # We also always add the service to talk to Prometheus self.addNodeExporterService(config) return config
[docs] def addVolumesService(self, config: InstanceConfiguration): """ Add a service to prepare and mount local scratch volumes. """ # TODO: when # # describes how to collect all the ephemeral disks declaratively and # make Ignition RAID them, stop doing it manually. Might depend on real # solution for # # TODO: check what kind of instance this is, and what ephemeral volumes # *should* be there, and declaratively RAID and mount them. config.addFile("/home/core/", contents=textwrap.dedent("""\ #!/bin/bash set -x ephemeral_count=0 drives=() # Directories are relative to /var directories=(lib/toil lib/mesos lib/docker lib/kubelet lib/cwl tmp) for drive in /dev/xvd{a..z} /dev/nvme{0..26}n1; do echo "checking for ${drive}" if [ -b $drive ]; then echo "found it" while [ "$(readlink -f "${drive}")" != "${drive}" ] ; do drive="$(readlink -f "${drive}")" echo "was a symlink to ${drive}" done seen=0 for other_drive in "${drives[@]}" ; do if [ "${other_drive}" == "${drive}" ] ; then seen=1 break fi done if (( "${seen}" == "1" )) ; then echo "already discovered via another name" continue fi if mount | grep "^${drive}"; then echo "already mounted, likely a root device" else ephemeral_count=$((ephemeral_count + 1 )) drives+=("${drive}") echo "increased ephemeral count by one" fi fi done if (("$ephemeral_count" == "0" )); then echo "no ephemeral drive" for directory in "${directories[@]}"; do sudo mkdir -p /var/$directory done exit 0 fi sudo mkdir /mnt/ephemeral if (("$ephemeral_count" == "1" )); then echo "one ephemeral drive to mount" sudo mkfs.ext4 -F "${drives[@]}" sudo mount "${drives[@]}" /mnt/ephemeral fi if (("$ephemeral_count" > "1" )); then echo "multiple drives" for drive in "${drives[@]}"; do sudo dd if=/dev/zero of=$drive bs=4096 count=1024 done # determine force flag sudo mdadm --create -f --verbose /dev/md0 --level=0 --raid-devices=$ephemeral_count "${drives[@]}" sudo mkfs.ext4 -F /dev/md0 sudo mount /dev/md0 /mnt/ephemeral fi for directory in "${directories[@]}"; do sudo mkdir -p /mnt/ephemeral/var/$directory sudo mkdir -p /var/$directory sudo mount --bind /mnt/ephemeral/var/$directory /var/$directory done """)) # TODO: Make this retry? config.addUnit("volume-mounting.service", contents=textwrap.dedent("""\ [Unit] Description=mounts ephemeral volumes & bind mounts toil directories Before=docker.service [Service] Type=oneshot Restart=no ExecStart=/usr/bin/bash /home/core/ [Install] """))
[docs] def addNodeExporterService(self, config: InstanceConfiguration): """ Add the node exporter service for Prometheus to an instance configuration. """ config.addUnit("node-exporter.service", contents=textwrap.dedent('''\ [Unit] Description=node-exporter container After=docker.service [Service] Restart=on-failure RestartSec=2 ExecStartPre=-/usr/bin/docker rm node_exporter ExecStart=/usr/bin/docker run \\ -p 9100:9100 \\ -v /proc:/host/proc \\ -v /sys:/host/sys \\ -v /:/rootfs \\ --name node-exporter \\ --restart always \\ \\ --path.procfs /host/proc \\ --path.sysfs /host/sys \\ --collector.filesystem.ignored-mount-points ^/(sys|proc|dev|host|etc)($|/) [Install] '''))
[docs] def toil_service_env_options(self) -> str: return "-e TMPDIR=/var/tmp"
[docs] def add_toil_service(self, config: InstanceConfiguration, role: str, keyPath: str = None, preemptible: bool = False): """ Add the Toil leader or worker service to an instance configuration. Will run Mesos master or agent as appropriate in Mesos clusters. For Kubernetes clusters, will just sleep to provide a place to shell into on the leader, and shouldn't run on the worker. :param role: Should be 'leader' or 'worker'. Will not work for 'worker' until leader credentials have been collected. :param keyPath: path on the node to a server-side encryption key that will be added to the node after it starts. The service will wait until the key is present before starting. :param preemptible: Whether a worker should identify itself as preemptible or not to the scheduler. """ # If keys are rsynced, then the mesos-agent needs to be started after the keys have been # transferred. The script loops on the new VM until it finds the keyPath file, then it starts the # mesos-agent. If there are multiple keys to be transferred, then the last one to be transferred must be # set to keyPath. MESOS_LOG_DIR = '--log_dir=/var/lib/mesos ' LEADER_DOCKER_ARGS = '--registry=in_memory --cluster={name}' # --no-systemd_enable_support is necessary in Ubuntu 16.04 (otherwise, # Mesos attempts to contact systemd but can't find its run file) WORKER_DOCKER_ARGS = '--work_dir=/var/lib/mesos --master={ip}:5050 --attributes=preemptible:{preemptible} --no-hostname_lookup --no-systemd_enable_support' if self.clusterType == 'mesos': if role == 'leader': entryPoint = 'mesos-master' entryPointArgs = MESOS_LOG_DIR + LEADER_DOCKER_ARGS.format(name=self.clusterName) elif role == 'worker': entryPoint = 'mesos-agent' entryPointArgs = MESOS_LOG_DIR + WORKER_DOCKER_ARGS.format(ip=self._leaderPrivateIP, preemptible=preemptible) else: raise RuntimeError("Unknown role %s" % role) elif self.clusterType == 'kubernetes': if role == 'leader': # We need *an* entry point or the leader container will finish # and go away, and thus not be available to take user logins. entryPoint = 'sleep' entryPointArgs = 'infinity' else: raise RuntimeError('Toil service not needed for %s nodes in a %s cluster', role, self.clusterType) else: raise RuntimeError('Toil service not needed in a %s cluster', self.clusterType) if keyPath: entryPointArgs = keyPath + ' ' + entryPointArgs entryPoint = "" customDockerInitCommand = customDockerInitCmd() if customDockerInitCommand: entryPointArgs = " ".join(["'" + customDockerInitCommand + "'", entryPoint, entryPointArgs]) entryPoint = "" # Set up the service. Make sure to make it default to using the # actually-big temp directory of /var/tmp (see # config.addUnit(f"toil-{role}.service", contents=textwrap.dedent(f'''\ [Unit] Description=toil-{role} container After=docker.service After=create-kubernetes-cluster.service [Service] Restart=on-failure RestartSec=2 ExecStartPre=-/usr/bin/docker rm toil_{role} ExecStartPre=-/usr/bin/bash -c '{customInitCmd()}' ExecStart=/usr/bin/docker run \\ {self.toil_service_env_options()} \\ --entrypoint={entryPoint} \\ --net=host \\ --init \\ -v /var/run/docker.sock:/var/run/docker.sock \\ -v /run/lock:/run/lock \\ -v /var/lib/mesos:/var/lib/mesos \\ -v /var/lib/docker:/var/lib/docker \\ -v /var/lib/toil:/var/lib/toil \\ -v /var/lib/cwl:/var/lib/cwl \\ -v /var/tmp:/var/tmp \\ -v /tmp:/tmp \\ -v /opt:/opt \\ -v /etc/kubernetes:/etc/kubernetes \\ -v /etc/kubernetes/admin.conf:/root/.kube/config \\ --name=toil_{role} \\ {applianceSelf()} \\ {entryPointArgs} [Install] '''))
[docs] def getKubernetesValues(self, architecture: str = 'amd64'): """ Returns a dict of Kubernetes component versions and paths for formatting into Kubernetes-related templates. """ cloud_provider = self.getKubernetesCloudProvider() return dict( ARCHITECTURE=architecture, CNI_VERSION="v0.8.2", CRICTL_VERSION="v1.17.0", CNI_DIR="/opt/cni/bin", DOWNLOAD_DIR="/opt/bin", SETUP_STATE_DIR="/etc/toil/kubernetes", # This is the version of Kubernetes to use # Get current from: curl -sSL # Make sure it is compatible with the kubelet.service unit we ship, or update that too. KUBERNETES_VERSION="v1.19.3", # Now we need the basic cluster services # Version of Flannel networking to get the YAML from FLANNEL_VERSION="v0.13.0", # Version of node CSR signign bot to run RUBBER_STAMP_VERSION="v0.3.1", # Version of the autoscaler to run AUTOSCALER_VERSION="1.19.0", # Version of metrics service to install for `kubectl top nodes` METRICS_API_VERSION="v0.3.7", CLUSTER_NAME=self.clusterName, # YAML line that tells the Kubelet to use a cloud provider, if we need one. CLOUD_PROVIDER_SPEC=('cloud-provider: ' + cloud_provider) if cloud_provider else '' )
[docs] def addKubernetesServices(self, config: InstanceConfiguration, architecture: str = 'amd64'): """ Add installing Kubernetes and Kubeadm and setting up the Kubelet to run when configured to an instance configuration. The same process applies to leaders and workers. """ values = self.getKubernetesValues(architecture) # We're going to ship the Kubelet service from Kubernetes' release pipeline via cloud-config config.addUnit("kubelet.service", contents=textwrap.dedent('''\ # This came from # It has been modified to replace /usr/bin with {DOWNLOAD_DIR} # License: [Unit] Description=kubelet: The Kubernetes Node Agent Documentation= [Service] ExecStart={DOWNLOAD_DIR}/kubelet Restart=always StartLimitInterval=0 RestartSec=10 [Install] ''').format(**values)) # It needs this config file config.addFile("/etc/systemd/system/kubelet.service.d/10-kubeadm.conf", mode='0644', contents=textwrap.dedent('''\ # This came from # It has been modified to replace /usr/bin with {DOWNLOAD_DIR} # License: # Note: This dropin only works with kubeadm and kubelet v1.11+ [Service] Environment="KUBELET_KUBECONFIG_ARGS=--bootstrap-kubeconfig=/etc/kubernetes/bootstrap-kubelet.conf --kubeconfig=/etc/kubernetes/kubelet.conf" Environment="KUBELET_CONFIG_ARGS=--config=/var/lib/kubelet/config.yaml" # This is a file that "kubeadm init" and "kubeadm join" generates at runtime, populating the KUBELET_KUBEADM_ARGS variable dynamically EnvironmentFile=-/var/lib/kubelet/kubeadm-flags.env # This is a file that the user can use for overrides of the kubelet args as a last resort. Preferably, the user should use # the .NodeRegistration.KubeletExtraArgs object in the configuration files instead. KUBELET_EXTRA_ARGS should be sourced from this file. EnvironmentFile=-/etc/default/kubelet ExecStart= ExecStart={DOWNLOAD_DIR}/kubelet $KUBELET_KUBECONFIG_ARGS $KUBELET_CONFIG_ARGS $KUBELET_KUBEADM_ARGS $KUBELET_EXTRA_ARGS ''').format(**values)) # Before we let the kubelet try to start, we have to actually download it (and kubeadm) # We set up this service so it can restart on failure despite not # leaving a process running, see # <> and # <> # We use a forking service with RemainAfterExit, since that lets # restarts work if the script fails. We also use a condition which # treats the service as successful and skips it if it made a file to # say it already ran. config.addFile("/home/core/", contents=textwrap.dedent('''\ #!/usr/bin/env bash set -e FLAG_FILE="{SETUP_STATE_DIR}/install-kubernetes.done" # Make sure we have Docker enabled; Kubeadm later might complain it isn't. systemctl enable docker.service mkdir -p {CNI_DIR} curl -L "{CNI_VERSION}/cni-plugins-linux-{ARCHITECTURE}-{CNI_VERSION}.tgz" | tar -C {CNI_DIR} -xz mkdir -p {DOWNLOAD_DIR} curl -L "{CRICTL_VERSION}/crictl-{CRICTL_VERSION}-linux-{ARCHITECTURE}.tar.gz" | tar -C {DOWNLOAD_DIR} -xz cd {DOWNLOAD_DIR} curl -L --remote-name-all{KUBERNETES_VERSION}/bin/linux/{ARCHITECTURE}/{{kubeadm,kubelet,kubectl}} chmod +x {{kubeadm,kubelet,kubectl}} mkdir -p "{SETUP_STATE_DIR}" touch "$FLAG_FILE" ''').format(**values)) config.addUnit("install-kubernetes.service", contents=textwrap.dedent('''\ [Unit] Description=base Kubernetes installation Before=kubelet.service ConditionPathExists=!{SETUP_STATE_DIR}/install-kubernetes.done [Service] ExecStart=/usr/bin/bash /home/core/ Type=forking RemainAfterExit=yes Restart=on-failure RestartSec=5s [Install] RequiredBy=kubelet.service ''').format(**values))
# Now we should have the kubeadm command, and the bootlooping kubelet # waiting for kubeadm to configure it.
[docs] def getKubernetesAutoscalerSetupCommands(self, values: Dict[str, str]) -> str: """ Return Bash commands that set up the Kubernetes cluster autoscaler for provisioning from the environment supported by this provisioner. Should only be implemented if Kubernetes clusters are supported. :param values: Contains definitions of cluster variables, like AUTOSCALER_VERSION and CLUSTER_NAME. :returns: Bash snippet """ raise NotImplementedError()
[docs] def getKubernetesCloudProvider(self) -> Optional[str]: """ Return the Kubernetes cloud provider (for example, 'aws'), to pass to the kubelets in a Kubernetes cluster provisioned using this provisioner. Defaults to None if not overridden, in which case no cloud provider integration will be used. :returns: Cloud provider name, or None """ return None
[docs] def addKubernetesLeader(self, config: InstanceConfiguration): """ Add services to configure as a Kubernetes leader, if Kubernetes is already set to be installed. """ values = self.getKubernetesValues() # Customize scheduler to pack jobs into as few nodes as possible # See: config.addFile("/home/core/scheduler-config.yml", mode='0644', contents=textwrap.dedent('''\ apiVersion: kind: KubeSchedulerConfiguration clientConnection: kubeconfig: /etc/kubernetes/scheduler.conf profiles: - schedulerName: default-scheduler plugins: score: disabled: - name: NodeResourcesLeastAllocated enabled: - name: NodeResourcesMostAllocated weight: 1 '''.format(**values))) # Main kubeadm cluster configuration. # Make sure to mount the scheduler config where the scheduler can see # it, which is undocumented but inferred from # config.addFile("/home/core/kubernetes-leader.yml", mode='0644', contents=textwrap.dedent('''\ apiVersion: kind: InitConfiguration nodeRegistration: kubeletExtraArgs: volume-plugin-dir: "/opt/libexec/kubernetes/kubelet-plugins/volume/exec/" {CLOUD_PROVIDER_SPEC} --- apiVersion: kind: ClusterConfiguration controllerManager: extraArgs: flex-volume-plugin-dir: "/opt/libexec/kubernetes/kubelet-plugins/volume/exec/" scheduler: extraArgs: config: "/etc/kubernetes/scheduler-config.yml" extraVolumes: - name: schedulerconfig hostPath: "/home/core/scheduler-config.yml" mountPath: "/etc/kubernetes/scheduler-config.yml" readOnly: true pathType: "File" networking: serviceSubnet: "" podSubnet: "" dnsDomain: "cluster.local" --- apiVersion: kind: KubeletConfiguration serverTLSBootstrap: true rotateCertificates: true cgroupDriver: systemd '''.format(**values))) # Make a script to apply that and the other cluster components # Note that we're escaping {{thing}} as {{{{thing}}}} because we need to match mustaches in a yaml we hack up. config.addFile("/home/core/", contents=textwrap.dedent('''\ #!/usr/bin/env bash set -e FLAG_FILE="{SETUP_STATE_DIR}/create-kubernetes-cluster.done" export PATH="$PATH:{DOWNLOAD_DIR}" # We need the kubelet being restarted constantly by systemd while kubeadm is setting up. # Systemd doesn't really let us say that in the unit file. systemctl start kubelet # We also need to set the hostname for 'kubeadm init' to work properly. /bin/sh -c "/usr/bin/hostnamectl set-hostname $(curl -s" if [[ ! -e /etc/kubernetes/admin.conf ]] ; then # Only run this once, it isn't idempotent kubeadm init --config /home/core/kubernetes-leader.yml fi mkdir -p $HOME/.kube cp /etc/kubernetes/admin.conf $HOME/.kube/config chown $(id -u):$(id -g) $HOME/.kube/config # Install network kubectl apply -f{FLANNEL_VERSION}/Documentation/kube-flannel.yml # Deploy rubber stamp CSR signing bot kubectl apply -f{RUBBER_STAMP_VERSION}/deploy/service_account.yaml kubectl apply -f{RUBBER_STAMP_VERSION}/deploy/role.yaml kubectl apply -f{RUBBER_STAMP_VERSION}/deploy/role_binding.yaml kubectl apply -f{RUBBER_STAMP_VERSION}/deploy/operator.yaml ''').format(**values) + self.getKubernetesAutoscalerSetupCommands(values) + textwrap.dedent('''\ # Set up metrics server, which needs serverTLSBootstrap and rubber stamp, and insists on running on a worker curl -sSL{METRICS_API_VERSION}/components.yaml | \\ sed 's/ - --secure-port=4443/ - --secure-port=4443\\n - --kubelet-preferred-address-types=Hostname/' | \\ kubectl apply -f - # Grab some joining info and make a file we can parse later with configparser echo "[DEFAULT]" >/etc/kubernetes/worker.ini echo "JOIN_TOKEN=$(kubeadm token create --ttl 0)" >>/etc/kubernetes/worker.ini echo "JOIN_CERT_HASH=sha256:$(openssl x509 -pubkey -in /etc/kubernetes/pki/ca.crt | openssl rsa -pubin -outform der 2>/dev/null | openssl dgst -sha256 -hex | sed 's/^.* //')" >>/etc/kubernetes/worker.ini echo "JOIN_ENDPOINT=$(hostname):6443" >>/etc/kubernetes/worker.ini mkdir -p "{SETUP_STATE_DIR}" touch "$FLAG_FILE" ''').format(**values)) config.addUnit("create-kubernetes-cluster.service", contents=textwrap.dedent('''\ [Unit] Description=Kubernetes cluster bootstrap After=install-kubernetes.service After=docker.service Before=toil-leader.service # Can't be before kubelet.service because Kubelet has to come up as we run this. ConditionPathExists=!{SETUP_STATE_DIR}/create-kubernetes-cluster.done [Service] ExecStart=/usr/bin/bash /home/core/ Type=forking RemainAfterExit=yes Restart=on-failure RestartSec=5s [Install] RequiredBy=toil-leader.service ''').format(**values)) # We also need a node cleaner service config.addFile("/home/core/", contents=textwrap.dedent('''\ #!/usr/bin/env bash # constantly clean up NotReady nodes that are tainted as having been deleted set -e export PATH="$PATH:{DOWNLOAD_DIR}" while true ; do echo "$(date | tr -d '\\n'): Checking for scaled-in nodes..." for NODE_NAME in $(kubectl --kubeconfig /etc/kubernetes/admin.conf get nodes -o json | jq -r '.items[] | select(.spec.taints) | select(.spec.taints[] | select(.key == "ToBeDeletedByClusterAutoscaler")) | select(.spec.taints[] | select(.key == "")) | select(.status.conditions[] | select(.type == "Ready" and .status == "Unknown")) |' | tr '\\n' ' ') ; do # For every node that's tainted as ToBeDeletedByClusterAutoscaler, and # as, and hasn't dialed in recently (and # is thus in readiness state Unknown) echo "Node $NODE_NAME is supposed to be scaled away and also gone. Removing from cluster..." # Drop it if possible kubectl --kubeconfig /etc/kubernetes/admin.conf delete node "$NODE_NAME" || true done sleep 300 done ''').format(**values)) config.addUnit("cleanup-nodes.service", contents=textwrap.dedent('''\ [Unit] Description=Remove scaled-in nodes After=create-kubernetes-cluster.service Requires=create-kubernetes-cluster.service [Service] ExecStart=/home/core/ Restart=always StartLimitInterval=0 RestartSec=10 [Install] '''))
[docs] def addKubernetesWorker(self, config: InstanceConfiguration, authVars: Dict[str, str], preemptible: bool = False): """ Add services to configure as a Kubernetes worker, if Kubernetes is already set to be installed. Authenticate back to the leader using the JOIN_TOKEN, JOIN_CERT_HASH, and JOIN_ENDPOINT set in the given authentication data dict. :param config: The configuration to add services to :param authVars: Dict with authentication info :param preemptible: Whether the worker should be labeled as preemptible or not """ # Collect one combined set of auth and general settings. values = dict(**self.getKubernetesValues(), **authVars) # Mark the node as preemptible if it is. # TODO: We use the same label that EKS uses here, because nothing is standardized. # This won't be quite appropriate as we aren't on EKS and we might not # even be on AWS, but the batch system should understand it. values['WORKER_LABEL_SPEC'] = 'node-labels: ""' if preemptible else '' # Kubeadm worker configuration config.addFile("/home/core/kubernetes-worker.yml", mode='0644', contents=textwrap.dedent('''\ apiVersion: kind: JoinConfiguration nodeRegistration: kubeletExtraArgs: volume-plugin-dir: "/opt/libexec/kubernetes/kubelet-plugins/volume/exec/" {CLOUD_PROVIDER_SPEC} {WORKER_LABEL_SPEC} discovery: bootstrapToken: apiServerEndpoint: {JOIN_ENDPOINT} token: {JOIN_TOKEN} caCertHashes: - "{JOIN_CERT_HASH}" --- apiVersion: kind: KubeletConfiguration cgroupDriver: systemd '''.format(**values))) # Make a script to join the cluster using that configuration config.addFile("/home/core/", contents=textwrap.dedent('''\ #!/usr/bin/env bash set -e FLAG_FILE="{SETUP_STATE_DIR}/join-kubernetes-cluster.done" export PATH="$PATH:{DOWNLOAD_DIR}" # We need the kubelet being restarted constantly by systemd while kubeadm is setting up. # Systemd doesn't really let us say that in the unit file. systemctl start kubelet kubeadm join {JOIN_ENDPOINT} --config /home/core/kubernetes-worker.yml mkdir -p "{SETUP_STATE_DIR}" touch "$FLAG_FILE" ''').format(**values)) config.addUnit("join-kubernetes-cluster.service", contents=textwrap.dedent('''\ [Unit] Description=Kubernetes cluster membership After=install-kubernetes.service After=docker.service # Can't be before kubelet.service because Kubelet has to come up as we run this. Requires=install-kubernetes.service ConditionPathExists=!{SETUP_STATE_DIR}/join-kubernetes-cluster.done [Service] ExecStart=/usr/bin/bash /home/core/ Type=forking RemainAfterExit=yes Restart=on-failure RestartSec=5s [Install] ''').format(**values))
def _getIgnitionUserData(self, role, keyPath=None, preemptible=False, architecture='amd64'): """ Return the text (not bytes) user data to pass to a provisioned node. If leader-worker authentication is currently stored, uses it to connect the worker to the leader. :param str keyPath: The path of a secret key for the worker to wait for the leader to create on it. :param bool preemptible: Set to true for a worker node to identify itself as preemptible in the cluster. """ # Start with a base config config = self.getBaseInstanceConfiguration() if self.clusterType == 'kubernetes': # Install Kubernetes self.addKubernetesServices(config, architecture) if role == 'leader': # Set up the cluster self.addKubernetesLeader(config) # We can't actually set up a Kubernetes worker without credentials # to connect back to the leader. if self.clusterType == 'mesos' or role == 'leader': # Leaders, and all nodes in a Mesos cluster, need a Toil service self.add_toil_service(config, role, keyPath, preemptible) if role == 'worker' and self._leaderWorkerAuthentication is not None: # We need to connect the worker to the leader. if self.clusterType == 'mesos': # This involves an SSH public key form the leader config.addSSHRSAKey(self._leaderWorkerAuthentication) elif self.clusterType == 'kubernetes': # We can install the Kubernetes worker and make it phone home # to the leader. # TODO: this puts sufficient info to fake a malicious worker # into the worker config, which probably is accessible by # anyone in the cloud account. self.addKubernetesWorker(config, self._leaderWorkerAuthentication, preemptible=preemptible) # Make it into a string for Ignition user_data = config.toIgnitionConfig() # Check if the config size exceeds the user data limit. If so, we'll # write it to the cloud and let Ignition fetch it during startup. user_data_limit: int = self._get_user_data_limit() if len(user_data) > user_data_limit: logger.warning(f"Ignition config size exceeds the user data limit ({len(user_data)} > {user_data_limit}). " "Writing to cloud storage...") src = self._write_file_to_cloud(f'configs/{role}/config-{uuid4()}.ign', contents=user_data.encode('utf-8')) return json.dumps({ 'ignition': { 'version': '2.2.0', # See: 'config': { 'replace': { 'source': src, } } } }, separators=(',', ':')) return user_data