python-dcos 0.2.0-2 / usr / lib / python2.7 / dist-packages / dcos / marathon.py

import json
from distutils.version import LooseVersion

from dcos import http, util
from dcos.errors import DCOSException, DCOSHTTPException

from six.moves import urllib

logger = util.get_logger(__name__)


def create_client(config=None):
    """Creates a Marathon client with the supplied configuration.

    :param config: configuration dictionary
    :type config: config.Toml
    :returns: Marathon client
    :rtype: dcos.marathon.Client
    """

    if config is None:
        config = util.get_config()

    marathon_url = _get_marathon_url(config)
    timeout = config.get('core.timeout', http.DEFAULT_TIMEOUT)

    logger.info('Creating marathon client with: %r', marathon_url)
    return Client(marathon_url, timeout=timeout)


def _get_marathon_url(config):
    """
    :param config: configuration dictionary
    :type config: config.Toml
    :returns: marathon base url
    :rtype: str
    """

    marathon_url = config.get('marathon.url')
    if marathon_url is None:
        dcos_url = util.get_config_vals(['core.dcos_url'], config)[0]
        marathon_url = urllib.parse.urljoin(dcos_url, 'marathon/')

    return marathon_url


def _to_exception(response):
    """
    :param response: HTTP response object or Exception
    :type response: requests.Response | Exception
    :returns: An exception with the message from the response JSON
    :rtype: Exception
    """

    if response.status_code == 400:
        msg = 'Error on request [{0} {1}]: HTTP {2}: {3}'.format(
            response.request.method,
            response.request.url,
            response.status_code,
            response.reason)

        # Marathon is buggy and sometimes return JSON, and sometimes
        # HTML.  We only include the error message if it's JSON.
        try:
            json_msg = response.json()
            msg += ':\n' + json.dumps(json_msg,
                                      indent=2,
                                      sort_keys=True,
                                      separators=(',', ': '))
        except ValueError:
            pass

        return DCOSException(msg)
    elif response.status_code == 409:
        return DCOSException(
            'App or group is locked by one or more deployments. '
            'Override with --force.')

    try:
        response_json = response.json()
    except Exception:
        logger.exception(
            'Unable to decode response body as a JSON value: %r',
            response)

        return DCOSException(
            'Error decoding response from [{0}]: HTTP {1}: {2}'.format(
                response.request.url, response.status_code, response.reason))
    message = response_json.get('message')
    if message is None:
        errs = response_json.get('errors')
        if errs is None:
            logger.error(
                'Marathon server did not return a message: %s',
                response_json)
            return DCOSException(_default_marathon_error())

        msg = '\n'.join(error['error'] for error in errs)
        return DCOSException(_default_marathon_error(msg))

    return DCOSException('Error: {}'.format(message))


def _http_req(fn, *args, **kwargs):
    """Make an HTTP request, and raise a marathon-specific exception for
    HTTP error codes.

    :param fn: function to call
    :type fn: function
    :param args: args to pass to `fn`
    :type args: [object]
    :param kwargs: kwargs to pass to `fn`
    :type kwargs: dict
    :returns: `fn` return value
    :rtype: object

    """
    try:
        return fn(*args, **kwargs)
    except DCOSHTTPException as e:
        raise _to_exception(e.response)


class Client(object):
    """Class for talking to the Marathon server.

    :param marathon_url: the base URL for the Marathon server
    :type marathon_url: str
    """

    def __init__(self, marathon_url, timeout=http.DEFAULT_TIMEOUT):
        self._base_url = marathon_url
        self._timeout = timeout

        min_version = "0.8.1"
        version = LooseVersion(self.get_about()["version"])
        self._version = version
        if version < LooseVersion(min_version):
            msg = ("The configured Marathon with version {0} is outdated. " +
                   "Please use version {1} or later.").format(
                       version,
                       min_version)
            raise DCOSException(msg)

    def _create_url(self, path):
        """Creates the url from the provided path.
        :param path: url path
        :type path: str
        :returns: constructed url
        :rtype: str
        """

        return urllib.parse.urljoin(self._base_url, path)

    def get_version(self):
        """Get marathon version
        :returns: marathon version
        rtype: LooseVersion
        """

        return self._version

    def get_about(self):
        """Returns info about Marathon instance

        :returns Marathon information
        :rtype: dict
        """

        url = self._create_url('v2/info')
        response = _http_req(http.get, url, timeout=self._timeout)

        return response.json()

    def get_app(self, app_id, version=None):
        """Returns a representation of the requested application version. If
        version is None the return the latest version.

        :param app_id: the ID of the application
        :type app_id: str
        :param version: application version as a ISO8601 datetime
        :type version: str
        :returns: the requested Marathon application
        :rtype: dict
        """

        app_id = self.normalize_app_id(app_id)
        if version is None:
            url = self._create_url('v2/apps{}'.format(app_id))
        else:
            url = self._create_url(
                'v2/apps{}/versions/{}'.format(app_id, version))

        response = _http_req(http.get, url, timeout=self._timeout)

        # Looks like Marathon return different JSON for versions
        if version is None:
            return response.json()['app']
        else:
            return response.json()

    def get_groups(self):
        """Get a list of known groups.

        :returns: list of known groups
        :rtype: list of dict
        """

        url = self._create_url('v2/groups')
        response = _http_req(http.get, url, timeout=self._timeout)
        return response.json()['groups']

    def get_group(self, group_id, version=None):
        """Returns a representation of the requested group version. If
        version is None the return the latest version.

        :param group_id: the ID of the application
        :type group_id: str
        :param version: application version as a ISO8601 datetime
        :type version: str
        :returns: the requested Marathon application
        :rtype: dict
        """

        group_id = self.normalize_app_id(group_id)
        if version is None:
            url = self._create_url('v2/groups{}'.format(group_id))
        else:
            url = self._create_url(
                'v2/groups{}/versions/{}'.format(group_id, version))

        response = _http_req(http.get, url, timeout=self._timeout)
        return response.json()

    def get_app_versions(self, app_id, max_count=None):
        """Asks Marathon for all the versions of the Application up to a
        maximum count.

        :param app_id: the ID of the application or group
        :type app_id: str
        :param id_type: type of the id ("apps" or "groups")
        :type app_id: str
        :param max_count: the maximum number of version to fetch
        :type max_count: int
        :returns: a list of all the version of the application
        :rtype: [str]
        """

        if max_count is not None and max_count <= 0:
            raise DCOSException(
                'Maximum count must be a positive number: {}'.format(max_count)
            )

        app_id = self.normalize_app_id(app_id)

        url = self._create_url('v2/apps{}/versions'.format(app_id))

        response = _http_req(http.get, url, timeout=self._timeout)

        if max_count is None:
            return response.json()['versions']
        else:
            return response.json()['versions'][:max_count]

    def get_apps(self):
        """Get a list of known applications.

        :returns: list of known applications
        :rtype: [dict]
        """

        url = self._create_url('v2/apps')
        response = _http_req(http.get, url, timeout=self._timeout)
        return response.json()['apps']

    def add_app(self, app_resource):
        """Add a new application.

        :param app_resource: application resource
        :type app_resource: dict, bytes or file
        :returns: the application description
        :rtype: dict
        """

        url = self._create_url('v2/apps')

        # The file type exists only in Python 2, preventing type(...) is file.
        if hasattr(app_resource, 'read'):
            app_json = json.load(app_resource)
        else:
            app_json = app_resource

        response = _http_req(http.post, url,
                             json=app_json,
                             timeout=self._timeout)

        return response.json()

    def _update(self, resource_id, payload, force=None, url_endpoint="apps"):
        """Update an application or group.

        :param resource_id: the app or group id
        :type resource_id: str
        :param payload: the json payload
        :type payload: dict
        :param force: whether to override running deployments
        :type force: bool
        :param url_endpoint: resource type to update ("apps" or "groups")
        :type url_endpoint: str
        :returns: the resulting deployment ID
        :rtype: str
        """

        resource_id = self.normalize_app_id(resource_id)

        if not force:
            params = None
        else:
            params = {'force': 'true'}

        url = self._create_url('v2/{}{}'.format(url_endpoint, resource_id))

        response = _http_req(http.put, url,
                             params=params,
                             json=payload,
                             timeout=self._timeout)

        return response.json().get('deploymentId')

    def update_app(self, app_id, payload, force=None):
        """Update an application.

        :param app_id: the application id
        :type app_id: str
        :param payload: the json payload
        :type payload: dict
        :param force: whether to override running deployments
        :type force: bool
        :returns: the resulting deployment ID
        :rtype: str
        """

        return self._update(app_id, payload, force)

    def update_group(self, group_id, payload, force=None):
        """Update a group.

        :param group_id: the group id
        :type group_id: str
        :param payload: the json payload
        :type payload: dict
        :param force: whether to override running deployments
        :type force: bool
        :returns: the resulting deployment ID
        :rtype: str
        """

        return self._update(group_id, payload, force, "groups")

    def scale_app(self, app_id, instances, force=None):
        """Scales an application to the requested number of instances.

        :param app_id: the ID of the application to scale
        :type app_id: str
        :param instances: the requested number of instances
        :type instances: int
        :param force: whether to override running deployments
        :type force: bool
        :returns: the resulting deployment ID
        :rtype: str
        """

        app_id = self.normalize_app_id(app_id)

        if not force:
            params = None
        else:
            params = {'force': 'true'}

        url = self._create_url('v2/apps{}'.format(app_id))

        response = _http_req(http.put,
                             url,
                             params=params,
                             json={'instances': int(instances)},
                             timeout=self._timeout)

        deployment = response.json()['deploymentId']
        return deployment

    def scale_group(self, group_id, scale_factor, force=None):
        """Scales a group with the requested scale-factor.
        :param group_id: the ID of the group to scale
        :type group_id: str
        :param scale_factor: the requested value of scale-factor
        :type scale_factor: float
        :param force: whether to override running deployments
        :type force: bool
        :returns: the resulting deployment ID
        :rtype: bool
        """

        group_id = self.normalize_app_id(group_id)

        if not force:
            params = None
        else:
            params = {'force': 'true'}

        url = self._create_url('v2/groups{}'.format(group_id))

        response = http.put(url,
                            params=params,
                            json={'scaleBy': scale_factor},
                            timeout=self._timeout)

        deployment = response.json()['deploymentId']
        return deployment

    def stop_app(self, app_id, force=None):
        """Scales an application to zero instances.

        :param app_id: the ID of the application to stop
        :type app_id: str
        :param force: whether to override running deployments
        :type force: bool
        :returns: the resulting deployment ID
        :rtype: bool
        """

        return self.scale_app(app_id, 0, force)

    def remove_app(self, app_id, force=None):
        """Completely removes the requested application.

        :param app_id: the ID of the application to remove
        :type app_id: str
        :param force: whether to override running deployments
        :type force: bool
        :rtype: None
        """

        app_id = self.normalize_app_id(app_id)

        if not force:
            params = None
        else:
            params = {'force': 'true'}

        url = self._create_url('v2/apps{}'.format(app_id))
        _http_req(http.delete, url, params=params, timeout=self._timeout)

    def remove_group(self, group_id, force=None):
        """Completely removes the requested application.

        :param group_id: the ID of the application to remove
        :type group_id: str
        :param force: whether to override running deployments
        :type force: bool
        :rtype: None
        """

        group_id = self.normalize_app_id(group_id)

        if not force:
            params = None
        else:
            params = {'force': 'true'}

        url = self._create_url('v2/groups{}'.format(group_id))

        _http_req(http.delete, url, params=params, timeout=self._timeout)

    def restart_app(self, app_id, force=None):
        """Performs a rolling restart of all of the tasks.

        :param app_id: the id of the application to restart
        :type app_id: str
        :param force: whether to override running deployments
        :type force: bool
        :returns: the deployment id and version
        :rtype: dict
        """

        app_id = self.normalize_app_id(app_id)

        if not force:
            params = None
        else:
            params = {'force': 'true'}

        url = self._create_url('v2/apps{}/restart'.format(app_id))

        response = _http_req(http.post, url,
                             params=params,
                             timeout=self._timeout)
        return response.json()

    def get_deployment(self, deployment_id):
        """Returns a deployment.

        :param deployment_id: the deployment id
        :type deployment_id: str
        :returns: a deployment
        :rtype: dict
        """

        url = self._create_url('v2/deployments')

        response = _http_req(http.get, url, timeout=self._timeout)
        deployment = next(
            (deployment for deployment in response.json()
             if deployment_id == deployment['id']),
            None)

        return deployment

    def get_deployments(self, app_id=None):
        """Returns a list of deployments, optionally limited to an app.

        :param app_id: the id of the application
        :type app_id: str
        :returns: a list of deployments
        :rtype: list of dict
        """

        url = self._create_url('v2/deployments')

        response = _http_req(http.get, url, timeout=self._timeout)

        if app_id is not None:
            app_id = self.normalize_app_id(app_id)
            deployments = [
                deployment for deployment in response.json()
                if app_id in deployment['affectedApps']
            ]
        else:
            deployments = response.json()

        return deployments

    def _cancel_deployment(self, deployment_id, force):
        """Cancels an application deployment.

        :param deployment_id: the deployment id
        :type deployment_id: str
        :param force: if set to `False`, stop the deployment and
                      create a new rollback deployment to reinstate the
                      previous configuration. If set to `True`, simply stop the
                      deployment.
        :type force: bool
        :returns: cancelation deployment
        :rtype: dict
        """

        if not force:
            params = None
        else:
            params = {'force': 'true'}

        url = self._create_url('v2/deployments/{}'.format(deployment_id))

        response = _http_req(http.delete, url,
                             params=params,
                             timeout=self._timeout)

        if force:
            return None
        else:
            return response.json()

    def rollback_deployment(self, deployment_id):
        """Rolls back an application deployment.

        :param deployment_id: the deployment id
        :type deployment_id: str
        :returns: cancelation deployment
        :rtype: dict
        """

        return self._cancel_deployment(deployment_id, False)

    def stop_deployment(self, deployment_id):
        """Stops an application deployment.

        :param deployment_id: the deployment id
        :type deployment_id: str
        :rtype: None
        """

        self._cancel_deployment(deployment_id, True)

    def get_tasks(self, app_id):
        """Returns a list of tasks, optionally limited to an app.

        :param app_id: the id of the application to restart
        :type app_id: str
        :returns: a list of tasks
        :rtype: [dict]
        """

        url = self._create_url('v2/tasks')

        response = _http_req(http.get, url, timeout=self._timeout)

        if app_id is not None:
            app_id = self.normalize_app_id(app_id)
            tasks = [
                task for task in response.json()['tasks']
                if app_id == task['appId']
            ]
        else:
            tasks = response.json()['tasks']

        return tasks

    def get_task(self, task_id):
        """Returns a task

        :param task_id: the id of the task
        :type task_id: str
        :returns: a tasks
        :rtype: dict
        """

        url = self._create_url('v2/tasks')

        response = _http_req(http.get, url, timeout=self._timeout)

        task = next(
            (task for task in response.json()['tasks']
             if task_id == task['id']),
            None)

        return task

    def get_app_schema(self):
        """Returns app json schema

        :returns: application json schema
        :rtype: json schema or None if endpoint doesn't exist
        """

        version = self.get_version()
        schema_version = LooseVersion("0.9.0")
        if version < schema_version:
            return None

        url = self._create_url('v2/schemas/app')
        response = _http_req(http.get, url, timeout=self._timeout)

        return response.json()

    def normalize_app_id(self, app_id):
        """Normalizes the application id.

        :param app_id: raw application ID
        :type app_id: str
        :returns: normalized application ID
        :rtype: str
        """

        return urllib.parse.quote('/' + app_id.strip('/'))

    def create_group(self, group_resource):
        """Add a new group.

        :param group_resource: grouplication resource
        :type group_resource: dict, bytes or file
        :returns: the group description
        :rtype: dict
        """
        url = self._create_url('v2/groups')

        # The file type exists only in Python 2, preventing type(...) is file.
        if hasattr(group_resource, 'read'):
            group_json = json.load(group_resource)
        else:
            group_json = group_resource

        response = _http_req(http.post, url,
                             json=group_json,
                             timeout=self._timeout)
        return response.json()

    def get_leader(self):
        """ Get the leading marathon instance.

        :returns: string of the form <ip>:<port>
        :rtype: str
        """

        url = self._create_url('v2/leader')
        response = _http_req(http.get, url, timeout=self._timeout)
        return response.json()['leader']


def _default_marathon_error(message=""):
    """
    :param message: additional message
    :type message: str
    :returns: marathon specific error message
    :rtype: str
    """

    return ("Marathon likely misconfigured. Please check your proxy or "
            "Marathon URL settings. See dcos config --help. {}").format(
                message)