Synnefo » snf-ganeti » Ganeti ExtStorage providers

# vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright 2013 GRNET S.A.

# Copyright 2013 IBM Corp.

# Copyright 2012 OpenStack LLC.

# 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.

#

# Authors:

#   Stratos Psomadakis <psomas@grnet.gr>

#   Ronen Kat <ronenkat@il.ibm.com>

#   Avishay Traeger <avishay@il.ibm.com>

""" Ganeti extstorage provider for IBM Storwize family and SVC storage systems

(ported from the Cinder Storwize / SVC volume driver).

The script takes its input from environment variables. Specifically the

following variables should be present:

 - VOL_NAME: The name of the new Image file

 - EXTP_SVC_SAN: The provider's configuration file section / identifier for the

                 SVC SAN

The following environmental variables are optional and should be present only

for specific commands:

 - VOL_SIZE: The size of the Image file

Based on the EXTP_SVC_SAN variable, the script will also read the SVC

configration parameters / options from the corresponding section of the INI SVC

configuration file. Currently, the config file path is hardcoded in

SVC_CONFIG_PATH (/etc/ganeti/extstorage/svc.conf). Each SAN / SVC section of

the INI file, which should set the following options:

 - host: The management hostname / IP for the SVC

 - port: The SSH daemon port for the SVC

 - user: The user to use to connect (SSH) to the SVC

 - password: The SSH password for the user

 - pool: The SVC volume pool to use to create vdisks / volumes

 - host_id: The id of the host in the SVC hosts table

Example SAN SVC INI config file:

    [san1]

    host = mysvchost1

    port = mysvcport1

    user = mysvcuser1

    password = mysvcpass1

    pool = mypool1

    host_id = mysvchostid1

    [san2]

    host = mysvchost2

    port = mysvcport2

    user = mysvcuser2

    password = mysvcpass2

    pool = mypool2

    host_id = mysvchostid2

The code branches to the correct function, depending on the name (sys.argv[0])

of the executed script (attach, create, etc.).

Returns O after successful completion, 1 on failure.

Limitations:

1. The provider expects CLI output in English, error messages may be in a

   localized format.

2. iSCSI support, public key authentication and detailed configuration for the

   SVC pool, volumes etc. not implemented yet.

3. The current version of the provider makes lots of assumptions about the

   running environment. It needs to be run on a Debian-like distro with various

   SCSI tools installed and multipath support as well. Multipath udev triggers

   should be removed, to ensure reliable operation of the provider.

"""

import ConfigParser

import glob

import os

import re

import subprocess

import sys

import traceback

import paramiko

# Hardcoded path to the SVC (Ganeti) config file

_SVC_CONFIG_PATH = "/etc/ganeti/extstorage/svc.conf"

# Hardcoded pool name, used for assertions

_SVC_POOL_NAME = "VM_POOL_SYNNEFO"

# Hardcoded host id range for HDIKA Synnefo nodes, used for assertions

_SVC_VALID_RANGE = range(7, 12) + range(19, 23)

# Hardcoded timeout for SSH (paramiko), in seconds

SVC_SSH_TIMEOUT = 30

# sysfs FC-related paths and tools, needed for adding (rescanning) and deleting

# FC disks / volumes.

_RESCAN_SCSI_CMD = 'rescan-scsi-bus'

_MULTIPATH_CMD = 'multipath'

_DMSETUP_CMD = 'dmsetup'

# Might be needing these below, in the future

#_SYSTOOL_CMD = 'systool'

#_SYSFS_FCHOSTS_PATH = '/sys/class/fc_host/'

#_SYSFS_FCSCAN_PATH = '/sys/class/scsi_host/%(fchost)s/scan'

#_SYSFS_FCSCAN_COMMAND = '- - -'

_SYSFS_BLOCK_PATH = '/sys/block/%(blodkdev)s/'

_SYSFS_FCDELETE_PATH = os.path.join(_SYSFS_BLOCK_PATH, '/device/delete')

_SYSFS_SLAVES_PATH = os.path.join(_SYSFS_BLOCK_PATH, '/slaves/')

_DEV_BYPATH = '/dev/disk/by-path/'

_SCSI_DEV_PREFIX = '3'

class SVCError(Exception):

    """ Base exception for SVC extstorage provider.

    """

    pass

class SVCConnection(object):

    """ SSH connection (paramiko) abstraction class.

    """

    def __init__(self, host, port, user, password):

        """ Initialize the SVCConnection object

        @type host: string

        @param host: host to connect to

        @type port: integer

        @param port: SSH port to connect to

        @type user: string

        @param user: SSH user to use

        @type password: string

        @param password: SSH password to use

        """

        super(SVCConnection, self).__init__()

        self._host = host

        self._port = port

        self._user = user

        self._password = password

        self._conn = None

    @staticmethod

    def _check_ssh_injection(cmd_list):

        """ Check for ssh injection patterns in the cmd list.

        @type cmd_list: list of strings

        @param cmd_list: a list containing the command to be executed plus any

                         arguments to the command

        """

        ssh_injection_pattern = ['`', '$', '|', '||', ';', '&', '&&', '>',

                                 '>>', '<']

        # Check whether injection attacks exist

        for arg in cmd_list:

            arg = arg.strip()

            # First, check no space in the middle of arg

            arg_len = len(arg.split())

            if arg_len > 1:

                raise SVCError("SSH injection detected: %s" % cmd_list)

            # Second, check whether danger character in command. So the shell

            # special operator must be a single argument.

            for char in ssh_injection_pattern:

                if arg == char:

                    continue

                result = arg.find(char)

                if not result == -1:

                    if result == 0 or not arg[result - 1] == '\\':

                        raise SVCError("SSH injection detected: %s" % cmd_list)

    def _create_connection(self):

        """ Create a paramiko SSH connection to the SVC.

        """

        self._conn = paramiko.SSHClient()

        self._conn.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        self._conn.connect(self._host, port=self._port,

                           username=self._user,

                           password=self._password,

                           timeout=SVC_SSH_TIMEOUT,

                           look_for_keys=False,

                           allow_agent=False)

        transport = self._conn.get_transport()

        transport.sock.settimeout(None)

        transport.set_keepalive(SVC_SSH_TIMEOUT)

    def run_ssh_cmd(self, cmdlist):

        """ Connect to host and run the command specified in cmdlist.

        @type cmd_list: list of strings

        @param cmd_list: a list containing the command to be executed plus any

                         arguments to the command

        """

        self._check_ssh_injection(cmdlist)

        cmd = ' '.join(cmdlist)

        try:

            if self._conn:

                if not self._conn.get_transport().is_active():

                    self._conn.close()

                    self._create_connection()

            else:

                self._create_connection()

        except Exception as exc:

            raise SVCError("%sCannot create SSH connection to SVC: %s"

                           % (traceback.format_exc(), exc))

        conn = self._conn

        stderr = ''

        try:

            stdin_stream, stdout_stream, stderr_stream = conn.exec_command(cmd)

            channel = stdout_stream.channel

            # NOTE(justinsb): This seems suspicious...

            # ...other SSH clients have buffering issues with this approach

            stdout = stdout_stream.read()

            stderr = stderr_stream.read()

            stdin_stream.close()

            exit_status = channel.recv_exit_status()

            if exit_status != 0:

                raise Exception("Non-zero exit status %s" % exit_status)

            return (stdout, stderr)

        except Exception as exc:

            raise SVCError("%sCannot run SSH command '%s' on SVC :%s (%s)" %

                           (traceback.format_exc(), cmd, stderr, exc))

class SVCProvider(object):

    """ SVC provider class.

    """

    def __init__(self, config_file):

        """ Initializes the SVC provider.

        The method will read env variables and the SVC ini config file to fill

        in all the needed info for the SVC provider ops and initialize the

        SSH connection to the SVC (L{SVCConnection}).

        @type config_file: string

        @param config_file: path to the SVC INI configuration file

        """

        super(SVCProvider, self).__init__()

        self._svc_pool = None

        self._host_id = None

        self._vol_name = None

        self._vol_size = None

        (host, port, user, password) = self._get_params(config_file)

        self._svc_connection = SVCConnection(host, port, user, password)

    def _parse_svc_config(self, config_file, san):

        """"Parse the SVC (Ganeti) configuration file.

        The config should be in 'INI' format.

        @type config_file: string

        @param config_file: path to the INI config file

        @type san: string

        @param san: SVC SAN identifier / section in the INI file

        @rtype: list of the SVC config file options

        @return: list of the SVC config file options

        """

        config = ConfigParser.SafeConfigParser()

        try:

            if not config.read(config_file):

                raise ConfigParser.Error("Unable to read config file")

            expected_opts = frozenset(['host', 'port', 'user', 'password',

                                       'pool', 'host_id'])

            config_opts = config.options(san)

            if frozenset(config_opts) != expected_opts:

                raise ConfigParser.Error('Malformed config file')

            opts = []

            host = config.get(san, 'host')

            if host == '':

                raise ConfigParser.Error('Bad value for host')

            opts.append(host)

            port = config.get(san, 'port')

            try:

                port = int(port)

                if port <= 0 or port > 65535:

                    raise ValueError

            except ValueError:

                raise ConfigParser.Error('Bad value for port')

            opts.append(port)

            user = config.get(san, 'user')

            if user == '':

                raise ConfigParser.Error('Bad value for user')

            opts.append(user)

            password = config.get(san, 'password')

            if password == '':

                raise ConfigParser.Error('Bad value for password')

            opts.append(password)

            pool = config.get(san, 'pool')

            if pool == '':

                raise ConfigParser.Error('Bad value for pool')

            self._svc_pool = pool

            host_id = config.get(san, 'host_id')

            try:

                host_id = int(host_id)

                if host_id <= 0:

                    raise ValueError

            except ValueError:

                raise ConfigParser.Error('Bad value for host_id')

            self._host_id = host_id

        except ConfigParser.Error as config_error:

            raise SVCError('Error while reading svc config file: %s.\n' %

                           config_error)

        return opts

    def _get_params(self, config_file):

        """ Read env variables and the INI config file.

        Return the values of the environmental variables, set by Ganeti, and

        the SVC configuration parameters set in the SVC (Ganeti) config file.

        @type config_file: string

        @param config_file: path to the INI config file

        @rtype: list of the SVC config file options

        @return: list of the SVC config file options

        """

        self._vol_name = os.getenv("VOL_NAME")

        if self._vol_name is None:

            raise SVCError('The environment variable VOL_NAME is missing.\n')

        if self._vol_name == '':

            raise SVCError('Volume name cannot be empty.\n')

        self._vol_size = os.getenv("VOL_SIZE")

        if self._vol_size is not None:

            try:

                self._vol_size = int(self._vol_size)

                if self._vol_size <= 0:

                    raise ValueError

            except ValueError:

                raise SVCError('Volume size must be positive integer')

        san = os.getenv("EXTP_SVC_SAN")

        if san is None:

            raise SVCError('The environment variable EXTP_SVC_SAN is'

                           'missing.\n')

        if san == '':

            raise SVCError('EXTP_SVC_SAN cannot be none')

        return self._parse_svc_config(config_file, san)

    @staticmethod

    def _driver_assert(assert_condition, exception_message):

        """ Internal assertion mechanism for CLI output.

        @type assert_condition: boolean

        @param assert_condition: condition to assert against

        @type exception_message: string

        @param exception_message: exception message to print

        """

        if not assert_condition:

            raise SVCError(exception_message)

    def _assert_ssh_return(self, test, fun, ssh_cmd, out, err):

        """ Call internal assertion mechanism for SSH cmds.

        @type test: boolean

        @param test: condition to assert against

        @type fun: object

        @param fun: function / method which called us

        @type ssh_cmd: list of strings

        @param ssh_cmd: list of strings (command plus command arguments)

        @type out: buffer

        @param out: stdout buffer

        @type err: buffer

        @param err: stderr buffer

        """

        self._driver_assert(

            test,

            '%(fun)s: Failed with unexpected CLI output.\n '

            'Command: %(cmd)s\n stdout: %(out)s\n stderr: %(err)s'

            % {'fun': fun,

                'cmd': ssh_cmd,

                'out': str(out),

                'err': str(err)})

    def _exec_cmd_and_parse_attrs(self, ssh_cmd):

        """Execute command on the Storwize/SVC and parse attributes.

        Exception is raised if the information from the system

        can not be obtained.

        @type ssh_cmd: list of strings

        @param ssh_cmd: a list of the command and additional arguments

        @rtype: dict

        @return: a dictionary containing the attributes returned by ssh cmd

        """

        out, err = self._svc_connection.run_ssh_cmd(ssh_cmd)

        self._assert_ssh_return(len(out),

                                '_exec_cmd_and_parse_attrs',

                                ssh_cmd, out, err)

        attributes = {}

        for attrib_line in out.split('\n'):

            # If '!' not found, return the string and two empty strings

            attrib_name, _, attrib_value = attrib_line.partition('!')

            if attrib_name is not None and len(attrib_name.strip()):

                attributes[attrib_name] = attrib_value

        return attributes

    def _get_hdr_dic(self, header, row, delim):

        """Return CLI row data as a dictionary indexed by names from header.

        The strings are converted to columns using the delimiter in

        delim.

        @type header: string

        @param header: header string

        @type row: string

        @param row: row string

        @type delim: string

        @param delim: column delimiter

        @rtype: dict

        @return: dictionary containing CLI row data

        """

        attributes = header.split(delim)

        values = row.split(delim)

        self._driver_assert(

            len(values) ==

            len(attributes),

            '_get_hdr_dic: attribute headers and values do not match.\n '

            'Headers: %(header)s\n Values: %(row)s'

            % {'header': str(header),

               'row': str(row)})

        # Is it the same as dict(zip())?

        #dic = dict((a, v) for a, v in map(None, attributes, values))

        return dict(zip(attributes, values))

    def _get_hostvdisk_mappings(self, host_id):

        """ Return the defined storage mappings for a host.

        @type host_id: integer

        @param host_id: id of the host in the SVC

        @rtype: dict

        @return: dictionary containing the storage mappings for a host

        """

        return_data = {}

        ssh_cmd = ['svcinfo', 'lshostvdiskmap', '-delim', '!', str(host_id)]

        out, _ = self._svc_connection.run_ssh_cmd(ssh_cmd)

        mappings = out.strip().split('\n')

        if len(mappings):

            header = mappings.pop(0)

            for mapping_line in mappings:

                mapping_data = self._get_hdr_dic(header, mapping_line, '!')

                return_data[mapping_data['vdisk_name']] = mapping_data

        return return_data

    def _map_vol_to_host(self, volume_name, host_id):

        """ Create a mapping between a volume to a host.

        @type volume_name: string

        @param volume_name: volume name

        @type host_id: integer

        @param host_id: id of the host in the SVC

        """

        assert(host_id in _SVC_VALID_RANGE)

        ssh_cmd = ['svctask', 'mkvdiskhostmap', '-host', str(host_id),

                   volume_name]

        out, err = self._svc_connection.run_ssh_cmd(ssh_cmd)

        self._assert_ssh_return('successfully created' in out,

                                '_map_vol_to_host', ssh_cmd, out, err)

    def _unmap_vol_from_host(self, volume_name, host_id):

        """ Delete a mapping between a volume and a host.

        @type volume_name: string

        @param volume_name: volume name

        @type host_id: integer

        @param host_id: id of the host in the SVC

        """

        assert(host_id in _SVC_VALID_RANGE)

        ssh_cmd = ['svctask', 'rmvdiskhostmap', '-host', str(host_id),

                   volume_name]

        out, err = self._svc_connection.run_ssh_cmd(ssh_cmd)

        # Verify CLI behaviour - no output is returned from

        # rmvdiskhostmap

        self._assert_ssh_return(len(out.strip()) == 0,

                                'unmap_vol_from_host', ssh_cmd, out, err)

    def _get_vdisk_attributes(self, vdisk_name):

        """ Return vdisk attributes, or None if vdisk does not exist.

        Exception is raised if the information from system can not be

        parsed/matched to a single vdisk.

        @type vdisk_name: string

        @param vdisk_name: vdisk name

        @rtype: dict

        @return: a dictionary containing the vdisk attributes

        """

        ssh_cmd = ['svcinfo', 'lsvdisk', '-bytes', '-delim', '!', vdisk_name]

        return self._exec_cmd_and_parse_attrs(ssh_cmd)

    def _is_vdisk_defined(self, vdisk_name):

        """ Check if vdisk is defined.

        @type vdisk_name: string

        @param vdisk_name: vdisk name

        @rtype: bool

        @return: vdisk defined or not

        """

        # Maybe get list of disks and search instead of relying on empty

        # attr reply

        try:

            self._get_vdisk_attributes(vdisk_name)

            return True

        except SVCError:

            return False

    def _is_vdisk_mapped(self, vdisk_name, host_id):

        """ Check if vdisk is mapped to the host. """

        return vdisk_name in self._get_hostvdisk_mappings(host_id)

    @staticmethod

    def _check_call(cmdlist):

        """ Wrapper around subprocess.check_call.

        @type cmdlist: list of strings

        @param cmdlist: list of command plus arguments

        """

        try:

            subprocess.check_call(cmdlist, stderr=subprocess.STDOUT)

        except subprocess.CalledProcessError as exc:

            raise SVCError("Command %s failed: %s" % (' '.join(cmdlist),

                                                      exc.output))

    @staticmethod

    def _check_output(cmdlist):

        """ Wrapper around subprocess.check_output.

        @type cmdlist: list of strings

        @param cmdlist: list of command plus arguments

        @rtype: string

        @return: output of cmdlist

        """

        try:

            return subprocess.check_output(cmdlist, stderr=subprocess.STDOUT)

        except subprocess.CalledProcessError as exc:

            raise SVCError("Command %s failed: %s" % (' '.join(cmdlist),

                                                      exc.output))

    @staticmethod

    def _get_devices(lun):

        """ Get the devices mapped to host, corresponding to LUN id lun

        @type lun: integer

        @param lun: LUN id

        @rtype: list of strings

        @return: devices discovered by host or not

        """

        return glob.glob("%s/*lun%s" % (_DEV_BYPATH, lun))

    def _devices_discovered(self, lun):

        """ Make sure the devices get created on the host.

        @type lun: integer

        @param lun: LUN id

        @rtype: bool

        @return: devices discovered by host or not

        """

        return len(self._get_devices(lun)) == 4

    def _devices_add(self, lun):

        """ Rescan SCSI / FC bus for new devices.

        @type lun: integer

        @param lun: LUN id

        """

        self._check_call([_RESCAN_SCSI_CMD])

        if self._devices_discovered(lun) is False:

            raise SVCError("SCSI bus rescan failed to properly add the new "

                           "devices")

    def _devices_remove(self, lun):

        """ Remove scanned devices.

        @type device: string

        @param device: device string

        @type lun: integer

        @param lun: LUN id

        """

        # Which one is better?:

        #dm_device = os.path.split(os.readlink(device))[-1]

        #dm_slaves = os.listdir(_SYSFS_SLAVES_PATH % {'blockdev': dm_device})

        dm_slaves = self._get_devices(lun)

        # Delete the disks

        try:

            for slave in dm_slaves:

                delete_path = _SYSFS_FCDELETE_PATH % {'blockdev': slave}

                with open(delete_path, 'w') as delete_bdev_file:

                    delete_bdev_file.write('1')

        except IOError as exc:

            raise SVCError("Cannot delete FC devices: %s" % exc)

        if self._devices_discovered(lun) is True:

            raise SVCError('Failed to remove FC devices')

    def _multipath_ok(self, device):

        """ Check if multipath device is set up correctly.

        @type device: string

        @param device: device string

        @rtype: bool

        @return: multipath device ok or not

        """

        self._check_call([_DMSETUP_CMD, 'info', device])

        return self._check_output([_MULTIPATH_CMD, '-ll', device]) != ""

    def _multipath_setup(self, device):

        """ Setup multpath / dm for newly discovered devices.

        @type device: string

        @param device: device string

        """

        self._check_call([_MULTIPATH_CMD])

        if self._multipath_ok(device) is False:

            raise SVCError("Multipath / DM setup failed")

    def _multipath_remove(self, device):

        """ Tear down multipath.

        @type device: string

        @param device: device string

        """

        self._check_call([_MULTIPATH_CMD, '-f', device])

        if self._multipath_ok(device) is True:

            raise SVCError("Unable to reove multipath device")

    def _host_device_remove(self, device, lun):

        """ Tear down multipath and remove the scanned devices.

        @type device: string

        @param device: device string

        @type lun: integer

        @type lun: LUN id

        """

        self._multipath_remove(device)

        self._devices_remove(lun)

    def _host_device_setup(self, device, lun):

        """ Discover and setup the devices on the host machine.

        @type device: string

        @param device: device string

        @type lun: integer

        @type lun: LUN id

        """

        try:

            self._devices_add(lun)

            self._multipath_setup(device)

        except:

            # Cleanup any devices added by SCSI bus rescan

            # and remove device mappings

            self._host_device_remove(device, lun)

            raise