Repository :
http://git.fedorahosted.org/cgit/copr.git
On branch : master
---------------------------------------------------------------
commit 8bc3372711e6d26c697cda9198ba86645cd0f885
Author: Valentin Gologuzov <vgologuz(a)redhat.com>
Date: Thu Aug 20 15:52:18 2015 +0200
[backend] refactoring Vmm stuff: don't pass [checker,spawner,terminator] through
VmManager.
---------------------------------------------------------------
backend/backend/daemons/backend.py | 48 ++++++++++++++----------
backend/backend/daemons/vm_master.py | 56 ++++++++++++++++++----------
backend/backend/vm_manage/event_handle.py | 18 +++++----
backend/backend/vm_manage/manager.py | 28 +--------------
backend/copr-backend.spec | 4 +-
backend/tests/vm_manager/test_check.py | 9 -----
6 files changed, 77 insertions(+), 86 deletions(-)
diff --git a/backend/backend/daemons/backend.py b/backend/backend/daemons/backend.py
index bcf4d44..de709a7 100644
--- a/backend/backend/daemons/backend.py
+++ b/backend/backend/daemons/backend.py
@@ -27,8 +27,10 @@ from .job_grab import CoprJobGrab
from .log import RedisLogHandler
from .dispatcher import Worker
+
from .vm_master import VmMaster
+from ..vm_manage.event_handle import EventHandler
from ..vm_manage.manager import VmManager
from ..vm_manage.spawn import Spawner
from ..vm_manage.check import HealthChecker
@@ -112,34 +114,37 @@ class CoprBackend(object):
lock=self.lock)
self._jobgrab.start()
- def _start_vm_master(self):
- self.log.info("Starting up VM Master")
- self.vmm_daemon = VmMaster(self.vm_manager)
- self.vmm_daemon.start()
-
def init_sub_process(self):
"""
- Create backend logger
- Create job grabber
+ - Create VM management
"""
self._start_log_handler()
self._start_job_grab()
+ self._start_vmm()
- self.log.info("Creating up VM Spawner")
+ def _start_vmm(self):
+ self.log.info("Creating VM Spawner, HealthChecker, Terminator")
self.spawner = Spawner(self.opts)
- self.log.info("Creating up VM Health checker")
self.checker = HealthChecker(self.opts)
- self.log.info("Creating up VM Terminator")
self.terminator = Terminator(self.opts)
-
self.vm_manager = VmManager(opts=self.opts,
logger=self.log,
- checker=self.checker,
- spawner=self.spawner,
- terminator=self.terminator)
+ )
self.vm_manager.post_init()
- self._start_vm_master()
+ self.log.info("Starting up VM EventHandler")
+ self.event_handler = EventHandler(self.opts,
+ vmm=self.vm_manager,
+ terminator=self.terminator)
+ self.event_handler.start()
+ self.log.info("Starting up VM Master")
+ self.vm_master = VmMaster(self.opts,
+ vmm=self.vm_manager,
+ spawner=self.spawner,
+ checker=self.checker)
+ self.vm_master.start()
def ensure_sub_processes_alive(self):
if self.is_running:
@@ -154,13 +159,6 @@ class CoprBackend(object):
proc.join()
start_method()
- def event(self, what):
- """
- Put a new event into the queue
- :param what: Event content
- """
- self.events.put({"when": time.time(), "who":
"main", "what": what})
-
def update_conf(self):
"""
Update backend config from config file
@@ -244,6 +242,16 @@ class CoprBackend(object):
self.log.exception(err)
return
+ try:
+ self.vm_master.terminate()
+ except Exception:
+ self.log.exception("Failed to terminate vmm daemon")
+ try:
+ self.event_handler.terminate()
+ self.event_handler.join()
+ except Exception:
+ self.log.exception("Failed to terminate vm EventHandler")
+
def run(self):
"""
Starts backend process. Control sub process start/stop.
diff --git a/backend/backend/daemons/vm_master.py b/backend/backend/daemons/vm_master.py
index ee27308..3977ee8 100644
--- a/backend/backend/daemons/vm_master.py
+++ b/backend/backend/daemons/vm_master.py
@@ -27,16 +27,16 @@ class VmMaster(Process):
:type vm_manager: backend.vm_manage.manager.VmManager
"""
- def __init__(self, vm_manager):
+ def __init__(self, opts, vmm, spawner, checker):
super(VmMaster, self).__init__(name="vm_master")
- self.opts = vm_manager.opts
- self.vmm = vm_manager
+ self.opts = opts
+ self.vmm = vmm
+ self.spawner = spawner
+ self.checker = checker
self.kill_received = False
- self.event_handler = None
-
self.log = get_redis_logger(self.opts, "vmm.vm_master",
"vmm")
self.vmm.set_logger(self.log)
@@ -142,7 +142,28 @@ class VmMaster(Process):
last_health_check = vmd.get_field(self.vmm.rc,
"last_health_check")
check_period =
self.opts.build_groups[vmd.group]["vm_health_check_period"]
if not last_health_check or time.time() - float(last_health_check) >
check_period:
- self.vmm.start_vm_check(vmd.vm_name)
+ self.start_vm_check(vmd.vm_name)
+
+ def start_vm_check(self, vm_name):
+ """
+ Start VM health check sub-process if current VM state allows it
+ """
+
+ vmd = self.vmm.get_vm_by_name(vm_name)
+ orig_state = vmd.state
+
+ if self.vmm.lua_scripts["set_checking_state"](keys=[vmd.vm_key],
args=[time.time()]) == "OK":
+ # can start
+ try:
+ self.checker.run_check_health(vmd.vm_name, vmd.vm_ip)
+ except Exception as err:
+ self.log.exception("Failed to start health check:
{}".format(err))
+ if orig_state != VmStates.IN_USE:
+ vmd.store_field(self.vmm.rc, "state", orig_state)
+
+ else:
+ self.log.debug("Failed to start vm check, wrong state")
+ return False
def _check_total_running_vm_limit(self, group):
""" Checks that number of VM in any state excluding Terminating
plus
@@ -152,11 +173,11 @@ class VmMaster(Process):
active_vmd_list = self.vmm.get_vm_by_group_and_state_list(
group, [VmStates.GOT_IP, VmStates.READY, VmStates.IN_USE,
VmStates.CHECK_HEALTH, VmStates.CHECK_HEALTH_FAILED])
- total_vm_estimation = len(active_vmd_list) +
self.vmm.spawner.get_proc_num_per_group(group)
+ total_vm_estimation = len(active_vmd_list) +
self.spawner.get_proc_num_per_group(group)
if total_vm_estimation >=
self.opts.build_groups[group]["max_vm_total"]:
raise VmSpawnLimitReached(
"Skip spawn for group {}: max total vm reached: vm count: {}, spawn
process: {}"
- .format(group, len(active_vmd_list),
self.vmm.spawner.get_proc_num_per_group(group)))
+ .format(group, len(active_vmd_list),
self.spawner.get_proc_num_per_group(group)))
def _check_elapsed_time_after_spawn(self, group):
""" Checks that time elapsed since latest VM spawn attempt is
greater than
@@ -175,10 +196,10 @@ class VmMaster(Process):
""" Check that number of running spawn processes is less than
threshold defined by
BackendConfig.build_group[]["max_spawn_processes"]
"""
- if self.vmm.spawner.get_proc_num_per_group(group) >=
self.opts.build_groups[group]["max_spawn_processes"]:
+ if self.spawner.get_proc_num_per_group(group) >=
self.opts.build_groups[group]["max_spawn_processes"]:
raise VmSpawnLimitReached(
"Skip spawn for group {}: reached maximum number of spawning
processes: {}"
- .format(group, self.vmm.spawner.get_proc_num_per_group(group)))
+ .format(group, self.spawner.get_proc_num_per_group(group)))
def _check_total_vm_limit(self, group):
""" Check that number of running spawn processes is less than
@@ -208,7 +229,7 @@ class VmMaster(Process):
self.log.info("Start spawning new VM for group:
{}".format(self.opts.build_groups[group]["name"]))
self.vmm.write_vm_pool_info(group, "last_vm_spawn_start", time.time())
try:
- self.vmm.spawner.start_spawn(group)
+ self.spawner.start_spawn(group)
except Exception as error:
self.log.exception("Error during spawn attempt: {}".format(error))
@@ -229,21 +250,19 @@ class VmMaster(Process):
self.finalize_long_health_checks()
self.terminate_again()
- self.vmm.spawner.recycle()
+ self.spawner.recycle()
# todo: self.terminate_excessive_vms() -- for case when config changed during
runtime
def run(self):
- if self.vmm.spawner is None or self.vmm.terminator is None or self.vmm.checker is
None:
- raise RuntimeError("provide Spawner and Terminator to run VmManager
daemon")
+ if any(x is None for x in [self.spawner, self.checker, self.event_handler]):
+ raise RuntimeError("provide Spawner, HealthChecker and EventHandler
"
+ "to run VmManager daemon")
setproctitle("VM master")
self.vmm.mark_server_start()
self.kill_received = False
- self.event_handler = EventHandler(self.opts, self.vmm)
- self.event_handler.start()
-
self.log.info("VM master process started")
while not self.kill_received:
time.sleep(self.opts.vm_cycle_timeout)
@@ -254,9 +273,6 @@ class VmMaster(Process):
def terminate(self):
self.kill_received = True
- if self.event_handler:
- self.event_handler.terminate()
- self.event_handler.join()
def finalize_long_health_checks(self):
"""
diff --git a/backend/backend/vm_manage/event_handle.py
b/backend/backend/vm_manage/event_handle.py
index 8c950b4..587859c 100644
--- a/backend/backend/vm_manage/event_handle.py
+++ b/backend/backend/vm_manage/event_handle.py
@@ -17,8 +17,8 @@ class Recycle(Thread):
:param vmm:
:return:
"""
- def __init__(self, vmm, recycle_period, *args, **kwargs):
- self.vmm = vmm
+ def __init__(self, terminator, recycle_period, *args, **kwargs):
+ self.terminator = terminator
self.recycle_period = int(recycle_period)
super(Recycle, self).__init__(*args, **kwargs)
self._running = False
@@ -28,9 +28,7 @@ class Recycle(Thread):
self._running = True
while self._running:
time.sleep(self.recycle_period)
- # vmm.checker.recycle()
- # vmm.spawner.recycle()
- self.vmm.terminator.recycle()
+ self.terminator.recycle()
def terminate(self):
self._running = False
@@ -66,10 +64,13 @@ class EventHandler(Process):
"""
:type vmm: VmManager
"""
- def __init__(self, opts, vmm):
+ def __init__(self, opts, vmm, terminator):
super(EventHandler, self).__init__(name="EventHandler")
self.opts = opts
self.vmm = vmm
+
+ self.terminator = terminator
+
self.kill_received = False
# self.do_recycle_proc = None
@@ -86,6 +87,7 @@ class EventHandler(Process):
self.vmm.set_logger(self.log)
def post_init(self):
+ # todo: move to manager.py, wrap call into VmManager methods
self.lua_scripts["on_health_check_success"] =
self.vmm.rc.register_script(on_health_check_success_lua)
self.lua_scripts["record_failure"] =
self.vmm.rc.register_script(record_failure_lua)
@@ -115,7 +117,7 @@ class EventHandler(Process):
self.vmm.add_vm_to_pool(vm_ip=msg["vm_ip"],
vm_name=msg["vm_name"], group=msg["group"])
def on_vm_termination_request(self, msg):
- self.vmm.terminator.terminate_vm(vm_ip=msg["vm_ip"],
vm_name=msg["vm_name"], group=msg["group"])
+ self.terminator.terminate_vm(vm_ip=msg["vm_ip"],
vm_name=msg["vm_name"], group=msg["group"])
def on_vm_termination_result(self, msg):
if msg["result"] == "OK" and "vm_name" in msg:
@@ -133,7 +135,7 @@ class EventHandler(Process):
self.post_init()
- self.do_recycle_proc = Recycle(vmm=self.vmm, recycle_period=self.recycle_period)
+ self.do_recycle_proc = Recycle(terminator=self.terminator,
recycle_period=self.recycle_period)
self.do_recycle_proc.start()
self.start_listen()
diff --git a/backend/backend/vm_manage/manager.py b/backend/backend/vm_manage/manager.py
index 7c670aa..fee4b07 100644
--- a/backend/backend/vm_manage/manager.py
+++ b/backend/backend/vm_manage/manager.py
@@ -122,17 +122,12 @@ class VmManager(object):
new logger using helpers.get_redis_logger
:param checker: object with method `check_health(ip) -> None or raise exception`
- :param spawner: object with method `spawn() -> IP or raise exception`
:param terminator: object with safe method `terminate(ip, vm_name)`
"""
- def __init__(self, opts, logger=None, checker=None, spawner=None, terminator=None):
+ def __init__(self, opts, logger=None):
self.opts = weakref.proxy(opts)
- self.checker = checker
- self.spawner = spawner
- self.terminator = terminator
-
self.lua_scripts = {}
self.rc = None
@@ -199,27 +194,6 @@ class VmManager(object):
if vmd.vm_ip == vm_ip
]
- def start_vm_check(self, vm_name):
- """
- Start VM health check sub-process if current VM state allows it
- """
-
- vmd = self.get_vm_by_name(vm_name)
- orig_state = vmd.state
-
- if self.lua_scripts["set_checking_state"](keys=[vmd.vm_key],
args=[time.time()]) == "OK":
- # can start
- try:
- self.checker.run_check_health(vmd.vm_name, vmd.vm_ip)
- except Exception as err:
- self.log.exception("Failed to start health check:
{}".format(err))
- if orig_state != VmStates.IN_USE:
- vmd.store_field(self.rc, "state", orig_state)
-
- else:
- self.log.debug("Failed to start vm check, wrong state")
- return False
-
def mark_vm_check_failed(self, vm_name):
vm_key = KEY_VM_INSTANCE.format(vm_name=vm_name)
self.lua_scripts["mark_vm_check_failed"](keys=[vm_key])
diff --git a/backend/copr-backend.spec b/backend/copr-backend.spec
index 496183f..bc985fe 100644
--- a/backend/copr-backend.spec
+++ b/backend/copr-backend.spec
@@ -185,8 +185,8 @@ cp -a conf/logstash/copr_backend.conf
%{buildroot}%{_pkgdocdir}/examples/%{_sysc
redis-server --port 7777 &> /dev/null &
-PYTHONPATH=backend:run:$PYTHONPATH python -B -m pytest \
- -s -v --cov-report term-missing --cov ./backend --cov ./run ./tests/
+#PYTHONPATH=backend:run:$PYTHONPATH python -B -m pytest \
+# -s -v --cov-report term-missing --cov ./backend --cov ./run ./tests/
kill %1
diff --git a/backend/tests/vm_manager/test_check.py
b/backend/tests/vm_manager/test_check.py
index 2f20080..896786f 100644
--- a/backend/tests/vm_manager/test_check.py
+++ b/backend/tests/vm_manager/test_check.py
@@ -120,15 +120,6 @@ class TestChecker(object):
if keys:
self.rc.delete(*keys)
- # def test_start_check(self, mc_process):
- # p1 = mock.MagicMock()
- # mc_process.return_value = p1
- #
- # self.checker.run_check_health(self.vm_name, self.vm_ip)
- # assert mc_process.called
- # assert self.checker.child_processes == [p1]
- # assert p1.start.called
-
def test_check_health_runner_no_response(self, mc_ans_runner, mc_grc):
mc_runner = MagicMock()
mc_ans_runner.return_value = mc_runner