util
/
rq
mirror of https://github.com/peter4431/rq.git
4
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use correct logger name for scheduler and lower level for heartbeat logs (#1311)

Browse Source
main
Aaron Tainter 4 years ago committed by GitHub
parent 167ab96070
commit 5cee02d83f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 7 additions and 6 deletions
Show Stats Download Patch File Download Diff File

13
rq/scheduler.py
Unescape Escape View File

@ -19,6 +19,7 @@ from redis import Redis
SCHEDULER_KEY_TEMPLATE = 'rq:scheduler:%s' SCHEDULER_KEY_TEMPLATE = 'rq:scheduler:%s'
SCHEDULER_LOCKING_KEY_TEMPLATE = 'rq:scheduler-lock:%s' SCHEDULER_LOCKING_KEY_TEMPLATE = 'rq:scheduler-lock:%s'
logger = logging.getLogger(__name__)
setup_loghandlers( setup_loghandlers(
level=logging.INFO, level=logging.INFO,
name="rq.scheduler", name="rq.scheduler",
@ -80,7 +81,7 @@ class RQScheduler(object):
"""Returns names of queue it successfully acquires lock on""" """Returns names of queue it successfully acquires lock on"""
successful_locks = set() successful_locks = set()
pid = os.getpid() pid = os.getpid()
logging.info("Trying to acquire locks for %s", ", ".join(self._queue_names)) logger.info("Trying to acquire locks for %s", ", ".join(self._queue_names))
for name in self._queue_names: for name in self._queue_names:
if self.connection.set(self.get_locking_key(name), pid, nx=True, ex=5): if self.connection.set(self.get_locking_key(name), pid, nx=True, ex=5):
successful_locks.add(name) successful_locks.add(name)
@ -154,7 +155,7 @@ class RQScheduler(object):
def heartbeat(self): def heartbeat(self):
"""Updates the TTL on scheduler keys and the locks""" """Updates the TTL on scheduler keys and the locks"""
logging.info("Scheduler sending heartbeat to %s", ", ".join(self.acquired_locks)) logger.debug("Scheduler sending heartbeat to %s", ", ".join(self.acquired_locks))
if len(self._queue_names) > 1: if len(self._queue_names) > 1:
with self.connection.pipeline() as pipeline: with self.connection.pipeline() as pipeline:
for name in self._queue_names: for name in self._queue_names:
@ -166,7 +167,7 @@ class RQScheduler(object):
self.connection.expire(key, self.interval + 5) self.connection.expire(key, self.interval + 5)
def stop(self): def stop(self):
logging.info("Scheduler stopping, releasing locks for %s...", logger.info("Scheduler stopping, releasing locks for %s...",
','.join(self._queue_names)) ','.join(self._queue_names))
keys = [self.get_locking_key(name) for name in self._queue_names] keys = [self.get_locking_key(name) for name in self._queue_names]
self.connection.delete(*keys) self.connection.delete(*keys)
@ -198,17 +199,17 @@ class RQScheduler(object):
def run(scheduler): def run(scheduler):
logging.info("Scheduler for %s started with PID %s", logger.info("Scheduler for %s started with PID %s",
','.join(scheduler._queue_names), os.getpid()) ','.join(scheduler._queue_names), os.getpid())
try: try:
scheduler.work() scheduler.work()
except: # noqa except: # noqa
logging.error( logger.error(
'Scheduler [PID %s] raised an exception.\n%s', 'Scheduler [PID %s] raised an exception.\n%s',
os.getpid(), traceback.format_exc() os.getpid(), traceback.format_exc()
) )
raise raise
logging.info("Scheduler with PID %s has stopped", os.getpid()) logger.info("Scheduler with PID %s has stopped", os.getpid())
def parse_names(queues_or_names): def parse_names(queues_or_names):

Write Preview
Loading…
Cancel
Save