deb-cinder/cinder/volume/drivers/datera.py

# Copyright 2016 Datera
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import json

from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import excutils
from oslo_utils import units
import requests
import six

from cinder import context
from cinder import exception
from cinder.i18n import _, _LE, _LI
from cinder import utils
from cinder.volume.drivers.san import san
from cinder.volume import qos_specs
from cinder.volume import volume_types

LOG = logging.getLogger(__name__)

d_opts = [
    cfg.StrOpt('datera_api_port',
               default='7717',
               help='Datera API port.'),
    cfg.StrOpt('datera_api_version',
               default='2',
               help='Datera API version.'),
    cfg.StrOpt('datera_num_replicas',
               default='1',
               help='Number of replicas to create of an inode.')
]


CONF = cfg.CONF
CONF.import_opt('driver_use_ssl', 'cinder.volume.driver')
CONF.register_opts(d_opts)

DEFAULT_STORAGE_NAME = 'storage-1'
DEFAULT_VOLUME_NAME = 'volume-1'


def _authenticated(func):
    """Ensure the driver is authenticated to make a request.

    In do_setup() we fetch an auth token and store it. If that expires when
    we do API request, we'll fetch a new one.
    """

    def func_wrapper(self, *args, **kwargs):
        try:
            return func(self, *args, **kwargs)
        except exception.NotAuthorized:
            # Prevent recursion loop. After the self arg is the
            # resource_type arg from _issue_api_request(). If attempt to
            # login failed, we should just give up.
            if args[0] == 'login':
                raise

            # Token might've expired, get a new one, try again.
            self._login()
            return func(self, *args, **kwargs)
    return func_wrapper


class DateraDriver(san.SanISCSIDriver):

    """The OpenStack Datera Driver

    Version history:
        1.0 - Initial driver
        1.1 - Look for lun-0 instead of lun-1.
        2.0 - Update For Datera API v2
    """
    VERSION = '2.0'

    def __init__(self, *args, **kwargs):
        super(DateraDriver, self).__init__(*args, **kwargs)
        self.configuration.append_config_values(d_opts)
        self.num_replicas = self.configuration.datera_num_replicas
        self.username = self.configuration.san_login
        self.password = self.configuration.san_password
        self.auth_token = None
        self.cluster_stats = {}
        self.datera_api_token = None

    def _login(self):
        """Use the san_login and san_password to set self.auth_token."""
        body = {
            'name': self.username,
            'password': self.password
        }

        # Unset token now, otherwise potential expired token will be sent
        # along to be used for authorization when trying to login.
        self.auth_token = None

        try:
            LOG.debug('Getting Datera auth token.')
            results = self._issue_api_request('login', 'put', body=body,
                                              sensitive=True)
            self.datera_api_token = results['key']
        except exception.NotAuthorized:
            with excutils.save_and_reraise_exception():
                LOG.error(_LE('Logging into the Datera cluster failed. Please '
                              'check your username and password set in the '
                              'cinder.conf and start the cinder-volume '
                              'service again.'))

    def _get_lunid(self):
        return 0

    def do_setup(self, context):
        # If we can't authenticate through the old and new method, just fail
        # now.
        if not all([self.username, self.password]):
            msg = _("san_login and/or san_password is not set for Datera "
                    "driver in the cinder.conf. Set this information and "
                    "start the cinder-volume service again.")
            LOG.error(msg)
            raise exception.InvalidInput(msg)

        self._login()

    @utils.retry(exception.VolumeDriverException, retries=3)
    def _wait_for_resource(self, id, resource_type):
        result = self._issue_api_request(resource_type, 'get', id)
        if result['storage_instances'][DEFAULT_STORAGE_NAME]['volumes'][
                DEFAULT_VOLUME_NAME]['op_state'] == 'available':
            return
        else:
            raise exception.VolumeDriverException(
                message=_('Resource not ready.'))

    def _create_resource(self, resource, resource_type, body):
        type_id = resource.get('volume_type_id', None)

        result = None
        try:
            result = self._issue_api_request(resource_type, 'post', body=body)
        except exception.Invalid:
            if resource_type == 'volumes' and type_id:
                LOG.error(_LE("Creation request failed. Please verify the "
                              "extra-specs set for your volume types are "
                              "entered correctly."))
            raise
        else:
            # Handle updating QOS Policies
            if resource_type == 'app_instances':
                url = ('app_instances/{}/storage_instances/{}/volumes/{'
                       '}/performance_policy')
                url = url.format(
                    resource['id'],
                    DEFAULT_STORAGE_NAME,
                    DEFAULT_VOLUME_NAME)
                if type_id is not None:
                    policies = self._get_policies_by_volume_type(type_id)
                    if policies:
                        self._issue_api_request(url, 'post', body=policies)
            if result['storage_instances'][DEFAULT_STORAGE_NAME]['volumes'][
                    DEFAULT_VOLUME_NAME]['op_state'] == 'available':
                return
            self._wait_for_resource(resource['id'], resource_type)

    def create_volume(self, volume):
        """Create a logical volume."""
        # Generate App Instance, Storage Instance and Volume
        # Volume ID will be used as the App Instance Name
        # Storage Instance and Volumes will have standard names
        app_params = (
            {
                'create_mode': "openstack",
                'uuid': str(volume['id']),
                'name': str(volume['id']),
                'access_control_mode': 'allow_all',
                'storage_instances': {
                    DEFAULT_STORAGE_NAME: {
                        'name': DEFAULT_STORAGE_NAME,
                        'volumes': {
                            DEFAULT_VOLUME_NAME: {
                                'name': DEFAULT_VOLUME_NAME,
                                'size': volume['size'],
                                'replica_count': int(self.num_replicas),
                                'snapshot_policies': {
                                }
                            }
                        }
                    }
                }
            })
        self._create_resource(volume, 'app_instances', body=app_params)

    def extend_volume(self, volume, new_size):
        # Offline App Instance, if necessary
        reonline = False
        app_inst = self._issue_api_request(
            "app_instances/{}".format(volume['id']))
        if app_inst['admin_state'] == 'online':
            reonline = True
            self.detach_volume(None, volume)
        # Change Volume Size
        app_inst = volume['id']
        storage_inst = DEFAULT_STORAGE_NAME
        data = {
            'size': new_size
        }
        self._issue_api_request(
            'app_instances/{}/storage_instances/{}/volumes/{}'.format(
                app_inst, storage_inst, DEFAULT_VOLUME_NAME),
            method='put', body=data)
        # Online Volume, if it was online before
        if reonline:
            self.create_export(None, volume)

    def create_cloned_volume(self, volume, src_vref):
        clone_src_template = ("/app_instances/{}/storage_instances/{"
                              "}/volumes/{}")
        src = clone_src_template.format(src_vref['id'], DEFAULT_STORAGE_NAME,
                                        DEFAULT_VOLUME_NAME)
        data = {
            'create_mode': 'openstack',
            'name': str(volume['id']),
            'uuid': str(volume['id']),
            'clone_src': src,
            'access_control_mode': 'allow_all'
        }
        self._issue_api_request('app_instances', 'post', body=data)

        if volume['size'] > src_vref['size']:
            self.extend_volume(volume, volume['size'])

    def delete_volume(self, volume):
        self.detach_volume(None, volume)
        app_inst = volume['id']
        try:
            self._issue_api_request('app_instances/{}'.format(app_inst),
                                    method='delete')
        except exception.NotFound:
            msg = _LI("Tried to delete volume %s, but it was not found in the "
                      "Datera cluster. Continuing with delete.")
            LOG.info(msg, volume['id'])

    def ensure_export(self, context, volume, connector):
        """Gets the associated account, retrieves CHAP info and updates."""
        return self.create_export(context, volume, connector)

    def create_export(self, context, volume, connector):
        url = "app_instances/{}".format(volume['id'])
        data = {
            'admin_state': 'online'
        }
        app_inst = self._issue_api_request(url, method='put', body=data)
        storage_instance = app_inst['storage_instances'][
            DEFAULT_STORAGE_NAME]

        portal = storage_instance['access']['ips'][0] + ':3260'
        iqn = storage_instance['access']['iqn']

        # Portal, IQN, LUNID
        provider_location = '%s %s %s' % (portal, iqn, self._get_lunid())
        return {'provider_location': provider_location}

    def detach_volume(self, context, volume, attachment=None):
        url = "app_instances/{}".format(volume['id'])
        data = {
            'admin_state': 'offline',
            'force': True
        }
        try:
            self._issue_api_request(url, method='put', body=data)
        except exception.NotFound:
            msg = _LI("Tried to detach volume %s, but it was not found in the "
                      "Datera cluster. Continuing with detach.")
            LOG.info(msg, volume['id'])

    def create_snapshot(self, snapshot):
        url_template = ('app_instances/{}/storage_instances/{}/volumes/{'
                        '}/snapshots')
        url = url_template.format(snapshot['volume_id'],
                                  DEFAULT_STORAGE_NAME,
                                  DEFAULT_VOLUME_NAME)

        snap_params = {
            'uuid': snapshot['id'],
        }
        self._issue_api_request(url, method='post', body=snap_params)

    def delete_snapshot(self, snapshot):
        snap_temp = ('app_instances/{}/storage_instances/{}/volumes/{'
                     '}/snapshots')
        snapu = snap_temp.format(snapshot['volume_id'],
                                 DEFAULT_STORAGE_NAME,
                                 DEFAULT_VOLUME_NAME)

        snapshots = self._issue_api_request(snapu, method='get')

        try:
            for ts, snap in snapshots.items():
                if snap['uuid'] == snapshot['id']:
                    url_template = snapu + '/{}'
                    url = url_template.format(ts)
                    self._issue_api_request(url, method='delete')
                    break
            else:
                raise exception.NotFound
        except exception.NotFound:
            msg = _LI("Tried to delete snapshot %s, but was not found in "
                      "Datera cluster. Continuing with delete.")
            LOG.info(msg, snapshot['id'])

    def create_volume_from_snapshot(self, volume, snapshot):
        snap_temp = ('app_instances/{}/storage_instances/{}/volumes/{'
                     '}/snapshots')
        snapu = snap_temp.format(snapshot['volume_id'],
                                 DEFAULT_STORAGE_NAME,
                                 DEFAULT_VOLUME_NAME)

        snapshots = self._issue_api_request(snapu, method='get')
        for ts, snap in snapshots.items():
            if snap['uuid'] == snapshot['id']:
                found_ts = ts
                break
        else:
            raise exception.NotFound

        src = ('/app_instances/{}/storage_instances/{}/volumes/{'
               '}/snapshots/{}'.format(
                   snapshot['volume_id'],
                   DEFAULT_STORAGE_NAME,
                   DEFAULT_VOLUME_NAME,
                   found_ts))
        app_params = (
            {
                'create_mode': 'openstack',
                'uuid': str(volume['id']),
                'name': str(volume['id']),
                'clone_src': src,
                'access_control_mode': 'allow_all'
            })
        self._issue_api_request(
            'app_instances',
            method='post',
            body=app_params)

    def get_volume_stats(self, refresh=False):
        """Get volume stats.

        If 'refresh' is True, run update first.
        The name is a bit misleading as
        the majority of the data here is cluster
        data.
        """
        if refresh or not self.cluster_stats:
            try:
                self._update_cluster_stats()
            except exception.DateraAPIException:
                LOG.error(_LE('Failed to get updated stats from Datera '
                              'cluster.'))
        return self.cluster_stats

    def _update_cluster_stats(self):
        LOG.debug("Updating cluster stats info.")

        results = self._issue_api_request('system')

        if 'uuid' not in results:
            LOG.error(_LE('Failed to get updated stats from Datera Cluster.'))

        backend_name = self.configuration.safe_get('volume_backend_name')
        stats = {
            'volume_backend_name': backend_name or 'Datera',
            'vendor_name': 'Datera',
            'driver_version': self.VERSION,
            'storage_protocol': 'iSCSI',
            'total_capacity_gb': int(results['total_capacity']) / units.Gi,
            'free_capacity_gb': int(results['available_capacity']) / units.Gi,
            'reserved_percentage': 0,
        }

        self.cluster_stats = stats

    def _get_policies_by_volume_type(self, type_id):
        """Get extra_specs and qos_specs of a volume_type.

        This fetches the scoped keys from the volume type. Anything set from
         qos_specs will override key/values set from extra_specs.
        """
        ctxt = context.get_admin_context()
        volume_type = volume_types.get_volume_type(ctxt, type_id)
        specs = volume_type.get('extra_specs')

        policies = {}
        for key, value in specs.items():
            if ':' in key:
                fields = key.split(':')
                key = fields[1]
                policies[key] = value

        qos_specs_id = volume_type.get('qos_specs_id')
        if qos_specs_id is not None:
            qos_kvs = qos_specs.get_qos_specs(ctxt, qos_specs_id)['specs']
            if qos_kvs:
                policies.update(qos_kvs)
        return policies

    @_authenticated
    def _issue_api_request(self, resource_type, method='get', resource=None,
                           body=None, action=None, sensitive=False):
        """All API requests to Datera cluster go through this method.

        :param resource_type: the type of the resource
        :param method: the request verb
        :param resource: the identifier of the resource
        :param body: a dict with options for the action_type
        :param action: the action to perform
        :returns: a dict of the response from the Datera cluster
        """
        host = self.configuration.san_ip
        port = self.configuration.datera_api_port
        api_token = self.datera_api_token
        api_version = self.configuration.datera_api_version

        payload = json.dumps(body, ensure_ascii=False)
        payload.encode('utf-8')

        if not sensitive:
            LOG.debug("Payload for Datera API call: %s", payload)

        header = {'Content-Type': 'application/json; charset=utf-8'}

        protocol = 'http'
        if self.configuration.driver_use_ssl:
            protocol = 'https'

        # TODO(thingee): Auth method through Auth-Token is deprecated. Remove
        # this and client cert verification stuff in the Liberty release.
        if api_token:
            header['Auth-Token'] = api_token

        client_cert = self.configuration.driver_client_cert
        client_cert_key = self.configuration.driver_client_cert_key
        cert_data = None

        if client_cert:
            protocol = 'https'
            cert_data = (client_cert, client_cert_key)

        connection_string = '%s://%s:%s/v%s/%s' % (protocol, host, port,
                                                   api_version, resource_type)

        if resource is not None:
            connection_string += '/%s' % resource
        if action is not None:
            connection_string += '/%s' % action

        LOG.debug("Endpoint for Datera API call: %s", connection_string)
        try:
            response = getattr(requests, method)(connection_string,
                                                 data=payload, headers=header,
                                                 verify=False, cert=cert_data)
        except requests.exceptions.RequestException as ex:
            msg = _(
                'Failed to make a request to Datera cluster endpoint due '
                'to the following reason: %s') % six.text_type(
                ex.message)
            LOG.error(msg)
            raise exception.DateraAPIException(msg)

        data = response.json()
        if not sensitive:
            LOG.debug("Results of Datera API call: %s", data)

        if not response.ok:
            LOG.debug(("Datera Response URL: %s\n"
                       "Datera Response Payload: %s\n"
                       "Response Object: %s\n"),
                      response.url,
                      payload,
                      vars(response))
            if response.status_code == 404:
                raise exception.NotFound(data['message'])
            elif response.status_code in [403, 401]:
                raise exception.NotAuthorized()
            elif response.status_code == 400 and 'invalidArgs' in data:
                msg = _('Bad request sent to Datera cluster:'
                        'Invalid args: %(args)s | %(message)s') % {
                            'args': data['invalidArgs']['invalidAttrs'],
                            'message': data['message']}
                raise exception.Invalid(msg)
            else:
                msg = _('Request to Datera cluster returned bad status:'
                        ' %(status)s | %(reason)s') % {
                            'status': response.status_code,
                            'reason': response.reason}
                LOG.error(msg)
                raise exception.DateraAPIException(msg)

        return data