Bogdanp/django_dramatiq

Using dramatiq, apscheduler with RabbitMQresults in missing heartbeats on RabbitMQ

Open

#44 opened on Jul 5, 2019

 (9 comments) (2 reactions) (0 assignees)Python (86 forks)github user discovery
help wanted

Repository metrics

Stars
 (384 stars)
PR merge metrics
 (No merged PRs in 30d)

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?

Contributor guide