Change in vdsm[master]: WIP: lib: clientIF: move vmContainer away

fromani at redhat.com fromani at redhat.com
Wed Feb 24 14:57:45 UTC 2016


Francesco Romani has uploaded a new change for review.

Change subject: WIP: lib: clientIF: move vmContainer away
......................................................................

WIP: lib: clientIF: move vmContainer away

Proof Of Concept/Work in Progress.

Change-Id: Iacd2ae6c5e9ca6a73c0fed978c78c9ebb001c46d
Signed-off-by: Francesco Romani <fromani at redhat.com>
---
M lib/vdsm/libvirtconnection.py
A lib/vdsm/vmdict.py
M vdsm/API.py
M vdsm/clientIF.py
M vdsm/virt/periodic.py
M vdsm/virt/vm.py
6 files changed, 168 insertions(+), 102 deletions(-)


  git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/01/53101/3

diff --git a/lib/vdsm/libvirtconnection.py b/lib/vdsm/libvirtconnection.py
index 6e522f4..b156ad0 100644
--- a/lib/vdsm/libvirtconnection.py
+++ b/lib/vdsm/libvirtconnection.py
@@ -29,6 +29,7 @@
 import libvirt
 from . import concurrent
 from . import utils
+from . import vmdict
 from .password import ProtectedPassword
 from .tool.configurators import passwd
 
@@ -110,8 +111,6 @@
 def get(target=None, killOnFailure=True):
     """Return current connection to libvirt or open a new one.
     Use target to get/create the connection object linked to that object.
-    target must have a callable attribute named 'dispatchLibvirtEvents' which
-    will be registered as a callback on libvirt events.
 
     Wrap methods of connection object so that they catch disconnection, and
     take the current process down.
@@ -179,7 +178,7 @@
                            libvirt.VIR_DOMAIN_EVENT_ID_WATCHDOG):
                     conn.domainEventRegisterAny(None,
                                                 ev,
-                                                target.dispatchLibvirtEvents,
+                                                vmdict.on_libvirt_event,
                                                 ev)
             # In case we're running into troubles with keeping the connections
             # alive we should place here:
diff --git a/lib/vdsm/vmdict.py b/lib/vdsm/vmdict.py
new file mode 100644
index 0000000..e1d95e0
--- /dev/null
+++ b/lib/vdsm/vmdict.py
@@ -0,0 +1,133 @@
+#
+# Copyright 2011-2016 Red Hat, Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
+#
+# Refer to the README and COPYING files for full details of the license
+#
+
+from __future__ import absolute_import
+
+import logging
+import threading
+
+import libvirt
+
+
+_log = logging.getLogger('vdsm.vmdict')
+_lock = threading.Lock()
+_vms = {}
+
+
+class UnknownVM(Exception):
+    pass
+
+
+class DuplicateVM(Exception):
+    pass
+
+
+def num_vms():
+    with _lock:
+        return len(_vms)
+
+
+def get_all_items():
+    """
+    Get a snapshot of the currently registered VMs.
+    Return value will be a dict of {vmUUID: VM_object}
+    """
+    with _lock:
+        return _vms.copy()
+
+
+def get_all_vms():
+    with _lock:
+        return _vms.values()
+
+
+def get_all_stats(self):
+    with _lock:
+        return [v.getStats() for v in _vms.values()]
+
+
+def add(vm):
+    with _lock:
+        if vm.id in _vms:
+            raise DuplicateVM()
+        _vms[vm.id] = vm
+
+
+def remove(vm):
+    with _lock:
+        if vm.id in _vms:
+            del _vms[vm.id]
+        else:
+            raise UnknownVM()
+
+
+def on_eio_cont(libvirt_vms, sdUUID):
+    with _lock:
+        for libvirt_vm in libvirt_vms:
+            state = libvirt_vm.state(0)
+            if state[1] == libvirt.VIR_DOMAIN_PAUSED_IOERROR:
+                vm_id = libvirt_vm.UUIDString()
+                vm_obj = _vms[vm_id]
+                if sdUUID in vm_obj.sdIds:
+                    _log.info("Cont vm %s in EIO", vm_id)
+                    vm_obj.cont()
+
+
+def on_libvirt_event(conn, dom, *args):
+    vmid = dom.UUIDString()
+    with _lock:
+        if vmid in _vms:
+            v = _vms[vmid]
+        else:
+            raise UnknownVM()
+
+    try:
+        eventid = args[-1]
+        if eventid == libvirt.VIR_DOMAIN_EVENT_ID_LIFECYCLE:
+            event, detail = args[:-1]
+            v.onLibvirtLifecycleEvent(event, detail, None)
+        elif eventid == libvirt.VIR_DOMAIN_EVENT_ID_REBOOT:
+            v.onReboot()
+        elif eventid == libvirt.VIR_DOMAIN_EVENT_ID_RTC_CHANGE:
+            utcoffset, = args[:-1]
+            v.onRTCUpdate(utcoffset)
+        elif eventid == libvirt.VIR_DOMAIN_EVENT_ID_IO_ERROR_REASON:
+            srcPath, devAlias, action, reason = args[:-1]
+            v.onIOError(devAlias, reason, action)
+        elif eventid == libvirt.VIR_DOMAIN_EVENT_ID_GRAPHICS:
+            phase, localAddr, remoteAddr, authScheme, subject = args[:-1]
+            v.log.debug('graphics event phase '
+                        '%s localAddr %s remoteAddr %s'
+                        'authScheme %s subject %s',
+                        phase, localAddr, remoteAddr, authScheme, subject)
+            if phase == libvirt.VIR_DOMAIN_EVENT_GRAPHICS_INITIALIZE:
+                v.onConnect(remoteAddr['node'], remoteAddr['service'])
+            elif phase == libvirt.VIR_DOMAIN_EVENT_GRAPHICS_DISCONNECT:
+                v.onDisconnect(clientIp=remoteAddr['node'],
+                               clientPort=remoteAddr['service'])
+        elif eventid == libvirt.VIR_DOMAIN_EVENT_ID_WATCHDOG:
+            action, = args[:-1]
+            v.onWatchdogEvent(action)
+        else:
+            v.log.warning('unknown event for VM %s id %s args %s',
+                          vmid, eventid, args)
+
+    except:
+        _log.exception("Error running VM %s callback", vmid)
diff --git a/vdsm/API.py b/vdsm/API.py
index 479c00b..8c624ce 100644
--- a/vdsm/API.py
+++ b/vdsm/API.py
@@ -58,6 +58,7 @@
 from vdsm.config import config
 import hostdev
 from caps import PAGE_SIZE_BYTES
+import vmdict
 
 
 haClient = None  # Define here to work around pyflakes issue #13
@@ -1349,7 +1350,7 @@
         Get statistics of all running VMs.
         """
         hooks.before_get_all_vm_stats()
-        statsList = self._cif.getAllVmStats()
+        statsList = vmdict.get_all_stats()
         statsList = hooks.after_get_all_vm_stats(statsList)
         return {'status': doneCode, 'statsList': statsList}
 
diff --git a/vdsm/clientIF.py b/vdsm/clientIF.py
index 0d1e235..ae3d22c 100644
--- a/vdsm/clientIF.py
+++ b/vdsm/clientIF.py
@@ -25,8 +25,6 @@
 import time
 import threading
 import uuid
-from functools import partial
-from weakref import proxy
 from collections import defaultdict
 
 from yajsonrpc.betterAsyncore import Reactor
@@ -40,11 +38,13 @@
 import libvirt
 from vdsm import libvirtconnection
 from vdsm import concurrent
+from vdsm import response
 from vdsm import utils
 from vdsm import supervdsm
 import caps
 import blkid
 from protocoldetector import MultiProtocolAcceptor
+import vmdict
 
 from virt import migration
 from virt import recovery
@@ -80,13 +80,11 @@
         :param log: a log object to be used for this object's logging.
         :type log: :class:`logging.Logger`
         """
-        self.vmContainerLock = threading.Lock()
         self._networkSemaphore = threading.Semaphore()
         self._shutdownSemaphore = threading.Semaphore()
         self.irs = irs
         if self.irs:
-            self._contEIOVmsCB = partial(clientIF.contEIOVms, proxy(self))
-            self.irs.registerDomainStateChangeCallback(self._contEIOVmsCB)
+            self.irs.registerDomainStateChangeCallback(_cont_eio_vms)
         self.log = log
         self._recovery = True
         self.channelListener = Listener(self.log)
@@ -101,7 +99,6 @@
         else:
             self.gluster = None
         try:
-            self.vmContainer = {}
             self.lastRemoteAccess = 0
             self._enabled = True
             self._netConfigDirty = False
@@ -137,14 +134,6 @@
                 self.irs.prepareForShutdown()
             raise
 
-    def getVMs(self):
-        """
-        Get a snapshot of the currently registered VMs.
-        Return value will be a dict of {vmUUID: VM_object}
-        """
-        with self.vmContainerLock:
-            return self.vmContainer.copy()
-
     @property
     def ready(self):
         return (self.irs is None or self.irs.ready) and not self._recovery
@@ -165,28 +154,6 @@
         self.bindings['jsonrpc'].reactor.server.send(message,
                                                      config.get('addresses',
                                                                 'event_queue'))
-
-    def contEIOVms(self, sdUUID, isDomainStateValid):
-        # This method is called everytime the onDomainStateChange
-        # event is emitted, this event is emitted even when a domain goes
-        # INVALID if this happens there is nothing to do
-        if not isDomainStateValid:
-            return
-
-        libvirtCon = libvirtconnection.get()
-        libvirtVms = libvirtCon.listAllDomains(
-            libvirt.VIR_CONNECT_LIST_DOMAINS_PAUSED)
-
-        with self.vmContainerLock:
-            self.log.info("vmContainerLock acquired")
-            for libvirtVm in libvirtVms:
-                state = libvirtVm.state(0)
-                if state[1] == libvirt.VIR_DOMAIN_PAUSED_IOERROR:
-                    vmId = libvirtVm.UUIDString()
-                    vmObj = self.vmContainer[vmId]
-                    if sdUUID in vmObj.sdIds:
-                        self.log.info("Cont vm %s in EIO", vmId)
-                        vmObj.cont()
 
     @classmethod
     def getInstance(cls, irs=None, log=None, scheduler=None):
@@ -433,17 +400,10 @@
         return {'status': doneCode, 'alignment': aligning}
 
     def createVm(self, vmParams, vmRecover=False):
-        with self.vmContainerLock:
-            if not vmRecover:
-                if vmParams['vmId'] in self.vmContainer:
-                    return errCode['exist']
-            vm = Vm(self, vmParams, vmRecover)
-            self.vmContainer[vmParams['vmId']] = vm
-        vm.run()
-        return {'status': doneCode, 'vmList': vm.status()}
-
-    def getAllVmStats(self):
-        return [v.getStats() for v in self.vmContainer.values()]
+        v = Vm(self, vmParams, vmRecover)
+        vmdict.add(v)
+        v.run()
+        return response.success(vmList=v.status())
 
     def createStompClient(self, client_socket):
         if 'jsonrpc' in self.bindings:
@@ -495,52 +455,10 @@
             self.log.exception("recovery: failed")
             raise
 
-    def dispatchLibvirtEvents(self, conn, dom, *args):
-        try:
-            eventid = args[-1]
-            vmid = dom.UUIDString()
-            v = self.vmContainer.get(vmid)
-
-            if not v:
-                self.log.debug('unknown vm %s eventid %s args %s',
-                               vmid, eventid, args)
-                return
-
-            if eventid == libvirt.VIR_DOMAIN_EVENT_ID_LIFECYCLE:
-                event, detail = args[:-1]
-                v.onLibvirtLifecycleEvent(event, detail, None)
-            elif eventid == libvirt.VIR_DOMAIN_EVENT_ID_REBOOT:
-                v.onReboot()
-            elif eventid == libvirt.VIR_DOMAIN_EVENT_ID_RTC_CHANGE:
-                utcoffset, = args[:-1]
-                v.onRTCUpdate(utcoffset)
-            elif eventid == libvirt.VIR_DOMAIN_EVENT_ID_IO_ERROR_REASON:
-                srcPath, devAlias, action, reason = args[:-1]
-                v.onIOError(devAlias, reason, action)
-            elif eventid == libvirt.VIR_DOMAIN_EVENT_ID_GRAPHICS:
-                phase, localAddr, remoteAddr, authScheme, subject = args[:-1]
-                v.log.debug('graphics event phase '
-                            '%s localAddr %s remoteAddr %s'
-                            'authScheme %s subject %s',
-                            phase, localAddr, remoteAddr, authScheme, subject)
-                if phase == libvirt.VIR_DOMAIN_EVENT_GRAPHICS_INITIALIZE:
-                    v.onConnect(remoteAddr['node'], remoteAddr['service'])
-                elif phase == libvirt.VIR_DOMAIN_EVENT_GRAPHICS_DISCONNECT:
-                    v.onDisconnect(clientIp=remoteAddr['node'],
-                                   clientPort=remoteAddr['service'])
-            elif eventid == libvirt.VIR_DOMAIN_EVENT_ID_WATCHDOG:
-                action, = args[:-1]
-                v.onWatchdogEvent(action)
-            else:
-                v.log.warning('unknown eventid %s args %s', eventid, args)
-
-        except:
-            self.log.error("Error running VM callback", exc_info=True)
-
     def _waitForDomainsUp(self):
         while self._enabled:
             launching = sum(int(v.lastStatus == vmstatus.WAIT_FOR_LAUNCH)
-                            for v in self.vmContainer.values())
+                            for v in vmdict.get_all_vms())
             if not launching:
                 break
             else:
@@ -550,13 +468,13 @@
             time.sleep(1)
 
     def _waitForStoragePool(self):
-        while (self._enabled and self.vmContainer and
+        while (self._enabled and vmdict.num_vms() > 0 and
                not self.irs.getConnectedStoragePoolsList()['poollist']):
             self.log.info('recovery: waiting for storage pool to go up')
             time.sleep(5)
 
     def _preparePathsForRecoveredVMs(self):
-        vm_objects = self.vmContainer.values()
+        vm_objects = vmdict.get_all_vms()
         num_vm_objects = len(vm_objects)
         for idx, vm_obj in enumerate(vm_objects):
             # Let's recover as much VMs as possible
@@ -572,3 +490,16 @@
                 self.log.exception(
                     "recovery [%d/%d]: failed for vm %s",
                     idx+1, num_vm_objects, vm_obj.id)
+
+
+def _cont_eio_vms(sdUUID, isDomainStateValid):
+    # This method is called everytime the onDomainStateChange
+    # event is emitted, this event is emitted even when a domain goes
+    # INVALID if this happens there is nothing to do
+    if not isDomainStateValid:
+        return
+
+    libvirtCon = libvirtconnection.get()
+    libvirtVms = libvirtCon.listAllDomains(
+        libvirt.VIR_CONNECT_LIST_DOMAINS_PAUSED)
+    vmdict.on_eio_cont(libvirtVms, sdUUID)
diff --git a/vdsm/virt/periodic.py b/vdsm/virt/periodic.py
index dee1b92..b3d8dc5 100644
--- a/vdsm/virt/periodic.py
+++ b/vdsm/virt/periodic.py
@@ -29,6 +29,7 @@
 
 from vdsm import executor
 from vdsm import libvirtconnection
+from vdsm import vmdict
 from vdsm.config import config
 
 from . import hoststats
@@ -66,8 +67,8 @@
     _executor.start()
 
     def per_vm_operation(func, period):
-        disp = VmDispatcher(
-            cif.getVMs, _executor, func, _timeout_from(period))
+        disp = VmDispatcher(vmdict.get_all_items,
+                            _executor, func, _timeout_from(period))
         return Operation(disp, period, scheduler)
 
     _operations = [
@@ -93,7 +94,7 @@
         Operation(
             sampling.VMBulkSampler(
                 libvirtconnection.get(cif),
-                cif.getVMs,
+                vmdict.get_all_items,
                 sampling.stats_cache),
             config.getint('vars', 'vm_sample_interval'),
             scheduler),
diff --git a/vdsm/virt/vm.py b/vdsm/virt/vm.py
index df58d1b..16eb406 100644
--- a/vdsm/virt/vm.py
+++ b/vdsm/virt/vm.py
@@ -47,6 +47,7 @@
 from vdsm import response
 from vdsm import supervdsm
 from vdsm import utils
+from vdsm import vmdict
 from vdsm.compat import pickle
 from vdsm.config import config
 from vdsm.define import ERROR, NORMAL, doneCode, errCode
@@ -3863,13 +3864,13 @@
         Clean VM from the system
         """
         try:
-            del self.cif.vmContainer[self.id]
-        except KeyError:
+            vmdict.remove(self)
+        except vmdict.UnknownVM:
             self.log.exception("Failed to delete VM %s", self.id)
         else:
             self._cleanupRecoveryFile()
             self.log.debug("Total desktops after destroy of %s is %d",
-                           self.conf['vmId'], len(self.cif.vmContainer))
+                           self.conf['vmId'], vmdict.num_vms())
 
     def destroy(self):
         self.log.debug('destroy Called')


-- 
To view, visit https://gerrit.ovirt.org/53101
To unsubscribe, visit https://gerrit.ovirt.org/settings

Gerrit-MessageType: newchange
Gerrit-Change-Id: Iacd2ae6c5e9ca6a73c0fed978c78c9ebb001c46d
Gerrit-PatchSet: 3
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Francesco Romani <fromani at redhat.com>
Gerrit-Reviewer: gerrit-hooks <automation at ovirt.org>


More information about the vdsm-patches mailing list