Merge "Add volume migration code to Nexenta iSCSI volume driver"

cinder/tests/test_nexenta.py

@@ -43,12 +43,14 @@ class TestNexentaISCSIDriver(test.TestCase):
     TEST_VOLUME_REF = {
         'name': TEST_VOLUME_NAME,
         'size': 1,
-        'id': '1'
+        'id': '1',
+        'status': 'available'
     }
     TEST_VOLUME_REF2 = {
         'name': TEST_VOLUME_NAME2,
         'size': 1,
-        'id': '2'
+        'id': '2',
+        'status': 'in-use'
     }
     TEST_SNAPSHOT_REF = {
         'name': TEST_SNAPSHOT_NAME,
@@ -72,6 +74,9 @@ def setUp(self):
         self.configuration.nexenta_target_group_prefix = 'cinder/'
         self.configuration.nexenta_blocksize = '8K'
         self.configuration.nexenta_sparse = True
+        self.configuration.nexenta_rrmgr_compression = 1
+        self.configuration.nexenta_rrmgr_tcp_buf_size = 1024
+        self.configuration.nexenta_rrmgr_connections = 2
         self.nms_mock = self.mox.CreateMockAnything()
         for mod in ['volume', 'zvol', 'iscsitarget', 'appliance',
                     'stmf', 'scsidisk', 'snapshot']:
@@ -150,6 +155,43 @@ def test_create_cloned_volume(self):
         self.mox.ReplayAll()
         self.drv.create_cloned_volume(vol, src_vref)
 
+    def test_migrate_volume(self):
+        volume = self.TEST_VOLUME_REF
+        host = {
+            'capabilities': {
+                'vendor_name': 'Nexenta',
+                'location_info': 'NexentaISCSIDriver:1.1.1.1:cinder',
+                'free_capacity_gb': 1
+            }
+        }
+        snapshot = {
+            'volume_name': volume['name'],
+            'name': 'cinder-migrate-snapshot-%s' % volume['id'],
+        }
+        self.nms_mock.appliance.ssh_list_bindings().AndReturn([])
+        self.nms_mock.zvol.create_snapshot('cinder/%s' % volume['name'],
+                                           snapshot['name'], '')
+
+        src = '%(volume)s/%(zvol)s@%(snapshot)s' % {
+            'volume': 'cinder',
+            'zvol': volume['name'],
+            'snapshot': snapshot['name']}
+        dst = '1.1.1.1:cinder'
+        cmd = ' '.join(['rrmgr -s zfs -c 1 -q -e -w 1024 -n 2', src, dst])
+
+        self.nms_mock.appliance.execute(cmd)
+
+        self.nms_mock.snapshot.destroy('cinder/%(volume)s@%(snapshot)s' % {
+                                       'volume': volume['name'],
+                                       'snapshot': snapshot['name']}, '')
+        volume_name = 'cinder/%s' % volume['name']
+        self.nms_mock.zvol.get_child_props(volume_name,
+                                           'origin').AndReturn(None)
+        self.nms_mock.zvol.destroy(volume_name, '')
+
+        self.mox.ReplayAll()
+        self.drv.migrate_volume(None, volume, host)
+
     def test_create_snapshot(self):
         self.nms_mock.zvol.create_snapshot('cinder/volume1', 'snapshot1', '')
         self.mox.ReplayAll()

cinder/volume/drivers/nexenta/iscsi.py

@@ -69,12 +69,17 @@ def __init__(self, *args, **kwargs):
                 options.NEXENTA_ISCSI_OPTIONS)
             self.configuration.append_config_values(
                 options.NEXENTA_VOLUME_OPTIONS)
+            self.configuration.append_config_values(
+                options.NEXENTA_RRMGR_OPTIONS)
         self.nms_protocol = self.configuration.nexenta_rest_protocol
         self.nms_host = self.configuration.nexenta_host
         self.nms_port = self.configuration.nexenta_rest_port
         self.nms_user = self.configuration.nexenta_user
         self.nms_password = self.configuration.nexenta_password
         self.volume = self.configuration.nexenta_volume
+        self.rrmgr_compression = self.configuration.nexenta_rrmgr_compression
+        self.rrmgr_tcp_buf_size = self.configuration.nexenta_rrmgr_tcp_buf_size
+        self.rrmgr_connections = self.configuration.nexenta_rrmgr_connections
 
     @property
     def backend_name(self):
@@ -127,6 +132,11 @@ def _is_clone_snapshot_name(snapshot):
         name = snapshot.split('@')[-1]
         return name.startswith('cinder-clone-snapshot-')
 
+    @staticmethod
+    def _get_migrate_snapshot_name(volume):
+        """Return name for snapshot that will be used to migrate the volume."""
+        return 'cinder-migrate-snapshot-%(id)s' % volume
+
     def create_volume(self, volume):
         """Create a zvol on appliance.
 
@@ -205,6 +215,84 @@ def create_cloned_volume(self, volume, src_vref):
                               '%(volume_name)s@%(name)s'), snapshot)
             raise
 
+    def _get_zfs_send_recv_cmd(self, src, dst):
+        """Returns rrmgr command for source and destination."""
+        return utils.get_rrmgr_cmd(src, dst,
+                                   compression=self.rrmgr_compression,
+                                   tcp_buf_size=self.rrmgr_tcp_buf_size,
+                                   connections=self.rrmgr_connections)
+
+    def migrate_volume(self, ctxt, volume, host):
+        """Migrate if volume and host are managed by Nexenta appliance.
+
+        :param ctxt: context
+        :param volume: a dictionary describing the volume to migrate
+        :param host: a dictionary describing the host to migrate to
+        """
+        LOG.debug(_('Enter: migrate_volume: id=%(id)s, host=%(host)s') %
+                  {'id': volume['id'], 'host': host})
+
+        false_ret = (False, None)
+
+        if volume['status'] != 'available':
+            return false_ret
+
+        if 'location_info' not in host['capabilities']:
+            return false_ret
+
+        dst_parts = host['capabilities']['location_info'].split(':')
+
+        if host['capabilities']['vendor_name'] != 'Nexenta' or \
+           dst_parts[0] != self.__class__.__name__ or \
+           host['capabilities']['free_capacity_gb'] < volume['size']:
+            return false_ret
+
+        dst_host, dst_volume = dst_parts[1:]
+
+        ssh_bound = False
+        ssh_bindings = self.nms.appliance.ssh_list_bindings()
+        for bind in ssh_bindings:
+            if bind.index(dst_host) != -1:
+                ssh_bound = True
+                break
+        if not(ssh_bound):
+            LOG.warning(_("Remote NexentaStor appliance at %s should be "
+                          "SSH-bound."), dst_host)
+
+        # Create temporary snapshot of volume on NexentaStor Appliance.
+        snapshot = {'volume_name': volume['name'],
+                    'name': self._get_migrate_snapshot_name(volume)}
+        self.create_snapshot(snapshot)
+
+        src = '%(volume)s/%(zvol)s@%(snapshot)s' % {
+            'volume': self.volume,
+            'zvol': volume['name'],
+            'snapshot': snapshot['name']}
+        dst = ':'.join([dst_host, dst_volume])
+
+        try:
+            self.nms.appliance.execute(self._get_zfs_send_recv_cmd(src, dst))
+        except nexenta.NexentaException as exc:
+            LOG.warning(_("Cannot send source snapshot %(src)s to "
+                          "destination %(dst)s. Reason: %(exc)s"),
+                        {'src': src, 'dst': dst, 'exc': exc})
+            return false_ret
+        finally:
+            try:
+                self.delete_snapshot(snapshot)
+            except nexenta.NexentaException as exc:
+                LOG.warning(_("Cannot delete temporary source snapshot "
+                              "%(src)s on NexentaStor Appliance: %(exc)s"),
+                            {'src': src, 'exc': exc})
+        try:
+            self.delete_volume(volume)
+        except nexenta.NexentaException as exc:
+            LOG.warning(_("Cannot delete source volume %(volume)s on "
+                          "NexentaStor Appliance: %(exc)s"),
+                        {'volume': volume['name'], 'exc': exc})
+
+        return (True, None)
+
     def create_snapshot(self, snapshot):
         """Create snapshot of existing zvol on appliance.
 

cinder/volume/drivers/nexenta/options.py

@@ -103,3 +103,16 @@
                 default=False,
                 help='flag to create sparse volumes'),
 ]
+
+NEXENTA_RRMGR_OPTIONS = [
+    cfg.IntOpt('nexenta_rrmgr_compression',
+               default=0,
+               help=('Enable stream compression, level 1..9. 1 - gives best '
+                     'speed; 9 - gives best compression.')),
+    cfg.IntOpt('nexenta_rrmgr_tcp_buf_size',
+               default=4096,
+               help='TCP Buffer size in KiloBytes.'),
+    cfg.IntOpt('nexenta_rrmgr_connections',
+               default=2,
+               help='Number of TCP connections.'),
+]

cinder/volume/drivers/nexenta/utils.py

@@ -20,7 +20,7 @@
 
 .. automodule:: nexenta.utils
 .. moduleauthor:: Victor Rodionov <victor.rodionov@nexenta.com>
-.. moduleauthor:: Mikhail Khodos <hodosmb@gmail.com>
+.. moduleauthor:: Mikhail Khodos <mikhail.khodos@nexenta.com>
 """
 
 import re
@@ -63,6 +63,22 @@ def str2gib_size(s):
     return size_in_bytes / units.GiB
 
 
+def get_rrmgr_cmd(src, dst, compression=None, tcp_buf_size=None,
+                  connections=None):
+    """Returns rrmgr command for source and destination."""
+    cmd = ['rrmgr', '-s', 'zfs']
+    if compression:
+        cmd.extend(['-c', '%s' % str(compression)])
+    cmd.append('-q')
+    cmd.append('-e')
+    if tcp_buf_size:
+        cmd.extend(['-w', str(tcp_buf_size)])
+    if connections:
+        cmd.extend(['-n', str(connections)])
+    cmd.extend([src, dst])
+    return ' '.join(cmd)
+
+
 def parse_nms_url(url):
     """Parse NMS url into normalized parts like scheme, user, host and others.
 

etc/cinder/cinder.conf.sample

@@ -1377,6 +1377,16 @@
 # value. (boolean value)
 #nexenta_nms_cache_volroot=true
 
+# Enable stream compression, level 1..9. 1 - gives best speed;
+# 9 - gives best compression. (integer value)
+#nexenta_rrmgr_compression=0
+
+# TCP Buffer size in KiloBytes. (integer value)
+#nexenta_rrmgr_tcp_buf_size=4096
+
+# Number of TCP connections. (integer value)
+#nexenta_rrmgr_connections=2
+
 # block size for volumes (blank=default,8KB) (string value)
 #nexenta_blocksize=