819 lines
30 KiB
Python
819 lines
30 KiB
Python
# Copyright 2018 The Armada Authors.
|
|
#
|
|
# 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
|
|
#
|
|
# 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 abc import ABC, abstractmethod
|
|
import collections
|
|
import copy
|
|
import math
|
|
import re
|
|
import subprocess # nosec
|
|
import time
|
|
|
|
from kubernetes import watch
|
|
from oslo_config import cfg
|
|
from oslo_log import log as logging
|
|
from retry import retry
|
|
import urllib3.exceptions
|
|
|
|
from armada import const
|
|
from armada.exceptions import k8s_exceptions
|
|
from armada.exceptions import manifest_exceptions
|
|
from armada.exceptions import armada_exceptions
|
|
from armada.handlers.schema import get_schema_info
|
|
from armada.utils.helm import is_test_pod
|
|
from armada.utils.release import label_selectors
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
CONF = cfg.CONF
|
|
|
|
ROLLING_UPDATE_STRATEGY_TYPE = 'RollingUpdate'
|
|
ASYNC_UPDATE_NOT_ALLOWED_MSG = 'Async update not allowed: '
|
|
|
|
|
|
def get_wait_labels(chart):
|
|
wait_config = chart.get('wait', {})
|
|
return wait_config.get('labels', {})
|
|
|
|
|
|
# TODO: Validate this object up front in armada validate flow.
|
|
class ChartWait():
|
|
def __init__(
|
|
self, k8s, release_id, chart, k8s_wait_attempts,
|
|
k8s_wait_attempt_sleep, timeout):
|
|
self.k8s = k8s
|
|
self.release_id = release_id
|
|
self.chart = chart
|
|
chart_data = self.chart[const.KEYWORD_DATA]
|
|
self.chart_data = chart_data
|
|
self.wait_config = self.chart_data.get('wait', {})
|
|
self.k8s_wait_attempts = max(k8s_wait_attempts, 1)
|
|
self.k8s_wait_attempt_sleep = max(k8s_wait_attempt_sleep, 1)
|
|
|
|
schema_info = get_schema_info(self.chart['schema'])
|
|
|
|
resources = self.wait_config.get('resources')
|
|
if isinstance(resources, list):
|
|
# Explicit resource config list provided.
|
|
resources_list = resources
|
|
else:
|
|
# TODO: Remove when v1 doc support is removed.
|
|
if schema_info.version < 2:
|
|
resources_list = [
|
|
{
|
|
'type': 'job',
|
|
'required': False
|
|
}, {
|
|
'type': 'pod'
|
|
}
|
|
]
|
|
else:
|
|
resources_list = self.get_resources_list(resources)
|
|
|
|
chart_labels = get_wait_labels(self.chart_data)
|
|
for resource_config in resources_list:
|
|
# Use chart labels as base labels for each config.
|
|
labels = dict(chart_labels)
|
|
resource_labels = resource_config.get('labels', {})
|
|
# Merge in any resource-specific labels.
|
|
if resource_labels:
|
|
labels.update(resource_labels)
|
|
resource_config['labels'] = labels
|
|
|
|
LOG.debug('Resolved `wait.resources` list: %s', resources_list)
|
|
|
|
self.waits = [self.get_resource_wait(conf) for conf in resources_list]
|
|
|
|
# Calculate timeout
|
|
wait_timeout = timeout
|
|
if wait_timeout is None:
|
|
wait_timeout = self.wait_config.get('timeout')
|
|
|
|
# TODO: Remove when v1 doc support is removed.
|
|
deprecated_timeout = self.chart_data.get('timeout')
|
|
if deprecated_timeout is not None:
|
|
LOG.warn(
|
|
'The `timeout` key is deprecated and support '
|
|
'for this will be removed soon. Use '
|
|
'`wait.timeout` instead.')
|
|
if wait_timeout is None:
|
|
wait_timeout = deprecated_timeout
|
|
|
|
if wait_timeout is None:
|
|
LOG.info(
|
|
'No Chart timeout specified, using default: %ss',
|
|
const.DEFAULT_CHART_TIMEOUT)
|
|
wait_timeout = const.DEFAULT_CHART_TIMEOUT
|
|
|
|
self.timeout = wait_timeout
|
|
|
|
# Determine whether to enable native wait.
|
|
native = self.wait_config.get('native', {})
|
|
|
|
# TODO: Remove when v1 doc support is removed.
|
|
default_native = schema_info.version < 2
|
|
|
|
self.native_enabled = native.get('enabled', default_native)
|
|
|
|
def get_timeout(self):
|
|
return self.timeout
|
|
|
|
def is_native_enabled(self):
|
|
return self.native_enabled
|
|
|
|
def wait(self, timeout):
|
|
deadline = time.time() + timeout
|
|
# TODO(seaneagan): Parallelize waits
|
|
for wait in self.waits:
|
|
wait.wait(timeout=timeout)
|
|
timeout = int(round(deadline - time.time()))
|
|
|
|
def get_resources_list(self, resources):
|
|
# Use default resource configs, with any provided resource type
|
|
# overrides merged in.
|
|
|
|
# By default, wait on all supported resource types.
|
|
resource_order = [
|
|
# Jobs may perform initialization so add them first.
|
|
'job',
|
|
'daemonset',
|
|
'statefulset',
|
|
'deployment',
|
|
'pod'
|
|
]
|
|
base_resource_config = {
|
|
# By default, skip if none found so we don't fail on charts
|
|
# which don't contain resources of a given type.
|
|
'required': False
|
|
}
|
|
# Create a map of resource types to default configs.
|
|
resource_configs = collections.OrderedDict(
|
|
[(type, base_resource_config) for type in resource_order])
|
|
|
|
# Handle any overrides and/or removals of resource type configs.
|
|
if resources:
|
|
for resource_type, v in resources.items():
|
|
if v is False:
|
|
# Remove this type.
|
|
resource_configs.pop(resource_type)
|
|
else:
|
|
# Override config for this type.
|
|
resource_configs[resource_type] = v
|
|
|
|
resources_list = []
|
|
# Convert the resource type map to a list of fully baked resource
|
|
# configs with type included.
|
|
for resource_type, config in resource_configs.items():
|
|
if isinstance(config, list):
|
|
configs = config
|
|
else:
|
|
configs = [config]
|
|
|
|
for conf in configs:
|
|
resource_config = copy.deepcopy(conf)
|
|
resource_config['type'] = resource_type
|
|
resources_list.append(resource_config)
|
|
|
|
return resources_list
|
|
|
|
def get_resource_wait(self, resource_config):
|
|
|
|
kwargs = dict(resource_config)
|
|
resource_type = kwargs.pop('type')
|
|
labels = kwargs.pop('labels')
|
|
|
|
try:
|
|
if resource_type == 'pod':
|
|
return PodWait(resource_type, self, labels, **kwargs)
|
|
elif resource_type == 'job':
|
|
return JobWait(resource_type, self, labels, **kwargs)
|
|
if resource_type == 'deployment':
|
|
return DeploymentWait(resource_type, self, labels, **kwargs)
|
|
elif resource_type == 'daemonset':
|
|
return DaemonSetWait(resource_type, self, labels, **kwargs)
|
|
elif resource_type == 'statefulset':
|
|
return StatefulSetWait(resource_type, self, labels, **kwargs)
|
|
except TypeError:
|
|
raise manifest_exceptions.ManifestException(
|
|
'invalid config for item in `wait.resources`: {}'.format(
|
|
resource_config))
|
|
|
|
raise manifest_exceptions.ManifestException(
|
|
'invalid `type` for item in `wait.resources`: {}'.format(
|
|
resource_config['type']))
|
|
|
|
|
|
class ResourceWait(ABC):
|
|
def __init__(
|
|
self, resource_type, chart_wait, labels, get_resources,
|
|
required=True):
|
|
self.resource_type = resource_type
|
|
self.chart_wait = chart_wait
|
|
self.label_selector = label_selectors(labels)
|
|
self.get_resources = get_resources
|
|
self.required = required
|
|
|
|
@abstractmethod
|
|
def is_resource_ready(self, resource):
|
|
'''
|
|
:param resource: resource to check readiness of.
|
|
:returns: 2-tuple of (status message, ready bool).
|
|
:raises: WaitException
|
|
'''
|
|
pass
|
|
|
|
def get_exclude_reason(self, resource):
|
|
'''
|
|
If a resource should be excluded from the wait, returns a message
|
|
to explain why. Unless overridden, all resources are included.
|
|
:param resource: resource to test
|
|
:returns: string representing exclude reason
|
|
'''
|
|
return None
|
|
|
|
def include_resource(self, resource):
|
|
exclude_reason = self.get_exclude_reason(resource)
|
|
|
|
if exclude_reason:
|
|
LOG.debug(
|
|
'Excluding %s %s from wait: %s', self.resource_type,
|
|
resource.metadata.name, exclude_reason)
|
|
|
|
return not exclude_reason
|
|
|
|
def handle_resource(self, resource):
|
|
resource_name = resource.metadata.name
|
|
resource_desc = '{} {}'.format(self.resource_type, resource_name)
|
|
|
|
try:
|
|
message, resource_ready = self.is_resource_ready(resource)
|
|
|
|
if resource_ready:
|
|
LOG.debug('%s is ready!', resource_desc)
|
|
else:
|
|
LOG.debug('%s not ready: %s', resource_desc, message)
|
|
|
|
return resource_ready
|
|
except armada_exceptions.WaitException as e:
|
|
LOG.warn('%s unlikely to become ready: %s', resource_desc, e)
|
|
return False
|
|
|
|
def wait(self, timeout):
|
|
'''
|
|
:param timeout: time before disconnecting ``Watch`` stream
|
|
'''
|
|
|
|
min_ready_msg = ', min_ready={}'.format(
|
|
self.min_ready.source) if isinstance(self, ControllerWait) else ''
|
|
LOG.info(
|
|
"Waiting for resource type=%s, namespace=%s labels=%s "
|
|
"required=%s%s for %ss", self.resource_type,
|
|
self.chart_wait.release_id.namespace, self.label_selector,
|
|
self.required, min_ready_msg, timeout)
|
|
if not self.label_selector:
|
|
LOG.warn(
|
|
'"label_selector" not specified, waiting with no labels '
|
|
'may cause unintended consequences.')
|
|
|
|
# Track the overall deadline for timing out during waits
|
|
deadline = time.time() + timeout
|
|
|
|
schema_info = get_schema_info(self.chart_wait.chart['schema'])
|
|
# TODO: Remove when v1 doc support is removed.
|
|
if schema_info.version < 2:
|
|
# NOTE(mark-burnett): Attempt to wait multiple times without
|
|
# modification, in case new resources appear after our watch exits.
|
|
successes = 0
|
|
while True:
|
|
modified = self._wait(deadline)
|
|
if modified is None:
|
|
break
|
|
if modified:
|
|
successes = 0
|
|
LOG.debug('Found modified resources: %s', sorted(modified))
|
|
else:
|
|
successes += 1
|
|
LOG.debug('Found no modified resources.')
|
|
|
|
if successes >= self.chart_wait.k8s_wait_attempts:
|
|
return
|
|
|
|
LOG.debug(
|
|
'Continuing to wait: %s consecutive attempts without '
|
|
'modified resources of %s required.', successes,
|
|
self.chart_wait.k8s_wait_attempts)
|
|
time.sleep(self.chart_wait.k8s_wait_attempt_sleep)
|
|
else:
|
|
self._wait(deadline)
|
|
|
|
# The Kubernetes Python Client does not always recover from broken
|
|
# connections to the k8s apiserver, and the resulting uncaught exceptions
|
|
# in the Watch.stream method cause the chart installation to fail. As long
|
|
# as the wait deadline has not passed, it is better to retry the entire
|
|
# wait operation.
|
|
@retry(
|
|
exceptions=(
|
|
urllib3.exceptions.ProtocolError,
|
|
urllib3.exceptions.MaxRetryError),
|
|
delay=1)
|
|
def _wait(self, deadline):
|
|
'''
|
|
Waits for resources to become ready.
|
|
Returns whether resources were modified, or `None` if that is to be
|
|
ignored.
|
|
'''
|
|
|
|
deadline_remaining = int(round(deadline - time.time()))
|
|
if deadline_remaining <= 0:
|
|
error = (
|
|
"Timed out waiting for resource type={}, namespace={}, "
|
|
"labels={}".format(
|
|
self.resource_type, self.chart_wait.release_id.namespace,
|
|
self.label_selector))
|
|
LOG.error(error)
|
|
raise k8s_exceptions.KubernetesWatchTimeoutException(error)
|
|
|
|
timed_out, modified, unready, found_resources = (
|
|
self._watch_resource_completions(timeout=deadline_remaining))
|
|
|
|
if (not found_resources) and not self.required:
|
|
return None
|
|
|
|
if timed_out:
|
|
if not found_resources:
|
|
details = (
|
|
'None found! Are `wait.labels` correct? Does '
|
|
'`wait.resources` need to exclude `type: {}`?'.format(
|
|
self.resource_type))
|
|
else:
|
|
details = (
|
|
'These {}s were not ready={}'.format(
|
|
self.resource_type, sorted(unready)))
|
|
error = (
|
|
'Timed out waiting for {}s (namespace={}, labels=({})). {}'.
|
|
format(
|
|
self.resource_type, self.chart_wait.release_id.namespace,
|
|
self.label_selector, details))
|
|
LOG.error(error)
|
|
raise k8s_exceptions.KubernetesWatchTimeoutException(error)
|
|
|
|
return modified
|
|
|
|
def _watch_resource_completions(self, timeout):
|
|
'''
|
|
Watch and wait for resource completions.
|
|
Returns lists of resources in various conditions for the calling
|
|
function to handle.
|
|
'''
|
|
LOG.debug(
|
|
'Starting to wait on: namespace=%s, resource type=%s, '
|
|
'label_selector=(%s), timeout=%s',
|
|
self.chart_wait.release_id.namespace, self.resource_type,
|
|
self.label_selector, timeout)
|
|
ready = {}
|
|
modified = set()
|
|
found_resources = False
|
|
|
|
if CONF.go_wait:
|
|
command = [
|
|
'armada-go', 'wait', '--resource-type',
|
|
"{}s".format(self.resource_type), '--namespace',
|
|
self.chart_wait.release_id.namespace, '--label-selector',
|
|
self.label_selector, '--timeout', "{}s".format(timeout)
|
|
]
|
|
if hasattr(self, "min_ready"):
|
|
_, _, m_ready = self.min_ready
|
|
command.extend(['--min-ready', "{}".format(m_ready)])
|
|
LOG.info('Running command=%s', command)
|
|
try:
|
|
with subprocess.Popen( # nosec
|
|
command, stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT, bufsize=1,
|
|
universal_newlines=True) as sp:
|
|
for line in sp.stdout:
|
|
LOG.info(line.rstrip())
|
|
sp.wait()
|
|
if sp.returncode != 0:
|
|
raise subprocess.CalledProcessError(
|
|
sp.returncode, command, output=sp.stdout)
|
|
return False, [], [], False
|
|
except subprocess.CalledProcessError as e:
|
|
LOG.info("armada-go wait exception: %s", e)
|
|
raise armada_exceptions.WaitException(e)
|
|
|
|
kwargs = {
|
|
'namespace': self.chart_wait.release_id.namespace,
|
|
'label_selector': self.label_selector,
|
|
'timeout_seconds': timeout
|
|
}
|
|
|
|
resource_list = self.get_resources(**kwargs)
|
|
for resource in resource_list.items:
|
|
# Only include resources that should be included in wait ops
|
|
if self.include_resource(resource):
|
|
ready[resource.metadata.name] = self.handle_resource(resource)
|
|
if not resource_list.items:
|
|
if not self.required:
|
|
msg = 'Skipping non-required wait, no %s resources found.'
|
|
LOG.debug(msg, self.resource_type)
|
|
return (False, modified, [], found_resources)
|
|
else:
|
|
found_resources = True
|
|
if all(ready.values()):
|
|
return (False, modified, [], found_resources)
|
|
|
|
# Only watch new events.
|
|
kwargs['resource_version'] = resource_list.metadata.resource_version
|
|
|
|
w = watch.Watch()
|
|
for event in w.stream(self.get_resources, **kwargs):
|
|
event_type = event['type'].upper()
|
|
resource = event['object']
|
|
resource_name = resource.metadata.name
|
|
resource_version = resource.metadata.resource_version
|
|
|
|
# Skip resources that should be excluded from wait operations
|
|
if not self.include_resource(resource):
|
|
continue
|
|
|
|
msg = (
|
|
'Watch event: type=%s, name=%s, namespace=%s, '
|
|
'resource_version=%s')
|
|
LOG.debug(
|
|
msg, event_type, resource_name,
|
|
self.chart_wait.release_id.namespace, resource_version)
|
|
|
|
if event_type in {'ADDED', 'MODIFIED'}:
|
|
found_resources = True
|
|
resource_ready = self.handle_resource(resource)
|
|
ready[resource_name] = resource_ready
|
|
|
|
if event_type == 'MODIFIED':
|
|
modified.add(resource_name)
|
|
|
|
elif event_type == 'DELETED':
|
|
LOG.debug('Resource %s: removed from tracking', resource_name)
|
|
ready.pop(resource_name)
|
|
|
|
elif event_type == 'ERROR':
|
|
LOG.error(
|
|
'Resource %s: Got error event %s', resource_name,
|
|
event['object'].to_dict())
|
|
raise k8s_exceptions.KubernetesErrorEventException(
|
|
'Got error event for resource: %s' % event['object'])
|
|
|
|
else:
|
|
LOG.error(
|
|
'Unrecognized event type (%s) for resource: %s',
|
|
event_type, event['object'])
|
|
raise (
|
|
k8s_exceptions.
|
|
KubernetesUnknownStreamingEventTypeException(
|
|
'Got unknown event type (%s) for resource: %s' %
|
|
(event_type, event['object'])))
|
|
|
|
if all(ready.values()):
|
|
return (False, modified, [], found_resources)
|
|
|
|
return (
|
|
True, modified,
|
|
[name for name, is_ready in ready.items()
|
|
if not is_ready], found_resources)
|
|
|
|
def _get_resource_condition(self, resource_conditions, condition_type):
|
|
for pc in resource_conditions:
|
|
if pc.type == condition_type:
|
|
return pc
|
|
|
|
|
|
class PodWait(ResourceWait):
|
|
def __init__(self, resource_type, chart_wait, labels, **kwargs):
|
|
super(PodWait, self).__init__(
|
|
resource_type, chart_wait, labels,
|
|
chart_wait.k8s.client.list_namespaced_pod, **kwargs)
|
|
|
|
def get_exclude_reason(self, resource):
|
|
pod = resource
|
|
|
|
# Exclude helm test pods
|
|
# TODO: Possibly exclude other helm hook pods/jobs (besides tests)?
|
|
if is_test_pod(pod):
|
|
return 'helm test pod'
|
|
|
|
if pod.status.phase == 'Evicted':
|
|
return "pod was evicted"
|
|
|
|
schema_info = get_schema_info(self.chart_wait.chart['schema'])
|
|
# TODO: Remove when v1 doc support is removed.
|
|
if schema_info.version < 2:
|
|
# Exclude job pods
|
|
if has_owner(pod, 'Job'):
|
|
return 'owned by job'
|
|
else:
|
|
# Exclude all pods with an owner (only include raw pods)
|
|
if has_owner(pod):
|
|
return 'owned by another resource'
|
|
|
|
return None
|
|
|
|
def is_resource_ready(self, resource):
|
|
pod = resource
|
|
name = pod.metadata.name
|
|
|
|
status = pod.status
|
|
phase = status.phase
|
|
|
|
if phase == 'Succeeded':
|
|
return ("Pod {} succeeded".format(name), True)
|
|
|
|
if phase == 'Running':
|
|
cond = self._get_resource_condition(status.conditions, 'Ready')
|
|
if cond and cond.status == 'True':
|
|
return ("Pod {} ready".format(name), True)
|
|
|
|
msg = "Waiting for pod {} to be ready..."
|
|
return (msg.format(name), False)
|
|
|
|
|
|
class JobWait(ResourceWait):
|
|
def __init__(self, resource_type, chart_wait, labels, **kwargs):
|
|
super(JobWait, self).__init__(
|
|
resource_type, chart_wait, labels,
|
|
chart_wait.k8s.batch_api.list_namespaced_job, **kwargs)
|
|
|
|
def get_exclude_reason(self, resource):
|
|
job = resource
|
|
|
|
# Exclude cronjob jobs
|
|
if has_owner(job, 'CronJob'):
|
|
return 'owned by cronjob (not part of release)'
|
|
|
|
return None
|
|
|
|
def is_resource_ready(self, resource):
|
|
job = resource
|
|
name = job.metadata.name
|
|
|
|
expected = job.spec.completions
|
|
completed = job.status.succeeded
|
|
|
|
if expected != completed:
|
|
msg = "Waiting for job {} to be successfully completed..."
|
|
return (msg.format(name), False)
|
|
msg = "job {} successfully completed"
|
|
return (msg.format(name), True)
|
|
|
|
|
|
def has_owner(resource, kind=None):
|
|
owner_references = resource.metadata.owner_references or []
|
|
|
|
for owner in owner_references:
|
|
if not kind or kind == owner.kind:
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
CountOrPercent = collections.namedtuple(
|
|
'CountOrPercent', 'number is_percent source')
|
|
|
|
# Controller logic (Deployment, DaemonSet, StatefulSet) is adapted from
|
|
# `kubectl rollout status`:
|
|
# https://github.com/kubernetes/kubernetes/blob/master/pkg/kubectl/rollout_status.go
|
|
|
|
|
|
class ControllerWait(ResourceWait):
|
|
def __init__(
|
|
self,
|
|
resource_type,
|
|
chart_wait,
|
|
labels,
|
|
get_resources,
|
|
min_ready="100%",
|
|
**kwargs):
|
|
super(ControllerWait, self).__init__(
|
|
resource_type, chart_wait, labels, get_resources, **kwargs)
|
|
|
|
if isinstance(min_ready, str):
|
|
match = re.match('(.*)%$', min_ready)
|
|
if match:
|
|
min_ready_percent = int(match.group(1))
|
|
self.min_ready = CountOrPercent(
|
|
number=min_ready_percent,
|
|
is_percent=True,
|
|
source=min_ready)
|
|
else:
|
|
raise manifest_exceptions.ManifestException(
|
|
"`min_ready` as string must be formatted as a percent "
|
|
"e.g. '80%'")
|
|
else:
|
|
self.min_ready = CountOrPercent(
|
|
number=min_ready, is_percent=False, source=min_ready)
|
|
|
|
def _is_min_ready(self, ready, total):
|
|
if self.min_ready.is_percent:
|
|
min_ready = math.ceil(total * (self.min_ready.number / 100))
|
|
else:
|
|
min_ready = self.min_ready.number
|
|
return ready >= min_ready
|
|
|
|
|
|
class DeploymentWait(ControllerWait):
|
|
def __init__(self, resource_type, chart_wait, labels, **kwargs):
|
|
super(DeploymentWait, self).__init__(
|
|
resource_type, chart_wait, labels,
|
|
chart_wait.k8s.apps_v1_api.list_namespaced_deployment, **kwargs)
|
|
|
|
def is_resource_ready(self, resource):
|
|
deployment = resource
|
|
name = deployment.metadata.name
|
|
spec = deployment.spec
|
|
status = deployment.status
|
|
gen = deployment.metadata.generation or 0
|
|
observed_gen = status.observed_generation or 0
|
|
|
|
if gen <= observed_gen:
|
|
# TODO: Don't fail for lack of progress if `min_ready` is met.
|
|
# TODO: Consider continuing after `min_ready` is met, so long as
|
|
# progress is being made.
|
|
cond = self._get_resource_condition(
|
|
status.conditions, 'Progressing')
|
|
if cond and (cond.reason or '') == 'ProgressDeadlineExceeded':
|
|
msg = "deployment {} exceeded its progress deadline"
|
|
return (msg.format(name), False)
|
|
|
|
replicas = spec.replicas or 0
|
|
updated_replicas = status.updated_replicas or 0
|
|
available_replicas = status.available_replicas or 0
|
|
if updated_replicas < replicas:
|
|
msg = (
|
|
"Waiting for deployment {} rollout to finish: {} out "
|
|
"of {} new replicas have been updated...")
|
|
return (msg.format(name, updated_replicas, replicas), False)
|
|
|
|
if replicas > updated_replicas:
|
|
msg = (
|
|
"Waiting for deployment {} rollout to finish: {} old "
|
|
"replicas are pending termination...")
|
|
pending = replicas - updated_replicas
|
|
return (msg.format(name, pending), False)
|
|
|
|
if not self._is_min_ready(available_replicas, updated_replicas):
|
|
msg = (
|
|
"Waiting for deployment {} rollout to finish: {} of {} "
|
|
"updated replicas are available, with min_ready={}")
|
|
return (
|
|
msg.format(
|
|
name, available_replicas, updated_replicas,
|
|
self.min_ready.source), False)
|
|
msg = "deployment {} successfully rolled out\n"
|
|
return (msg.format(name), True)
|
|
|
|
msg = "Waiting for deployment spec update to be observed..."
|
|
return (msg.format(), False)
|
|
|
|
|
|
class DaemonSetWait(ControllerWait):
|
|
def __init__(
|
|
self,
|
|
resource_type,
|
|
chart_wait,
|
|
labels,
|
|
allow_async_updates=False,
|
|
**kwargs):
|
|
super(DaemonSetWait, self).__init__(
|
|
resource_type, chart_wait, labels,
|
|
chart_wait.k8s.apps_v1_api.list_namespaced_daemon_set, **kwargs)
|
|
|
|
self.allow_async_updates = allow_async_updates
|
|
|
|
def is_resource_ready(self, resource):
|
|
daemon = resource
|
|
name = daemon.metadata.name
|
|
spec = daemon.spec
|
|
status = daemon.status
|
|
gen = daemon.metadata.generation or 0
|
|
observed_gen = status.observed_generation or 0
|
|
|
|
if not self.allow_async_updates:
|
|
is_update = observed_gen > 1
|
|
if is_update:
|
|
strategy = spec.update_strategy.type or ''
|
|
is_rolling = strategy == ROLLING_UPDATE_STRATEGY_TYPE
|
|
if not is_rolling:
|
|
msg = "{}: update strategy type = {}"
|
|
|
|
raise armada_exceptions.WaitException(
|
|
msg.format(ASYNC_UPDATE_NOT_ALLOWED_MSG, strategy))
|
|
|
|
if gen <= observed_gen:
|
|
updated_number_scheduled = status.updated_number_scheduled or 0
|
|
desired_number_scheduled = status.desired_number_scheduled or 0
|
|
number_available = status.number_available or 0
|
|
|
|
if (updated_number_scheduled < desired_number_scheduled):
|
|
msg = (
|
|
"Waiting for daemon set {} rollout to finish: {} out "
|
|
"of {} new pods have been updated...")
|
|
return (
|
|
msg.format(
|
|
name, updated_number_scheduled,
|
|
desired_number_scheduled), False)
|
|
|
|
if not self._is_min_ready(number_available,
|
|
desired_number_scheduled):
|
|
msg = (
|
|
"Waiting for daemon set {} rollout to finish: {} of {} "
|
|
"updated pods are available, with min_ready={}")
|
|
return (
|
|
msg.format(
|
|
name, number_available, desired_number_scheduled,
|
|
self.min_ready.source), False)
|
|
|
|
msg = "daemon set {} successfully rolled out"
|
|
return (msg.format(name), True)
|
|
|
|
msg = "Waiting for daemon set spec update to be observed..."
|
|
return (msg.format(), False)
|
|
|
|
|
|
class StatefulSetWait(ControllerWait):
|
|
def __init__(
|
|
self,
|
|
resource_type,
|
|
chart_wait,
|
|
labels,
|
|
allow_async_updates=False,
|
|
**kwargs):
|
|
super(StatefulSetWait, self).__init__(
|
|
resource_type, chart_wait, labels,
|
|
chart_wait.k8s.apps_v1_api.list_namespaced_stateful_set, **kwargs)
|
|
|
|
self.allow_async_updates = allow_async_updates
|
|
|
|
def is_resource_ready(self, resource):
|
|
sts = resource
|
|
name = sts.metadata.name
|
|
spec = sts.spec
|
|
status = sts.status
|
|
gen = sts.metadata.generation or 0
|
|
observed_gen = status.observed_generation or 0
|
|
replicas = spec.replicas or 0
|
|
ready_replicas = status.ready_replicas or 0
|
|
updated_replicas = status.updated_replicas or 0
|
|
current_replicas = status.current_replicas or 0
|
|
|
|
if not self.allow_async_updates:
|
|
is_update = observed_gen > 1
|
|
if is_update:
|
|
strategy = spec.update_strategy.type or ''
|
|
is_rolling = strategy == ROLLING_UPDATE_STRATEGY_TYPE
|
|
if not is_rolling:
|
|
msg = "{}: update strategy type = {}"
|
|
|
|
raise armada_exceptions.WaitException(
|
|
msg.format(ASYNC_UPDATE_NOT_ALLOWED_MSG, strategy))
|
|
|
|
if (is_rolling and replicas
|
|
and spec.update_strategy.rolling_update
|
|
and spec.update_strategy.rolling_update.partition):
|
|
msg = "{}: partitioned rollout"
|
|
|
|
raise armada_exceptions.WaitException(
|
|
msg.format(ASYNC_UPDATE_NOT_ALLOWED_MSG))
|
|
|
|
if (observed_gen == 0 or gen > observed_gen):
|
|
msg = "Waiting for statefulset spec update to be observed..."
|
|
return (msg, False)
|
|
|
|
if replicas and not self._is_min_ready(ready_replicas, replicas):
|
|
msg = (
|
|
"Waiting for statefulset {} rollout to finish: {} of {} "
|
|
"pods are ready, with min_ready={}")
|
|
return (
|
|
msg.format(
|
|
name, ready_replicas, replicas,
|
|
self.min_ready.source), False)
|
|
|
|
update_revision = status.update_revision or 0
|
|
current_revision = status.current_revision or 0
|
|
|
|
if update_revision != current_revision:
|
|
msg = (
|
|
"waiting for statefulset rolling update to complete {} "
|
|
"pods at revision {}...")
|
|
return (msg.format(updated_replicas, update_revision), False)
|
|
|
|
msg = "statefulset rolling update complete {} pods at revision {}..."
|
|
return (msg.format(current_replicas, current_revision), True)
|