First try on implementing Python3 support. This was done using futurize (http://python-future.org/) and then manually fixing a few issues.

Changes made:

Adds python-future as dependency which only supports python >=2.6. Most of the small changes should be pretty straightforward e.g. python3 style print function and no .iteritems() in python3

Probably the most problematic area is strings. Unicodes were converted to use str from python-future library and now that python3 has separate bytes type there are few places where those are converted. There is also few places where it is explicitly checking if python version is 3 and doing different stuff e.g. logger just skips encode/decode in python3 and probably there is better way to do it.

Redis is now using decode_responses flag which may or may not be the good solution but at least it allowed me to get it working w/o decoding all the responses manually.

Also changed urllib2 to use urllib trough python-future. And needed to add monkey.patch_all(subprocess=False) to mrq_worker to get that running fine. Didn't really dig too deep in this monkeypatching stuff so not sure if this is a problem.

There are few test cases failing but couldn't get them to pass even with master. There is also no automated test runner for python3 ( I was locally testing with modified Dockerfile ).

requirements.txt also now contains different packages based on python version so installing with recent enough pip works but no idea about setuptools(?)

Any feedback on how I could improve this would be great.

( I have been running this with some "real" python3 jobs which seem to work but there is most likely some corner cases which I might be missing and also didn't run any benchmarks so no idea about the performance )

Hi - we're seeing a condition where we have some number of jobs in the "started" state, but no workers are running (they have all exited due to "burst mode" and the input queue being empty).

Any ideas what could cause this? For instance what if the worker consumes a job but then quits unexpectedly? And any suggestions how to defend against it, like requeue a job after some timeout, or via some manual call from the top-level script.

Thanks!

Instead of queue_job(path, data), we should consider to use similar to celery style: task.delay(data) or task.enqueue(data).

Also, since it is intended to be just one function in task, maybe we should do it... celery-way? Create decorator and write simple function - just as short way. Saves 1 indent and makes code easier to understand.

Not an issue, really, just wondering, why mrq uses mongodb?

I mean, e.g. celery and python-rq use only one broker (redis or rmq) without additional databases.

I believe there is a pretty good reason for using two storages, so I wanted to ask about it :)

Thanks in advance.

Hi @ all, installed mrq as readme tells, everything went well, but trying to launch the dashboard results in the following error. Does anyone knows, how to fix it. I'd like to use the dashboard alone. Redis-server and mongodb server are running. I'm using latest python 2.7 interpreter.

Thanks in advance...

Traceback (most recent call last): File "/home/tom3kk/.conda/envs/py27_mrq/bin/mrq-dashboard", line 7, in <module> from mrq.dashboard.app import main File "/home/tom3kk/.conda/envs/py27_mrq/lib/python2.7/site-packages/mrq/dashboard/app.py", line 29, in <module> set_current_config(cfg) File "/home/tom3kk/.conda/envs/py27_mrq/lib/python2.7/site-packages/mrq/context.py", line 82, in set_current_config patch_import() File "/home/tom3kk/.conda/envs/py27_mrq/lib/python2.7/site-packages/mrq/monkey.py", line 127, in patch_import import gevent.coros File "/home/tom3kk/.conda/envs/py27_mrq/lib/python2.7/site-packages/gevent/builtins.py", line 93, in __import__ result = _import(*args, **kwargs) ImportError: No module named coros

Since I use docker containers for development via boot2docker, it is common to force application host to 0.0.0.0. Since mrq-dashboard runs just like...

run_simple('', int(os.environ.get("PORT", 5555)), app)

It is imposible to even look at dashboard for me. Yea, I can create container with proxy setup like nginx, but really, dude.

Hello PricingAssistant,

We are implementing a task-based processing for our infrastructure management API requests. We are looking for Python queue frameworks. At first, we were interested in using Celery+Redis for it. Then we remembered that RQ was a lighter alternative. After digging into the subject, we discovered your project.

From what says your README, it looks like the proper solution for us: simple as RQ, stores results in a MongoDB instance, comes with a beautiful dashboard.

So my question is can we try to use MrQ for our infrastructure API? Is it easy to start with?

So first the good: I've ported the basic functionalty of our job system from RQ to MRQ and generally it went well, at least so far. Dashboard is certainly a huge improvement. The burst mode you added also works well (note as such I'm using latest code from github, rather than the pip install version).

Some observations/suggestions:

• I couldn't actually find any example of queueing jobs from Python (vs command line). Eventually I reverse-engineered mrq-run.py and figured out I had to do this to get jop.queue_jobs() to work: mrq.context.set_current_config(mrq.config.get_config())
• Python being untyped, it would be good to document params better. E.g. in RQ enqueue() takes a callable, but MRQ job.queue_job(s) takes a string. This isn't immediately obvious.
• Dashboard frequently reads -ve jobs/second.

However big issue is that the system starts to fail above a certain number of workers, somewhere in the region of 1000. It seems like database updates fail or time out and aren't retried. In some cases the job fails with an exception, but mostly just the status is wrong, ie it gets permanently stuck in either "queued" or "started" state in the DB. Note that workers are still running (variously in "wait" or "full" status) but the queues are not consumed, ie there's an inconsistency between the worker process and/or mongo and/or redis which never resolves (I've just been wiping mongo+redis between tests).

'Failed' is perhaps fixable with the retry operations, but the stuck jobs where presumably some other part of the socket communications failed/timed out are more of an issue.

Here's a code snippet:

    for job_id in job_ids:
        job_res = mrq.job.get_job_result(job_id)
        if job_res["status"] == "queued": num_queued += 1
        elif job_res["status"] == "started": num_started += 1
        elif job_res["status"] == "success": num_finished += 1
        else: num_failed += 1

    log("num_queued: %d, num_started: %d, num_finished:%d, num_failed: %d" % 
            (num_queued, num_started, num_finished, num_failed))

Here's the final output for 50k jobs (number of jobs is less important than number of workers): 09:19:26.906279: num_queued: 32, num_started: 8, num_finished:49957, num_failed: 3

Different runs give different numbers of stuck or failed (sometimes it will complete ok). Fewer workers (<500) and it's always ok, it goes to zero queued+started, and my script exits. Note the stats also agree on the dashboard "Statuses" tab.

For the failed jobs, here's the backtrace:

Traceback (most recent call last): 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/mrq/worker.py", line 540, in perform_job 
    job.perform() 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/mrq/job.py", line 279, in perform 
    self.save_success(result) 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/mrq/job.py", line 364, in save_success 
    self._save_status("success", updates) 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/mrq/job.py", line 437, in _save_status 
    }, {"$set": db_updates}, w=w, j=j, manipulate=False) 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/pymongo/collection.py", line 1956, in update 
    with self._socket_for_writes() as sock_info: 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/contextlib.py", line 17, in __enter__ 
    return self.gen.next() 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/pymongo/mongo_client.py", line 665, in _get_socket 
    with server.get_socket(self.__all_credentials) as sock_info: 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/contextlib.py", line 17, in __enter__ 
    return self.gen.next() 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/pymongo/server.py", line 102, in get_socket 
    with self.pool.get_socket(all_credentials, checkout) as sock_info: 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/contextlib.py", line 17, in __enter__ 
    return self.gen.next() 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/pymongo/pool.py", line 509, in get_socket 
    sock_info = self._get_socket_no_auth() 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/pymongo/pool.py", line 543, in _get_socket_no_auth 
    sock_info, from_pool = self.connect(), False 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/pymongo/pool.py", line 475, in connect 
    DEFAULT_CODEC_OPTIONS)) 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/pymongo/network.py", line 48, in command 
    response = receive_message(sock, 1, request_id) 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/pymongo/network.py", line 60, in receive_message 
    header = _receive_data_on_socket(sock, 16) 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/pymongo/network.py", line 84, in _receive_data_on_socket 
    raise AutoReconnect("connection closed") 
AutoReconnect: connection closed 

Any ideas? Thanks.

Whilst upgrading to 0.1.10 we noticed an issue in the dashboard where queues are not showing in the workers tab..

Can confirm this information isn't in the database

db.mrq_workers.find({}, {name:1, config:1})
{ "_id" : ObjectId("54e5d55f2431ff5cc80a00d7"), "config" : { "max_jobs" : 0, "processes" : 0, "name" : "", "queues" : [  "st_queue_a" ], "greenlets" : 20, "scheduler" : false }, "name" : "ip-" }
{ "_id" : ObjectId("54e5d5682431ff5cf648adb2"), "config" : { "max_jobs" : 0, "processes" : 0, "name" : "", "queues" : [  "st_queue_b" ], "greenlets" : 8, "scheduler" : false }, "name" : "ip-" }
{ "_id" : ObjectId("54e5d5882431ff5d3f2ba1c9"), "config" : { "max_jobs" : 0, "processes" : 0, "name" : "", "queues" : [  "st_queue_c" ], "greenlets" : 8, "scheduler" : false }, "name" : "ip-" }
{ "_id" : ObjectId("54e5d55c2431ff5cb9bc719a"), "config" : { "max_jobs" : 0, "processes" : 0, "scheduler" : false, "greenlets" : 50, "name" : "" }, "name" : "ip-" }
{ "_id" : ObjectId("54e5d5642431ff5cdef8ce38"), "config" : { "max_jobs" : 0, "processes" : 0, "scheduler" : false, "greenlets" : 30, "name" : "" }, "name" : "ip-" }

We run a customer worker script instead of mrq-worker as it allows us to work with our own config setup.. It's almost identical to mrq-worker with a couple of tweaks

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import os

# Needed to make getaddrinfo() work in pymongo on Mac OS X
# Docs mention it's a better choice for Linux as well.
# This must be done asap in the worker
if "GEVENT_RESOLVER" not in os.environ:
    os.environ["GEVENT_RESOLVER"] = "ares"

from gevent import monkey
monkey.patch_all()

import config
import sys

sys.path.insert(0, os.getcwd())

from mrq.utils import load_class_by_path
from mrq.context import set_current_config

....

    worker_class = load_class_by_path(cfg["worker_class"])
    set_current_config(cfg)

    w = worker_class()
    exitcode = w.work_loop()
    sys.exit(exitcode)

I have verified cfg is valid and contains 'queues', i.e cfg['queues'] = ["queue_names"] We have also changed the redis_prefix and mongo db name to ensure a fresh test for this.. But problem persists..

Any help would be appreciated on this :)

Update: From tests, it looks like the workers that are the busiest don't get updated... Will keep looking

Hi! In the docs says:

You can queue jobs from your Python code to avoid using mrq-run from the command-line.

But in the crawling example it uses the command-line.

How could I add jobs to a queue without the command-line?

Thaks

I get this error:

D:\myproject>mrq-run a.Fetch url http://www.google.com
Monkey-patching MongoDB methods...
Traceback (most recent call last):
  File "c:\miniconda3\envs\dgpy-dev\lib\runpy.py", line 193, in _run_module_as_main
    "__main__", mod_spec)
  File "c:\miniconda3\envs\dgpy-dev\lib\runpy.py", line 85, in _run_code
    exec(code, run_globals)
  File "C:\miniconda3\envs\dgpy-dev\Scripts\mrq-run.exe\__main__.py", line 9, in <module>
  File "c:\miniconda3\envs\dgpy-dev\lib\site-packages\mrq\bin\mrq_run.py", line 56, in main
    worker_class = load_class_by_path(cfg["worker_class"])
  File "c:\miniconda3\envs\dgpy-dev\lib\site-packages\mrq\utils.py", line 99, in __missing__
    ret = self[key] = f(key)
  File "c:\miniconda3\envs\dgpy-dev\lib\site-packages\mrq\utils.py", line 113, in load_class_by_path
    taskpath)),
  File "c:\miniconda3\envs\dgpy-dev\lib\importlib\__init__.py", line 127, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
  File "<frozen importlib._bootstrap>", line 1006, in _gcd_import
  File "<frozen importlib._bootstrap>", line 983, in _find_and_load
  File "<frozen importlib._bootstrap>", line 967, in _find_and_load_unlocked
  File "<frozen importlib._bootstrap>", line 677, in _load_unlocked
  File "<frozen importlib._bootstrap_external>", line 728, in exec_module
  File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
  File "c:\miniconda3\envs\dgpy-dev\lib\site-packages\mrq\worker.py", line 18, in <module>
    from redis.lock import LuaLock
ImportError: cannot import name 'LuaLock' from 'redis.lock' (c:\miniconda3\envs\dgpy-dev\lib\site-packages\redis\lock.py)

There is a small typo in docs/jobs.md.

Should read instantiate rather than instanciate.

Semi-automated pull request generated by https://github.com/timgates42/meticulous/blob/master/docs/NOTE.md

Bumps lxml from 3.4.2 to 4.9.1.

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting @dependabot rebase.

While reading the documents, noticed a spelling typo.

Did a keep sweep of the rest of the markdown documents and made a couple other corrections.

Please let me know if there are additional actions to be taken as part of this pull request. There's not really any contribution process I could find in the docs.

I'm getting this error when I try to browse the "statuses" page in the dashboard and I have a large (~ 200.000) number of jobs.

172.17.0.1 - - [17/Feb/2021 17:13:11] "GET /api/datatables/status?sEcho=1&iColumns=4&sColumns=&iDisplayStart=0&iDisplayLength=20&mDataProp_0=function&mDataProp_1=function&mDataProp_2=function&mDataProp_3=function&sSearch=&bRegex=false&sSearch_0=&bRegex_0=false&bSearchable_0=true&sSearch_1=&bRegex_1=false&bSearchable_1=true&sSearch_2=&bRegex_2=false&bSearchable_2=true&sSearch_3=&bRegex_3=false&bSearchable_3=true&iSortCol_0=0&sSortDir_0=asc&iSortingCols=1&bSortable_0=true&bSortable_1=true&bSortable_2=true&bSortable_3=true HTTP/1.1" 500 -
Error on request:
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/dist-packages/werkzeug/serving.py", line 304, in run_wsgi
    execute(self.server.app)
  File "/usr/local/lib/python3.6/dist-packages/werkzeug/serving.py", line 292, in execute
    application_iter = app(environ, start_response)
  File "/usr/local/lib/python3.6/dist-packages/flask/app.py", line 2463, in __call__
    return self.wsgi_app(environ, start_response)
  File "/usr/local/lib/python3.6/dist-packages/flask/app.py", line 2449, in wsgi_app
    response = self.handle_exception(e)
  File "/usr/local/lib/python3.6/dist-packages/flask/app.py", line 1866, in handle_exception
    reraise(exc_type, exc_value, tb)
  File "/usr/local/lib/python3.6/dist-packages/flask/_compat.py", line 39, in reraise
    raise value
  File "/usr/local/lib/python3.6/dist-packages/flask/app.py", line 2446, in wsgi_app
    response = self.full_dispatch_request()
  File "/usr/local/lib/python3.6/dist-packages/flask/app.py", line 1951, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/usr/local/lib/python3.6/dist-packages/flask/app.py", line 1820, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/usr/local/lib/python3.6/dist-packages/flask/_compat.py", line 39, in reraise
    raise value
  File "/usr/local/lib/python3.6/dist-packages/flask/app.py", line 1949, in full_dispatch_request
    rv = self.dispatch_request()
  File "/usr/local/lib/python3.6/dist-packages/flask/app.py", line 1935, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/usr/local/lib/python3.6/dist-packages/mrq/dashboard/app.py", line 82, in api_jobstatuses
    {"$group": {"_id": "$status", "jobs": {"$sum": 1}}}
  File "/usr/local/lib/python3.6/dist-packages/mrq/monkey.py", line 99, in mrq_monkey_patched
    ret = base_method(self, *args, **kwargs)
  File "/home/thomas/.local/lib/python3.6/site-packages/pymongo/collection.py", line 2458, in aggregate
    **kwargs)
  File "/home/thomas/.local/lib/python3.6/site-packages/pymongo/collection.py", line 2377, in _aggregate
    retryable=not cmd._performs_write)
  File "/home/thomas/.local/lib/python3.6/site-packages/pymongo/mongo_client.py", line 1471, in _retryable_read
    return func(session, server, sock_info, slave_ok)
  File "/home/thomas/.local/lib/python3.6/site-packages/pymongo/aggregation.py", line 148, in get_cursor
    user_fields=self._user_fields)
  File "/home/thomas/.local/lib/python3.6/site-packages/pymongo/pool.py", line 694, in command
    exhaust_allowed=exhaust_allowed)
  File "/home/thomas/.local/lib/python3.6/site-packages/pymongo/network.py", line 161, in command
    parse_write_concern_error=parse_write_concern_error)
  File "/home/thomas/.local/lib/python3.6/site-packages/pymongo/helpers.py", line 160, in _check_command_response
    raise OperationFailure(errmsg, code, response, max_wire_version)
pymongo.errors.OperationFailure: Sort exceeded memory limit of 104857600 bytes, but did not opt in to external sorting., full error: {'ok': 0.0, 'errmsg': 'Sort exceeded memory limit of 104857600 bytes, but did not opt in to external sorting.', 'code': 16819, 'codeName': 'Location16819'}

add support to delay a job and enqueue it N seconds later. those jobs are going to have the delayed status. e.g: delay a job 1 hour.

queue_job(main_task_path, params, delay=3600)