Change in vdsm[master]: WIP: sample in a separate process

fromani at redhat.com fromani at redhat.com
Mon Jan 19 15:41:01 UTC 2015


Francesco Romani has uploaded a new change for review.

Change subject: WIP: sample in a separate process
......................................................................

WIP: sample in a separate process

DO NOT MERGE WORK IN PROGRESS

this patch is meant as live demo, it
is NOT done and NOT ready for review.

Change-Id: Iaff616a65f7e816a52ce7f5db5a5dfddd7770175
Signed-off-by: Francesco Romani <fromani at redhat.com>
---
M lib/vdsm/config.py.in
M tests/vmfakelib.py
M vdsm/virt/sampling.py
M vdsm/virt/vm.py
M vdsm/virt/vmstats.py
5 files changed, 155 insertions(+), 80 deletions(-)


  git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/82/37082/1

diff --git a/lib/vdsm/config.py.in b/lib/vdsm/config.py.in
index aa9e7f5..b76de8d 100644
--- a/lib/vdsm/config.py.in
+++ b/lib/vdsm/config.py.in
@@ -145,8 +145,9 @@
         ('vm_watermark_interval', '2',
             'How often should we sample each vm for statistics (seconds).'),
 
-        ('vm_sample_bulk', 'false',
-            'Use libvirt bulk stats to sampling. Requires libvirt >= 1.2.9'),
+        ('vm_sample_process', 'false',
+            'Use separate process and libvirt bulk stats forsampling. '
+            'Requires libvirt >= 1.2.9'),
 
         ('vm_sample_cpu_interval', '15', None),
 
diff --git a/tests/vmfakelib.py b/tests/vmfakelib.py
index 42c6f1f..607cc84 100644
--- a/tests/vmfakelib.py
+++ b/tests/vmfakelib.py
@@ -228,6 +228,9 @@
         self._dom = dom
         self._bulkStatsFormat = bulkStatsFormat
 
+    def refresh(self, now):
+        return True
+
     def getLastSampleTime(self):
         return time.time()
 
diff --git a/vdsm/virt/sampling.py b/vdsm/virt/sampling.py
index 92593ef..172c1d6 100644
--- a/vdsm/virt/sampling.py
+++ b/vdsm/virt/sampling.py
@@ -25,16 +25,18 @@
 from collections import deque
 import errno
 import logging
+import multiprocessing
 import os
 import re
 import threading
 import time
 
+import libvirt
+
 from vdsm.config import config
 from vdsm.constants import P_VDSM_RUN, P_VDSM_CLIENT_LOG
 from vdsm import executor
 from vdsm import ipwrapper
-from vdsm import libvirtconnection
 from vdsm import netinfo
 from vdsm import schedule
 from vdsm import utils
@@ -52,29 +54,11 @@
 _SAMPLING_TASKS = 100  # just 'high enough' number
 
 
-_scheduler = None
+_pipe = None
 
-_executor = None
+_proc = None
 
-_sampler = None
-
-
-def _dispatch_stats(vm, stats):
-    # vm.stats may be None due a benign race with the creation thread.
-    # Don't worry and go ahead.
-    if vm.stats:
-        vm.stats.sampleBulk.append(stats)
-        vm.stats.last_sample_time = time.time()
-
-
-def _collect_stats(vm, bulk_stats):
-    vm.stats.sampleVmJobs()
-    vm.stats.sampleCpuTune()
-
-    vm.extendDrivesIfNeeded(bulk_stats)
-
-    for vmDrive in vm.getDiskDevices():
-        vm.updateDriveVolume(vmDrive)
+_stats = {}
 
 
 class Sampler(object):
@@ -84,13 +68,12 @@
 
     SAMPLE_DELAY = 2  # seconds. must be > SAMPLE_TIMEOUT
 
-    def __init__(self, cif,
-                 conn=None, scheduler=_scheduler, executor=_executor,
+    def __init__(self, scheduler, executor,
                  timefn=utils.monotonic_time):
-        self._cif = cif
-        self._conn = libvirtconnection.get(cif) if conn is None else conn
         self._scheduler = scheduler
         self._executor = executor
+        self._timefn = timefn
+        self._conn = libvirt.openReadOnly('qemu:///system')
         self._skip_doms = ExpiringSet(self.STUCK_TIMEOUT)
         self._log = logging.getLogger("sampling.sampler")
         self._lock = threading.Lock()
@@ -98,6 +81,7 @@
         self._call = None
         self._sampling = False
         self._last_sampling_time = None
+        self._bulk = SampleWindow(2)
 
     def start(self):
         with self._lock:
@@ -117,19 +101,32 @@
                     self._call = None
                 self._iterator = None
 
+    def get(self, vm_id):
+        if vm_id in self._skip_doms:
+            self._log.debug('vm %s is not responsive: skipping', vm_id)
+            return ({}, {}, 1.0)
+        self._log.debug('vm %s is responsive', vm_id)
+        first_sample, last_sample, interval = self._bulk.stats()
+        res = (first_sample.get(vm_id, {}),
+               last_sample.get(vm_id, {}),
+               interval)
+        self._log.debug('getting stats for %s:\n%s', vm_id, res)
+        return res
+
     # Task interface
 
     def __call__(self):
         try:
-            vms = self._cif.getVMs()
             sampling_time = self._timefn()
 
             if not self._sampling:
                 self._sampling = True
                 bulk_stats = self._conn.getAllDomainStats()
+                self._log.debug('sampling all domains')
             else:
                 # preempted a stuck sampling call
-                doms = _get_responsive_doms(vms, self._skip_doms)
+                doms = _get_responsive_doms(self._conn.listAllDomains(),
+                                            self._skip_doms)
                 self._log.debug('sampling %d domains - %d skipped',
                                 len(doms), len(self._skip_doms))
                 if doms:
@@ -139,7 +136,9 @@
 
             if self._is_fresher(sampling_time):
                 self._log.debug('updating %d stats', len(bulk_stats))
-                _put_stats_into_vms(vms, self._skip_doms, bulk_stats)
+
+                self._bulk.append(_update_stats(bulk_stats, self._skip_doms))
+
             else:
                 self._log.warning('skipped update with stale data')
         except Exception:
@@ -176,62 +175,121 @@
             self.SAMPLE_DELAY, self._dispatch)
 
 
-def _get_responsive_doms(vms, skip_doms):
+def _get_responsive_doms(all_doms, skip_doms):
     doms = []
-    for vmId, vmObj in vms.iteritems():
+    for dom in all_doms:
+        vmId = dom.UUIDString()
         if vmId in skip_doms:
             continue
-        elif not vmObj.isDomainResponsive():
-            skip_doms.add(vmId)
-            vmObj.setUnresponsive()
         else:
-            doms.append(vmObj._dom._dom)
+            state, details, stateTime = dom.controlInfo()
+            if state != libvirt.VIR_DOMAIN_CONTROL_OK:
+                skip_doms.add(vmId)
+            else:
+                doms.append(dom)
 
     return doms
 
 
-def _put_stats_into_vms(vms, skip_doms, bulk_stats):
+def _update_stats(bulk_stats, skip_doms):
+    snapshot = {}
     for dom, stats in bulk_stats:
-        vmObj = vms.get(dom.UUIDString(), None)
-        if vmObj is None:
-            continue
-
-        skip_doms.remove(vmObj.id)
-        vmObj.setResponsive()
-
-        _dispatch_stats(vmObj, stats)
-        # special case handled last
-        _collect_stats(vmObj, bulk_stats)
+        vmId = dom.UUIDString()
+        skip_doms.remove(vmId)
+        snapshot[vmId] = stats
+    logging.debug('snapshot: %s', snapshot)
+    return snapshot
 
 
 def start(cif):
     """ Called during application startup """
-    global _executor
-    global _sampler
-    global _scheduler
+    global _pipe
+    global _proc
 
-    if config.getboolean('vars', 'vm_sample_bulk'):
-        sched = schedule.Scheduler("sampling.scheduler")
-        sched.start()
-        _scheduler = sched
+    if config.getboolean('vars', 'vm_sample_process'):
+        cmd, res = multiprocessing.Queue(), multiprocessing.Queue()
 
-        exc = executor.Executor(
-            _SAMPLING_THREADS, _SAMPLING_TASKS, sched)
-        exc.start()
-        _executor = exc
+        proc = multiprocessing.Process(target=_sampling, args=(cmd, res))
+        proc.start()
 
-        _sampler = Sampler(cif, None, sched, exc)
-        _sampler.start()
+    _pipe = (cmd, res)
+    _proc = proc
 
 
 def stop():
     """ Called during application shutdown """
-    if _sampler:
-        _sampler.stop()
-    if _executor:
-        _executor.stop(wait=True)
-    if _scheduler:
-        _scheduler.stop()
+    if _proc:
+        _proc.terminate()
+
+
+def _sampling(cmd, res):
+    LOG_CONF_PATH = "/etc/vdsm/sampling.logger.conf"
+
+    try:
+        logging.config.fileConfig(LOG_CONF_PATH, disable_existing_loggers=True)
+    except:
+        logging.basicConfig(filename='/dev/stdout', filemode='w+',
+                            level=logging.DEBUG)
+        log = logging.getLogger("SuperVdsm.Server")
+        log.warn("Could not init proper logging", exc_info=True)
+
+    sched = schedule.Scheduler("sampling.scheduler")
+    sched.start()
+
+    exc = executor.Executor(
+        _SAMPLING_THREADS, _SAMPLING_TASKS, sched)
+    exc.start()
+
+    smp = Sampler(sched, exc)
+    smp.start()
+
+    logging.debug('Sampling loop starts')
+
+    while True:
+        try:
+            vmId = cmd.get()
+        except EOFError:
+            break
+        else:
+            logging.debug('sending stats for: %s', vmId)
+            res.put(smp.get(vmId))
+
+    logging.debug('Sampling loop ends')
+
+    smp.stop()
+    exc.stop(wait=True)
+    sched.stop()
+
+
+def _get_cached_stats(vmId, now):
+    try:
+        timestamp, bulk_stats = _stats[vmId]
+    except KeyError:
+        return None
+    else:
+        if (now - timestamp) >= Sampler.SAMPLE_DELAY:
+            return None
+        return bulk_stats
+
+
+def _update_cached_stats(vmId, now):
+    cmd, res = _pipe
+    logging.debug('asking sampling data for: %s', vmId)
+    cmd.put(vmId)
+    bulk_stats = res.get()
+    logging.debug('received sampling data: %s', bulk_stats)
+    _stats[vmId] = (now, bulk_stats)
+    return bulk_stats
+
+
+def get(vmId, now):
+    stats = _get_cached_stats(vmId, now)
+    if stats is None:
+        logging.debug('updating cached data for %s', vmId)
+        stats = _update_cached_stats(vmId, now)
+    else:
+        logging.debug('using cached data for %s', vmId)
+    return stats
 
 
 class InterfaceSample(object):
diff --git a/vdsm/virt/vm.py b/vdsm/virt/vm.py
index 6150fd9..7d39fda 100644
--- a/vdsm/virt/vm.py
+++ b/vdsm/virt/vm.py
@@ -67,8 +67,8 @@
 from .vmtune import io_tune_values_to_dom, io_tune_dom_to_values
 from . import vmxml
 
+from . import sampling
 from .sampling import AdvancedStatsFunction, AdvancedStatsThread
-from .sampling import SampleWindow
 from .sampling import translateCpuSample, translateNetSample
 from .sampling import translateDiskSample, translateVcpuSample
 from .utils import isVdsmImage
@@ -463,6 +463,9 @@
     def getSampleCpuTune(self):
         return self.sampleCpuTune.getLastSample()
 
+    def refresh(self, now):
+        return True
+
     def handleStatsException(self, ex):
         # We currently handle only libvirt exceptions
         if not hasattr(ex, "get_error_code"):
@@ -495,10 +498,9 @@
 
         self.last_sample_time = None
 
-        self.sampleBulk = SampleWindow(self.SAMPLE_NUM)
-
-        self._sampleVmJobs = SampleWindow(self.SAMPLE_NUM)
-        self._sampleCpuTune = SampleWindow(self.SAMPLE_NUM)
+        self._bulkStats = None
+        self._sampleVmJobs = sampling.SampleWindow(self.SAMPLE_NUM)
+        self._sampleCpuTune = sampling.SampleWindow(self.SAMPLE_NUM)
 
     def sampleVmJobs(self):
         # compatibility with VmStatsThread
@@ -541,26 +543,32 @@
         self._sampleCpuTune.append(infos)
 
     # vmstats.Collector interface
+    def refresh(self, now):
+        self.last_sample_time = now
+        self._bulkStats = sampling.get(self._vm.id, now)
+        return bool(self._bulkStats)
+
     def getLastSampleTime(self):
         return self.last_sample_time
 
     def getSampleCpu(self):
-        return self.sampleBulk.stats()
+        return self._bulkStats
 
     def getSampleNet(self):
-        return self.sampleBulk.stats()
+        return self._bulkStats
 
     def getSampleDisk(self):
-        return self.sampleBulk.stats()
+        return self._bulkStats
 
     def getSampleBalloon(self):
-        return self.sampleBulk.stats()
+        return self._bulkStats
 
     def getSampleVmJobs(self):
         return self._sampleVmJobs.last()
 
     def getSampleVcpu(self):
-        return self.sampleBulk.last()
+        first_sample, last_sample, interval = self._bulkStats
+        return first_sample or last_sample
 
     def getSampleCpuTune(self):
         return self._sampleCpuTune.last()
@@ -1907,7 +1915,7 @@
         return domxml.toxml()
 
     def _initVmStats(self):
-        if config.getboolean('vars', 'vm_sample_bulk'):
+        if config.getboolean('vars', 'vm_sample_process'):
             self.stats = VmStats(self)
         else:
             self.stats = VmStatsThread(self)
diff --git a/vdsm/virt/vmstats.py b/vdsm/virt/vmstats.py
index d07677d..5fe734f 100644
--- a/vdsm/virt/vmstats.py
+++ b/vdsm/virt/vmstats.py
@@ -47,12 +47,18 @@
     def get(self):
         stats = {}
 
+        now = time.time()
         try:
             stats['statsAge'] = (
-                time.time() - self._collector.getLastSampleTime())
+                now - self._collector.getLastSampleTime())
         except TypeError:
             self._log.debug("Stats age not available")
             stats['statsAge'] = -1.0
+
+        if self._collector.refresh(now):
+            self._vm.setResponsive()
+        else:
+            self._vm.setUnresponsive()
 
         sCpu, eCpu, cpuInterval = self._collector.getSampleCpu()
         self._getCpuStats(stats, sCpu, eCpu, cpuInterval)
@@ -71,7 +77,6 @@
         self._getCpuTuneInfo(stats, cpuTuneSample)
         self._getCpuCount(stats, cpuTuneSample)
         self._getUserCpuTuneInfo(stats, cpuTuneSample)
-
         self._getIoTuneStats(stats)
 
         return stats


-- 
To view, visit http://gerrit.ovirt.org/37082
To unsubscribe, visit http://gerrit.ovirt.org/settings

Gerrit-MessageType: newchange
Gerrit-Change-Id: Iaff616a65f7e816a52ce7f5db5a5dfddd7770175
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Francesco Romani <fromani at redhat.com>


More information about the vdsm-patches mailing list