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