diff --git a/.gitignore b/.gitignore index 6d66da24..9b6e7c85 100644 --- a/.gitignore +++ b/.gitignore @@ -17,6 +17,7 @@ develop-eggs lib lib64 __pycache__ +.cache # Installer logs pip-log.txt @@ -44,3 +45,5 @@ mrq-config.py dump.rdb supervisord.pid memory_traces +mrq/dashboard/static/node_modules/ +.vscode diff --git a/.travis.yml b/.travis.yml index f4c10126..55ae5964 100644 --- a/.travis.yml +++ b/.travis.yml @@ -6,6 +6,7 @@ services: env: - PYTHON_BIN=python - PYTHON_BIN=python3 + - PYTHON_BIN=/pypy/bin/pypy before_install: - docker ps @@ -15,5 +16,6 @@ before_install: # TODO: coveralls? script: - - docker run -i -t -v `pwd`:/app:rw -w /app mrq_local $PYTHON_BIN -m pylint --errors-only --init-hook="import sys; sys.path.append('.')" -d E1103 --rcfile .pylintrc mrq - - docker run -i -t -v `pwd`:/app:rw -w /app mrq_local $PYTHON_BIN -m pytest tests/ -v --junitxml=pytest-report.xml --cov mrq --cov-report term + - docker run -i -t -v `pwd`:/app:rw -w /app pricingassistant/mrq $PYTHON_BIN -m pylint --errors-only --init-hook="import sys; sys.path.append('.')" -d E1103 --rcfile .pylintrc mrq +# - docker run -i -t -v `pwd`:/app:rw -w /app mrq_local $PYTHON_BIN -m pytest tests/ --collect-only + - docker run -i -t -v `pwd`:/app:rw -w /app pricingassistant/mrq $PYTHON_BIN -m pytest tests/ -v --junitxml=pytest-report.xml --cov mrq --cov-report term --timeout-method=thread --timeout=240 diff --git a/Dockerfile b/Dockerfile index 29ab4fdf..4f39dbf0 100644 --- a/Dockerfile +++ b/Dockerfile @@ -5,14 +5,14 @@ FROM debian:jessie # https://github.com/docker-library/buildpack-deps/issues/40 # -RUN echo \ - 'deb ftp://ftp.us.debian.org/debian/ jessie main\n \ - deb ftp://ftp.us.debian.org/debian/ jessie-updates main\n \ - deb http://security.debian.org jessie/updates main\n' \ - > /etc/apt/sources.list - -RUN apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv 7F0CEB10 -RUN echo "deb http://repo.mongodb.org/apt/debian wheezy/mongodb-org/3.0 main" > /etc/apt/sources.list.d/mongodb-org-3.0.list +# RUN echo \ +# 'deb ftp://ftp.us.debian.org/debian/ jessie main\n \ +# deb ftp://ftp.us.debian.org/debian/ jessie-updates main\n \ +# deb http://security.debian.org jessie/updates main\n' \ +# > /etc/apt/sources.list + +RUN apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv 0C49F3730359A14518585931BC711F9BA15703C6 +RUN echo "deb http://repo.mongodb.org/apt/debian jessie/mongodb-org/3.4 main" > /etc/apt/sources.list.d/mongodb-org-3.4.list RUN apt-get update && \ apt-get install -y --no-install-recommends \ curl \ @@ -21,17 +21,27 @@ RUN apt-get update && \ python-pip \ python3-pip \ python3-dev \ + make \ git \ vim \ - mongodb-org-server \ + bzip2 \ + mongodb-org \ nginx redis-server \ + g++ \ && \ apt-get clean -y && \ rm -rf /var/lib/apt/lists/* +RUN curl -sL https://deb.nodesource.com/setup_7.x | bash - +RUN apt-get install -y --no-install-recommends nodejs + +# Download pypy +RUN curl -sL 'https://bitbucket.org/squeaky/portable-pypy/downloads/pypy-5.8-1-linux_x86_64-portable.tar.bz2' > /pypy.tar.bz2 && tar jxvf /pypy.tar.bz2 && rm -rf /pypy.tar.bz2 && mv /pypy-* /pypy + # Upgrade pip RUN pip install --upgrade --ignore-installed pip RUN pip3 install --upgrade --ignore-installed pip +RUN /pypy/bin/pypy -m ensurepip ADD requirements-heroku.txt /app/requirements-heroku.txt ADD requirements-base.txt /app/requirements-base.txt @@ -50,8 +60,21 @@ RUN pip install -r /app/requirements-heroku.txt && \ pip install -r /app/requirements-dashboard.txt && \ rm -rf ~/.cache +RUN /pypy/bin/pip install -r /app/requirements-heroku.txt && \ + /pypy/bin/pip install -r /app/requirements-base.txt && \ + /pypy/bin/pip install -r /app/requirements-dev.txt && \ + /pypy/bin/pip install -r /app/requirements-dashboard.txt && \ + rm -rf ~/.cache + RUN mkdir -p /data/db +RUN ln -s /app/mrq/bin/mrq_run.py /usr/bin/mrq-run +RUN ln -s /app/mrq/bin/mrq_worker.py /usr/bin/mrq-worker +RUN ln -s /app/mrq/bin/mrq_agent.py /usr/bin/mrq-agent +RUN ln -s /app/mrq/dashboard/app.py /usr/bin/mrq-dashboard + +ENV PYTHONPATH /app + VOLUME ["/data"] WORKDIR /app diff --git a/Dockerfile-with-code b/Dockerfile-with-code new file mode 100644 index 00000000..812364ff --- /dev/null +++ b/Dockerfile-with-code @@ -0,0 +1,3 @@ +FROM pricingassistant/mrq-env:latest + +ADD ./mrq /app/mrq diff --git a/MANIFEST.in b/MANIFEST.in index 2ab06571..7555e7e6 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -1,5 +1,4 @@ include *.md -recursive-include mrq/supervisord_templates * include requirements* recursive-include mrq/dashboard/static * recursive-include mrq/dashboard/templates * diff --git a/Makefile b/Makefile index 2f9d048e..73d32d52 100644 --- a/Makefile +++ b/Makefile @@ -1,52 +1,57 @@ docker: - docker build -t mrq_local . + docker build -t pricingassistant/mrq-env . + docker build -t pricingassistant/mrq -f Dockerfile-with-code . + +docker_push: + docker push pricingassistant/mrq-env:latest + docker push pricingassistant/mrq:latest test: docker - sh -c "docker run --rm -i -t -p 27017:27017 -p 6379:6379 -p 5555:5555 -p 20020:20020 -v `pwd`:/app:rw -w /app mrq_local python -m pytest tests/ -v --instafail" + sh -c "docker run --rm -i -t -p 27017:27017 -p 6379:6379 -p 5555:5555 -p 20020:20020 -v `pwd`:/app:rw -w /app pricingassistant/mrq-env python -m pytest tests/ -v --instafail" test3: docker - sh -c "docker run --rm -i -t -p 27017:27017 -p 6379:6379 -p 5555:5555 -p 20020:20020 -v `pwd`:/app:rw -w /app mrq_local python3 -m pytest tests/ -v --instafail" + sh -c "docker run --rm -i -t -p 27017:27017 -p 6379:6379 -p 5555:5555 -p 20020:20020 -v `pwd`:/app:rw -w /app pricingassistant/mrq-env python3 -m pytest tests/ -v --instafail" + +testpypy: docker + sh -c "docker run --rm -i -t -p 27017:27017 -p 6379:6379 -p 5555:5555 -p 20020:20020 -v `pwd`:/app:rw -w /app pricingassistant/mrq-env /pypy/bin/pypy -m pytest tests/ -v --instafail" shell: - sh -c "docker run --rm -i -t -p 27017:27017 -p 6379:6379 -p 5555:5555 -p 20020:20020 -p 8000:8000 -v `pwd`:/app:rw -w /app mrq_local bash" + sh -c "docker run --rm -i -t -p 27017:27017 -p 6379:6379 -p 5555:5555 -p 20020:20020 -p 8000:8000 -v `pwd`:/app:rw -w /app pricingassistant/mrq-env bash" + +reshell: + # Reconnect in the current taskqueue container + sh -c 'docker exec -t -i `docker ps | grep pricingassistant/mrq-env | cut -f 1 -d " "` bash' shell_noport: - sh -c "docker run --rm -i -t -v `pwd`:/app:rw -w /app mrq_local bash" + sh -c "docker run --rm -i -t -v `pwd`:/app:rw -w /app pricingassistant/mrq-env bash" docs_serve: - sh -c "docker run --rm -i -t-p 8000:8000 -v `pwd`:/app:rw -w /app mrq_local mkdocs serve" + sh -c "docker run --rm -i -t -p 8000:8000 -v `pwd`:/app:rw -w /app pricingassistant/mrq-env mkdocs serve" lint: docker - docker run -i -t -v `pwd`:/app:rw -w /app mrq_local pylint --init-hook="import sys; sys.path.append('.')" --rcfile .pylintrc mrq + docker run -i -t -v `pwd`:/app:rw -w /app pricingassistant/mrq-env pylint -j 0 --init-hook="import sys; sys.path.append('.')" --rcfile .pylintrc mrq linterrors: docker - docker run -i -t -v `pwd`:/app:rw -w /app mrq_local pylint --errors-only --init-hook="import sys; sys.path.append('.')" -d E1103 --rcfile .pylintrc mrq + docker run -i -t -v `pwd`:/app:rw -w /app pricingassistant/mrq-env pylint -j 0 --errors-only --init-hook="import sys; sys.path.append('.')" --rcfile .pylintrc mrq linterrors3: docker - docker run -i -t -v `pwd`:/app:rw -w /app mrq_local python3 -m pylint --errors-only --init-hook="import sys; sys.path.append('.')" -d E1103 --rcfile .pylintrc mrq + docker run -i -t -v `pwd`:/app:rw -w /app pricingassistant/mrq-env python3 -m pylint -j 0 --errors-only --init-hook="import sys; sys.path.append('.')" --rcfile .pylintrc mrq virtualenv: - virtualenv venv --distribute - -virtualenv_pypy: - virtualenv -p /usr/bin/pypy pypy --distribute + virtualenv venv --distribute --python=python2.7 deps: pip install -r requirements.txt pip install -r requirements-dev.txt pip install -r requirements-dashboard.txt -deps_pypy: - pip install git+git://github.com/schmir/gevent@pypy-hacks - pip install cffi - pip install git+git://github.com/gevent-on-pypy/pypycore - export GEVENT_LOOP=pypycore.loop - pip install -r requirements-pypy.txt - clean: find . -path ./venv -prune -o -name "*.pyc" -exec rm {} \; find . -name __pycache__ | xargs rm -r +build_dashboard: + cd mrq/dashboard/static && npm install && mkdir -p bin && npm run build + dashboard: python mrq/dashboard/app.py @@ -62,8 +67,11 @@ pep8: autopep8: autopep8 --max-line-length 99 -aaaaaaaa --in-place --recursive mrq -pypi: linterrors +pypi: linterrors linterrors3 python setup.py sdist upload build_docs: python scripts/propagate_docs.py + +ensureindexes: + mrq-run mrq.basetasks.indexes.EnsureIndexes diff --git a/docs/command-line.md b/docs/command-line.md index 786ad2e0..218126f7 100644 --- a/docs/command-line.md +++ b/docs/command-line.md @@ -16,7 +16,6 @@ The following general flags can be passed as command-line arguments to either ** - `--mongodb_jobs, --mongodb`: MongoDB URI for the jobs, scheduled_jobs & workers database. Defaults to **mongodb://127.0.0.1:27017/mrq**. - `--mongodb_logs` :MongoDB URI for the logs database."0" will disable remote logs, "1" will use main MongoDB. Defaults to **1** - `--mongodb_logs_size`: If provided, sets the log collection to capped to that amount of bytes. - - `--no_mongodb_ensure_indexes`: If provided, skip the creation of MongoDB indexes at worker startup. - `--redis`: Redis URI. Defaults to **redis://127.0.0.1:6379**. - `--redis_prefix`: Redis key prefix. Defaults to "mrq". - `--redis_max_connections`: Redis max connection pool size. Defaults to **1000**. @@ -46,16 +45,14 @@ You can pass additional configuration flags: - `--max_jobs`: Gevent:max number of jobs to do before quitting. Use as a workaround for memory leaks in your tasks. Defaults to **0** - `--max_memory`: Max memory (in Mb) after which the process will be shut down. Use with `--processes [1-N]` - to have supervisord automatically respawn the worker when this happens. Defaults to **1** + to have the worker automatically respawn when this happens. Defaults to **1** - `--grenlets, --gevent, --g`: Max number of greenlets to use. Defaults to **1**. - - `--processes, --p`: Number of processes to launch with supervisord. Defaults to **0** (no supervisord). - - `--supervisord_template`: Path of supervisord template to use. Defaults to **supervisord_templates/default.conf**. + - `--processes, --p`: Number of processes to launch . Defaults to **0**. - `--scheduler`: Run the scheduler. Defaults to **false**. - `--scheduler_interval`: Seconds between scheduler checks. Defaults to **60** seconds, only ints are acceptable. - `--report_interval`: Seconds between worker reports to MongoDB. Defaults to **10** seconds, floats are acceptable too. - `--report_file`: Filepath of a json dump of the worker status. Disabled if none. - `--subqueues_refresh_interval`: Seconds between worker refreshes of the known subqueues. - - `--subqueues_delimiter`: Delimiter between main queue and subqueue names. - `--paused_queues_refresh_interval`: Seconds between worker refreshes of the paused queues list. - `--admin_port`: Start an admin server on this port, if provided. Incompatible with --processes. Defaults to **0** - `--admin_ip`: IP for the admin server to listen on. Use "0.0.0.0" to allow access from outside. Defaults to **127.0.0.1**. @@ -71,7 +68,7 @@ The default is to run tasks one at a time. You should obviously change this beha This will start 30 greenlets over 3 UNIX processes. Each of them will run 10 jobs at the same time. -As soon as you use the `--processes` option (even with `--processes=1`) then supervisord will be used to control the processes. It is quite useful to manage long-running instances. +The worker is autonomous to handle its processes. It is quite useful to manage long-running instances. ### Simulating network latency diff --git a/docs/configuration.md b/docs/configuration.md index 061bbcff..30852d86 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -20,7 +20,6 @@ Remember, put mrq-config.py in your workers directory. MONGODB_JOBS = "mongodb://127.0.0.1:27017/mrq" # MongoDB URI for the jobs, scheduled_jobs & workers database.Defaults to mongodb://127.0.0.1:27017/mrq MONGODB_LOGS = 1 #MongoDB URI for the logs database."0" will disable remote logs, "1" will use main MongoDB.Defaults to 1 MONGODB_LOGS_SIZE = None #If provided, sets the log collection to capped to that amount of bytes. -NO_MONGODB_ENSURE_INDEXES = None #If provided, skip the creation of MongoDB indexes at worker startup. #Redis settings REDIS = "redis://127.0.0.1:6379" #Redis URI.Defaults to redis://127.0.0.1:6379 @@ -57,10 +56,9 @@ USE_LARGE_JOB_IDS = False #Do not use compacted job IDs in Redis. For compatibil QUEUES = ("default",) # The queues to listen on.Defaults to default , which will listen on all queues. MAX_JOBS = 0 #Gevent:max number of jobs to do before quitting. Workaround for memory leaks in your tasks. Defaults to 0 MAX_TIME = 0 # max number of seconds a worker runs before quitting -MAX_MEMORY = 1 #Max memory (in Mb) after which the process will be shut down. Use with PROCESS = [1-N] to have supervisord automatically respawn the worker when this happens.Defaults to 1 +MAX_MEMORY = 1 #Max memory (in Mb) after which the process will be shut down. Use with PROCESS = [1-N] to have the worker automatically respawned when this happens.Defaults to 1 GRENLETS = 1 #Max number of greenlets to use.Defaults to 1. -PROCESSES = 0 #Number of processes to launch with supervisord.Defaults to 0. -SUPERVISORD_TEMPLATE = "supervisord_templates/default.conf" #Path of supervisord template to use. Defaults to supervisord_templates/default.conf. +PROCESSES = 0 #Number of processes to launch.Defaults to 0. SCHEDULER = False #Run the scheduler.Defaults to False. SCHEDULER_INTERVAL = 60 #Seconds between scheduler checks.Defaults to 60 seconds, only ints are acceptable. REPORT_INTERVAL = 10.5 #Seconds between worker reports to MongoDB.Defaults to 10 seconds, floats are acceptable too. diff --git a/docs/design.md b/docs/design.md deleted file mode 100644 index bce07d69..00000000 --- a/docs/design.md +++ /dev/null @@ -1,8 +0,0 @@ -# Design - -A talk with some slides about MRQ's design is upcoming. - -A couple things to know: - -- We use Redis as a main queue for task IDs -- We store metadata on the tasks in MongoDB so they can be browsable and managed more easily. diff --git a/docs/get-started.md b/docs/get-started.md index 89804385..2bd06fbe 100644 --- a/docs/get-started.md +++ b/docs/get-started.md @@ -108,5 +108,3 @@ This was a preview on the very basic features of MRQ. What makes it actually use * You can run multiple workers in parallel. Each worker can also run multiple greenlets in parallel. * Workers can dequeue from multiple queues * You can queue jobs from your Python code to avoid using `mrq-run` from the command-line. - -These features will be demonstrated in a future example of a simple web crawler. diff --git a/docs/index.md b/docs/index.md index e8353bb7..ce5bf6e5 100644 --- a/docs/index.md +++ b/docs/index.md @@ -16,7 +16,6 @@ MRQ was first developed at [Pricing Assistant](http://pricingassistant.com) and * **Great [dashboard](http://mrq.readthedocs.org/en/latest/dashboard/):** Have visibility and control on everything: queued jobs, current jobs, worker status, ... * **Per-job logs:** Get the log output of each task separately in the dashboard * **Gevent worker:** IO-bound tasks can be done in parallel in the same UNIX process for maximum throughput - * **Supervisord integration:** CPU-bound tasks can be split across several UNIX processes with a single command-line flag * **Job management:** You can retry, requeue, cancel jobs from the code or the dashboard. * **Performance:** Bulk job queueing, easy job profiling * **Easy [configuration](http://mrq.readthedocs.org/en/latest/configuration):** Every aspect of MRQ is configurable through command-line flags or a configuration file diff --git a/docs/jobs-maintenance.md b/docs/jobs-maintenance.md index 86ca746e..d1d1822e 100644 --- a/docs/jobs-maintenance.md +++ b/docs/jobs-maintenance.md @@ -30,18 +30,9 @@ SCHEDULER_TASKS = [ "interval": 3600 }, - # This will requeue jobs 'lost' between redis.blpop() and mongo.update(status=started). - # This can happen only when the worker is killed brutally in the middle of dequeue_jobs() + # This will make sure MRQ's indexes are built { - "path": "mrq.basetasks.cleaning.RequeueLostJobs", - "params": {}, - "interval": 24 * 3600 - }, - - # This will clean the list of known queues in Redis. It will mostly remove empty queues - # so that they are not displayed in the dashboard anymore. - { - "path": "mrq.basetasks.cleaning.CleanKnownQueues", + "path": "mrq.basetasks.indexes.EnsureIndexes", "params": {}, "interval": 24 * 3600 } diff --git a/docs/jobs.md b/docs/jobs.md index 83bfc982..ffc25a81 100644 --- a/docs/jobs.md +++ b/docs/jobs.md @@ -24,7 +24,7 @@ However, to be reliable a task queue needs to prepare for everything that can go * ```retry```: The method `task.retry()` was called to interrupt the job but mark it for being retried later. This may be useful when calling unreliable 3rd-party services. * ```maxretries```: The task was retried too many times. Max retries default to 3 and can be configured globally or per task. At this point it should be up to you to cancel them or requeue them again. -Only jobs in statuses `success` and `cancel` will be cleaned from MongoDB after a delay of `result_ttl` seconds (see [Task configuration](configuration.md)) +Jobs in status `success` will be cleaned from MongoDB after a delay of `result_ttl` seconds (see [Task configuration](configuration.md)) ## Task API diff --git a/docs/performance.md b/docs/performance.md index c380babd..9867981e 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -1,4 +1,4 @@ -# Performance +# Worker performance Performance is an explicit goal of MRQ as it was first developed at [Pricing Assistant](http://www.pricingassistant.com/) for crawling billions of web pages. @@ -8,6 +8,8 @@ On a regular Macbook Pro, we see 1300 jobs/second in a single worker process wit However what we are really measuring there is MongoDB's write performance. An install of MRQ with properly scaled MongoDB and Redis instances is be capable of much more. +For more, see our tutorial on [Queue performance](queue-performance.md). + ## PyPy support Earlier in its development MRQ was tested successfully on PyPy but we are waiting for better PyPy+gevent support to continue working on it, as performance was worse than CPython. diff --git a/docs/queue-performance.md b/docs/queue-performance.md new file mode 100644 index 00000000..062b0458 --- /dev/null +++ b/docs/queue-performance.md @@ -0,0 +1,211 @@ +This tutorial will guide you through the configuration of a MRQ queue for maximum performance. + +Code is available in the `examples/queue_performance` folder. To be able to run the commands below, you should enter the container first: + +``` +make shell +make stack +cd examples/queue_performance +``` + + + +## Regular queue + + + + +### Default setup + +Let's start with a simple task that squares integers, from the `tasks.py` file: + +``` +class Square(Task): + def run(self, params): + return int(params["n"]) ** 2 +``` + +You can enqueue it 200 times on a regular, MongoDB-backed queue named `square` with this code: + +``` +from mrq.job import queue_jobs +queue_jobs("tasks.Square", [{"n": 42} for _ in range(200)], queue="square") +``` + +For convenience, we will use the `enqueue.py` file to do this. Here is the command to enqueue the jobs and launch a worker to dequeue them: + +``` +./enqueue.py square 200 && mrq-worker square +``` + +You should see the output of the worker, with a line like this one at the end (performance numbers from a 2015 MacBook Pro): + +``` +[INFO] Worker spent 2.398 seconds performing 200 jobs (83.403 jobs/second) +``` + +As we have `DEQUEUE_STRATEGY = "burst"` in the `mrq-config.py` file, the worker exits as soon as there are no jobs left on the queue, which is more convenient for this tutorial. + +80 jobs per second is rather slow. The main bottleneck is that by default, `mrq-worker` uses a single process and a single greenlet. With this setup, jobs are executed sequentially and between each, the worker must fetch the next one from MongoDB. As a consequence, most of the time of the worker is spent on blocking I/O to MongoDB: not good! + + + + +### Multi-greenlet worker + +Fortunately, MRQ uses [gevent](http://gevent.org) and allows us to start many greenlets at once in the same worker. Let's try with 5 greenlets: + +``` +./enqueue.py square 200 && mrq-worker square --greenlets 5 +... +[INFO] Worker spent 0.652 seconds performing 200 jobs (306.554 jobs/second) +``` + +We got an almost linear increase in performance! What if we tried 50 greenlets? + +``` +./enqueue.py square 200 && mrq-worker square --greenlets 50 +... +[INFO] Worker spent 0.382 seconds performing 200 jobs (523.174 jobs/second) +``` + +A nice increase again, but definitely not linear anymore. Depending on your workload, the performance gains will stop at some point either because you hit a CPU bottleneck on the worker, or the concurrency limit of your MongoDB server. + +If MongoDB is the limiting factor, you have 2 choices to go further: + + - Scale your MongoDB instance ([many options](https://docs.mongodb.com/manual/administration/analyzing-mongodb-performance/) are available, including [sharding](https://docs.mongodb.com/manual/sharding/)) + - Switch to a Redis-backed queue (also called a *raw queue* in MRQ). + + + +## Raw queue + + + + +### Default setup + +A raw queue must be configured in `mrq-config.py` with its job factory function, which will transform a "raw" parameter string into a complete job definition: + +``` +RAW_QUEUES = { + "square_raw": { + "job_factory": lambda rawparam: { + "path": "tasks.Square", + "params": { + "n": rawparam + } + } + } +} +``` + +The only thing that will be queued in redis will be the raw parameter. This has the benefit of using much less storage than MongoDB-backed queues, but also of being faster to dequeue: + +``` +./enqueue.py square_raw 2000 && mrq-worker square_raw --greenlets 30 +... +[INFO] Worker spent 1.804 seconds performing 2000 jobs (1108.419 jobs/second) +``` + +Much better! If you use `top` while launching these commands (you can open a second shell in the same container with the `make reshell` command from the host), you will see that the python worker process is now maxing-out a CPU. + + + + +### Multi-process worker + +As you know, a single Python process can only use a single CPU. Let's try to use all the cores you have at your disposal to get better performance! + +mrq-worker can start multiple processes with the ```--processes``` flag. The worker handles its own processes. If you use this option you will have to manually terminate the worker with a `ctrl-C` keystroke once it is finished: + +``` +./enqueue.py square_raw 20000 && mrq-worker square_raw --greenlets 30 --processes 5 +... +[INFO] Worker spent 6.697 seconds performing 4239 jobs (632.986 jobs/second) +... +[INFO] Worker spent 6.505 seconds performing 4307 jobs (662.075 jobs/second) +... +``` + +Each of the 5 worker processes handled its share of the jobs. The performance numbers aren't aggregated but you can see that the global throughput is now more than 3000 jobs per second. + +`top` reveals that the bottleneck is once again MongoDB. We are using a Redis-backed queue so jobs are not queued in MongoDB anymore but by default they are still inserted there once they are started. This is done to be able to see them in MRQ's dashboard as well as to store their results once they reach the `success` state. + + + + +### Redis-only queue + +If you don't need visibility on started jobs or on their results, you can actually bypass MongoDB altogether with this configuration: + +``` +RAW_QUEUES = { + "square_nostorage_raw": { + "statuses_no_storage": ("started", "success"), + "job_factory": lambda rawparam: { + "path": "tasks.Square", + "params": { + "n": rawparam + } + } + } +} +``` + +Let's try that with a single-process worker: + +``` +./enqueue.py square_nostorage_raw 20000 && mrq-worker square_nostorage_raw --greenlets 50 +... +[INFO] Worker spent 8.449 seconds performing 20000 jobs (2367.030 jobs/second) +``` + +Redis should be at less than 1% CPU load, so we can definitely keep adding processes: + +``` +./enqueue.py square_nostorage_raw 20000 && mrq-worker square_nostorage_raw --greenlets 50 --processes 5 +... +[INFO] Worker spent 11.884 seconds performing 20850 jobs (1754.444 jobs/second) +... +[INFO] Worker spent 11.851 seconds performing 20950 jobs (1767.750 jobs/second) +... +``` + +We are now close to 9000 jobs per second, maxing-out the local CPUs again! + +From there on, the sky is the limit! You should be able to run thousands of workers accross hundreds of machines before maxing-out a high-performance Redis instance. + +Beyond that, using using multiple queues on a [Redis Cluster](https://redis.io/topics/cluster-tutorial) will definitely allow you to run several million jobs per second. If you do, please drop us a line ;-) + + +## Choosing the right kind of queue + +### Queue types + +With the different settings explored in this tutorial, MRQ allows you to choose how much data you want to store in MongoDB and Redis. + +By choosing the right kind of queue for your jobs, you will strike a balance between performance, visibility in the dashboard, and safety guarantees. + +Here is a table to sum up the choices: + +| **Queue type** | **Regular** | **Raw** | **Raw with no_storage config** | +|----------------------------------------|-------------|-------------|--------------------------------| +| **Storage for queued jobs** | MongoDB | Redis | Redis | +| **Storage for started & success jobs** | MongoDB | MongoDB | None | +| **Performance** | + | ++ | +++ | +| **Visibility in the dashboard** | Full | After start | Job counts & failed jobs | +| **Safety** | +++ | ++ | + | + + + + +### Job safety + +A regular queue is guaranteed not to lose any jobs once they have been inserted in MongoDB. + +A raw queue can lose jobs if the worker abruptly exits in a short time window, between the dequeue from Redis and the insertion in MongoDB. + +A raw queue backed by Redis only won't be able to guarantee that a job is finished once it has been dequeued, if the worker abruptly exists. + +There are several ways to make raw queues safer. The easiest one is to use a `timed_set` raw queue backed by a Redis ZSET. We'll expand on this in an upcoming tutorial! diff --git a/docs/queues.md b/docs/queues.md index c77df9c2..c388a9ef 100644 --- a/docs/queues.md +++ b/docs/queues.md @@ -1,6 +1,6 @@ # Regular queues -With regular queues, MRQ stores the task metadata in MongoDB and the task IDs in a Redis list. This design allows a good compromise between performance and visibility. +With regular queues, MRQ stores the tasks in MongoDB. You can transform a queue into a [pile](https://en.wikipedia.org/wiki/LIFO_(computing)) by appending `_reverse` to its name: @@ -68,4 +68,6 @@ queue_raw_jobs("myqueue_timed_set", { }) ``` -For more examples of raw queue configuration, check [the tests](https://github.com/pricingassistant/mrq/blob/master/tests/fixtures/config-raw1.py) +For more examples of raw queue configuration, check [the tests](https://github.com/pricingassistant/mrq/blob/master/tests/fixtures/config-raw1.py). + +You should also read our tutorial on [Queue performance](queue-performance.md) to get a good overview of the different queue types. diff --git a/docs/recurring-jobs.md b/docs/recurring-jobs.md index 4bdeae92..c24944ad 100644 --- a/docs/recurring-jobs.md +++ b/docs/recurring-jobs.md @@ -4,4 +4,6 @@ MRQ provides a simple scheduler to help you run tasks every X units of time like See the [tests](https://github.com/pricingassistant/mrq/blob/master/tests/test_scheduler.py) -Please note that scheduling jobs once (setting a precise time for them to be executed in the future) is supported by `timed_set` [raw queues](queues.md#raw-queues). \ No newline at end of file +Please note that scheduling jobs once (setting a precise time for them to be executed in the future) is supported by `timed_set` [raw queues](queues.md#raw-queues). + +Be sure to do `mrq-run mrq.basetasks.indexes.EnsureIndexes` at least once to build the indexes for MRQ, because the scheduler depends on a unique index on the `hash` field. \ No newline at end of file diff --git a/docs/tests.md b/docs/tests.md index 2f53973b..0c2e933d 100644 --- a/docs/tests.md +++ b/docs/tests.md @@ -15,5 +15,6 @@ You can also open a shell inside the docker (just like you would enter in a virt ``` $ make docker -$ make ssh +$ make shell +$ py.test tests/ -v ``` diff --git a/examples/queue_performance/enqueue.py b/examples/queue_performance/enqueue.py new file mode 100755 index 00000000..bd2848a0 --- /dev/null +++ b/examples/queue_performance/enqueue.py @@ -0,0 +1,15 @@ +#!/usr/bin/env python +import sys +from mrq.context import setup_context +from mrq.job import queue_jobs, queue_raw_jobs + +setup_context() + +queue = sys.argv[1] +n = int(sys.argv[2]) + +if queue == "square": + queue_jobs("tasks.Square", [{"n": 42} for _ in range(n)], queue=queue) + +elif queue in ("square_raw", "square_nostorage_raw"): + queue_raw_jobs(queue, [42 for _ in range(n)]) diff --git a/examples/queue_performance/tasks.py b/examples/queue_performance/tasks.py new file mode 100644 index 00000000..40fe6d01 --- /dev/null +++ b/examples/queue_performance/tasks.py @@ -0,0 +1,23 @@ +from mrq.task import Task +import time + + +class Square(Task): + """ Returns the square of an integer """ + def run(self, params): + return int(params["n"]) ** 2 + + +class CPU(Task): + """ A CPU-intensive task """ + def run(self, params): + for n in range(int(params["n"])): + n ** n + return params["a"] + + +class IO(Task): + """ An IO-intensive task """ + def run(self, params): + time.sleep(float(params["sleep"])) + return params["a"] diff --git a/examples/scheduler/README.md b/examples/scheduler/README.md new file mode 100644 index 00000000..7cef4828 --- /dev/null +++ b/examples/scheduler/README.md @@ -0,0 +1,61 @@ +Simple task scheduler with MRQ +=========================== + +This is a simple exemple of a recurring task using MRQ. + +We have a simple task that print text to the terminal, and we want to launch it every 10 seconds. That is what this exemple is demonstrating. + +How to use +========== + +First, get into the docker image at the root of this directory: +``` +docker run -t -i -v `pwd`:/src -w /src pricingassistant/mrq bash +``` +Don't forget to `cd` in the directory of this exemple! + +Launch MongoDB & Redis if they are not already started: +``` +$ mongod & +$ redis-server & +``` + +Then launch a scheduler worker, feeding him the config. +``` +mrq-worker --scheduler --config config.py +``` + +In the config. we have described two time the task, with different parameters +You should then see the task printing its parameters every 10 seconds on the terminal. + +``` +2018-03-09 11:16:03.927182 [DEBUG] Scheduler: added tasks.Print 1 None None None None [["x","Another test."]] +2018-03-09 11:16:03.930823 [DEBUG] Scheduler: added tasks.Print 1 None None None None [["x","Test."]] +2018-03-09 11:16:03.960213 [DEBUG] Scheduler: queued tasks.Print 1 None None None None [["x","Another test."]] +2018-03-09 11:16:03.970537 [DEBUG] Scheduler: queued tasks.Print 1 None None None None [["x","Test."]] +2018-03-09 11:16:04.887654 [DEBUG] Starting tasks.Print({u'x': u'Another test.'}) +Hello world ! +Another test. +2018-03-09 11:16:04.901856 [DEBUG] Job 5aa26cf322f9db001dfdbdb0 success: 0.014454s total +2018-03-09 11:16:04.903789 [DEBUG] Starting tasks.Print({u'x': u'Test.'}) +Hello world ! +Test. +2018-03-09 11:16:04.905972 [DEBUG] Job 5aa26cf322f9db001dfdbdb1 success: 0.002258s total +2018-03-09 11:16:14.985899 [DEBUG] Scheduler: queued tasks.Print 1 None None None None [["x","Another test."]] +2018-03-09 11:16:14.988513 [DEBUG] Scheduler: queued tasks.Print 1 None None None None [["x","Test."]] +2018-03-09 11:16:15.961432 [DEBUG] Starting tasks.Print({u'x': u'Another test.'}) +Hello world ! +Another test. +2018-03-09 11:16:15.964429 [DEBUG] Job 5aa26cfe22f9db001dfdbdb4 success: 0.004088s total +2018-03-09 11:16:15.967952 [DEBUG] Starting tasks.Print({u'x': u'Test.'}) +Hello world ! +Test. +2018-03-09 11:16:15.970359 [DEBUG] Job 5aa26cfe22f9db001dfdbdb5 success: 0.002809s total +``` + +Other features +============== + +It is also possible to launch tasks everyday, or on specifics days of the week or month. + +We advise you to look at the [tests](https://github.com/pricingassistant/mrq/blob/master/tests/test_scheduler.py) for these use-cases. \ No newline at end of file diff --git a/examples/scheduler/config.py b/examples/scheduler/config.py new file mode 100644 index 00000000..ee03f256 --- /dev/null +++ b/examples/scheduler/config.py @@ -0,0 +1,19 @@ + +SCHEDULER_TASKS = [ + { + "path": "tasks.Print", + "params": { + "x": "Test." + }, + "interval": 1 + }, + { + "path": "tasks.Print", + "params": { + "x": "Another test." + }, + "interval": 1 + } +] + +SCHEDULER_INTERVAL = 10 diff --git a/examples/scheduler/tasks.py b/examples/scheduler/tasks.py new file mode 100644 index 00000000..91fa52e9 --- /dev/null +++ b/examples/scheduler/tasks.py @@ -0,0 +1,10 @@ + +from mrq.task import Task + + +class Print(Task): + + def run(self, params): + + print("Hello world !") + print(params["x"]) diff --git a/examples/simple_crawler/README.md b/examples/simple_crawler/README.md index 757f9dd6..79c4c2b2 100644 --- a/examples/simple_crawler/README.md +++ b/examples/simple_crawler/README.md @@ -7,7 +7,11 @@ This is a simple demo app that crawls a website, to demo some MRQ features. How to use ========== -First, get into a Python virtualenv (`make virtualenv`) or into the docker image at the root of this directory (`make ssh`) +First, get into the docker image at the root of this directory: +``` +docker run -t -i -v `pwd`:/src -w /src pricingassistant/mrq bash +``` +Don't forget to `cd` in the directory of this exemple! Then install MRQ and the packages needed for this example: ``` @@ -51,13 +55,13 @@ This is obviously a very simple crawler, production systems will be much more co Expected result for crawler.Report ================================== -As of 2015-02-20: +As of 2018-03-06: ``` Crawl stats =========== -URLs queued: 26 -URLs successfully crawled: 23 +URLs queued: 81 +URLs successfully crawled: 81 URLs redirected: 1 -Bytes fetched: 608131 -``` \ No newline at end of file +Bytes fetched: 4099658 +``` diff --git a/examples/simple_crawler/crawler.py b/examples/simple_crawler/crawler.py index d60d0574..391a7582 100644 --- a/examples/simple_crawler/crawler.py +++ b/examples/simple_crawler/crawler.py @@ -87,14 +87,14 @@ def run(self, params): collection = connections.mongodb_jobs.simple_crawler_urls print() - print( "Crawl stats") - print( "===========") - print( "URLs queued: %s" % collection.find().count()) - print( "URLs successfully crawled: %s" % collection.find({"fetched_date": {"$exists": True}}).count()) - print( "URLs redirected: %s" % collection.find({"redirected_to": {"$exists": True}}).count()) - print( "Bytes fetched: %s" % (list(collection.aggregate( + print("Crawl stats") + print("===========") + print("URLs queued: %s" % collection.find().count()) + print("URLs successfully crawled: %s" % collection.find({"fetched_date": {"$exists": True}}).count()) + print("URLs redirected: %s" % collection.find({"redirected_to": {"$exists": True}}).count()) + print("Bytes fetched: %s" % (list(collection.aggregate([ {"$group": {"_id": None, "sum": {"$sum": "$html_length"}}} - )) or [{}])[0].get("sum", 0)) + ])) or [{}])[0].get("sum", 0)) print() diff --git a/examples/simple_crawler/requirements.txt b/examples/simple_crawler/requirements.txt index 8c6e2f06..ac36c9e1 100644 --- a/examples/simple_crawler/requirements.txt +++ b/examples/simple_crawler/requirements.txt @@ -1,3 +1,2 @@ -mrq==0.1.11 lxml==3.4.2 -requests==2.4.3 \ No newline at end of file +requests==2.4.3 diff --git a/examples/timed_set/README.md b/examples/timed_set/README.md new file mode 100644 index 00000000..468c348b --- /dev/null +++ b/examples/timed_set/README.md @@ -0,0 +1,119 @@ +Simple Web Crawler with MRQ +=========================== + +This is a simple demo app that uses raw timed set queues. + + +How to use +========== + +First, get into a Python virtualenv (`make virtualenv`) or into the docker image at the root of this directory (`make ssh`) + +Then install MRQ and the packages needed for this example: +``` +$ cd examples/timed_set +``` + +Launch MongoDB & Redis if they are not already started: +``` +$ mongod & +$ redis-server & +``` + +Enqueue raw jobs with python script: + +``` +$ python enqueue_raw_jobs.py example_timed_set 4 10 +``` + +You can check redis entries: + +``` +$ redis-cli +$ ZRANGE mrq:q:example_timed_set 0 -1 WITHSCORES +``` + +You should see the following lines : (except timestamp of course) + +``` +1) "task_0" +2) "1520457493.2588561" +3) "task_1" +4) "1520457503.2588561" +5) "task_2" +6) "1520457513.2588561" +7) "task_3" +8) "1520457523.2588561" +``` + +You should also launch a dashboard to monitor the progress: +``` +$ mrq-dashboard +``` + +Then spawn a worker listenig to your timed_set queue example_timed_set: +``` +$ mrq-worker example_timed_set --config=/app/examples/timed_set/config.py +``` + +This is obviously a very simple example, production systems will be much more complex but it gives you an overview of timed set queues and a good starting point. + + +Expected result for mrq-worker example_timed_set --config=/app/examples/timed_set/config.py +================================== + +``` +[DEBUG] Starting example.Print({'test': 'task_0'}) +Hello World +Given params test is task_0 +Bye + +[DEBUG] Starting example.Print({'test': 'task_1'}) +Hello World +Time of last tasks execution 10.04 seconds +Given params test is task_1 +Bye + +[DEBUG] Starting example.Print({'test': 'task_2'}) +Hello World +Time of last tasks execution 10.03 seconds +Given params test is task_2 +Bye + +[DEBUG] Starting example.Print({'test': 'task_3'}) +Hello World +Time of last tasks execution 10.03 seconds +Given params test is task_3 +Bye +``` + +Limitation +================================== + +Note that the duration beetween enqueued tasks execution depends on when you start your worker. For example if you enqueue tasks every 10 seconds from now and you're waiting 20 seconds before spawning you worker, first 2 tasks we'll be executed directly as the expected execution time is in the past. Here is the expected output of that case : + +``` +[DEBUG] Starting example.Print({'test': 'task_0'}) +Hello World +Given params test is task_0 +Bye + +[DEBUG] Starting example.Print({'test': 'task_1'}) +Hello World +Last task was executed 3.13 seconds ago +Given params test is task_1 +Bye + +[DEBUG] Starting example.Print({'test': 'task_2'}) +Hello World +Last task was executed 10.03 seconds ago +Given params test is task_2 +Bye + +[DEBUG] Starting example.Print({'test': 'task_3'}) +Hello World +Last task was executed 10.03 seconds ago +Given params test is task_3 +Bye +``` + diff --git a/examples/timed_set/config.py b/examples/timed_set/config.py new file mode 100644 index 00000000..e294e026 --- /dev/null +++ b/examples/timed_set/config.py @@ -0,0 +1,10 @@ +RAW_QUEUES = { + "example_timed_set": { + "job_factory": lambda rawparam: { + "path": "example.Print", + "params": { + "test": rawparam + } + } + } +} \ No newline at end of file diff --git a/examples/timed_set/enqueue_raw_jobs.py b/examples/timed_set/enqueue_raw_jobs.py new file mode 100644 index 00000000..e43a6bcd --- /dev/null +++ b/examples/timed_set/enqueue_raw_jobs.py @@ -0,0 +1,16 @@ +#!/usr/bin/env python +import sys +import time +from mrq.context import setup_context +from mrq.job import queue_raw_jobs + +setup_context() + +queue = sys.argv[1] +n = int(sys.argv[2]) +t = int(sys.argv[3]) + +if queue in ("example_timed_set"): + now = time.time() + # every 10 seconds + queue_raw_jobs(queue, {"task_%s" % _: now + (_ + 1) * t for _ in range(n)}) diff --git a/examples/timed_set/example.py b/examples/timed_set/example.py new file mode 100644 index 00000000..c35c11b4 --- /dev/null +++ b/examples/timed_set/example.py @@ -0,0 +1,28 @@ +from mrq.task import Task +from datetime import datetime +import mrq.context as context +import time + + +class Print(Task): + + def run(self, params): + + print "Hello World" + + last_task = context.connections.redis.get("test:print") + if last_task: + print "Last task was executed %.2f seconds ago" % (time.time() - float(last_task)) + + context.connections.redis.set("test:print", time.time()) + + print "Given params test is", params["test"] + + print "Bye" + + +class RemoveRedisEntry(Task): + + def run(self, params): + + context.connections.redis.delete("test:print") \ No newline at end of file diff --git a/mkdocs.yml b/mkdocs.yml index d6d28661..96fab71a 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -17,8 +17,8 @@ pages: - ["metrics.md", "Visibility", "Metrics"] - ["io-monitoring.md", "Visibility", "I/O Monitoring"] -- ["design.md", "Advanced", "Design and architecture"] -- ["performance.md", "Advanced", "Performance"] +- ["performance.md", "Advanced", "Worker performance"] +- ["queue-performance.md", "Advanced", "Queue performance"] - ["recurring-jobs.md", "Advanced", "Recurring jobs"] - ["jobs-maintenance.md", "Advanced", "Jobs maintenance"] - ["best-practices.md", "Advanced", "Best practices"] diff --git a/mrq/agent.py b/mrq/agent.py new file mode 100644 index 00000000..3910c5cf --- /dev/null +++ b/mrq/agent.py @@ -0,0 +1,203 @@ +from .context import get_current_config, connections, log, run_task, metric +import time +import datetime +import gevent +import argparse +import random +import shlex +import traceback +from collections import defaultdict +from bson import ObjectId +from redis.lock import LuaLock +from .processes import Process, ProcessPool +from .utils import MovingETA, normalize_command +from .queue import Queue + + +class Agent(Process): + """ MRQ Agent manages its local worker pool and takes turns in orchestrating the others in its group. """ + + def __init__(self, worker_group=None): + self.greenlets = {} + self.id = ObjectId() + self.worker_group = worker_group or get_current_config()["worker_group"] + self.pool = ProcessPool(extra_env={ + "MRQ_AGENT_ID": str(self.id), + "MRQ_WORKER_GROUP": self.worker_group + }) + self.config = get_current_config() + self.status = "started" + metric("agent", data={"worker_group": self.worker_group, "agent_id": self.id}) + self.dateorchestrated = None + + # global redis key used to ensure only one agent orchestrator runs at a time + self.redis_queuestats_lock_key = "%s:queuestatslock" % (self.config["redis_prefix"]) + + # global HSET redis key used to store queue stats + self.redis_queuestats_key= "%s:queuestats" % (self.config["redis_prefix"]) + + def work(self): + + self.install_signal_handlers() + self.datestarted = datetime.datetime.utcnow() + + self.pool.start() + self.manage() + + self.greenlets["orchestrate"] = gevent.spawn(self.greenlet_orchestrate) + self.greenlets["orchestrate"].start() + + self.greenlets["manage"] = gevent.spawn(self.greenlet_manage) + self.greenlets["manage"].start() + + # Disabled for now + # self.greenlets["queuestats"] = gevent.spawn(self.greenlet_queuestats) + # self.greenlets["queuestats"].start() + + try: + self.pool.wait() + finally: + self.shutdown_now() + self.status = "stop" + self.manage() + + def shutdown_now(self): + self.pool.terminate() + + for g in self.greenlets.values(): + g.kill() + + def shutdown_graceful(self): + self.pool.stop(timeout=None) + + def greenlet_manage(self): + """ This greenlet always runs in background to update current status + in MongoDB every N seconds. + """ + + while True: + try: + self.manage() + except Exception as e: # pylint: disable=broad-except + log.error("When reporting: %s" % e) + finally: + time.sleep(self.config["report_interval"]) + + def manage(self): + + report = self.get_agent_report() + + try: + db = connections.mongodb_jobs.mrq_agents.find_and_modify({ + "_id": ObjectId(self.id) + }, {"$set": report}, upsert=True) + if not db: + return + except Exception as e: # pylint: disable=broad-except + log.debug("Agent report failed: %s" % e) + return + + # If the desired_workers was changed by an orchestrator, apply the changes locally + if self.status != "stop" and sorted(db.get("desired_workers", [])) != sorted(self.pool.desired_commands): + + group = self.fetch_worker_group_definition() + process_termination_timeout = float(group.get("process_termination_timeout") or 60) + self.pool.set_commands(db.get("desired_workers", []), timeout=process_termination_timeout) + + def get_agent_report(self): + report = { + "current_workers": [p["command"] for p in self.pool.processes], + "total_cpu": get_current_config()["total_cpu"], + "total_memory": get_current_config()["total_memory"], + "worker_group": self.worker_group, + "status": self.status, + "dateorchestrated": self.dateorchestrated, + "datestarted": self.datestarted, + "datereported": datetime.datetime.utcnow(), + "dateexpires": datetime.datetime.utcnow() + datetime.timedelta(seconds=(self.config["report_interval"] * 3) + 5) + } + metric("agent", data={"worker_group": self.worker_group, "agent_id": self.id, "worker_count": len(self.pool.processes)}) + return report + + def greenlet_orchestrate(self): + while True: + try: + self.orchestrate() + except Exception as e: + log.error("Orchestration error! %s" % e) + traceback.print_exc() + + time.sleep(self.config["orchestrate_interval"]) + + def orchestrate(self): + run_task("mrq.basetasks.orchestrator.Orchestrate", {}) + + def greenlet_queuestats(self): + + interval = min(self.config["orchestrate_interval"], 1 * 60) + lock_timeout = 5 * 60 + (interval * 2) + + while True: + lock = LuaLock(connections.redis, self.redis_queuestats_lock_key, + timeout=lock_timeout, thread_local=False, blocking=False) + with lock: + lock_expires = time.time() + lock_timeout + self.queue_etas = defaultdict(lambda: MovingETA(5)) + + while True: + self.queuestats() + + # Because queue stats can be expensive, we try to keep the lock on the same agent + lock_extend = (time.time() + lock_timeout) - lock_expires + lock_expires += lock_extend + lock.extend(lock_extend) + + time.sleep(interval) + + time.sleep(interval) + + def queuestats(self): + """ Compute ETAs for every known queue & subqueue """ + + start_time = time.time() + log.debug("Starting queue stats...") + + # Fetch all known queues + queues = [Queue(q) for q in Queue.all_known()] + + new_queues = {queue.id for queue in queues} + old_queues = set(self.queue_etas.keys()) + + for deleted_queue in old_queues.difference(new_queues): + self.queue_etas.pop(deleted_queue) + + t = time.time() + stats = {} + + for queue in queues: + cnt = queue.count_jobs_to_dequeue() + eta = self.queue_etas[queue.id].next(cnt, t=t) + + # Number of jobs to dequeue, ETA, Time of stats + stats[queue.id] = "%d %s %d" % (cnt, eta if eta is not None else "N", int(t)) + + with connections.redis.pipeline(transaction=True) as pipe: + if random.randint(0, 100) == 0 or len(stats) == 0: + pipe.delete(self.redis_queuestats_key) + if len(stats) > 0: + pipe.hmset(self.redis_queuestats_key, stats) + pipe.execute() + + log.debug("... done queue stats in %0.4fs" % (time.time() - start_time)) + + def fetch_worker_group_definition(self): + definition = connections.mongodb_jobs.mrq_workergroups.find_one({"_id": self.worker_group}) + + # Prepend all commands by their worker profile. + commands = [] + for command in definition.get("commands", []): + simplified_command, worker_count = normalize_command(command, self.worker_group) + commands.extend([simplified_command] * worker_count) + + definition["commands"] = commands + return definition diff --git a/mrq/basetasks/cleaning.py b/mrq/basetasks/cleaning.py index 31f9eb35..41551505 100644 --- a/mrq/basetasks/cleaning.py +++ b/mrq/basetasks/cleaning.py @@ -1,4 +1,4 @@ -from builtins import str +from future.builtins import str from mrq.queue import Queue from mrq.task import Task from mrq.job import Job @@ -73,188 +73,3 @@ def run(self, params): stats["requeued"] += 1 return stats - - -class RequeueRedisStartedJobs(Task): - - """ Requeue jobs that were started in Redis but not in Mongo. - - They could have been lost by a worker interrupt between - redis.lpop and mongodb.update - """ - - max_concurrency = 1 - - def run(self, params): - - redis_key_started = Queue.redis_key_started() - - stats = { - "fetched": 0, - "requeued": 0 - } - - # Fetch all the jobs started more than a minute ago - they should not - # be in redis:started anymore - job_ids = connections.redis.zrangebyscore( - redis_key_started, "-inf", time.time() - params.get("timeout", 60)) - - # TODO this should be wrapped inside Queue or Worker - # we shouldn't access these internals here - queue_obj = Queue("default") - unserialized_job_ids = queue_obj.unserialize_job_ids(job_ids) - - for i, job_id in enumerate(job_ids): - - queue = Job(unserialized_job_ids[i], start=False, fetch=False).fetch( - full_data=True).data["queue"] - - queue_obj = Queue(queue) - - stats["fetched"] += 1 - - log.info("Requeueing %s on %s" % (unserialized_job_ids[i], queue)) - - # TODO LUA script & don't rpush if not in zset anymore. - with connections.redis.pipeline(transaction=True) as pipeline: - pipeline.zrem(redis_key_started, job_id) - pipeline.rpush(queue_obj.redis_key, job_id) - pipeline.execute() - - stats["requeued"] += 1 - - return stats - - -class RequeueLostJobs(Task): - - """ Requeue jobs that were queued but don't appear in Redis anymore. - - They could have been lost by a Redis flush or another severe issue - """ - - max_concurrency = 1 - - def run(self, params): - - # If there are more than this much items on the queue, we don't try to check if our mongodb - # jobs are still queued. - max_queue_items = params.get("max_queue_items", 1000) - - stats = { - "fetched": 0, - "requeued": 0 - } - - # This was only checking in Redis and wasn't resistant to a redis-wide flush. - # Doing Queue.all() is slower but covers more edge cases. - # all_queues = Queue.all_known() - - all_queues = Queue.all() - - log.info("Checking %s queues" % len(all_queues)) - - for queue_name in all_queues: - - queue = Queue(queue_name) - queue_size = queue.size() - - # If the queue is raw, the jobs were only stored in redis so they are lost for good. - if queue.is_raw: - continue - - log.info("Checking queue %s" % queue_name) - - if queue_size > max_queue_items: - log.info("Stopping because queue %s has %s items" % - (queue_name, queue_size)) - continue - - queue_jobs_ids = set(queue.list_job_ids(limit=max_queue_items + 1)) - if len(queue_jobs_ids) >= max_queue_items: - log.info( - "Stopping because queue %s actually had more than %s items" % - (queue_name, len(queue_jobs_ids))) - continue - - for job_data in connections.mongodb_jobs.mrq_jobs.find({ - "queue": queue_name, - "status": "queued" - }, projection={"_id": 1}).sort([["_id", 1]]): - - stats["fetched"] += 1 - - if str(job_data["_id"]) in queue_jobs_ids: - log.info("Found job %s on queue %s. Stopping" % (job_data["_id"], queue.id)) - break - - # At this point, this job is not on the queue and we're sure - # the queue is less than max_queue_items - # We can safely requeue the job. - log.info("Requeueing %s on %s" % (job_data["_id"], queue.id)) - - stats["requeued"] += 1 - job = Job(job_data["_id"]) - job.requeue(queue=queue_name) - - return stats - - -class MigrateKnownQueues(Task): - """ - Migrate known_queues from old set format to new zset - """ - - max_concurrency = 1 - - def run(self, params): - key = "%s:known_queues" % get_current_config()["redis_prefix"] - for queue in connections.redis.smembers(key): - Queue(queue).add_to_known_queues() - - -class CleanKnownQueues(Task): - - """ - Cleans the known queues in Redis. - - To be deleted, a queue must: - - not have been used in the last 7 days - - be empty - """ - - max_concurrency = 1 - - def run(self, params): - - max_age = int(params.get("max_age") or (7 * 86400)) - pretend = bool(params.get("pretend")) - check_mongo = bool(params.get("check_mongo")) - - known_queues = Queue.redis_known_queues() - - removed_queues = [] - - queues_from_config = Queue.all_known_from_config() - - print("Found %s known queues & %s from config" % (len(known_queues), len(queues_from_config))) - - # Only clean queues older than N days - time_threshold = time.time() - max_age - for queue, time_last_used in known_queues.items(): - if queue in queues_from_config: - continue - if time_last_used < time_threshold: - q = Queue(queue, add_to_known_queues=False) - size = q.size() - if check_mongo: - size += connections.mongodb_jobs.mrq_jobs.count({"queue": queue}) - if size == 0: - removed_queues.append(queue) - print("Removing empty queue '%s' from known queues ..." % queue) - if not pretend: - q.remove_from_known_queues() - - print("Cleaned %s queues" % len(removed_queues)) - - return removed_queues diff --git a/mrq/basetasks/indexes.py b/mrq/basetasks/indexes.py new file mode 100644 index 00000000..a0ce9472 --- /dev/null +++ b/mrq/basetasks/indexes.py @@ -0,0 +1,45 @@ +from mrq.task import Task +from mrq.context import connections + + +class EnsureIndexes(Task): + + def run(self, params): + + if connections.mongodb_logs: + connections.mongodb_logs.mrq_logs.ensure_index( + [("job", 1)], background=True) + connections.mongodb_logs.mrq_logs.ensure_index( + [("worker", 1)], background=True, sparse=True) + + connections.mongodb_jobs.mrq_workers.ensure_index( + [("status", 1)], background=True) + connections.mongodb_jobs.mrq_workers.ensure_index( + [("datereported", 1)], background=True, expireAfterSeconds=3600) + + connections.mongodb_jobs.mrq_jobs.ensure_index( + [("status", 1)], background=True) + connections.mongodb_jobs.mrq_jobs.ensure_index( + [("path", 1)], background=True) + connections.mongodb_jobs.mrq_jobs.ensure_index( + [("worker", 1)], background=True, sparse=True) + connections.mongodb_jobs.mrq_jobs.ensure_index( + [("queue", 1)], background=True) + connections.mongodb_jobs.mrq_jobs.ensure_index( + [("dateexpires", 1)], sparse=True, background=True, expireAfterSeconds=0) + connections.mongodb_jobs.mrq_jobs.ensure_index( + [("dateretry", 1)], sparse=True, background=True) + connections.mongodb_jobs.mrq_jobs.ensure_index( + [("datequeued", 1)], background=True) + connections.mongodb_jobs.mrq_jobs.ensure_index( + [("queue", 1), ("status", 1), ("datequeued", 1), ("_id", 1)], background=True) + + connections.mongodb_jobs.mrq_scheduled_jobs.ensure_index( + [("hash", 1)], unique=True, background=False) + + connections.mongodb_jobs.mrq_agents.ensure_index( + [("datereported", 1)], background=True) + connections.mongodb_jobs.mrq_agents.ensure_index( + [("dateexpires", 1)], background=True, expireAfterSeconds=0) + connections.mongodb_jobs.mrq_agents.ensure_index( + [("worker_group", 1)], background=True) diff --git a/mrq/basetasks/orchestrator.py b/mrq/basetasks/orchestrator.py new file mode 100644 index 00000000..124502b9 --- /dev/null +++ b/mrq/basetasks/orchestrator.py @@ -0,0 +1,103 @@ +from future.builtins import str +from mrq.queue import Queue +from mrq.task import Task +from mrq.job import Job +from mrq.context import log, connections, run_task, get_current_config, subpool_map +from collections import defaultdict +import math +import shlex +import argparse +from ..config import add_parser_args +from ..utils import normalize_command +import traceback +import datetime +import re + + +class Orchestrate(Task): + + max_concurrency = 1 + + def run(self, params): + + self.config = get_current_config() + + concurrency = int(params.get("concurrency", 5)) + groups = self.fetch_worker_group_definitions() + if len(groups) == 0: + log.error("No worker group definition yet. Can't orchestrate!") + return + + subpool_map(concurrency, self.orchestrate, groups) + + def redis_orchestrator_lock_key(self, worker_group): + """ Returns the global redis key used to ensure only one agent orchestrator runs at a time """ + return "%s:orchestratorlock:%s" % (get_current_config()["redis_prefix"], worker_group) + + def orchestrate(self, worker_group): + try: + self.do_orchestrate(worker_group) + except Exception as e: + log.error("Orchestration error! %s" % e) + traceback.print_exc() + + def do_orchestrate(self, group): + """ Manage the desired workers of *all* the agents in the given group """ + + log.debug("Starting orchestration run for worker group %s" % group["_id"]) + + agents = self.fetch_worker_group_agents(group) + + # Evaluate what workers are currently, rightfully there. They won't be touched. + for agent in agents: + desired_workers = self.get_desired_workers_for_agent(group, agent) + agent["new_desired_workers"] = [] + agent["new_desired_workers"] = desired_workers + + for agent in agents: + if sorted(agent["new_desired_workers"]) != sorted(agent.get("desired_workers", [])): + connections.mongodb_jobs.mrq_agents.update_one({"_id": agent["_id"]}, {"$set": { + "desired_workers": agent["new_desired_workers"] + }}) + + # Remember the date of the last successful orchestration (will be reported) + self.dateorchestrated = datetime.datetime.utcnow() + + log.debug("Orchestration finished.") + + def redis_queuestats_key(self): + """ Returns the global HSET redis key used to store queue stats """ + return "%s:queuestats" % (get_current_config()["redis_prefix"]) + + def get_desired_workers_for_agent(self, group, agent): + return group.get("commands", []) + + def fetch_worker_group_reports(self, worker_group, projection=None): + return list(connections.mongodb_jobs.mrq_workers.find({ + "config.worker_group": worker_group["_id"] + }, projection=projection)) + + def fetch_worker_group_definitions(self): + + definitions = list(connections.mongodb_jobs.mrq_workergroups.find()) + + for definition in definitions: + commands = [] + # Prepend all commands by their worker group. + for command in definition.get("commands", []): + simplified_command, worker_count = normalize_command(command, definition["_id"]) + commands.extend([simplified_command] * worker_count) + definition["commands"] = commands + + return definitions + + def fetch_worker_group_agents(self, worker_group): + return list(connections.mongodb_jobs.mrq_agents.find({"worker_group": worker_group["_id"], "status": "started"})) + + def get_config_for_profile(self, profile): + parser = argparse.ArgumentParser() + add_parser_args(parser, "worker") + parts = shlex.split(profile["command"]) + if "mrq-worker" in parts: + parts = parts[parts.index("mrq-worker") + 1:] + return parser.parse_args(parts) diff --git a/mrq/basetasks/utils.py b/mrq/basetasks/utils.py index c6652a8f..f409d95a 100644 --- a/mrq/basetasks/utils.py +++ b/mrq/basetasks/utils.py @@ -1,10 +1,10 @@ from __future__ import print_function from future.utils import itervalues -from builtins import str +from future.builtins import str from mrq.task import Task from mrq.queue import Queue from bson import ObjectId -from mrq.context import connections, get_current_config +from mrq.context import connections, get_current_config, get_current_job from collections import defaultdict from mrq.utils import group_iter import datetime @@ -33,6 +33,8 @@ def run(self, params): def build_query(self): query = {} + current_job = get_current_job() + if self.params.get("id"): query["_id"] = ObjectId(self.params.get("id")) @@ -49,12 +51,17 @@ def build_query(self): query[k] = {"$in": list(self.params[k])} else: query[k] = self.params[k] + if query.get("worker"): + query["worker"] = ObjectId(query["worker"]) if self.params.get("params"): params_dict = json.loads(self.params.get("params")) # pylint: disable=no-member for key in params_dict: query["params.%s" % key] = params_dict[key] + + if current_job and "_id" not in query: + query["_id"] = {"$lte": current_job.id} return query @@ -120,6 +127,7 @@ def perform_action(self, action, query, destination_queue): updates = { "status": "queued", + "datequeued": datetime.datetime.utcnow(), "dateupdated": datetime.datetime.utcnow() } @@ -133,11 +141,4 @@ def perform_action(self, action, query, destination_queue): "_id": {"$in": jobs_by_queue[queue]} }, {"$set": updates}, multi=True) - # Between these two lines, jobs can become "lost" too. - - Queue(destination_queue or queue, add_to_known_queues=True).enqueue_job_ids( - [str(x) for x in jobs_by_queue[queue]]) - - print(stats) - return stats diff --git a/mrq/bin/mrq_agent.py b/mrq/bin/mrq_agent.py new file mode 100644 index 00000000..f9d2f94a --- /dev/null +++ b/mrq/bin/mrq_agent.py @@ -0,0 +1,40 @@ +#!/usr/bin/env python +import os +import sys +is_pypy = '__pypy__' in sys.builtin_module_names + +# 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 and not is_pypy: + os.environ["GEVENT_RESOLVER"] = "ares" + +from gevent import monkey +monkey.patch_all() + +import argparse + +sys.path.insert(0, os.getcwd()) + +from mrq import config +from mrq.agent import Agent +from mrq.context import set_current_config + + +def main(): + + parser = argparse.ArgumentParser(description='Start a MRQ agent') + + cfg = config.get_config(parser=parser, config_type="agent", sources=("file", "env", "args")) + + set_current_config(cfg) + + agent = Agent() + + agent.work() + + sys.exit(agent.exitcode) + + +if __name__ == "__main__": + main() diff --git a/mrq/bin/mrq_run.py b/mrq/bin/mrq_run.py index eedf6e4c..c38fec92 100755 --- a/mrq/bin/mrq_run.py +++ b/mrq/bin/mrq_run.py @@ -2,18 +2,20 @@ from __future__ import print_function import os +import sys +is_pypy = '__pypy__' in sys.builtin_module_names # 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: + +if "GEVENT_RESOLVER" not in os.environ and not is_pypy: os.environ["GEVENT_RESOLVER"] = "ares" # We must still monkey-patch the methods for job sub-pools. from gevent import monkey monkey.patch_all() -import sys import argparse import ujson as json import json as json_stdlib @@ -22,7 +24,7 @@ sys.path.insert(0, os.getcwd()) from mrq import config, utils -from mrq.context import set_current_config, set_current_job, connections +from mrq.context import set_current_config, set_logger_config, set_current_job, connections from mrq.job import queue_job from mrq.utils import load_class_by_path, MongoJSONEncoder @@ -33,6 +35,7 @@ def main(): cfg = config.get_config(parser=parser, config_type="run", sources=("file", "env", "args")) cfg["is_cli"] = True set_current_config(cfg) + set_logger_config() if len(cfg["taskargs"]) == 1: params = json.loads(cfg["taskargs"][0]) # pylint: disable=no-member diff --git a/mrq/bin/mrq_worker.py b/mrq/bin/mrq_worker.py index ec58575c..b7b6da8c 100755 --- a/mrq/bin/mrq_worker.py +++ b/mrq/bin/mrq_worker.py @@ -1,118 +1,57 @@ #!/usr/bin/env python import os -from builtins import str +from future.builtins import str +import sys +is_pypy = '__pypy__' in sys.builtin_module_names # 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: +if "GEVENT_RESOLVER" not in os.environ and not is_pypy: os.environ["GEVENT_RESOLVER"] = "ares" from gevent import monkey -monkey.patch_all(subprocess=False) +monkey.patch_all() -import sys import tempfile import signal import psutil import argparse - -try: - import subprocess32 as subprocess -except: - import subprocess +import pipes sys.path.insert(0, os.getcwd()) from mrq import config from mrq.utils import load_class_by_path -from mrq.context import set_current_config +from mrq.context import set_current_config, set_logger_config def main(): - parser = argparse.ArgumentParser(description='Start a RQ worker') + parser = argparse.ArgumentParser(description='Start a MRQ worker') cfg = config.get_config(parser=parser, config_type="worker", sources=("file", "env", "args")) - # If we are launching with a --processes option and without the SUPERVISOR_ENABLED env - # then we should just call supervisord. - if cfg["processes"] > 0 and not os.environ.get("SUPERVISOR_ENABLED"): - - # We wouldn't need to do all that if supervisord supported environment - # variables in all its config fields! - with open(cfg["supervisord_template"], "r") as f: - conf = f.read() - - fh, path = tempfile.mkstemp(prefix="mrqsupervisordconfig") - f = os.fdopen(fh, "w") - - # We basically relaunch ourselves, but the config will contain the - # MRQ_SUPERVISORD_ISWORKER env. - conf = conf.replace("{{ SUPERVISORD_COMMAND }}", " ".join(sys.argv)) - conf = conf.replace( - "{{ SUPERVISORD_PROCESSES }}", str(cfg["processes"])) - - f.write(conf) - f.close() - - try: - - # start_new_session=True avoids sending the current process' - # signals to the child. - process = subprocess.Popen( - ["supervisord", "-c", path], start_new_session=True) - - def sigint_handler(signum, frame): # pylint: disable=unused-argument - - # At this point we need to send SIGINT to all workers. Unfortunately supervisord - # doesn't support this, so we have to find all the children pids and send them the - # signal ourselves :-/ - # https://github.com/Supervisor/supervisor/issues/179 - # - psutil_process = psutil.Process(process.pid) - worker_processes = psutil_process.get_children(recursive=False) + set_current_config(cfg) + set_logger_config() - if len(worker_processes) == 0: - return process.send_signal(signal.SIGTERM) + # If we are launching with a --processes option and without MRQ_IS_SUBPROCESS, we are a manager process + if cfg["processes"] > 0 and not os.environ.get("MRQ_IS_SUBPROCESS"): - for child_process in worker_processes: - child_process.send_signal(signal.SIGINT) + from mrq.supervisor import Supervisor - # Second time sigint is used, we should terminate supervisord itself which - # will send SIGTERM to all the processes anyway. - signal.signal(signal.SIGINT, sigterm_handler) + command = " ".join(map(pipes.quote, sys.argv)) + w = Supervisor(command, numprocs=cfg["processes"]) + w.work() + sys.exit(w.exitcode) - # Wait for all the childs to finish - for child_process in worker_processes: - child_process.wait() - - # Then stop supervisord itself. - process.send_signal(signal.SIGTERM) - - def sigterm_handler(signum, frame): # pylint: disable=unused-argument - process.send_signal(signal.SIGTERM) - - signal.signal(signal.SIGINT, sigint_handler) - signal.signal(signal.SIGTERM, sigterm_handler) - - process.wait() - - finally: - os.remove(path) - - # If not, start the actual worker + # If not, start an actual worker else: worker_class = load_class_by_path(cfg["worker_class"]) - - set_current_config(cfg) - w = worker_class() - - exitcode = w.work() - - sys.exit(exitcode) + w.work() + sys.exit(w.exitcode) if __name__ == "__main__": main() diff --git a/mrq/config.py b/mrq/config.py index 1def7c86..7c53ec7b 100644 --- a/mrq/config.py +++ b/mrq/config.py @@ -1,12 +1,14 @@ from __future__ import print_function -from builtins import str +from future.builtins import str import argparse import os import sys import re +import psutil from .version import VERSION -from .utils import get_local_ip, DelimiterArgParser +from .utils import get_local_ip import atexit +import logging def add_parser_args(parser, config_type): @@ -69,21 +71,6 @@ def add_parser_args(parser, config_type): help='MongoDB URI for the logs database. ' + ' "0" will disable remote logs, "1" will use main MongoDB.') - parser.add_argument( - '--mongodb_logs_size', - action='store', - default=16 * - 1024 * - 1024, - type=int, - help='If provided, sets the log collection to capped to that amount of bytes.') - - parser.add_argument( - '--no_mongodb_ensure_indexes', - action='store_true', - default=False, - help='If provided, skip the creation of MongoDB indexes at worker startup.') - parser.add_argument( '--redis', action='store', @@ -116,12 +103,6 @@ def add_parser_args(parser, config_type): action='store', help='Specify a different name') - parser.add_argument( - '--quiet', - default=False, - action='store_true', - help='Don\'t output task logs') - parser.add_argument( '--config', '-c', @@ -135,6 +116,30 @@ def add_parser_args(parser, config_type): action="store", help='Path to a custom worker class') + parser.add_argument( + '--quiet', + default=False, + action='store_true', + help='Don\'t output task logs') + + parser.add_argument( + '--log_handler', + default="mrq.logger.MongoHandler", + action="store", + help='Path to a log handler class') + + parser.add_argument( + '--log_format', + default="%(asctime)s [%(levelname)s] %(message)s", + action="store", + help='log format') + + parser.add_argument( + '--log_level', + default="DEBUG", + action="store", + help='set the logging level') + parser.add_argument( '--version', '-v', @@ -155,6 +160,13 @@ def add_parser_args(parser, config_type): type=str, help='Adds random latency to the network calls, zero to N seconds. Can be a range (1-2)') + parser.add_argument( + '--default_job_ttl', + default=180 * 24 * 3600, + action='store', + type=float, + help='Seconds the tasks are kept in MongoDB when statuses are not success, abort, cancel and started') + parser.add_argument( '--default_job_result_ttl', default=7 * 24 * 3600, @@ -255,8 +267,52 @@ def add_parser_args(parser, config_type): type=str, help='Bind the dashboard to this IP. Default is "0.0.0.0", use "127.0.0.1" to restrict access.') - # Worker-specific args + # Agent-specific args + elif config_type == "agent": + + parser.add_argument( + '--worker_group', + default="default", + action="store", + type=str, + help='The name of the worker group to manage') + + parser.add_argument( + '--total_memory', + default=int(psutil.virtual_memory().total * 0.8 / (1024 * 1024)), + action="store", + type=int, + help="How much memory MB this agent's workers can use. Used for scheduling, not a hard limit.") + + parser.add_argument( + '--total_cpu', + default=psutil.cpu_count(logical=True) * 1024, + action="store", + type=int, + help="How much CPU units this agent's workers can use. We recommend using 1024 per CPU.") + + parser.add_argument( + '--orchestrate_interval', + default=30, + action="store", + type=float, + help="How much seconds to wait between orchestration runs.") + + parser.add_argument( + '--report_interval', + default=10, + action='store', + type=float, + help='Seconds between agent reports to MongoDB') + + parser.add_argument( + '--autoscaling_taskpath', + default=None, + action='store', + type=str, + help='Path to a task that can perform autoscaling actions during orchestration') + # Worker-specific args elif config_type == "worker": parser.add_argument( @@ -279,8 +335,8 @@ def add_parser_args(parser, config_type): default=0, type=int, action='store', - help='Max memory (in Mb) after which the process will be shut down. Use with --processes [1-N]' + - 'to have supervisord automatically respawn the worker when this happens') + help='Max memory (in Mb) after which the process will be shut down. Use with mrq-agent' + + 'to automatically respawn the worker when this happens') parser.add_argument( '--greenlets', @@ -297,18 +353,7 @@ def add_parser_args(parser, config_type): default=0, type=int, action='store', - help='Number of processes to launch with supervisord') - - default_template = os.path.abspath(os.path.join( - os.path.dirname(__file__), - "supervisord_templates/default.conf" - )) - - parser.add_argument( - '--supervisord_template', - default=default_template, - action='store', - help='Path of supervisord template to use') + help='Number of processes to launch') parser.add_argument( '--scheduler', @@ -316,6 +361,12 @@ def add_parser_args(parser, config_type): action='store_true', help='Run the scheduler') + parser.add_argument( + '--ensure_indexes', + default=False, + action='store_true', + help='Ensures the internal MongoDB indexes of MRQ are built, or does so in the background') + parser.add_argument( '--scheduler_interval', default=60, @@ -337,6 +388,41 @@ def add_parser_args(parser, config_type): type=str, help='Filepath of a json dump of the worker status. Disabled if none') + parser.add_argument( + '--agent_id', + default="", + action='store', + type=str, + help='ID of the Agent this worker process is linked to') + + parser.add_argument( + '--worker_id', + default="", + action='store', + type=str, + help='ID of this worker process. Should be left empty to be autogenerated in most cases.') + + parser.add_argument( + '--worker_group', + default="", + action='store', + type=str, + help='Worker group of the agent this worker was launched from') + + parser.add_argument( + '--task_whitelist', + default="", + action='store', + type=str, + help='Comma-separated list of task paths to do exclusively on this worker. Only for regular queues.') + + parser.add_argument( + '--task_blacklist', + default="", + action='store', + type=str, + help='Comma-separated list of task paths to do exclude from this worker. Only for regular queues.') + parser.add_argument( 'queues', nargs='*', @@ -345,7 +431,7 @@ def add_parser_args(parser, config_type): parser.add_argument( '--subqueues_refresh_interval', - default=10, + default=60, action='store', type=float, help="Seconds between worker refreshes of the known subqueues") @@ -357,12 +443,6 @@ def add_parser_args(parser, config_type): type=float, help="Seconds between worker refreshes of the paused queues list") - parser.add_argument( - '--subqueues_delimiter', - default='/', - help='Delimiter between main queue and subqueue names', - action=DelimiterArgParser) - parser.add_argument( '--admin_port', default=0, @@ -384,6 +464,13 @@ def add_parser_args(parser, config_type): type=str, help='Overwrite the local IP, to be displayed in the dashboard.') + parser.add_argument( + '--external_ip', + default=None, + action="store", + type=str, + help='Overwrite the external IP, to be displayed in the dashboard.') + parser.add_argument( '--max_latency', default=1., @@ -401,6 +488,12 @@ def add_parser_args(parser, config_type): 'to dequeue them in command-line order.') +class ArgumentParserIgnoringDefaults(argparse.ArgumentParser): + def add_argument(self, *args, **kwargs): + kwargs.pop("default", None) + return argparse.ArgumentParser.add_argument(self, *args, **kwargs) + + def get_config( sources=( "file", @@ -425,15 +518,15 @@ def get_config( # Keys that can't be passed from the command line default_config["tasks"] = {} + default_config["log_handlers"] = {} default_config["scheduled_tasks"] = {} - # Only keep values different from config, actually passed on the command - # line + # Only keep values actually passed on the command line from_args = {} if "args" in sources: - for k, v in parser.parse_args().__dict__.items(): - if default_config[k] != v: - from_args[k] = v + cmdline_parser = ArgumentParserIgnoringDefaults(argument_default=argparse.SUPPRESS) + add_parser_args(cmdline_parser, config_type) + from_args = cmdline_parser.parse_args().__dict__ # If we were given another config file, use it @@ -483,12 +576,12 @@ def get_config( merged_config.update(extra) if merged_config["profile"]: - import cProfile + import cProfile, pstats profiler = cProfile.Profile() profiler.enable() def print_profiling(): - profiler.print_stats(sort="cumulative") + pstats.Stats(profiler).sort_stats("cumulative").print_stats() atexit.register(print_profiling) diff --git a/mrq/context.py b/mrq/context.py index 6474224c..0a77f64d 100644 --- a/mrq/context.py +++ b/mrq/context.py @@ -1,18 +1,18 @@ from future import standard_library standard_library.install_aliases() -from builtins import next -from builtins import map +from future.builtins import next, map from past.builtins import basestring -from .logger import Logger +import logging import gevent import gevent.pool import urllib.parse +import sys import time import pymongo import traceback from .utils import LazyObject, load_class_by_path -from itertools import count as itertools_count from .config import get_config +from .subpool import subpool_map, subpool_imap # This should be MRQ's only Python object shared by all the jobs in the same process _GLOBAL_CONTEXT = { @@ -28,7 +28,7 @@ } # Global log object, usable from all jobs -log = Logger(None, job="current") +log = logging.getLogger("mrq.current") def setup_context(**kwargs): @@ -65,10 +65,30 @@ def set_current_worker(worker): def get_current_worker(): return _GLOBAL_CONTEXT["worker"] +def set_logger_config(): + config = _GLOBAL_CONTEXT["config"] + if config.get("quiet"): + log.disabled = True + else: + log_format = config["log_format"] + logging.basicConfig(format=log_format) + log.setLevel(getattr(logging, config["log_level"])) + + handlers = config["log_handlers"].keys() if config["log_handlers"] else [config["log_handler"]] + for handler in handlers: + handler_class = load_class_by_path(handler) + handler_config = config["log_handlers"].get(handler, {}) + handler_format = handler_config.pop("format", log_format) + handler_level = getattr(logging, handler_config.pop("level", config["log_level"])) + log_handler = handler_class(**handler_config) + formatter = logging.Formatter(handler_format) + log_handler.setFormatter(formatter) + log_handler.setLevel(handler_level) + log.addHandler(log_handler) + def set_current_config(config): _GLOBAL_CONTEXT["config"] = config - log.quiet = config["quiet"] if config["add_network_latency"] != "0" and config["add_network_latency"]: from mrq.monkey import patch_network_latency @@ -82,11 +102,11 @@ def set_current_config(config): from mrq.monkey import patch_io_all patch_io_all(config) - if config["mongodb_logs"] == "0": - log.handler.collection = False - def get_current_config(): + if not _GLOBAL_CONTEXT["config"]: + log.warning("get_current_config was called before setup of MRQ's environment. " + "Use context.setup_context() for setting up MRQ's environment.") return _GLOBAL_CONTEXT["config"] @@ -130,7 +150,7 @@ def versiontuple(v): password=redis_url.password, max_connections=int(config.get("redis_max_connections")), timeout=int(config.get("redis_timeout")), - decode_responses=True + decode_responses=False ) return pyredis.StrictRedis(connection_pool=redis_pool) @@ -203,121 +223,6 @@ def trace(*args): greenlet.settrace(trace) # pylint: disable=no-member -def subpool_map(pool_size, func, iterable): - """ Starts a Gevent pool and run a map. Takes care of setting current_job and cleaning up. """ - - if not pool_size: - return [func(*args) for args in iterable] - - counter = itertools_count() - - current_job = get_current_job() - - def inner_func(*args): - """ As each call to 'func' will be done in a random greenlet of the subpool, we need to - register their IDs with set_current_job() to make get_current_job() calls work properly - inside 'func'. - """ - next(counter) - if current_job: - set_current_job(current_job) - - try: - ret = func(*args) - except Exception as exc: - trace = traceback.format_exc() - log.error("Error in subpool: %s \n%s" % (exc, trace)) - raise - - if current_job: - set_current_job(None) - return ret - - def inner_iterable(): - """ This will be called inside the pool's main greenlet, which ID also needs to be registered """ - if current_job: - set_current_job(current_job) - - for x in iterable: - yield x - - if current_job: - set_current_job(None) - - start_time = time.time() - pool = gevent.pool.Pool(size=pool_size) - ret = pool.map(inner_func, inner_iterable()) - pool.join(raise_error=True) - total_time = time.time() - start_time - - log.debug("SubPool ran %s greenlets in %0.6fs" % (counter, total_time)) - - return ret - - -def subpool_imap(pool_size, func, iterable, flatten=False, unordered=False, buffer_size=None): - """ Generator version of subpool_map. Should be used with unordered=True for optimal performance """ - - if not pool_size: - for args in iterable: - yield func(*args) - - counter = itertools_count() - - current_job = get_current_job() - - def inner_func(*args): - """ As each call to 'func' will be done in a random greenlet of the subpool, we need to - register their IDs with set_current_job() to make get_current_job() calls work properly - inside 'func'. - """ - next(counter) - if current_job: - set_current_job(current_job) - - try: - ret = func(*args) - except Exception as exc: - trace = traceback.format_exc() - log.error("Error in subpool: %s \n%s" % (exc, trace)) - raise - - if current_job: - set_current_job(None) - return ret - - def inner_iterable(): - """ This will be called inside the pool's main greenlet, which ID also needs to be registered """ - if current_job: - set_current_job(current_job) - - for x in iterable: - yield x - - if current_job: - set_current_job(None) - - start_time = time.time() - pool = gevent.pool.Pool(size=pool_size) - - if unordered: - iterator = pool.imap_unordered(inner_func, inner_iterable(), maxsize=buffer_size or pool_size) - else: - iterator = pool.imap(inner_func, inner_iterable()) - - for x in iterator: - if flatten: - for y in x: - yield y - else: - yield x - - pool.join(raise_error=True) - total_time = time.time() - start_time - - log.debug("SubPool ran %s greenlets in %0.6fs" % (counter, total_time)) - - def run_task(path, params): """ Runs a task code synchronously """ task_class = load_class_by_path(path) diff --git a/mrq/dashboard/app.py b/mrq/dashboard/app.py index 7c1933cb..bbd4541c 100644 --- a/mrq/dashboard/app.py +++ b/mrq/dashboard/app.py @@ -15,6 +15,7 @@ import json import argparse from werkzeug.serving import run_simple +from future.builtins import str sys.path.insert(0, os.getcwd()) @@ -113,6 +114,7 @@ def api_taskpaths(): return jsonify(data) +# Route to be deprecated! @app.route('/workers') @requires_auth def get_workers(): @@ -122,6 +124,24 @@ def get_workers(): return jsonify(data) +@app.route('/api/workergroups', methods=["GET"]) +@requires_auth +def get_workergroups(): + collection = connections.mongodb_jobs.mrq_workergroups + data = {"workergroups": {str(row.pop("_id")): row for row in collection.find(sort=[("_id", 1)])}} + return jsonify(data) + + +@app.route('/api/workergroups', methods=["POST"]) +@requires_auth +def post_workergroups(): + workergroups = json.loads(request.form["workergroups"]) + for k, v in workergroups.iteritems(): + connections.mongodb_jobs.mrq_workergroups.update_one({"_id": k}, {"$set": v}, upsert=True) + + return jsonify({"status": "ok"}) + + def build_api_datatables_query(req): query = {} @@ -133,11 +153,15 @@ def build_api_datatables_query(req): for param in ["queue", "path", "exceptiontype"]: if req.args.get(param): if "*" in req.args[param]: - regexp = "^%s$" % re.escape(req.args[param]).replace("*", ".*") - query[param] = re.compile(regexp) + regexp = "^%s$" % req.args[param].replace("*", ".*") + query[param] = {"$regex": regexp} else: query[param] = req.args[param] + if req.args.get("queue") and req.args["queue"].endswith("/"): + subqueues = Queue(req.args["queue"]).get_known_subqueues() + query["queue"] = {"$in": list(subqueues)} + if req.args.get("status"): statuses = req.args["status"].split("-") if len(statuses) == 1: @@ -172,7 +196,6 @@ def api_datatables(unit): sort = None skip = int(request.args.get("iDisplayStart", 0)) limit = int(request.args.get("iDisplayLength", 20)) - with_mongodb_size = bool(request.args.get("with_mongodb_size")) if unit == "queues": @@ -180,16 +203,8 @@ def api_datatables(unit): for name in Queue.all_known(): queue = Queue(name) - jobs = None - if with_mongodb_size: - jobs = connections.mongodb_jobs.mrq_jobs.count({ - "queue": name, - "status": request.args.get("status") or "queued" - }) - q = { "name": name, - "jobs": jobs, # MongoDB size "size": queue.size(), # Redis size "is_sorted": queue.is_sorted, "is_timed": queue.is_timed, @@ -198,7 +213,7 @@ def api_datatables(unit): } if queue.is_sorted: - raw_config = cfg.get("raw_queues", {}).get(name, {}) + raw_config = queue.get_config() q["graph_config"] = raw_config.get("dashboard_graph", lambda: { "start": time.time() - (7 * 24 * 3600), "stop": time.time() + (7 * 24 * 3600), @@ -216,7 +231,7 @@ def api_datatables(unit): queues.append(q) - queues.sort(key=lambda x: -((x["jobs"] or 0) + x["size"])) + queues.sort(key=lambda x: -x["size"]) data = { "aaData": queues, @@ -225,10 +240,27 @@ def api_datatables(unit): elif unit == "workers": fields = None - query = {"status": {"$nin": ["stop"]}} collection = connections.mongodb_jobs.mrq_workers sort = [("datestarted", -1)] + query = {} + if request.args.get("id"): + query["_id"] = ObjectId(request.args["id"]) + else: + if request.args.get("status"): + statuses = request.args["status"].split("-") + query["status"] = {"$in": statuses} + if request.args.get("ip"): + query["$or"] = [{"config.local_ip": request.args["ip"]}, {"config.external_ip": request.args["ip"]}] + if request.args.get("queue"): + query["config.queues"] = request.args["queue"] + + elif unit == "agents": + fields = None + query = {"status": {"$nin": ["stop"]}} + collection = connections.mongodb_jobs.mrq_agents + sort = [("datestarted", -1)] + if request.args.get("showstopped"): query = {} @@ -241,7 +273,7 @@ def api_datatables(unit): fields = None query = build_api_datatables_query(request) - sort = [("_id", 1)] + sort = None # TODO [("_id", 1)] # We can't search easily params because we store it as decoded JSON in mongo :( # Add a string index? @@ -291,27 +323,14 @@ def api_job_result(job_id): @requires_auth def api_job_traceback(job_id): collection = connections.mongodb_jobs.mrq_jobs - if get_current_config().get("save_traceback_history"): - - field_sent = "traceback_history" - else: - field_sent = "traceback" job_data = collection.find_one( - {"_id": ObjectId(job_id)}, projection=[field_sent]) + {"_id": ObjectId(job_id)}, projection=["traceback_history", "traceback"]) if not job_data: - # If a job has no traceback history, we fallback onto traceback - if field_sent == "traceback_history": - field_sent = "traceback" - job_data = collection.find_one( - {"_id": ObjectId(job_id)}, projection=[field_sent]) - if not job_data: - job_data = {} + return jsonify({"traceback": "No exception raised"}) - return jsonify({ - field_sent: job_data.get(field_sent, "No exception raised") - }) + return jsonify(job_data) @app.route('/api/jobaction', methods=["POST"]) diff --git a/mrq/dashboard/static/bin/0.bundle.js b/mrq/dashboard/static/bin/0.bundle.js new file mode 100644 index 00000000..3b8171dd --- /dev/null +++ b/mrq/dashboard/static/bin/0.bundle.js @@ -0,0 +1,12 @@ +webpackJsonp([0],[,function(t,e,n){var i,o;/*! jQuery v2.1.0 | (c) 2005, 2014 jQuery Foundation, Inc. | jquery.org/license */ +!function(e,n){"object"==typeof t&&"object"==typeof t.exports?t.exports=e.document?n(e,!0):function(t){if(!t.document)throw new Error("jQuery requires a window with a document");return n(t)}:n(e)}("undefined"!=typeof window?window:this,function(r,a){function s(t){var e=t.length,n=rt.type(t);return"function"!==n&&!rt.isWindow(t)&&(!(1!==t.nodeType||!e)||("array"===n||0===e||"number"==typeof e&&e>0&&e-1 in t))}function l(t,e,n){if(rt.isFunction(e))return rt.grep(t,function(t,i){return!!e.call(t,i,t)!==n});if(e.nodeType)return rt.grep(t,function(t){return t===e!==n});if("string"==typeof e){if(ut.test(e))return rt.filter(e,t,n);e=rt.filter(e,t)}return rt.grep(t,function(t){return Q.call(e,t)>=0!==n})}function d(t,e){for(;(t=t[e])&&1!==t.nodeType;);return t}function c(t){var e=vt[t]={};return rt.each(t.match(mt)||[],function(t,n){e[n]=!0}),e}function p(){it.removeEventListener("DOMContentLoaded",p,!1),r.removeEventListener("load",p,!1),rt.ready()}function u(){Object.defineProperty(this.cache={},0,{get:function(){return{}}}),this.expando=rt.expando+Math.random()}function h(t,e,n){var i;if(void 0===n&&1===t.nodeType)if(i="data-"+e.replace(Tt,"-$1").toLowerCase(),"string"==typeof(n=t.getAttribute(i))){try{n="true"===n||"false"!==n&&("null"===n?null:+n+""===n?+n:Ct.test(n)?rt.parseJSON(n):n)}catch(t){}kt.set(t,e,n)}else n=void 0;return n}function f(){return!0}function g(){return!1}function b(){try{return it.activeElement}catch(t){}}function m(t,e){return rt.nodeName(t,"table")&&rt.nodeName(11!==e.nodeType?e:e.firstChild,"tr")?t.getElementsByTagName("tbody")[0]||t.appendChild(t.ownerDocument.createElement("tbody")):t}function v(t){return t.type=(null!==t.getAttribute("type"))+"/"+t.type,t}function y(t){var e=Wt.exec(t.type);return e?t.type=e[1]:t.removeAttribute("type"),t}function x(t,e){for(var n=0,i=t.length;i>n;n++)wt.set(t[n],"globalEval",!e||wt.get(e[n],"globalEval"))}function w(t,e){var n,i,o,r,a,s,l,d;if(1===e.nodeType){if(wt.hasData(t)&&(r=wt.access(t),a=wt.set(e,r),d=r.events)){delete a.handle,a.events={};for(o in d)for(n=0,i=d[o].length;i>n;n++)rt.event.add(e,o,d[o][n])}kt.hasData(t)&&(s=kt.access(t),l=rt.extend({},s),kt.set(e,l))}}function k(t,e){var n=t.getElementsByTagName?t.getElementsByTagName(e||"*"):t.querySelectorAll?t.querySelectorAll(e||"*"):[];return void 0===e||e&&rt.nodeName(t,e)?rt.merge([t],n):n}function C(t,e){var n=e.nodeName.toLowerCase();"input"===n&&jt.test(t.type)?e.checked=t.checked:("input"===n||"textarea"===n)&&(e.defaultValue=t.defaultValue)}function T(t,e){var n=rt(e.createElement(t)).appendTo(e.body),i=r.getDefaultComputedStyle?r.getDefaultComputedStyle(n[0]).display:rt.css(n[0],"display");return n.detach(),i}function S(t){var e=it,n=Bt[t];return n||(n=T(t,e),"none"!==n&&n||(zt=(zt||rt("