Merge of qemu-datapath into REQ477/master branch

ocaml/xapi/xapi_globs.ml

@@ -824,6 +824,8 @@ let nbd_firewall_config_script = ref "/opt/xensource/libexec/nbd-firewall-config
 
 let firewall_port_config_script = ref "/etc/xapi.d/plugins/firewall-port"
 
+let nbd_client_manager_script = ref "/opt/xensource/libexec/nbd_client_manager.py"
+
 let disable_logging_for= ref []
 
 let nvidia_whitelist = ref "/usr/share/nvidia/vgpu/vgpuConfig.xml"
@@ -1110,6 +1112,7 @@ module Resources = struct
     "killall", kill_process_script, "Executed to kill process";
     "nbd-firewall-config", nbd_firewall_config_script, "Executed after NBD-related networking changes to configure the firewall for NBD";
     "firewall-port-config", firewall_port_config_script, "Executed when starting/stopping xapi-clusterd to configure firewall port";
+    "nbd_client_manager", nbd_client_manager_script, "Executed to safely connect to and disconnect from NBD devices using nbd-client";
   ]
   let essential_files = [
     "pool_config_file", pool_config_file, "Pool configuration file";

ocaml/xapi/xapi_vbd.ml

@@ -31,6 +31,36 @@ let update_allowed_operations ~__context ~self : unit =
 let assert_attachable ~__context ~self : unit =
   assert_attachable ~__context ~self
 
+let print_fork_error f =
+  try
+    f ()
+  with Forkhelpers.Spawn_internal_error(stderr, stdout, status) as e ->
+    begin match status with
+      | Unix.WEXITED n ->
+        error "Forkhelpers.Spawn_internal_error(%s, %s, WEXITED %d)" stderr stdout n;
+        raise e
+      | Unix.WSIGNALED n ->
+        error "Forkhelpers.Spawn_internal_error(%s, %s, WSIGNALED %d)" stderr stdout n;
+        raise e
+      | Unix.WSTOPPED n ->
+        error "Forkhelpers.Spawn_internal_error(%s, %s, WSTOPPED %d)" stderr stdout n;
+        raise e
+    end
+
+let run_command cmd args =
+  debug "running %s %s" cmd (String.concat " " args);
+  let stdout, stderr = print_fork_error (fun () -> Forkhelpers.execute_command_get_output cmd args) in
+  stdout
+
+module NbdClient = struct
+  let start_nbd_client ~unix_socket_path ~export_name =
+    run_command !Xapi_globs.nbd_client_manager_script ["connect"; "--path"; unix_socket_path; "--exportname"; export_name]
+    |> String.trim
+
+  let stop_nbd_client ~nbd_device =
+    run_command !Xapi_globs.nbd_client_manager_script ["disconnect"; "--device"; nbd_device]
+    |> ignore
+end
 
 let set_mode ~__context ~self ~value =
   let vm = Db.VBD.get_VM ~__context ~self in
@@ -47,10 +77,24 @@ let plug ~__context ~self =
     Storage_access.attach_and_activate ~__context ~vbd:self ~domid
       (fun attach_info ->
          let params = attach_info.Storage_interface.params in
-         let prefix = "/dev/" in
-         let prefix_len = String.length prefix in
-         let path = String.sub params prefix_len (String.length params - prefix_len) in
-         Db.VBD.set_device ~__context ~self ~value:path;
+         let device_path =
+           let nbd_prefix = "hack|nbd:unix:" in
+           let is_nbd = Astring.String.is_prefix ~affix:nbd_prefix params in
+           if is_nbd then begin
+             debug "Using nbd-client for VBD.plug of VBD '%s', attach_info.params='%s'" (Ref.string_of self) params;
+             let export_name = "qemu_node" in
+             let unix_socket_path = params |> Astring.String.cuts ~empty:false ~sep:nbd_prefix |> List.hd |> String.split_on_char '|' |> List.hd in
+             NbdClient.start_nbd_client ~unix_socket_path ~export_name
+           end
+           else params
+         in
+         let device_path =
+           let prefix = "/dev/" in
+           let prefix_len = String.length prefix in
+           String.sub device_path prefix_len (String.length device_path - prefix_len)
+         in
+         debug "device path: %s" device_path;
+         Db.VBD.set_device ~__context ~self ~value:device_path;
          Db.VBD.set_currently_attached ~__context ~self ~value:true;
       )
   end
@@ -69,6 +113,10 @@ let unplug ~__context ~self =
   if System_domains.storage_driver_domain_of_vbd ~__context ~vbd:self = vm && not force_loopback_vbd then begin
     debug "VBD.unplug of loopback VBD '%s'" (Ref.string_of self);
     let domid = Int64.to_int (Db.VM.get_domid ~__context ~self:vm) in
+    let device = Db.VBD.get_device ~__context ~self in
+    let nbd_device_prefix = "nbd" in
+    let is_nbd = String.startswith nbd_device_prefix device in
+    if is_nbd then NbdClient.stop_nbd_client ~nbd_device:("/dev/" ^ device);
     Storage_access.deactivate_and_detach ~__context ~vbd:self ~domid;
     Db.VBD.set_currently_attached ~__context ~self ~value:false
   end

scripts/Makefile

@@ -25,6 +25,7 @@ install:
 	$(IPROG) xn_diagnostics $(DESTDIR)$(LIBEXECDIR)
 	$(IPROG) thread_diagnostics $(DESTDIR)$(LIBEXECDIR)
 	$(IPROG) list_plugins $(DESTDIR)$(LIBEXECDIR)
+	$(IPROG) nbd_client_manager.py $(DESTDIR)$(LIBEXECDIR)
 	mkdir -p $(DESTDIR)$(ETCDIR)/bugtool/xapi
 	mkdir -p $(DESTDIR)$(ETCDIR)/bugtool/xenopsd
 	$(IDATA) bugtool-plugin/xapi.xml $(DESTDIR)$(ETCDIR)/bugtool

scripts/nbd_client_manager.py

@@ -0,0 +1,202 @@
+#!/usr/bin/python
+
+"""
+Provides functions and a CLI for safely connecting to and disconnecting from
+NBD devices.
+"""
+
+import argparse
+import logging
+import logging.handlers
+import os
+import subprocess
+import time
+import fcntl
+
+
+LOGGER = logging.getLogger("nbd_client_manager")
+LOGGER.setLevel(logging.DEBUG)
+
+LOCK_FILE = '/var/run/nonpersistent/nbd_client_manager'
+
+
+class NbdDeviceNotFound(Exception):
+    """
+    The NBD device file does not exist. Raised when there are no free NBD
+    devices.
+    """
+    def __init__(self, nbd_device):
+        self.nbd_device = nbd_device
+
+
+class FileLock(object):
+    """Container for data relating to a file lock"""
+    def __init__(self, path):
+        self._path = path
+        self._lock_file = None
+
+    def _lock(self):
+        """Acquire the lock"""
+        flags = fcntl.LOCK_EX
+        self._lock_file = open(self._path, 'w+')
+        fcntl.flock(self._lock_file, flags)
+
+    def _unlock(self):
+        """Unlock and remove the lock file"""
+        if self._lock_file:
+            fcntl.flock(self._lock_file, fcntl.LOCK_UN)
+            self._lock_file.close()
+            self._lock_file = None
+
+    def __enter__(self):
+        self._lock()
+
+    def __exit__(self, *args):
+        self._unlock()
+
+
+FILE_LOCK = FileLock(path=LOCK_FILE)
+
+
+def _call(cmd_args, error=True):
+    """
+    [call cmd_args] executes [cmd_args] and returns the exit code.
+    If [error] and exit code != 0, log and throws a CalledProcessError.
+    """
+    LOGGER.debug("Running cmd %s", cmd_args)
+    p = subprocess.Popen(
+        cmd_args,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        close_fds=True
+    )
+
+    stdout, stderr = p.communicate()
+
+    if error and p.returncode != 0:
+        LOGGER.error(
+            "%s exitted with code %d: %s",
+            ' '.join(cmd_args),
+            p.returncode,
+            stderr)
+
+        raise subprocess.CalledProcessError(
+            returncode=p.returncode,
+            cmd=cmd_args,
+            output=stderr)
+
+    return p.returncode
+
+
+def _is_nbd_device_connected(nbd_device):
+    """
+    Checks whether the specified nbd device is connected according to
+    nbd-client.
+    """
+    # First check if the file exists, because "nbd-client -c" returns
+    # 1 for a non-existent file.
+    if not os.path.exists(nbd_device):
+        raise NbdDeviceNotFound(nbd_device)
+    cmd = ['nbd-client', '-check', nbd_device]
+    returncode = _call(cmd, error=False)
+    if returncode == 0:
+        return True
+    if returncode == 1:
+        return False
+    raise subprocess.CalledProcessError(returncode=returncode, cmd=cmd)
+
+
+def _find_unused_nbd_device():
+    """
+    Returns the path of the first /dev/nbdX device that is not
+    connected according to nbd-client.
+    """
+    for device_no in range(0, 100):
+        nbd_device = "/dev/nbd{}".format(device_no)
+        if not _is_nbd_device_connected(nbd_device=nbd_device):
+            return nbd_device
+
+
+def _wait_for_nbd_device(nbd_device, connected):
+    while _is_nbd_device_connected(nbd_device=nbd_device) != connected:
+        LOGGER.debug(
+            'Connection status of NBD device %s not yet %s, waiting',
+            nbd_device,
+            connected)
+        time.sleep(0.1)
+
+
+def connect_nbd(path, exportname):
+    """Connects to a free NBD device using nbd-client and returns its path"""
+    _call(['modprobe', 'nbd'])
+    with FILE_LOCK:
+        nbd_device = _find_unused_nbd_device()
+        cmd = ['nbd-client', '-unix', path, nbd_device, '-name', exportname]
+        _call(cmd)
+        _wait_for_nbd_device(nbd_device=nbd_device, connected=True)
+    return nbd_device
+
+
+def disconnect_nbd_device(nbd_device):
+    """
+    Disconnects the given device using nbd-client.
+    This function is idempotent: calling it on an already disconnected device
+    does nothing.
+    """
+    with FILE_LOCK:
+        if _is_nbd_device_connected(nbd_device=nbd_device):
+            cmd = ['nbd-client', '-disconnect', nbd_device]
+            _call(cmd)
+            _wait_for_nbd_device(nbd_device=nbd_device, connected=False)
+
+
+def _connect_cli(args):
+    device = connect_nbd(path=args.path, exportname=args.exportname)
+    print device
+
+
+def _disconnect_cli(args):
+    disconnect_nbd_device(nbd_device=args.device)
+
+
+if __name__ == '__main__':
+    # Configure the root logger to log into syslog
+    # (Specifically, into /var/log/user.log)
+    SYSLOG_HANDLER = logging.handlers.SysLogHandler(
+        address='/dev/log',
+        facility=logging.handlers.SysLogHandler.LOG_USER)
+    logging.getLogger().addHandler(SYSLOG_HANDLER)
+
+    try:
+        parser = argparse.ArgumentParser(
+            description="Connect to and disconnect from an NBD device")
+
+        subparsers = parser.add_subparsers(dest='command_name')
+
+        parser_connect = subparsers.add_parser(
+            'connect',
+            help='Connect to a free NBD device and return its path')
+        parser_connect.add_argument(
+            '--path',
+            required=True,
+            help="The path of the Unix domain socket of the NBD server")
+        parser_connect.add_argument(
+            '--exportname',
+            required=True,
+            help="The export name of the device to connect to")
+        parser_connect.set_defaults(func=_connect_cli)
+
+        parser_disconnect = subparsers.add_parser(
+            'disconnect',
+            help='Disconnect from the given NBD device')
+        parser_disconnect.add_argument(
+            '--device',
+            required=True,
+            help="The path of the NBD device to disconnect")
+        parser_disconnect.set_defaults(func=_disconnect_cli)
+
+        args = parser.parse_args()
+        args.func(args)
+    except Exception as e:
+        LOGGER.exception(e)
+        raise

scripts/static-vdis

@@ -213,6 +213,17 @@ def call_backend_detach(driver, config):
         res = xmlrpc[0][0]
     return res
 
+def connect_nbd(path, exportname):
+    return subprocess.check_output(
+        ['/opt/xensource/libexec/nbd_client_manager.py', 'connect',
+         '--path', path,
+         '--exportname', exportname]).strip()
+
+def disconnect_nbd_device(nbd_device):
+    subprocess.check_call(
+        ['/opt/xensource/libexec/nbd_client_manager.py', 'disconnect',
+         '--device', nbd_device])
+
 def attach(vdi_uuid):
     found = False
     for existing in list():
@@ -242,6 +253,12 @@ def attach(vdi_uuid):
                     attach = call_datapath_plugin(scheme, "Datapath.attach", [ vol_uri, "0" ])
                     path = attach['implementation'][1]
                     call_datapath_plugin(scheme, "Datapath.activate", [ vol_uri, "0" ])
+                    nbd_prefix = 'hack|nbd:unix:'
+                    nbd = path.startswith(nbd_prefix)
+                    if nbd:
+                        path = path[len(nbd_prefix):]
+                        socket_path = path.split('|')[0]
+                        path = connect_nbd(socket_path, 'qemu_node')
 
                 os.symlink(path, d + "/disk")
                 return d + "/disk"
@@ -256,6 +273,9 @@ def detach(vdi_uuid):
                 return
             found = True
             d = main_dir + "/" + existing['id']
+            disk = existing['disk']
+            if disk.startswith('/dev/nbd'):
+                disconnect_nbd_device(disk)
             if not (os.path.exists(d + "/sr-uri")):
                 # SMAPIv1
                 config = read_whole_file(d + "/config")