Bogdanp/django_dramatiq

Using dramatiq, apscheduler with RabbitMQresults in missing heartbeats on RabbitMQ

Ouverte

#44 ouverte le 5 juil. 2019

 (9 commentaires) (2 réactions) (0 personne assignée)Python (86 forks)github user discovery
help wanted

Métriques du dépôt

Stars
 (384 étoiles)
Métriques de merge PR
 (Aucune PR mergée en 30 j)

Description

I needed to add a scheduler to run some actors periodically and I added a django management command to run the scheduler, it goes like this:

import signal
import sys

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.interval import IntervalTrigger
from django.core.management.base import BaseCommand


class Command(BaseCommand):
    help = 'Run task scheduler'

    def handle(self, *args, **options):
        scheduler = BlockingScheduler()
        
        scheduler.add(
            trigger=IntervalTrigger(
                seconds=30,
            ),
            name='task1',
            func='project.apps.app.tasks.task1.send',  # Edit:  added '.send' 
        )

        def shutdown(*args):
            self.stdout.write("Exiting...")
            sys.exit(0)

        signal.signal(signal.SIGINT, shutdown)
        signal.signal(signal.SIGTERM, shutdown)

        self.stdout.write("Discovered tasks:")

        for s in scheduler.get_jobs():
            self.stdout.write(f"* {s.name} - {s.trigger}")

        self.stdout.write("\nStarting scheduler...")

        scheduler.start()

        return 0

It works but connection resets from time to time, with this:

[2019-07-05 21:40:49,396] [ERROR] pika.adapters.utils.io_services_utils: _AsyncBaseTransport._produce() failed, aborting connection: error=ConnectionResetError(104, 'Connection reset by peer'); sock=<socket.socket fd=36, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('10.42.6.131', 55836)>; Caller's stack: Traceback (most recent call last): File "/usr/lib/python3.7/site-packages/pika/adapters/utils/io_services_utils.py", line 1097, in _on_socket_writable self._produce() File "/usr/lib/python3.7/site-packages/pika/adapters/utils/io_services_utils.py", line 820, in _produce self._tx_buffers[0]) File "/usr/lib/python3.7/site-packages/pika/adapters/utils/io_services_utils.py", line 79, in retry_sigint_wrap return func(*args, **kwargs) File "/usr/lib/python3.7/site-packages/pika/adapters/utils/io_services_utils.py", line 861, in _sigint_safe_send return sock.send(data) ConnectionResetError: [Errno 104] Connection reset by peer Traceback (most recent call last): File "/usr/lib/python3.7/site-packages/pika/adapters/utils/io_services_utils.py", line 1097, in _on_socket_writable self._produce() File "/usr/lib/python3.7/site-packages/pika/adapters/utils/io_services_utils.py", line 820, in _produce self._tx_buffers[0]) File "/usr/lib/python3.7/site-packages/pika/adapters/utils/io_services_utils.py", line 79, in retry_sigint_wrap return func(*args, **kwargs) File "/usr/lib/python3.7/site-packages/pika/adapters/utils/io_services_utils.py", line 861, in _sigint_safe_send return sock.send(data) ConnectionResetError: [Errno 104] Connection reset by peer [2019-07-05 21:40:49,488] [ERROR] pika.adapters.base_connection: connection_lost: StreamLostError: ("Stream connection lost: ConnectionResetError(104, 'Connection reset by peer')",) [2019-07-05 21:40:49,555] [ERROR] pika.adapters.blocking_connection: Unexpected connection close detected: StreamLostError: ("Stream connection lost: ConnectionResetError(104, 'Connection reset by peer')",)

and here is the RabbitMQ log:

2019-07-05 16:24:40.426 [error] <0.31273.0> closing AMQP connection <0.31273.0> (10.42.6.131:36524 -> 10.42.6.93:5672): missed heartbeats from client, timeout: 60s

I tried to do something similar to one in @Bogdanp post. Am I missing something?

Guide contributeur