diff --git a/django_q/core.py b/django_q/core.py index afeef60..108b459 100644 --- a/django_q/core.py +++ b/django_q/core.py @@ -64,6 +64,7 @@ class Cluster(object): self.start_event = None self.stopped_event = None self.pid = current_process().pid + self.host = socket.gethostname() self.list_key = list_key signal.signal(signal.SIGTERM, self.sig_handler) signal.signal(signal.SIGINT, self.sig_handler) @@ -81,6 +82,8 @@ class Cluster(object): self.sentinel = Process(target=Sentinel, args=(self.stop_event, self.start_event, self.list_key)) self.sentinel.start() logger.info('Q Cluster-{} starting.'.format(self.pid)) + while not self.start_event.is_set(): + sleep(0.2) return self.pid def stop(self): @@ -329,7 +332,7 @@ def async(func, *args, **kwargs): class SignedPackage(object): """ - Wraps Django's signing module with custom JsonPickle serializer + Wraps Django's signing module with custom Pickle serializer """ @staticmethod @@ -350,7 +353,7 @@ class SignedPackage(object): class PickleSerializer(object): """ - Simple wrapper around JsonPickle for signing.dumps and + Simple wrapper around Pickle for signing.dumps and signing.loads. """ diff --git a/django_q/management/commands/qmonitor.py b/django_q/management/commands/qmonitor.py index 0abe8a9..505898b 100644 --- a/django_q/management/commands/qmonitor.py +++ b/django_q/management/commands/qmonitor.py @@ -12,51 +12,65 @@ from django_q.core import Stat, RUNNING, STOPPED class Command(BaseCommand): help = "Monitors cluster activity" + def add_arguments(self, parser): + parser.add_argument('--run-once', + action='store_true', + dest='run_once', + default=False, + help='Run once till first stat') + def handle(self, *args, **options): - term = Terminal() - with term.fullscreen(), term.hidden_cursor(), term.cbreak(): - val = None - start_width = int(term.width / 8) - while val not in (u'q', u'Q',): - col_width = int(term.width / 8) - # In case of resize - if col_width != start_width: - print(term.clear) - start_width = col_width - print(term.move(0, 0) + term.black_on_green(term.center('Host', width=col_width - 1))) - print(term.move(0, 1 * col_width) + term.black_on_green(term.center('Id', width=col_width - 1))) - print(term.move(0, 2 * col_width) + term.black_on_green(term.center('Status', width=col_width - 1))) - print(term.move(0, 3 * col_width) + term.black_on_green(term.center('Pool', width=col_width - 1))) - print(term.move(0, 4 * col_width) + term.black_on_green(term.center('TQ', width=col_width - 1))) - print(term.move(0, 5 * col_width) + term.black_on_green(term.center('RQ', width=col_width - 1))) - print(term.move(0, 6 * col_width) + term.black_on_green(term.center('Deaths', width=col_width - 1))) - print(term.move(0, 7 * col_width) + term.black_on_green(term.center('Uptime', width=col_width - 1))) - i = 2 - stats = Stat.get_all() - print(term.clear_eos()) - for stat in stats: - # color status - if stat.status == RUNNING: - status = term.green(RUNNING) - elif stat.status == STOPPED: - status = term.red(STOPPED) - else: - status = term.yellow(stat.status) - # format uptime - uptime = (timezone.now() - stat.tob).total_seconds() - hours, remainder = divmod(uptime, 3600) - minutes, seconds = divmod(remainder, 60) - uptime = '%d:%02d:%02d' % (hours, minutes, seconds) - print(term.move(i, 0) + term.center('{}'.format(stat.host), width=col_width - 1)) - print(term.move(i, 1 * col_width) + term.center('{}'.format(stat.cluster_id), width=col_width - 1)) - print(term.move(i, 2 * col_width) + term.center('{}'.format(status), width=col_width - 1)) - print( - term.move(i, 3 * col_width) + term.center('{}'.format(len(stat.workers)), width=col_width - 1)) - print(term.move(i, 4 * col_width) + term.center('{}'.format(stat.task_q_size), width=col_width - 1)) - print(term.move(i, 5 * col_width) + term.center('{}'.format(stat.done_q_size), width=col_width - 1)) - print(term.move(i, 6 * col_width) + term.center('{}'.format(stat.reincarnations), - width=col_width - 1)) - print(term.move(i, 7 * col_width) + term.center('{}'.format(uptime), width=col_width - 1)) - i += 1 - print(term.move(i + 2, 0) + term.center('[Press q to quit]')) - val = term.inkey(timeout=1) + monitor(run_once=options['run_once']) + + +def monitor(run_once=False): + term = Terminal() + with term.fullscreen(), term.hidden_cursor(), term.cbreak(): + val = None + start_width = int(term.width / 8) + while val not in (u'q', u'Q',): + col_width = int(term.width / 8) + # In case of resize + if col_width != start_width: + print(term.clear) + start_width = col_width + print(term.move(0, 0) + term.black_on_green(term.center('Host', width=col_width - 1))) + print(term.move(0, 1 * col_width) + term.black_on_green(term.center('Id', width=col_width - 1))) + print(term.move(0, 2 * col_width) + term.black_on_green(term.center('Status', width=col_width - 1))) + print(term.move(0, 3 * col_width) + term.black_on_green(term.center('Pool', width=col_width - 1))) + print(term.move(0, 4 * col_width) + term.black_on_green(term.center('TQ', width=col_width - 1))) + print(term.move(0, 5 * col_width) + term.black_on_green(term.center('RQ', width=col_width - 1))) + print(term.move(0, 6 * col_width) + term.black_on_green(term.center('Deaths', width=col_width - 1))) + print(term.move(0, 7 * col_width) + term.black_on_green(term.center('Uptime', width=col_width - 1))) + i = 2 + stats = Stat.get_all() + print(term.clear_eos()) + for stat in stats: + # color status + if stat.status == RUNNING: + status = term.green(RUNNING) + elif stat.status == STOPPED: + status = term.red(STOPPED) + else: + status = term.yellow(stat.status) + # format uptime + uptime = (timezone.now() - stat.tob).total_seconds() + hours, remainder = divmod(uptime, 3600) + minutes, seconds = divmod(remainder, 60) + uptime = '%d:%02d:%02d' % (hours, minutes, seconds) + print(term.move(i, 0) + term.center('{}'.format(stat.host), width=col_width - 1)) + print(term.move(i, 1 * col_width) + term.center('{}'.format(stat.cluster_id), width=col_width - 1)) + print(term.move(i, 2 * col_width) + term.center('{}'.format(status), width=col_width - 1)) + print( + term.move(i, 3 * col_width) + term.center('{}'.format(len(stat.workers)), width=col_width - 1)) + print(term.move(i, 4 * col_width) + term.center('{}'.format(stat.task_q_size), width=col_width - 1)) + print(term.move(i, 5 * col_width) + term.center('{}'.format(stat.done_q_size), width=col_width - 1)) + print(term.move(i, 6 * col_width) + term.center('{}'.format(stat.reincarnations), + width=col_width - 1)) + print(term.move(i, 7 * col_width) + term.center('{}'.format(uptime), width=col_width - 1)) + i += 1 + # for testing + if run_once: + return Stat.get_all() + print(term.move(i + 2, 0) + term.center('[Press q to quit]')) + val = term.inkey(timeout=1) diff --git a/django_q/tests/test_admin.py b/django_q/tests/test_admin.py new file mode 100644 index 0000000..4b9001f --- /dev/null +++ b/django_q/tests/test_admin.py @@ -0,0 +1,7 @@ +def test_admin_view(admin_client): + response = admin_client.get('/admin/django_q/') + assert response.status_code == 200 + response = admin_client.get('/admin/django_q/failure/') + assert response.status_code == 200 + response = admin_client.get('/admin/django_q/success/') + assert response.status_code == 200 \ No newline at end of file diff --git a/django_q/tests/test_q.py b/django_q/tests/test_cluster.py similarity index 89% rename from django_q/tests/test_q.py rename to django_q/tests/test_cluster.py index 462468c..d8abd9e 100644 --- a/django_q/tests/test_q.py +++ b/django_q/tests/test_cluster.py @@ -20,16 +20,6 @@ class WordClass(object): def get_words(self): return self.word_list - -def test_admin_view(admin_client): - response = admin_client.get('/admin/django_q/') - assert response.status_code == 200 - response = admin_client.get('/admin/django_q/failure/') - assert response.status_code == 200 - response = admin_client.get('/admin/django_q/success/') - assert response.status_code == 200 - - def test_redis_connection(): assert r.ping() is True @@ -39,13 +29,9 @@ def test_cluster_initial(): assert c.sentinel is None assert c.is_idle c.start() - while c.is_starting: - sleep(0.2) assert c.sentinel.is_alive() is True assert c.is_running c.stop() - while c.is_stopping: - sleep(0.2) assert c.sentinel.is_alive() is False assert c.has_stopped @@ -92,8 +78,6 @@ def run_cluster(): r.delete(list_key) c = Cluster(list_key=list_key) assert c.start() > 0 - while not c.is_running: - sleep(0.5) while c.stat.task_q_size > 0 and c.stat.done_q_size > 0: sleep(0.5) assert c.stop() is True diff --git a/django_q/tests/test_monitor.py b/django_q/tests/test_monitor.py new file mode 100644 index 0000000..ef07520 --- /dev/null +++ b/django_q/tests/test_monitor.py @@ -0,0 +1,11 @@ +from django_q.core import Cluster +from django_q.management.commands.qmonitor import monitor + + +def test_monitor(): + c = Cluster() + c.start() + stats = monitor(run_once=True) + c.stop() + assert len(stats) > 0 + assert stats[0].cluster_id == c.pid diff --git a/setup.py b/setup.py index a0a56ec..1028a88 100644 --- a/setup.py +++ b/setup.py @@ -36,6 +36,7 @@ setup( description='A multiprocessing task queue for Django', long_description=README, install_requires=['django>=1.7', 'redis', 'coloredlogs', 'django-picklefield', 'blessed'], + test_requires=['pytest', 'pytest-django', ], cmdclass={'test': PyTest}, classifiers=[ 'Development Status :: 2 - PreAlpha',