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