from ..exceptions import InvalidChoiceError from .base import MarathonResource, MarathonObject, assert_valid_path from .constraint import MarathonConstraint from .container import MarathonContainer from .deployment import MarathonDeployment from .task import MarathonTask from ..util import get_log from ..util import to_datetime log = get_log() class MarathonApp(MarathonResource): """Marathon Application resource. See: https://mesosphere.github.io/marathon/docs/rest-api.html#post-/v2/apps :param list[str] accepted_resource_roles: a list of resource roles (the resource offer must contain at least one of these for the app to be launched on that host) :param list[str] args: args form of the command to run :param int backoff_factor: multiplier for subsequent backoff :param int backoff_seconds: base time, in seconds, for exponential backoff :param str cmd: cmd form of the command to run :param constraints: placement constraints :type constraints: list[:class:`marathon.models.constraint.MarathonConstraint`] or list[tuple] :param container: container info :type container: :class:`marathon.models.container.MarathonContainer` or dict :param float cpus: cpus required per instance :param list[str] dependencies: services (app IDs) on which this app depends :param int disk: disk required per instance :param deployments: (read-only) currently running deployments that affect this app :type deployments: list[:class:`marathon.models.deployment.MarathonDeployment`] :param dict env: env vars :param str executor: executor :param int gpus: gpus required per instance :param health_checks: health checks :type health_checks: list[:class:`marathon.models.MarathonHealthCheck`] or list[dict] :param str id: app id :param str role: mesos role :param int instances: instances :param last_task_failure: last task failure :type last_task_failure: :class:`marathon.models.app.MarathonTaskFailure` or dict :param float mem: memory (in MB) required per instance :param dict secrets: A map with named secret declarations. :type port_definitions: list[:class:`marathon.models.app.PortDefinitions`] or list[dict] :param list[int] ports: ports :param bool require_ports: require the specified `ports` to be available in the resource offer :param list[str] store_urls: store URLs :param float task_rate_limit: (Removed in Marathon 0.7.0) maximum number of tasks launched per second :param tasks: (read-only) tasks :type tasks: list[:class:`marathon.models.task.MarathonTask`] :param int tasks_running: (read-only) the number of running tasks :param int tasks_staged: (read-only) the number of staged tasks :param int tasks_healthy: (read-only) the number of healthy tasks :param int tasks_unhealthy: (read-only) the number of unhealthy tasks :param upgrade_strategy: strategy by which app instances are replaced during a deployment :type upgrade_strategy: :class:`marathon.models.app.MarathonUpgradeStrategy` or dict :param list[str] uris: uris :param str user: user :param str version: version id :param version_info: time of last scaling, last config change :type version_info: :class:`marathon.models.app.MarathonAppVersionInfo` or dict :param task_stats: task statistics :type task_stats: :class:`marathon.models.app.MarathonTaskStats` or dict :param dict labels :type readiness_checks: list[:class:`marathon.models.app.ReadinessCheck`] or list[dict] :type residency: :class:`marathon.models.app.Residency` or dict :param int task_kill_grace_period_seconds: Configures the termination signal escalation behavior of executors when stopping tasks. :param list[dict] unreachable_strategy: Handling for unreachable instances. :param str kill_selection: Defines which instance should be killed first in case of e.g. rescaling. """ UPDATE_OK_ATTRIBUTES = [ 'args', 'backoff_factor', 'backoff_seconds', 'cmd', 'constraints', 'container', 'cpus', 'dependencies', 'disk', 'env', 'executor', 'gpus', 'health_checks', 'instances', 'kill_selection', 'labels', 'max_launch_delay_seconds', 'mem', 'ports', 'require_ports', 'store_urls', 'task_rate_limit', 'upgrade_strategy', 'unreachable_strategy', 'uris', 'user', 'version', 'role' ] """List of attributes which may be updated/changed after app creation""" CREATE_ONLY_ATTRIBUTES = ['id', 'accepted_resource_roles'] """List of attributes that should only be passed on creation""" READ_ONLY_ATTRIBUTES = [ 'deployments', 'tasks', 'tasks_running', 'tasks_staged', 'tasks_healthy', 'tasks_unhealthy'] """List of read-only attributes""" KILL_SELECTIONS = ["YOUNGEST_FIRST", "OLDEST_FIRST"] def __init__(self, accepted_resource_roles=None, args=None, backoff_factor=None, backoff_seconds=None, cmd=None, constraints=None, container=None, cpus=None, dependencies=None, deployments=None, disk=None, env=None, executor=None, health_checks=None, id=None, role=None, instances=None, kill_selection=None, labels=None, last_task_failure=None, max_launch_delay_seconds=None, mem=None, ports=None, require_ports=None, store_urls=None, task_rate_limit=None, tasks=None, tasks_running=None, tasks_staged=None, tasks_healthy=None, task_kill_grace_period_seconds=None, tasks_unhealthy=None, upgrade_strategy=None, unreachable_strategy=None, uris=None, user=None, version=None, version_info=None, ip_address=None, fetch=None, task_stats=None, readiness_checks=None, readiness_check_results=None, secrets=None, port_definitions=None, residency=None, gpus=None, networks=None): # self.args = args or [] self.accepted_resource_roles = accepted_resource_roles self.args = args # Marathon 0.7.0-RC1 throws a validation error if this is [] and cmd is passed: # "error": "AppDefinition must either contain a 'cmd' or a 'container'." self.backoff_factor = backoff_factor self.backoff_seconds = backoff_seconds self.cmd = cmd self.constraints = [ c if isinstance(c, MarathonConstraint) else MarathonConstraint(*c) for c in (constraints or []) ] self.container = container if (isinstance(container, MarathonContainer) or container is None) \ else MarathonContainer.from_json(container) self.cpus = cpus self.dependencies = dependencies or [] self.deployments = [ d if isinstance( d, MarathonDeployment) else MarathonDeployment().from_json(d) for d in (deployments or []) ] self.disk = disk self.env = env or dict() self.executor = executor self.gpus = gpus self.health_checks = health_checks or [] self.health_checks = [ hc if isinstance( hc, MarathonHealthCheck) else MarathonHealthCheck().from_json(hc) for hc in (health_checks or []) ] self.id = assert_valid_path(id) self.role = role self.instances = instances if kill_selection and kill_selection not in self.KILL_SELECTIONS: raise InvalidChoiceError( 'kill_selection', kill_selection, self.KILL_SELECTIONS) self.kill_selection = kill_selection self.labels = labels or {} self.last_task_failure = last_task_failure if (isinstance(last_task_failure, MarathonTaskFailure) or last_task_failure is None) \ else MarathonTaskFailure.from_json(last_task_failure) self.max_launch_delay_seconds = max_launch_delay_seconds self.mem = mem self.ports = ports or [] self.port_definitions = [ pd if isinstance( pd, PortDefinition) else PortDefinition.from_json(pd) for pd in (port_definitions or []) ] self.readiness_checks = [ rc if isinstance( rc, ReadinessCheck) else ReadinessCheck().from_json(rc) for rc in (readiness_checks or []) ] self.readiness_check_results = readiness_check_results or [] self.residency = residency self.require_ports = require_ports self.secrets = secrets or {} for k, s in self.secrets.items(): if not isinstance(s, Secret): self.secrets[k] = Secret().from_json(s) self.store_urls = store_urls or [] self.task_rate_limit = task_rate_limit self.tasks = [ t if isinstance(t, MarathonTask) else MarathonTask().from_json(t) for t in (tasks or []) ] self.tasks_running = tasks_running self.tasks_staged = tasks_staged self.tasks_healthy = tasks_healthy self.task_kill_grace_period_seconds = task_kill_grace_period_seconds self.tasks_unhealthy = tasks_unhealthy self.upgrade_strategy = upgrade_strategy if (isinstance(upgrade_strategy, MarathonUpgradeStrategy) or upgrade_strategy is None) \ else MarathonUpgradeStrategy.from_json(upgrade_strategy) self.unreachable_strategy = unreachable_strategy \ if (isinstance(unreachable_strategy, MarathonUnreachableStrategy) or unreachable_strategy is None) \ else MarathonUnreachableStrategy.from_json(unreachable_strategy) self.uris = uris or [] self.fetch = fetch or [] self.user = user self.version = version self.version_info = version_info if (isinstance(version_info, MarathonAppVersionInfo) or version_info is None) \ else MarathonAppVersionInfo.from_json(version_info) self.task_stats = task_stats if (isinstance(task_stats, MarathonTaskStats) or task_stats is None) \ else MarathonTaskStats.from_json(task_stats) self.networks = networks def add_env(self, key, value): self.env[key] = value class MarathonHealthCheck(MarathonObject): """Marathon health check. See https://mesosphere.github.io/marathon/docs/health-checks.html :param str command: health check command (if protocol == 'COMMAND') :param int grace_period_seconds: how long to ignore health check failures on initial task launch (before first healthy status) :param int interval_seconds: how long to wait between health checks :param int max_consecutive_failures: max number of consecutive failures before the task should be killed :param str path: health check target path (if protocol == 'HTTP') :param int port_index: target port as indexed in app's `ports` array :param str protocol: health check protocol ('HTTP', 'TCP', or 'COMMAND') :param int timeout_seconds: how long before a waiting health check is considered failed :param bool ignore_http1xx: Ignore HTTP informational status codes 100 to 199. :param dict kwargs: additional arguments for forward compatibility """ def __init__(self, command=None, grace_period_seconds=None, interval_seconds=None, max_consecutive_failures=None, path=None, port_index=None, protocol=None, timeout_seconds=None, ignore_http1xx=None, **kwargs): if command is None: self.command = None elif isinstance(command, str): self.command = { "value": command } elif type(command) is dict and 'value' in command: log.warn('Deprecated: Using command as dict instead of string is deprecated') self.command = { "value": command['value'] } else: raise ValueError(f'Invalid command format: {command}') self.grace_period_seconds = grace_period_seconds self.interval_seconds = interval_seconds self.max_consecutive_failures = max_consecutive_failures self.path = path self.port_index = port_index self.protocol = protocol self.timeout_seconds = timeout_seconds self.ignore_http1xx = ignore_http1xx # additional not previously known healthcheck attributes for k, v in kwargs.items(): setattr(self, k, v) class MarathonTaskFailure(MarathonObject): """Marathon Task Failure. :param str app_id: application id :param str host: mesos slave running the task :param str message: error message :param str task_id: task id :param str instance_id: instance id :param str state: task state :param timestamp: when this task failed :type timestamp: datetime or str :param str version: app version with which this task was started """ def __init__(self, app_id=None, host=None, message=None, task_id=None, instance_id=None, slave_id=None, state=None, timestamp=None, version=None): self.app_id = app_id self.host = host self.message = message self.task_id = task_id self.instance_id = instance_id self.slave_id = slave_id self.state = state self.timestamp = to_datetime(timestamp) self.version = version class MarathonUpgradeStrategy(MarathonObject): """Marathon health check. See https://mesosphere.github.io/marathon/docs/health-checks.html :param float minimum_health_capacity: minimum % of instances kept healthy on deploy """ def __init__(self, maximum_over_capacity=None, minimum_health_capacity=None): self.maximum_over_capacity = maximum_over_capacity self.minimum_health_capacity = minimum_health_capacity class MarathonUnreachableStrategy(MarathonObject): """Marathon unreachable Strategy. Define handling for unreachable instances. Given `unreachable_inactive_after_seconds = 60` and `unreachable_expunge_after = 120`, an instance will be expunged if it has been unreachable for more than 120 seconds or a second instance is started if it has been unreachable for more than 60 seconds.", See https://mesosphere.github.io/marathon/docs/? :param int unreachable_inactive_after_seconds: time an instance is unreachable for in seconds before marked as inactive. :param int unreachable_expunge_after_seconds: time an instance is unreachable for in seconds before expunged. :param int inactive_after_seconds :param int expunge_after_seconds """ DISABLED = 'disabled' def __init__(self, unreachable_inactive_after_seconds=None, unreachable_expunge_after_seconds=None, inactive_after_seconds=None, expunge_after_seconds=None): self.unreachable_inactive_after_seconds = unreachable_inactive_after_seconds self.unreachable_expunge_after_seconds = unreachable_expunge_after_seconds self.inactive_after_seconds = inactive_after_seconds self.expunge_after_seconds = expunge_after_seconds @classmethod def from_json(cls, attributes): if attributes == cls.DISABLED: return cls.DISABLED return super().from_json(attributes) class MarathonAppVersionInfo(MarathonObject): """Marathon App version info. See release notes for Marathon v0.11.0 https://github.com/mesosphere/marathon/releases/tag/v0.11.0 :param str app_id: application id :param str host: mesos slave running the task """ def __init__(self, last_scaling_at=None, last_config_change_at=None): self.last_scaling_at = to_datetime(last_scaling_at) self.last_config_change_at = to_datetime(last_config_change_at) class MarathonTaskStats(MarathonObject): """Marathon task statistics See https://mesosphere.github.io/marathon/docs/rest-api.html#taskstats-object-v0-11 :param started_after_last_scaling: contains statistics about all tasks that were started after the last scaling or restart operation. :type started_after_last_scaling: :class:`marathon.models.app.MarathonTaskStatsType` or dict :param with_latest_config: contains statistics about all tasks that run with the same config as the latest app version. :type with_latest_config: :class:`marathon.models.app.MarathonTaskStatsType` or dict :param with_outdated_config: contains statistics about all tasks that were started before the last config change which was not simply a restart or scaling operation. :type with_outdated_config: :class:`marathon.models.app.MarathonTaskStatsType` or dict :param total_summary: contains statistics about all tasks. :type total_summary: :class:`marathon.models.app.MarathonTaskStatsType` or dict """ def __init__(self, started_after_last_scaling=None, with_latest_config=None, with_outdated_config=None, total_summary=None): self.started_after_last_scaling = started_after_last_scaling if \ (isinstance(started_after_last_scaling, MarathonTaskStatsType) or started_after_last_scaling is None) \ else MarathonTaskStatsType.from_json(started_after_last_scaling) self.with_latest_config = with_latest_config if \ (isinstance(with_latest_config, MarathonTaskStatsType) or with_latest_config is None) \ else MarathonTaskStatsType.from_json(with_latest_config) self.with_outdated_config = with_outdated_config if \ (isinstance(with_outdated_config, MarathonTaskStatsType) or with_outdated_config is None) \ else MarathonTaskStatsType.from_json(with_outdated_config) self.total_summary = total_summary if \ (isinstance(total_summary, MarathonTaskStatsType) or total_summary is None) \ else MarathonTaskStatsType.from_json(total_summary) class MarathonTaskStatsType(MarathonObject): """Marathon app task stats :param stats: stast about app tasks :type stats: :class:`marathon.models.app.MarathonTaskStatsStats` or dict """ def __init__(self, stats=None): self.stats = stats if (isinstance(stats, MarathonTaskStatsStats) or stats is None)\ else MarathonTaskStatsStats.from_json(stats) class MarathonTaskStatsStats(MarathonObject): """Marathon app task stats :param counts: app task count breakdown :type counts: :class:`marathon.models.app.MarathonTaskStatsCounts` or dict :param life_time: app task life time stats :type life_time: :class:`marathon.models.app.MarathonTaskStatsLifeTime` or dict """ def __init__(self, counts=None, life_time=None): self.counts = counts if (isinstance(counts, MarathonTaskStatsCounts) or counts is None)\ else MarathonTaskStatsCounts.from_json(counts) self.life_time = life_time if (isinstance(life_time, MarathonTaskStatsLifeTime) or life_time is None)\ else MarathonTaskStatsLifeTime.from_json(life_time) class MarathonTaskStatsCounts(MarathonObject): """Marathon app task counts Equivalent to tasksStaged, tasksRunning, tasksHealthy, tasksUnhealthy. :param int staged: Staged task count :param int running: Running task count :param int healthy: Healthy task count :param int unhealthy: unhealthy task count """ def __init__(self, staged=None, running=None, healthy=None, unhealthy=None): self.staged = staged self.running = running self.healthy = healthy self.unhealthy = unhealthy class MarathonTaskStatsLifeTime(MarathonObject): """Marathon app life time statistics Measured from `"startedAt"` (timestamp of the Mesos TASK_RUNNING status update) of each running task until now :param float average_seconds: Average seconds :param float median_seconds: Median seconds """ def __init__(self, average_seconds=None, median_seconds=None): self.average_seconds = average_seconds self.median_seconds = median_seconds class ReadinessCheck(MarathonObject): """Marathon readiness check: https://mesosphere.github.io/marathon/docs/readiness-checks.html :param string name (Optional. Default: "readinessCheck"): The name used to identify this readiness check. :param string protocol (Optional. Default: "HTTP"): Protocol of the requests to be performed. Either HTTP or HTTPS. :param string path (Optional. Default: "/"): Path to the endpoint the task exposes to provide readiness status. Example: /path/to/readiness. :param string port_name (Optional. Default: "http-api"): Name of the port to query as described in the portDefinitions. Example: http-api. :param int interval_seconds (Optional. Default: 30 seconds): Number of seconds to wait between readiness checks. :param int timeout_seconds (Optional. Default: 10 seconds): Number of seconds after which a readiness check times out, regardless of the response. This value must be smaller than interval_seconds. :param list http_status_codes_for_ready (Optional. Default: [200]): The HTTP/HTTPS status code to treat as ready. :param bool preserve_last_response (Optional. Default: false): If true, the last readiness check response will be preserved and exposed in the API as part of a deployment. """ def __init__(self, name=None, protocol=None, path=None, port_name=None, interval_seconds=None, http_status_codes_for_ready=None, preserve_last_response=None, timeout_seconds=None): self.name = name self.protocol = protocol self.path = path self.port_name = port_name self.interval_seconds = interval_seconds self.http_status_codes_for_ready = http_status_codes_for_ready self.preserve_last_response = preserve_last_response self.timeout_seconds = timeout_seconds class PortDefinition(MarathonObject): """Marathon port definitions: https://mesosphere.github.io/marathon/docs/ports.html :param int port: The port :param string protocol: tcp or udp :param string name: (optional) the name of the port :param dict labels: undocumented """ def __init__(self, port=None, protocol=None, name=None, labels=None): self.port = port self.protocol = protocol self.name = name self.labels = labels class Residency(MarathonObject): """Declares how "resident" an app is: https://mesosphere.github.io/marathon/docs/persistent-volumes.html :param int relaunch_escalation_timeout_seconds: How long marathon will try to relaunch where the volumes is, defaults to 3600 :param string task_lost_behavior: What to do after a TASK_LOST. See the official Marathon docs for options """ def __init__(self, relaunch_escalation_timeout_seconds=None, task_lost_behavior=None): self.relaunch_escalation_timeout_seconds = relaunch_escalation_timeout_seconds self.task_lost_behavior = task_lost_behavior class Secret(MarathonObject): """Declares marathon secret object. :param str source: The source of the secret's value. The format depends on the secret store used by Mesos. """ def __init__(self, source=None): self.source = source