adjust

#!/usr/bin/env python3
from __future__ import print_function

import copy
import importlib
import sys
import os
import errno
import subprocess
import time
import datetime
import hashlib

from collections.abc import Iterable

import json
import yaml

# import signal

from adjust import Adjust, AdjustError

json_enc = json.JSONEncoder(separators=(",", ":")).encode

# === constants
DESC_FILE = "./config.yaml"
EXCLUDE_LABEL = "optune.ai/exclude"
Gi = 1024 * 1024 * 1024
MEM_STEP = 128 * 1024 * 1024  # minimal useful increment in mem limit/reserve, bytes
CPU_STEP = 0.0125  # 1.25% of a core (even though 1 millicore is the highest resolution supported by k8s)
MAX_MEM = 4 * Gi  # bytes, may be overridden to higher limit
MAX_CPU = 4.0  # cores
# MAX_REPLICAS = 1000 # arbitrary, TBD
FORCED_RESTART_ANN = "servo.opsani.com/forceRestartAt"  # pod annotation to set for forced restart

# the k8s obj to which we make queries/updates:
DEPLOYMENT = "deployment"
# DEPLOYMENT = "deployment.v1.apps"  # new, not supported in 1.8 (it has v1beta1)
RESOURCE_MAP = {"mem": "memory", "cpu": "cpu"}

# top-level keys in config data that are not printed on --query
EXCLUDE_FROM_QUERY = ["driver", "update_annotation", "force_restart"]


class ConfigError(Exception):  # user-provided descriptor not readable
    pass


def import_encoder_base():
    try:
        return importlib.import_module("encoders.base")
    except ImportError:
        raise ImportError("Unable to import base for encoders when handling `command` section.")


# === compute hash of arbitrary data struct
# (copied inline from skopos/.../plugins/spec_hash_helper.py)


def _dbg(*data):
    with open("/skopos/plugins/dbg.log", "a") as f:
        print(data, file=f)


def get_hash(data):
    """md5 hash of Python data. This is limited to scalars that are convertible to string and container
    structures (list, dict) containing such scalars. Some data items are not distinguishable, if they have
    the same representation as a string, e.g., hash(b'None') == hash('None') == hash(None)"""
    # _dbg("get_hash", data)
    hasher = hashlib.md5()
    dump_container(data, hasher.update)
    return hasher.hexdigest()


def dump_container(c, func):
    """stream the contents of a container as a string through a function
    in a repeatable order, suitable, e.g., for hashing
    """
    #
    if isinstance(c, dict):  # dict
        func("{".encode("utf-8"))
        for k in sorted(c):  # for all repeatable
            func("{}:".format(k).encode("utf-8"))
            dump_container(c[k], func)
            func(",".encode("utf-8"))
        func("}".encode("utf-8"))
    elif isinstance(c, list):  # list
        func("[".encode("utf-8"))
        for k in c:  # for all repeatable
            dump_container(k, func)
            func(",".encode("utf-8"))
        func("]".encode("utf-8"))
    else:  # everything else
        if isinstance(c, type(b"")):
            pass  # already a stream, keep as is
        elif isinstance(c, str):
            # encode to stream explicitly here to avoid implicit encoding to ascii
            c = c.encode("utf-8")
        else:
            c = str(c).encode("utf-8")  # convert to string (e.g., if integer)
        func(c)  # simple value, string or convertible-to-string


# ===


def kubectl(namespace, *args):
    cmd_args = ["kubectl"]
    if not bool(int(os.environ.get("OPTUNE_USE_DEFAULT_NAMESPACE", "0"))):
        cmd_args.append("--namespace=" + namespace)
    # append conditional args as provided by env vars
    if os.getenv("OPTUNE_K8S_SERVER") is not None:
        cmd_args.append("--server=" + os.getenv("OPTUNE_K8S_SERVER"))
    if os.getenv("OPTUNE_K8S_TOKEN") is not None:
        cmd_args.append("--token=" + os.getenv("OPTUNE_K8S_TOKEN"))
    if bool(int(os.getenv("OPTUNE_K8S_SKIP_TLS_VERIFY", "0"))):
        cmd_args.append("--insecure-skip-tls-verify=true")
    dbg_txt = "DEBUG: ns='{}', env='{}', r='{}', args='{}'".format(
        namespace, os.environ.get("OPTUNE_USE_DEFAULT_NAMESPACE", "???"), cmd_args, list(args)
    )
    if args[0] == "patch":
        print(dbg_txt, file=sys.stderr)
    else:
        dbg_log(dbg_txt)
    return cmd_args + list(args)


def k_get(namespace, qry):
    """run kubectl get and return parsed json output"""
    if not isinstance(qry, list):
        qry = [qry]
    # this will raise exception if it fails:
    output = subprocess.check_output(kubectl(namespace, "get", "--output=json", *qry))
    output = output.decode("utf-8")
    output = json.loads(output)
    return output


def k_patch(namespace, typ, obj, patchstr):
    """run kubectl patch and return parsed json output"""

    # this will raise exception if it fails:
    cmd = kubectl(namespace, "patch", "--output=json", typ, obj, "-p", patchstr)
    output = subprocess.check_output(cmd)
    output = output.decode("utf-8")
    output = json.loads(output)
    return output


def read_desc():
    """load the user-defined descriptor, returning a dictionary of the contents under the k8s top-level key, if any"""
    try:
        f = open(DESC_FILE)
        desc = yaml.safe_load(f)
    except IOError as e:
        if e.errno == errno.ENOENT:
            raise ConfigError("configuration file {} does not exist".format(DESC_FILE))
        raise ConfigError("cannot read configuration from {}: {}".format(DESC_FILE, e.strerror))
    except yaml.error.YAMLError as e:
        raise ConfigError("syntax error in {}: {}".format(DESC_FILE, str(e)))

    refer_tip = "You can refer to a sample configuration in README.md."
    assert bool(desc), "Configuration file is empty."
    driver_key = "k8s"
    if os.environ.get("OPTUNE_USE_DRIVER_NAME", False):
        driver_key = os.path.basename(__file__)
    assert driver_key in desc and desc[driver_key], (
        "No configuration is defined for K8s driver in config file {}. "
        'Please set up configuration for deployments under key "{}". '
        "{}".format(DESC_FILE, refer_tip, driver_key)
    )
    desc = desc[driver_key]

    assert (
        "application" in desc and desc["application"]
    ), 'Section "application" was not defined in a configuration file. {}'.format(refer_tip)
    assert (
        "components" in desc["application"] and desc["application"]["components"] is not None
    ), 'Section "components" was not defined in a configuration file section "application". {}'.format(refer_tip)
    assert desc["application"][
        "components"
    ], "No components were defined in a configuration file. " "Please define at least one component. {}".format(
        refer_tip
    )

    comps = desc["application"]["components"]
    replicas_tracker = {}
    for name, comp in comps.items():
        settings = comp.get("settings", {})
        # sub-setting validation
        validate_setting_configs(name, settings)

        # cross-component validation
        if "replicas" in settings:
            dep_name = comp.get("deployment", name)
            dep_name = dep_name.split("/")[0]  # if no '/', this just gets the whole name
            replicas_tracker.setdefault(dep_name, 0)
            replicas_tracker[dep_name] += 1

    if len(replicas_tracker) < sum(replicas_tracker.values()):
        rotten_deps = map(lambda d: d[0], filter(lambda c: c[1] > 1, replicas_tracker.items()))
        raise Exception(
            'Several components in the same deployment have "replicas" defined. Affected deployments: {}. '
            'Please, keep only one "replicas" per deployment.'.format(", ".join(rotten_deps))
        )

    ann_key = desc.get("update_annotation", None)
    if ann_key is not None:
        assert isinstance(ann_key, str), "'update_annotation' must have a string value"
    if "force_restart" in desc:
        v = desc["force_restart"]
        if isinstance(v, str):
            try:
                v = bool(int(v))
            except Exception:
                raise ConfigError("'force_restart' must be boolean or convertible to integer/boolean")
        desc["force_restart"] = v

    return desc


def validate_setting_configs(name, settings):
    for k, v in settings.items():
        if k in ["mem", "cpu"] and v.get("selector") == "request_min_limit" and not (v.get("limit_min", 0) > 0):
            err_str = "Component {name} configuration was malformed; limit_min > 0 required when selector == request_min_limit. Found: {val}"
            raise ConfigError(err_str.format(name=name, val=v.get("limit_min")))


def numval(v, minv, maxv, step=1.0, pinn=None):
    """shortcut for creating linear setting descriptions"""
    ret = {"value": v, "min": minv, "max": maxv, "step": step, "type": "range"}
    if pinn is not None:
        ret["pinned"] = bool(pinn)
    return ret


def cpuunits(s):
    """convert a string for CPU resource (with optional unit suffix) into a number"""
    if s[-1] == "m":  # there are no units other than 'm' (millicpu)
        return float(s[:-1]) / 1000.0
    return float(s)


# valid mem units: E, P, T, G, M, K, Ei, Pi, Ti, Gi, Mi, Ki
# nb: 'm' suffix found after setting 0.7Gi
mumap = {
    "E": 1000 ** 6,
    "P": 1000 ** 5,
    "T": 1000 ** 4,
    "G": 1000 ** 3,
    "M": 1000 ** 2,
    "K": 1000,
    "m": 1000 ** -1,
    "Ei": 1024 ** 6,
    "Pi": 1024 ** 5,
    "Ti": 1024 ** 4,
    "Gi": 1024 ** 3,
    "Mi": 1024 ** 2,
    "Ki": 1024,
}


def memunits(s):
    """convert a string for memory resource (with optional unit suffix) into a number"""
    for u, m in mumap.items():
        if s.endswith(u):
            return float(s[: -len(u)]) * m
    return float(s)


def check_setting(name, settings):
    assert isinstance(settings, Iterable), 'Object "settings" passed to check_setting() is not iterable.'
    assert name not in settings, (
        'Setting "{}" has been define more than once. '
        "Please, check other config sections for setting duplicates.".format(name)
    )


def encoder_setting_name(setting_name, encoder_config):
    prefix = encoder_config["setting_prefix"] if "setting_prefix" in encoder_config else ""
    return "{}{}".format(prefix, setting_name)


def describe_encoder(value, config, exception_context="a describe phase of an encoder"):
    encoder_base = import_encoder_base()
    try:
        settings = encoder_base.describe(config, value or "")
        for name, setting in settings.items():
            yield (encoder_setting_name(name, config), setting)
    except BaseException as e:
        raise Exception("Error while handling {}: {}".format(exception_context, str(e)))


def encode_encoder(settings, config, expected_type=None, exception_context="an encode phase of an encoder"):
    encoder_base = import_encoder_base()
    try:
        sanitized_settings = settings
        prefix = config.get("setting_prefix")
        if prefix:
            sanitized_settings = dict(
                map(lambda i: (i[0].lstrip(prefix), i[1]), filter(lambda i: i[0].startswith(prefix), settings.items()))
            )
        encoded_value, encoded_settings = encoder_base.encode(config, sanitized_settings, expected_type=expected_type)
        encoded_settings = list(map(lambda setting_name: encoder_setting_name(setting_name, config), encoded_settings))
        return encoded_value, encoded_settings
    except BaseException as e:
        raise Exception("Error while handling {}: {}".format(exception_context, str(e)))


def islookinglikerangesetting(s):
    return "min" in s or "max" in s or "step" in s


def islookinglikeenumsetting(s):
    return "values" in s


def israngesetting(s):
    return s.get("type") == "range" or islookinglikerangesetting(s)


def isenumsetting(s):
    return s.get("type") == "enum" or islookinglikeenumsetting(s)


def issetting(s):
    return isinstance(s, dict) and (israngesetting(s) or isenumsetting(s))


def get_rsrc(desc_settings, cont_resources, sn):
    rn = RESOURCE_MAP[sn]
    selector = desc_settings.get(sn, {}).get("selector", "both")
    if selector in ["request", "both", "request_min_limit"]:
        val = cont_resources.get("requests", {}).get(rn)
        if val is None:
            val = cont_resources.get("limits", {}).get(rn)
            if val is not None:
                Adjust.print_json_error(
                    error="warning",
                    cl=None,
                    message='Using the non-selected value "limit" for resource "{}" as the selected value is not set'.format(
                        sn
                    ),
                )
            else:
                val = "0"
    else:  # selector == 'limit'
        val = cont_resources.get("limits", {}).get(rn)
        if val is None:
            val = cont_resources.get("requests", {}).get(rn)
            if val is not None:
                if selector == "limit":
                    Adjust.print_json_error(
                        error="warning",
                        cl=None,
                        message='Using the non-selected value "request" for resource "{}" as the selected value is not set'.format(
                            sn
                        ),
                    )
                # else: don't print warning for 'both'
            else:
                val = "0"
    return val


def get_latest_rs(appname, labels, deployment):
    rs = k_get(appname, ["-l", labels, "rs"])["items"]
    dep_rs = [
        x
        for x in rs
        if any(
            y.get("uid") is not None and y.get("uid") == deployment.get("metadata", {}).get("uid")
            for y in x.get("metadata", {}).get("ownerReferences", [])
        )
    ]
    if not dep_rs:
        raise AdjustError("Unable to locate replicaset(s) for deployment. Found replica_sets: {}".format(rs))

    return max(
        dep_rs,
        key=lambda r: int(r.get("metadata", {}).get("annotations", {}).get("deployment.kubernetes.io/revision", -1)),
    )


def get_latest_pods(appname, labels, replicaset, pod_debug=False):
    if pod_debug:
        output = subprocess.check_output(kubectl(appname, "get", "-l", labels, "pods"))
        print("DEBUG pods: \n{}".format(output.decode("utf-8")), file=sys.stderr)

    pods = k_get(appname, ["-l", labels, "pods"])
    return [
        pod
        for pod in pods["items"]
        if any(
            ownRef.get("uid") is not None and ownRef.get("uid") == replicaset.get("metadata", {}).get("uid")
            for ownRef in pod.get("metadata", {}).get("ownerReferences", [])
        )
    ]


def raw_query(appname, desc, pod_debug=False):
    """
    Read the list of deployments in a namespace and fill in data into desc.
    Both the input 'desc' and the return value are in the 'settings query response' format.
    NOTE only 'cpu', 'memory' and 'replicas' settings are filled in even if not present in desc.
    Other settings must have a description in 'desc' to be returned.
    """
    desc = copy.deepcopy(desc)

    app = desc["application"]
    comps = app["components"]

    cfg = desc.pop(
        "control", {}
    )  # FIXME TODO - query doesn't receive data from remote, only the local cfg can be used; where in the data should the "control" section really be?? note, [userdata][deployment] sub-keys for specifying the 'reference' app means we have to have that 'reference' as a single deployment and it has to be excluded from enumeration as an 'adjustable' component, using the whitelist.
    refapp = cfg.get("userdata", {}).get("deployment", None)
    mon_data = {}
    if refapp:
        d2 = desc.copy()
        c2 = copy.deepcopy(cfg)
        c2["userdata"].pop("deployment", None)
        d2["control"] = c2
        if (
            len(comps) != 1
        ):  # 'reference app' works only with single-component (due to the use of deployment name as 'component name' and having both apps in the same namespace)
            raise AdjustError(
                "operation with reference app not possible when multiple components are defined",
                status="aborted",
                reason="ref-app-unavailable",
            )
        refcomps = {refapp: comps[list(comps.keys())[0]]}
        d2["application"] = {
            "components": refcomps
        }  # single component, renamed (so we pick the 'reference deployment' in the same namespace)
        try:
            refqry, _, _ = raw_query(appname, d2)
        except AdjustError as e:
            raise AdjustError(str(e), status="aborted", reason="ref-app-unavailable")
        # let other exceptions go unchanged

        # TODO: maybe something better than a sum is needed here, some multi-component scale events could end up modifying scale counts without changing the overall sum
        replicas_sum = sum((c["settings"]["replicas"]["value"] for c in refqry["application"]["components"].values()))
        refqry = refqry["monitoring"]  # we don't need other data from refqry any more
        mon_data = {
            "ref_spec_id": refqry["spec_id"],
            "ref_version_id": refqry["version_id"],
            "ref_runtime_count": replicas_sum,
        }
        if refqry.get("runtime_id"):
            mon_data["ref_runtime_id"] = refqry["runtime_id"]

    deployments = k_get(appname, DEPLOYMENT)
    # note d["Kind"] should be "List"
    deps_list = deployments["items"]
    if (
        not deps_list
    ):  # NOTE we don't distinguish the case when the namespace doesn't exist at all or is just empty (k8s will return an empty list whether or not it exists)
        raise AdjustError(
            "application '{}' does not exist or has no components".format(appname),
            status="aborted",
            reason="app-unavailable",
        )  # NOTE not a documented 'reason'
    deps_dict = {dep["metadata"]["name"]: dep for dep in deps_list}
    raw_specs = {}
    imgs = {}
    runtime_ids = {}
    restart_counts = {}
    # ?? TODO: is it possible to have an item in 'd' with "kind" other than "Deployment"? (likely no)
    #          is it possible to have replicas == 0 (and how do we represent that, if at all)
    for full_comp_name, comp_data in comps.items():
        dep_name = comp_data.get("deployment", full_comp_name)
        cont_name = None
        if "/" in dep_name:
            dep_name, cont_name = dep_name.split("/")
        assert (
            dep_name in deps_dict
        ), 'Could not find deployment "{}" defined for component "{}" in namespace "{}".' "".format(
            dep_name, full_comp_name, appname
        )
        dep = deps_dict[dep_name]
        conts = dep["spec"]["template"]["spec"]["containers"]
        if cont_name is not None:
            contsd = {c["name"]: c for c in conts}
            assert cont_name in contsd, (
                'Could not find container with name "{}" in deployment "{}" '
                'for component "{}" in namespace "{}".'
                "".format(cont_name, dep_name, full_comp_name, appname)
            )
            cont = contsd[cont_name]
        else:
            cont = conts[0]

        # skip if excluded by label
        try:
            if bool(int(dep["metadata"].get("labels", {}).get(EXCLUDE_LABEL, "0"))):  # string value of 1 (non-0)
                continue
        except ValueError as e:  # int() is the only thing that should trigger exceptions here
            # TODO add warning to annotations to be returned
            print(
                "failed to parse exclude label for deployment {}: {}: {}; ignored".format(
                    dep_name, type(e).__name__, str(e)
                ),
                file=sys.stderr,
            )
            # pass # fall through, ignore unparseable label

        # selector for pods, NOTE this relies on having a equality-based label selector,
        # k8s seems to support other types, I don't know what's being used in practice.
        try:
            sel = dep["spec"]["selector"]["matchLabels"]
        except KeyError:
            raise AdjustError(
                "only deployments with matchLabels selector are supported, found selector: {}".format(
                    repr(dep["spec"].get("selector", {}))
                ),
                status="aborted",
                reason="app-unavailable",
            )  # NOTE not a documented 'reason'
        # convert to string suitable for 'kubect -l labelsel'
        sel = ",".join(("{}={}".format(k, v) for k, v in sel.items()))

        # list of pods, for runtime_id
        try:
            latest_rs = get_latest_rs(appname=appname, labels=sel, deployment=dep)
            pods = get_latest_pods(appname, sel, latest_rs, pod_debug)
            # NOTE: "Terminating" is not an actual phase on the pod status. More info here: https://github.com/kubernetes/kubernetes/issues/22839
            non_terminating = [pod for pod in pods if not pod["metadata"].get("deletionTimestamp")]
            runtime_ids[dep_name] = [pod["metadata"]["uid"] for pod in non_terminating]
            restart_counts[dep_name] = [
                {
                    "pod+container": "{}+{}".format(pod["metadata"]["name"], cont_stat["name"]),
                    "restartCount": cont_stat["restartCount"],
                }
                for pod in pods
                for cont_stat in pod["status"].get("containerStatuses", [])
            ]
        except subprocess.CalledProcessError as e:
            # TODO: re-implement graceful failure
            # Adjust.print_json_error(error="warning", cl="CalledProcessError", message='Unable to retrieve pods: {}. Output: {}'.format(e, e.output))
            raise AdjustError(
                "Unable to get pods for deployment {}: rc {}, output: {}".format(dep_name, e.returncode, e.output),
                status="aborted",
                reason="app-unavailable",
            )

        # extract deployment settings
        # NOTE: generation, resourceVersion and uid can help detect changes
        # (also, to check PG's k8s code in oco)
        replicas = dep["spec"]["replicas"]
        tmplt_spec = dep["spec"]["template"]["spec"]
        raw_specs[dep_name] = tmplt_spec  # save for later, used to checksum all specs

        # name, env, resources (limits { cpu, memory }, requests { cpu, memory })
        # FIXME: what to do if there's no mem reserve or limits defined? (a namespace can have a default mem limit, but that's not necessarily set, either)
        # (for now, we give the limit as 0, treated as 'unlimited' - AFAIK)
        imgs[full_comp_name] = cont["image"]  # FIXME, is this always defined?
        comp = comps[full_comp_name] = comps[full_comp_name] or {}
        settings = comp["settings"] = comp.setdefault("settings", {}) or {}
        read_mem = settings and "mem" in settings
        read_cpu = settings and "cpu" in settings
        read_replicas = not settings or "replicas" in settings
        res = cont.get("resources")
        if res:
            if read_mem:
                mem_val = get_rsrc(desc_settings=settings, cont_resources=res, sn="mem")
                # (value, min, max, step) all in GiB
                settings["mem"] = numval(
                    v=memunits(mem_val) / Gi,
                    minv=(settings.get("mem") or {}).get("min", MEM_STEP / Gi),
                    maxv=(settings.get("mem") or {}).get("max", MAX_MEM / Gi),
                    step=(settings.get("mem") or {}).get("step", MEM_STEP / Gi),
                    pinn=(settings.get("mem") or {}).get("pinned", None),
                )
            if read_cpu:
                cpu_val = get_rsrc(desc_settings=settings, cont_resources=res, sn="cpu")
                # (value, min, max, step), all in CPU cores
                settings["cpu"] = numval(
                    v=cpuunits(cpu_val),
                    minv=(settings.get("cpu") or {}).get("min", CPU_STEP),
                    maxv=(settings.get("cpu") or {}).get("max", MAX_CPU),
                    step=(settings.get("cpu") or {}).get("step", CPU_STEP),
                    pinn=(settings.get("cpu") or {}).get("pinned", None),
                )
        else:
            if read_mem:
                settings["mem"]["type"] = "range"
                settings["mem"]["value"] = None
            if read_cpu:
                settings["cpu"]["type"] = "range"
                settings["cpu"]["value"] = None
            # TODO: adjust min/max to include current values, (e.g., increase mem_max to at least current if current > max)
        # set replicas: FIXME: can't actually be set for each container (the pod as a whole is replicated); for now we have no way of expressing this limitation in the setting descriptions
        # note: setting min=max=current replicas, since there is no way to know what is allowed; use override descriptor to loosen range
        if read_replicas:
            settings["replicas"] = numval(
                v=replicas,
                minv=(settings.get("replicas") or {}).get("min", replicas),
                maxv=(settings.get("replicas") or {}).get("max", replicas),
                step=(settings.get("replicas") or {}).get("step", 1),
                pinn=(settings.get("replicas") or {}).get("pinned", None),
            )

        # current settings of custom env vars (NB: type conv needed for numeric values!)
        cont_env_list = cont.get("env", [])
        # include only vars for which the keys 'name' and 'value' are defined
        cont_env_dict = {i["name"]: i["value"] for i in cont_env_list if "name" in i and "value" in i}

        env = comp.get("env")
        if env:
            for en, ev in env.items():
                check_setting(en, settings)
                assert isinstance(ev, dict), 'Setting "{}" in section "env" of a config file is not a dictionary.'
                if "encoder" in ev:
                    for name, setting in describe_encoder(
                        cont_env_dict.get(en),
                        ev["encoder"],
                        exception_context="an environment variable {}" "".format(en),
                    ):
                        check_setting(name, settings)
                        settings[name] = setting
                if issetting(ev):
                    defval = ev.pop("default", None)
                    val = cont_env_dict.get(en, defval)
                    val = float(val) if israngesetting(ev) and isinstance(val, (int, str)) else val
                    assert val is not None, (
                        'Environment variable "{}" does not have a current value defined and '
                        "neither it has a default value specified in a config file. "
                        "Please, set current value for this variable or adjust the "
                        "configuration file to include its default value."
                        "".format(en)
                    )
                    val = {**ev, "value": val}
                    settings[en] = val
            # Remove section "env" from final descriptor
            del comp["env"]

        command = comp.get("command")
        if command:
            if command.get("encoder"):
                for name, setting in describe_encoder(
                    cont.get("command", []), command["encoder"], exception_context="a command section"
                ):
                    check_setting(name, settings)
                    settings[name] = setting
                # Remove section "command" from final descriptor
            del comp["command"]

    # if runtime_ids:
    mon_data["runtime_id"] = get_hash(runtime_ids)

    # app state data
    # (NOTE we strip the component names because our (single-component) 'reference' app will necessarily have a different component name)
    # this should be resolved by complete re-work, if we are to support 'reference' app in a way that allows multiple components
    raw_specs = [raw_specs[k] for k in sorted(raw_specs.keys())]
    imgs = [imgs[k] for k in sorted(imgs.keys())]
    mon_data.update(
        {
            "spec_id": get_hash(raw_specs),
            "version_id": get_hash(imgs),
            # "runtime_count": replicas_sum
        }
    )

    desc["monitoring"] = mon_data

    return desc, deps_list, restart_counts


# DEBUG:
def ydump(fn, data):
    f = open(fn, "w")
    yaml.dump(data, f)
    f.close()


def dbg_log(*args):
    if os.getenv("TDR_DEBUG_LOG"):
        print(*args, file=sys.stderr)


def query(appname, desc):
    r, _, _ = raw_query(appname, desc)
    return r


class Waiter(object):
    """an object for use to poll and wait for a condition;
    use:
        w = Waiter(max_time, delay)
        while w.wait():
            if test_condition(): break
        if w.expired:
            raise Hell
    """

    def __init__(self, timeout, delay=1):
        self.timefn = time.time  # change that on windows to time.clock
        self.start = self.timefn()
        self.end = self.start + timeout
        self.delay = delay
        self.expired = False

    def wait(self):
        self.expired = self.end < self.timefn()
        waiting = not self.expired
        if waiting:
            time.sleep(self.delay)  # TODO: add support for increasing delay over time
        return waiting


def test_dep_generation(dep, g, ge=False):
    """check if the deployment status indicates it has been updated to the given generation number"""
    if ge:
        return dep["status"]["observedGeneration"] >= g
    return dep["status"]["observedGeneration"] == g


def test_dep_progress(dep):
    """check if the deployment object 'dep' has reached final successful status
    ('dep' should be the data returned by 'kubectl get deployment' or the equivalent API call, e.g.,
    GET /apis/(....)/namespaces/:ns/deployments/my-deployment-name).
    This tests the conditions[] array and the replica counts and converts the data to a simplified status, as follows:
    - if the deployment appears to be in progress and k8s is still waiting for updates from the controlled objects (replicasets and their pods),
      return a tuple (x, ""), where x is the fraction of the updated instances (0.0 .. 1.0, excluding 1.0).
    - if the deployment has completed, return (1.0, "")
    - if the deployment has stalled or failed, return (x, "(errormsg)"), with an indication of the
      detected failure (NOTE: in k8s, the 'stall' is never final and could be unblocked by change
      of resources or other modifications of the cluster not related to the deployment in question,
      but we assume that the system is operating under stable conditions and there won't be anyone
      or anything that can unblock such a stall)
    """
    dbg_log("test_dep_progress:")
    spec_replicas = dep["spec"]["replicas"]  # this is what we expect as target
    progress_final = None
    dep_status = dep["status"]

    sel = dep["spec"]["selector"]["matchLabels"]
    sel = ",".join(("{}={}".format(k, v) for k, v in sel.items()))
    latest_rs = get_latest_rs(dep["metadata"]["namespace"], sel, dep)
    rs_status = latest_rs["status"]

    for co in dep_status["conditions"]:
        dbg_log(
            "... condition type {}, reason {}, status {}, message {}".format(
                co.get("type"), co.get("reason"), co.get("status"), co.get("message")
            )
        )
        if co["type"] == "Progressing":
            if co["status"] == "True" and co["reason"] == "NewReplicaSetAvailable":
                # if the replica set was updated, test the replica counts
                if dep_status.get("updatedReplicas", None) == spec_replicas:  # update complete, check other counts
                    if (
                        rs_status.get("availableReplicas", None) == spec_replicas
                        and rs_status.get("readyReplicas", None) == spec_replicas
                    ):
                        return (1.0, "")  # done
            elif co["status"] == "False":  # failed
                return (
                    dep_status.get("updatedReplicas", 0) / spec_replicas,
                    co["reason"] + ", " + co.get("message", ""),
                )
            # otherwise, assume in-progress
        elif co["type"] == "ReplicaFailure":
            # note if this status is found, we report failure early here, before k8s times out
            return (dep_status.get("updatedReplicas", 0) / spec_replicas, co["reason"] + ", " + co.get("message", ""))

    # no errors and not complete yet, assume in-progress
    # (NOTE if "Progressing" condition isn't found, but updated replicas is good, we will return 100% progress; in this case check that other counts are correct, as well!
    if spec_replicas == 0:  # If dep is being destroyed
        progress = 1.0 if dep_status.get("replicas", 0) == 0 else 0.99 / dep_status.get("replicas", 0)
    else:
        progress = dep_status.get("updatedReplicas", 0) / spec_replicas
        if progress == 1.0:
            if (
                rs_status.get("availableReplicas", None) == spec_replicas
                and rs_status.get("readyReplicas", None) == spec_replicas
            ):
                return (1.0, "")  # all good
            progress = 0.99  # available/ready counts aren't there - don't report 100%, wait loop will contiune until ready or time out

    # check for pod restarts
    pods = get_latest_pods(dep["metadata"]["namespace"], sel, latest_rs)
    if progress == 1.0 and spec_replicas == 0:
        progress = 1.0 if len(pods) == 0 else 0.99 / len(pods)
    restart_counts = [
        {
            "pod+container": "{}+{}".format(pod["metadata"]["name"], cont_stat["name"]),
            "restartCount": cont_stat["restartCount"],
        }
        for pod in pods
        for cont_stat in pod["status"].get("containerStatuses", [])
        if cont_stat["restartCount"] > 0
    ]
    if restart_counts and spec_replicas > 0:
        return (
            progress,
            "component(s) crash restart detected on deployment {}: {}".format(dep["metadata"]["name"], restart_counts),
        )

    return (progress, "")


def compare_settings(patch, dep):
    """test select parts of a deployment patch against an actual deployment object,
    return None if they match, or a string detailing the difference otherwise.
    Only spec/template/spec/containers/:N:/resources/limits and .../resources/requests are compared. If a
    patch has 'None' setting for a resource, the corresponding value is the deployment is not checked
    (patching to None means 'delete', not sure if K8s deletes or sets to a default value in this case).
    """
    try:
        p_containers = patch["spec"]["template"]["spec"]["containers"]
    except KeyError:
        # patch does not set any resource, return OK
        return None
    # KeyError here not caught (propagate as fatal error)
    d_containers = dep["spec"]["template"]["spec"]["containers"]

    # convert arrays to maps, for easy matching
    p_containers = {x["name"]: x for x in p_containers}
    d_containers = {x["name"]: x for x in d_containers}

    # compare
    for k, v in p_containers.items():
        if not v.get("resources"):  # patch didn't set resources
            continue
        c = d_containers.get(k)
        if not c or not c.get("resources"):  # patch has resources, but dep doesn't: mismatch
            msg = "no resources in deployment for container '{}'".format(k)
            print("compare_settings: " + msg, file=sys.stderr)
            return msg
        p_rsrc = v["resources"]
        d_rsrc = c["resources"]
        print("compare_settings: cname={}, comparing: {} <> {}".format(k, repr(p_rsrc), repr(d_rsrc)), file=sys.stderr)
        for rtype, rd in p_rsrc.items():
            for rname, rvalue in rd.items():
                if rvalue is None:  # don't compare if patch says None
                    continue
                try:
                    dep_rvalue = d_rsrc[rtype][rname]
                except KeyError:  # patched value not present in dep - mismatch
                    msg = "{}.{}.{}: no value".format(k, rtype, rname)
                    print("compare_settings: " + msg, file=sys.stderr)
                    return msg
                if rname == "memory":
                    r = memunits(rvalue) == memunits(dep_rvalue)
                elif rname == "cpu":
                    r = cpuunits(rvalue) == cpuunits(dep_rvalue)
                if not r:
                    msg = "compare_settings: {}.{}.{}: {}!={}".format(k, rtype, rname, rvalue, dep_rvalue)
                    print("compare_settings: " + msg, file=sys.stderr)
                    return msg
    return None


# FIXME: observed a patch trigger spontaneous reduction in replica count! (happened when update was attempted without replica count changes and 2nd replica was not schedulable according to k8s)
# NOTE: update of 'observedGeneration' does not mean that the 'deployment' object is done updating; also checking readyReplicas or availableReplicas in status does not help (these numbers may be for OLD replicas, if the new replicas cannot be started at all). We check for a 'Progressing' condition with a specific 'reason' code as an indication that the deployment is fully updated.
# The 'kubectl rollout status' command relies only on the deployment object - therefore info in it should be sufficient to track progress.
# ? do we need to use --to-revision with the undo command?
# FIXME: cpu request above 0.05 fails for 2 replicas on minikube. Not understood. (NOTE also that setting cpu_limit without specifying request causes request to be set to the same value, except if limit is very low - in that case, request isn't set at all)


def wait_for_update(appname, obj, patch_gen, print_progress, c=0, t=1, wait_for_progress=40, phase="", cmp_=None):
    """wait for a patch to take effect. appname is the namespace, obj is the deployment name, patch_gen is the object generation immediately after the patch was applied (should be a k8s obj with "kind":"Deployment")"""
    wait_for_gen = 15  # time to wait for object update ('observedGeneration')
    # wait_for_progress = 40 # time to wait for rollout to complete

    part = 1.0 / float(t)
    m = "updating {}".format(obj)

    dbg_log("waiting for update: deployment {}, generation {}".format(obj, patch_gen))

    # NOTE: best to implement this with a 'watch', not using an API poll!

    # ?watch=1 & resourceVersion = metadata[resourceVersion], timeoutSeconds=t,
    # --raw=''
    # GET /apis/apps/v1/namespaces/{namespace}/deployments

    t0 = time.time()
    w = Waiter(wait_for_gen, 2)
    r = None
    while w.wait():
        # NOTE: no progress prints here, this wait should be short
        r = k_get(appname, DEPLOYMENT + "/" + obj)
        # ydump("tst_wait{}_output_{}.yaml".format(rc,obj),r) ; rc = rc+1

        if test_dep_generation(r, patch_gen, ge=True):
            break
    r0 = r

    if r:
        print(
            "DEBUG: waited {}s for k8s object update, expected g = {}, g now = {}".format(
                time.time() - t0, patch_gen, r["status"]["observedGeneration"]
            ),
            file=sys.stderr,
        )
    if w.expired:
        raise AdjustError(
            "update of {} failed, timed out waiting for k8s object update".format(obj),
            status="failed",
            reason="adjust-failed",
        )

    dbg_log("waiting for progress: deployment {}, generation {}".format(obj, patch_gen))

    p = 0.0  #

    m = "waiting for progress from k8s {}".format(obj)

    w = Waiter(wait_for_progress, 2)
    c = float(c)
    err = "(wait skipped)"
    while w.wait():
        r = k_get(appname, DEPLOYMENT + "/" + obj)
        print_progress(int((c + p) * part * 100), m)
        p, err = test_dep_progress(r)
        if p == 1.0:
            if not test_dep_generation(r0, patch_gen) and cmp_:
                # if generation did not match exactly, there has been another update besides ours,
                # compare the configuration to the expected one and fail if a controlled setting was changed
                print(
                    "WARNING: detected concurrent update during adjust, re-checking settings",
                    file=sys.stderr,
                    flush=True,
                )
                diff = compare_settings(cmp_, r)
                if diff:
                    raise AdjustError("deployment was modified unexpectedly: " + diff, reason="overwritten")
            return  # all done
        if err:
            break

    # loop ended, timed out:
    status = "rejected"
    reason = "start-failed"
    err_text = "during {}; update of {} failed: timed out waiting for replicas to come up, status: {}".format(
        phase, obj, err
    )
    if "component(s) crash restart detected" in err:
        reason = "unstable"
        err_text = "during {}; {}".format(phase, err)
    raise AdjustError(err_text, status=status, reason=reason)


def set_rsrc(cont, cp, sn, sv, desc_settings):
    rn = RESOURCE_MAP[sn]
    sv_str = rsrc_str(sn, sv)

    sel = desc_settings.get("selector", "both")
    if sel == "request":
        cp.setdefault("resources", {}).setdefault("requests", {})[rn] = sv_str
        if cont.get("resources", {}).get("limits", {}).get(rn) is not None:
            cp["resources"].setdefault("limits", {})[rn] = None  # Remove corresponding limit if exists
    elif sel == "limit":
        cp.setdefault("resources", {}).setdefault("limits", {})[rn] = sv_str
        if cont.get("resources", {}).get("requests", {}).get(rn) is not None:
            cp["resources"].setdefault("requests", {})[rn] = None  # Remove corresponding request if exists
    elif sel == "request_min_limit":
        lv = max(desc_settings["limit_min"], sv)
        cp.setdefault("resources", {}).setdefault("requests", {})[rn] = sv_str
        cp.setdefault("resources", {}).setdefault("limits", {})[rn] = rsrc_str(sn, lv)
    else:  # both
        cp.setdefault("resources", {}).setdefault("requests", {})[rn] = sv_str
        cp.setdefault("resources", {}).setdefault("limits", {})[rn] = sv_str


def rsrc_str(setting_name, value):
    if setting_name == "mem":
        return str(round(value, 5)) + "Gi"  # internal memory representation is in GiB
    return str(round(value, 5))


def _value(x):
    if isinstance(x, dict) and "value" in x:
        return x["value"]
    return x


def add_meta(patch, key, data):
    """add json-encoded data to 'metadata[key]' in patch"""
    ma = patch.setdefault("metadata", {}).setdefault("annotations", {})
    ma[key] = json_enc(data)


def add_pod_meta(patch, key, data):
    """add json-encoded data to 'spec.template.metadata[key]' in patch"""
    ma = (
        patch.setdefault("spec", {}).setdefault("template", {}).setdefault("metadata", {}).setdefault("annotations", {})
    )
    ma[key] = json_enc(data)


def update(appname, desc, data, print_progress):

    adjust_on = desc.get("adjust_on", False)

    if adjust_on:
        try:
            # nosec: ast.literal_eval would not work here and the eval has been constrained
            should_adjust = eval(adjust_on, {"__builtins__": None}, {"data": data})  # nosec
        except Exception:
            should_adjust = False
        if not should_adjust:
            return {"status": "ok", "reason": "Skipped due to 'adjust_on' condition"}

    # NOTE: we'll need the raw k8s api data to see the container names (setting names for a single-container
    #       pod will include only the deployment(=pod) name, not the container name)
    _, raw, _ = raw_query(appname, desc)

    # convert k8s list of deployments into map
    raw = {dep["metadata"]["name"]: dep for dep in raw}

    patchlst = {}
    # FIXME: NB: app-wide settings not supported

    cfg = data.get("control", {})

    # FIXME: off-spec; step-down in data if a 'state' key is provided at the top.
    if "state" in data:
        data = data["state"]

    for comp_name, comp_data in data.get("application", {}).get("components", {}).items():
        settings = comp_data.get("settings", {})
        if not settings:
            continue
        # copy of the settings in short form, for annotation:
        settings_ann = {name: _value(value) for name, value in settings.items()}
        patches = {}
        replicas = None
        comp_desc = desc["application"]["components"].get(comp_name) or {}

        # find deployment name and container name, and verify its existence
        cont_name = None
        dep_name = comp_desc.get("deployment", comp_name)
        if "/" in dep_name:
            dep_name, cont_name = dep_name.split("/", 1)
        if dep_name not in raw:
            raise AdjustError(
                'Cannot find deployment with name "{}" for component "{}" in namespace "{}"'.format(
                    dep_name, comp_name, appname
                ),
                status="failed",
                reason="unknown",
            )  # FIXME 'reason' code (either bad config or invalid input to update())
        cont_name = (
            cont_name or raw[dep_name]["spec"]["template"]["spec"]["containers"][0]["name"]
        )  # chk for KeyError FIXME
        tgt_container = next(
            (c for c in raw[dep_name]["spec"]["template"]["spec"]["containers"] if c["name"] == cont_name), None
        )
        if tgt_container is None:
            raise AdjustError(
                'Could not find container with name "{}" in deployment "{}" '
                'for component "{}" in namespace "{}".'.format(cont_name, dep_name, comp_name, appname),
                status="failed",
                reason="unknown",
            )  # see note above

        cont_patch = patches.setdefault(cont_name, {})

        command = comp_desc.get("command")
        if command:
            if command.get("encoder"):
                cont_patch["command"], encoded_settings = encode_encoder(
                    settings, command["encoder"], expected_type=list
                )

                # Prevent encoded settings from further processing
                for setting in encoded_settings:
                    del settings[setting]

        env = comp_desc.get("env")
        if env:
            for en, ev in env.items():
                if ev.get("encoder"):
                    val, encoded_settings = encode_encoder(settings, ev["encoder"], expected_type=str)
                    patch_env = cont_patch.setdefault("env", [])
                    patch_env.append({"name": en, "value": val})

                    # Prevent encoded settings from further processing
                    for setting in encoded_settings:
                        del settings[setting]
                elif issetting(ev):
                    patch_env = cont_patch.setdefault("env", [])
                    patch_env.append({"name": en, "value": str(settings[en]["value"])})
                    del settings[en]

        # Settings and env vars
        for name, value in settings.items():
            value = _value(value)  # compatibility: allow a scalar, but also work with {"value": {anything}}
            cont_patch = patches.setdefault(cont_name, {})
            if name in ("mem", "cpu"):
                set_rsrc(tgt_container, cont_patch, name, value, comp_desc.get("settings", {}).get(name, {}))
                continue
            if name == "replicas":
                replicas = int(value)

        patch = patchlst.setdefault(dep_name, {})
        if patches:  # convert to array
            cp = (
                patch.setdefault("spec", {})
                .setdefault("template", {})
                .setdefault("spec", {})
                .setdefault("containers", [])
            )
            for n, v in patches.items():
                v["name"] = n
                cp.append(v)
        if replicas is not None:
            patch.setdefault("spec", {})["replicas"] = replicas
        if desc.get("force_restart", False):
            # restart is forced simply by adding an annotation with a value that doesn't repeat - it causes
            # the pod spec to be 'different', so the deployment will re-create the pod.
            # (this is the same method as used by 'kubectl rollout restart')
            ann = (
                patch.setdefault("spec", {})
                .setdefault("template", {})
                .setdefault("metadata", {})
                .setdefault("annotations", {})
            )
            ann[FORCED_RESTART_ANN] = datetime.datetime.utcnow().isoformat(timespec="seconds") + "Z"

    if not patchlst:
        raise Exception("No components were defiend in a configuration file. Cannot proceed with an adjustment.")

    # add update annotation, if configured
    ann_key = desc.get("update_annotation", None)
    if ann_key is not None:
        if len(patchlst) == 1:
            v = next(iter(patchlst.values()))
            ts = datetime.datetime.utcnow().isoformat(timespec="seconds") + "Z"
            add_pod_meta(v, ann_key, {"settings": settings_ann, "time": ts})
        else:
            print(
                "WARNING: no annotation added on update: 'update_annotation' is not compatible with multiple components",
                file=sys.stderr,
            )

    # NOTE: optimization possible: apply all patches first, then wait for them to complete (significant if making many changes at once!)

    # NOTE: it seems there's no way to update multiple resources with one 'patch' command
    #       (though -f accepts a directory, not sure how -f=dir works; maybe all listed resources
    #        get the *same* patch from the cmd line - not what we want)

    # execute patch commands
    patched_count = 0
    for n, v in patchlst.items():
        # ydump("tst_before_output_{}.yaml".format(n), k_get(appname, DEPLOYMENT + "/" + n))
        # run: kubectl patch deployment[.v1.apps] $n -p "{jsondata}"
        patchstr = json_enc(v)
        try:
            patch_r = k_patch(appname, DEPLOYMENT, n, patchstr)
        except Exception as e:  # TODO: limit to expected errors
            raise AdjustError(str(e), status="failed", reason="adjust-failed")
        p, _ = test_dep_progress(patch_r)
        if test_dep_generation(patch_r, patch_r["metadata"]["generation"]) and p == 1.0:
            # patch made no changes, skip wait_for_update:
            patched_count = patched_count + 1
            continue

        # ydump("tst_patch_output_{}.yaml".format(n), patch_r)

        # wait for update to complete (and print progress)
        # timeout default is set to be slightly higher than the default K8s timeout (so we let k8s detect progress stall first)
        try:
            wait_for_update(
                appname,
                n,
                patch_r["metadata"]["generation"],
                print_progress,
                patched_count,
                len(patchlst),
                cfg.get("timeout", 630),
                "rollout",
                cmp_=v,
            )
        except AdjustError as e:
            if e.reason not in ["start-failed", "unstable"]:  # not undo-able
                raise
            onfail = desc.get("on_fail", "rollback")  # valid values: nop, destroy, rollback (destroy == scale-to-zero)
            if onfail in ("rollback", "destroy_new"):
                try:
                    subprocess.run(
                        kubectl(appname, "rollout", "undo", DEPLOYMENT + "/" + n),
                        stdout=subprocess.PIPE,
                        stderr=subprocess.PIPE,
                        check=True,
                    )
                    print("UNDONE", file=sys.stderr)
                    dep_r = k_get(appname, DEPLOYMENT + "/" + n)  # Get deployment after rollback for latest generation
                    wait_for_update(
                        appname,
                        n,
                        dep_r["metadata"]["generation"],
                        print_progress,
                        patched_count,
                        len(patchlst),
                        cfg.get("timeout", 630),
                        "rollback",
                    )
                except subprocess.CalledProcessError as se:
                    # progress msg with warning TODO
                    print("undo for {} failed: {}".format(n, e), file=sys.stderr)
                    e.args = tuple([e.args[0] + ". Rollback failed: {}".format(se)]) + e.args[1:]
                except AdjustError as se:
                    e.args = tuple([e.args[0] + ". Rollback failed: {}".format(se)]) + e.args[1:]
                except Exception as se:
                    e.args = tuple([e.args[0] + ". Rollback failed: {}".format(se)]) + e.args[1:]
                    raise
                else:
                    e.args = tuple([e.args[0] + ". Rollback succeeded"]) + e.args[1:]
            if onfail == "destroy":
                try:
                    destroy_r = k_patch(appname, DEPLOYMENT, n, '{ "spec": { "replicas": 0 } }')
                    print("DESTROYED", file=sys.stderr)
                    wait_for_update(
                        appname,
                        n,
                        destroy_r["metadata"]["generation"],
                        print_progress,
                        patched_count,
                        len(patchlst),
                        cfg.get("timeout", 630),
                        "destroy",
                    )
                except subprocess.CalledProcessError as se:
                    # progress msg with warning TODO
                    print("destroy for {} failed: {}".format(n, e), file=sys.stderr)
                    e.args = tuple([e.args[0] + ". Destroy failed: {}".format(se)]) + e.args[1:]
                except AdjustError as se:
                    e.args = tuple([e.args[0] + ". Destroy failed: {}".format(se)]) + e.args[1:]
                except Exception as se:
                    e.args = tuple([e.args[0] + ". Destroy failed: {}".format(se)]) + e.args[1:]
                    raise
                else:
                    e.args = tuple([e.args[0] + ". Destroy succeeded"]) + e.args[1:]
            raise
        patched_count = patched_count + 1

    # spec_id and version_id should be tested without settlement_time, too - TODO

    # post-adjust settlement, if enabled
    testdata0, raw, _ = raw_query(appname, desc, pod_debug=True)
    settlement_time = cfg.get("settlement", desc.get("settlement", 0))
    refapp = cfg.get("userdata", {}).get("deployment", None)
    mon0 = testdata0["monitoring"]

    if "ref_version_id" in mon0 and mon0["version_id"] != mon0["ref_version_id"]:
        raise AdjustError(
            "application version does not match reference version", status="aborted", reason="version-mismatch"
        )

    # aborted status reasons that aren't supported: ref-app-inconsistent, ref-app-unavailable

    if not settlement_time:
        return {"monitoring": mon0, "status": "ok", "reason": "success"}

    # TODO: adjust progress accounting when there is settlement_time!=0
    # wait and watch the app, checking for changes
    w = Waiter(
        settlement_time, delay=min(settlement_time, 5)
    )  # NOTE: delay between tests may be made longer than the delay between progress reports
    m = "waiting for k8s settlement"
    try:
        while w.wait():
            print_progress(99, m)
            testdata, raw, restart_counts = raw_query(appname, desc, pod_debug=True)
            # check for container restart
            restarted_containers = {}
            for dep_name, restarts_list in restart_counts.items():
                cur_conts = {v["pod+container"]: v["restartCount"] for v in restarts_list if v["restartCount"] > 0}
                if cur_conts:
                    restarted_containers[dep_name] = cur_conts

            if restarted_containers:
                raise AdjustError(
                    "during settlement; component(s) crash restart detected. Restarted deployments {{pod+container: restartCount}}: {}".format(
                        ", ".join("{} {}".format(k, v) for k, v in restarted_containers.items())
                    ),
                    status="rejected",
                    reason="unstable",
                )

            mon = testdata["monitoring"]
            # compare to initial mon data set
            if mon["runtime_id"] != mon0["runtime_id"]:  # restart detected
                raise AdjustError(
                    "during settlement; component(s) intentional restart detected",
                    status="transient-failure",
                    reason="app-restart",
                )
            # TODO: what to do with version change?
            # if mon["version_id"] != mon0["version_id"]:
            #     raise AdjustError("application was modified unexpectedly during settlement", status="transient-failure", reason="app-update")
            if mon["spec_id"] != mon0["spec_id"]:
                raise AdjustError(
                    "application configuration was modified unexpectedly during settlement",
                    status="transient-failure",
                    reason="app-update",
                )

            if refapp:
                if mon["ref_spec_id"] != mon0["ref_spec_id"]:
                    raise AdjustError(
                        "reference application configuration was modified unexpectedly during settlement",
                        status="transient-failure",
                        reason="ref-app-update",
                    )
                if mon["ref_runtime_count"] != mon0["ref_runtime_count"]:
                    raise AdjustError(
                        "reference application replicas count changed unexpectedly during settlement",
                        status="transient-failure",
                        reason="ref-app-scale",
                    )

        # Final readiness check
        unready_dep_pods = {}
        for n in patchlst.keys():
            dep = k_get(appname, DEPLOYMENT + "/" + n)
            sel = dep["spec"]["selector"]["matchLabels"]
            sel = ",".join(("{}={}".format(k, v) for k, v in sel.items()))
            latest_rs = get_latest_rs(appname, sel, dep)
            pods = get_latest_pods(appname, sel, latest_rs)

            unready_pods = [
                p["metadata"]["name"]
                for p in pods
                if not all(cs["ready"] for cs in p["status"].get("containerStatuses", []))
            ]
            if unready_pods:
                unready_dep_pods[n] = unready_pods

        if unready_dep_pods:
            if any(
                cont_stat.get("state", {}).get("waiting", {}).get("reason") == "ImagePullBackOff"
                for p in pods
                for cont_stat in p["status"].get("containerStatuses", [])
            ):
                raise AdjustError("container image pull failure detected", status="failed", reason="image-pull-failed")
            raise AdjustError(
                "Deployment pods not ready after settlement expired: {}".format(unready_dep_pods),
                status="rejected",
                reason="unstable",
            )

    except AdjustError as e:
        onfail = desc.get("on_fail", "rollback")  # valid values: nop, destroy, rollback (destroy == scale-to-zero)
        if onfail == "rollback":
            try:
                for n in patchlst.keys():
                    subprocess.run(
                        kubectl(appname, "rollout", "undo", DEPLOYMENT + "/" + n),
                        stdout=subprocess.PIPE,
                        stderr=subprocess.PIPE,
                        check=True,
                    )
                    dep_r = k_get(appname, DEPLOYMENT + "/" + n)  # Get deployment after rollback for latest generation
                    wait_for_update(
                        appname,
                        n,
                        dep_r["metadata"]["generation"],
                        print_progress,
                        patched_count,
                        len(patchlst),
                        cfg.get("timeout", 630),
                        "settlement rollback",
                    )
                print("UNDONE", file=sys.stderr)
            except subprocess.CalledProcessError as se:
                # progress msg with warning TODO
                print("undo for {} failed: {}".format(n, se), file=sys.stderr)
                e.args = tuple([e.args[0] + ". Rollback failed: {}".format(se)]) + e.args[1:]
            except AdjustError as se:
                e.args = tuple([e.args[0] + ". Rollback failed: {}".format(se)]) + e.args[1:]
            except Exception as se:
                e.args = tuple([e.args[0] + ". Rollback failed: {}".format(se)]) + e.args[1:]
                raise
            else:
                e.args = tuple([e.args[0] + ". Rollback succeeded"]) + e.args[1:]
        if onfail == "destroy" or onfail == "destroy_new":
            try:
                for n in patchlst.keys():
                    destroy_r = k_patch(appname, DEPLOYMENT, n, '{ "spec": { "replicas": 0 } }')
                    wait_for_update(
                        appname,
                        n,
                        destroy_r["metadata"]["generation"],
                        print_progress,
                        patched_count,
                        len(patchlst),
                        cfg.get("timeout", 630),
                        "settlement destroy",
                    )
                print("DESTROYED", file=sys.stderr)
            except subprocess.CalledProcessError as se:
                # progress msg with warning TODO
                print("destroy for {} failed: {}".format(n, e), file=sys.stderr)
                e.args = tuple([e.args[0] + ". Destroy failed: {}".format(se)]) + e.args[1:]
            except AdjustError as se:
                e.args = tuple([e.args[0] + ". Destroy failed: {}".format(se)]) + e.args[1:]
            except Exception as se:
                e.args = tuple([e.args[0] + ". Destroy failed: {}".format(se)]) + e.args[1:]
                raise
            else:
                e.args = tuple([e.args[0] + ". Destroy succeeded"]) + e.args[1:]
        # if e.status != 'rejected':
        raise

    # update() return
    return {"monitoring": mon0, "status": "ok", "reason": "success"}


# cancel not supported
# def cancel(signum, frame):
#     if not wait_in_progress:
#         sys_exit(1)
#     print("aborting operation...", file=sys.stderr)


VERSION = "1.2"


class K8sAdjust(Adjust):
    def _progress(self, progress, message):
        """adapter for the default base class implementation of progress message"""
        self.progress = progress
        self.print_progress(message=message)

    def query(self):
        try:
            desc = read_desc()
        except ConfigError as e:
            raise AdjustError(
                str(e), reason="unknown"
            )  # maybe we should introduce reason=config (or even a different status class, instead of 'failed')
        for k in EXCLUDE_FROM_QUERY:
            desc.pop(k, None)
        namespace = os.environ.get("OPTUNE_NAMESPACE", desc.get("namespace", self.app_id))
        r = query(namespace, desc)
        return r

    def adjust(self, data):
        try:
            desc = read_desc()
        except ConfigError as e:
            raise AdjustError(
                str(e), reason="unknown"
            )  # maybe we should introduce reason=config (or even a different status class, instead of 'failed')
        # all other exceptions: default handler - stack trace and sys.exit(1)
        namespace = os.environ.get("OPTUNE_NAMESPACE", desc.get("namespace", self.app_id))
        r = update(namespace, desc, data, self._progress)
        return r


if __name__ == "__main__":
    # signal.signal(signal.SIGUSR1, cancel)
    # signal.signal(signal.SIGTERM, cancel)
    # signal.signal(signal.SIGINT, cancel)

    K8sAdjust(
        VERSION,
        "K8s driver for OCO servo.\n" "Note: set OPTUNE_USE_DEFAULT_NAMESPACE=1 environment var when embedded into app",
        supports_cancel=False,
        progress_interval=None,
    ).run()

# TODO: TBD: no support for multiple apps with different descriptors (desc file /app.yaml is not app-specific)