vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Lint BuildManager

Browse source
This commit is contained in:
Jimmy Zelinskie 2014-11-18 15:45:56 -05:00
parent 043a30ee96
commit 6df6f28edf
9 changed files with 187 additions and 173 deletions
Download patch file Download diff file Expand all files Collapse all files

63
buildman/server.py
View file

@ -2,24 +2,23 @@ import logging
import trollius
from autobahn.asyncio.wamp import RouterFactory, RouterSessionFactory
from autobahn.asyncio.websocket import WampWebSocketServerFactory, WampWebSocketServerProtocol
from autobahn.asyncio.websocket import WampWebSocketServerFactory
from autobahn.wamp import types
from autobahn.wamp.exception import ApplicationError
from aiowsgi import create_server as create_wsgi_server
from flask import Flask
from threading import Event, Lock
from threading import Event
from trollius.coroutines import From
from buildjob import BuildJob, BuildJobLoadException
from buildman.buildjob import BuildJob, BuildJobLoadException
logger = logging.getLogger(__name__)
LOGGER = logging.getLogger(__name__)
WORK_CHECK_TIMEOUT = 10
TIMEOUT_PERIOD_MINUTES = 20
RESERVATION_SECONDS = (TIMEOUT_PERIOD_MINUTES + 5) * 60
class BUILD_JOB_RESULT(object):
class BuildJobResult(object):
""" Build job result enum """
INCOMPLETE = 'incomplete'
COMPLETE = 'complete'
@ -29,20 +28,22 @@ class BuilderServer(object):
""" Server which handles both HTTP and WAMP requests, managing the full state of the build
controller.
"""
_loop = None
_current_status = 'starting'
_current_components = []
_job_count = 0
def __init__(self, server_hostname, queue, build_logs, user_files, lifecycle_manager_klass):
self._session_factory = RouterSessionFactory(RouterFactory())
self._loop = None
self._current_status = 'starting'
self._current_components = []
self._job_count = 0
self._session_factory = RouterSessionFactory(RouterFactory())
self._server_hostname = server_hostname
self._queue = queue
self._build_logs = build_logs
self._user_files = user_files
self._lifecycle_manager = lifecycle_manager_klass(
self._register_component, self._unregister_component, self._job_complete)
self._register_component,
self._unregister_component,
self._job_complete
)
self._shutdown_event = Event()
self._current_status = 'running'
@ -69,11 +70,11 @@ class BuilderServer(object):
logging.debug('Starting server on port 8080, with controller on port 8181')
try:
loop.run_forever()
loop.run_forever()
except KeyboardInterrupt:
pass
pass
finally:
loop.close()
loop.close()
def close(self):
logging.debug('Requested server shutdown')
@ -88,7 +89,7 @@ class BuilderServer(object):
"""
logging.debug('Registering component with realm %s', realm)
component = component_klass(types.ComponentConfig(realm = realm), realm=realm, **kwargs)
component = component_klass(types.ComponentConfig(realm=realm), realm=realm, **kwargs)
component.server = self
component.parent_manager = self._lifecycle_manager
component.build_logs = self._build_logs
@ -101,15 +102,15 @@ class BuilderServer(object):
def _unregister_component(self, component):
logging.debug('Unregistering component with realm %s and token %s',
component.builder_realm, component.expected_token)
component.builder_realm, component.expected_token)
self._current_components.remove(component)
self._session_factory.remove(component)
def _job_complete(self, build_job, job_status):
if job_status == BUILD_JOB_RESULT.INCOMPLETE:
if job_status == BuildJobResult.INCOMPLETE:
self._queue.incomplete(build_job.job_item(), restore_retry=True, retry_after=30)
elif job_status == BUILD_JOB_RESULT.ERROR:
elif job_status == BuildJobResult.ERROR:
self._queue.incomplete(build_job.job_item(), restore_retry=False)
else:
self._queue.complete(build_job.job_item())
@ -119,42 +120,42 @@ class BuilderServer(object):
if self._current_status == 'shutting_down' and not self._job_count:
self._shutdown_event.set()
# TODO: check for work here?
# TODO:(jschorr) check for work here?
@trollius.coroutine
def _work_checker(self):
while self._current_status == 'running':
logger.debug('Checking for more work')
LOGGER.debug('Checking for more work')
job_item = self._queue.get(processing_time=RESERVATION_SECONDS)
if job_item is None:
logger.debug('No additional work found. Going to sleep for %s seconds', WORK_CHECK_TIMEOUT)
LOGGER.debug('No additional work found. Going to sleep for %s seconds', WORK_CHECK_TIMEOUT)
yield From(trollius.sleep(WORK_CHECK_TIMEOUT))
continue
try:
build_job = BuildJob(job_item)
except BuildJobLoadException as irbe:
logger.exception(irbe)
LOGGER.exception(irbe)
self._queue.incomplete(job_item, restore_retry=False)
logger.debug('Build job found. Checking for an avaliable worker.')
LOGGER.debug('Build job found. Checking for an avaliable worker.')
if self._lifecycle_manager.schedule(build_job, self._loop):
self._job_count = self._job_count + 1
logger.debug('Build job scheduled. Running: %s', self._job_count)
LOGGER.debug('Build job scheduled. Running: %s', self._job_count)
else:
logger.debug('All workers are busy. Requeuing.')
LOGGER.debug('All workers are busy. Requeuing.')
self._queue.incomplete(job_item, restore_retry=True, retry_after=0)
yield From(trollius.sleep(WORK_CHECK_TIMEOUT))
@trollius.coroutine
def _initialize(self, loop, host):
self._loop = loop
# Create the WAMP server.
transport_factory = WampWebSocketServerFactory(self._session_factory, debug_wamp = False)
transport_factory.setProtocolOptions(failByDrop = True)
transport_factory = WampWebSocketServerFactory(self._session_factory, debug_wamp=False)
transport_factory.setProtocolOptions(failByDrop=True)
# Initialize the controller server and the WAMP server
create_wsgi_server(self._controller_app, loop=loop, host=host, port=8181)