A3s 843 update pymongo

Dockerfile

@@ -1,4 +1,4 @@
-FROM debian:jessie
+FROM debian:buster-slim
 
 #
 # httpredir.debian.org is often unreliable
@@ -11,32 +11,36 @@ FROM debian:jessie
 # 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 \
  gcc \
  python-dev \
  python-pip \
+ python-setuptools \
  python3-pip \
  python3-dev \
+ python3-setuptools \
  make \
  git \
  vim \
  bzip2 \
- mongodb-org \
  nginx redis-server \
  g++ \
+ apt-utils \
  && \
  apt-get clean -y && \
  rm -rf /var/lib/apt/lists/*
 
-RUN curl -sL https://deb.nodesource.com/setup_7.x | bash -
+RUN curl -sL https://deb.nodesource.com/setup_12.x | bash -
 RUN apt-get install -y --no-install-recommends nodejs
 
+RUN curl -sL https://www.mongodb.org/static/pgp/server-4.4.asc | apt-key add -
+RUN echo "deb http://repo.mongodb.org/apt/debian buster/mongodb-org/4.4 main" > /etc/apt/sources.list.d/mongodb-org-4.4.list
+RUN apt-get update && apt-get install -y mongodb-org
+
 # 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
+RUN curl -sL 'https://github.com/squeaky-pl/portable-pypy/releases/download/pypy-7.2.0/pypy-7.2.0-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
@@ -48,16 +52,16 @@ ADD requirements-base.txt /app/requirements-base.txt
 ADD requirements-dev.txt /app/requirements-dev.txt
 ADD requirements-dashboard.txt /app/requirements-dashboard.txt
 
-RUN pip3 install -r /app/requirements-heroku.txt && \
- pip3 install -r /app/requirements-base.txt && \
- pip3 install -r /app/requirements-dev.txt && \
- pip3 install -r /app/requirements-dashboard.txt && \
+RUN python3 -m pip install -r /app/requirements-heroku.txt && \
+ python3 -m pip install -r /app/requirements-base.txt && \
+ python3 -m pip install -r /app/requirements-dev.txt && \
+ python3 -m pip install -r /app/requirements-dashboard.txt && \
  rm -rf ~/.cache
 
-RUN pip install -r /app/requirements-heroku.txt && \
- pip install -r /app/requirements-base.txt && \
- pip install -r /app/requirements-dev.txt && \
- pip install -r /app/requirements-dashboard.txt && \
+RUN python -m pip install -r /app/requirements-heroku.txt && \
+ python -m pip install -r /app/requirements-base.txt && \
+ python -m pip install -r /app/requirements-dev.txt && \
+ python -m pip install -r /app/requirements-dashboard.txt && \
  rm -rf ~/.cache
 
 RUN /pypy/bin/pip install -r /app/requirements-heroku.txt && \

README.md

@@ -150,6 +150,17 @@ This was a preview on the very basic features of MRQ. What makes it actually use
 These features will be demonstrated in a future example of a simple web crawler.
 
 
+## How to compile a new version of MRQ
+- Go to mrq/version.py file and update from 0.9.28.1 to 0.9.29 (for example)
+- Commit and pushes the last changes
+- Go to in the master branch (after merge pull request)
+- Execute the command `python setup.py sdist`
+- Make sure you have installed the python twine package (`pip install twine`).
+- Execute the twine command to upload the new version of mrq to pypi.org (`twine upload dist/mrq-custom-0.9.29.tar.gz`)
+- In the previous step you will be asked for the pypi.org credentials are in 1password, you must request them in #password_requests
+- Update all applications that use MRQ with this version in their requirements.txt
+
+
 # More
 
 Full documentation is available on [readthedocs](http://mrq.readthedocs.org/en/latest/)

docs/jobs-maintenance.md

@@ -7,6 +7,13 @@ To do that, you should add these recurring scheduled jobs to your mrq-config.py:
 ```
 SCHEDULER_TASKS = [
 
+ # This will queue jobs in the 'delayed' status.
+ {
+ "path": "mrq.basetasks.cleaning.QueueDelayedJobs",
+ "params": {},
+ "interval": 60
+ },
+
  # This will requeue jobs in the 'retry' status, until they reach their max_retries.
  {
  "path": "mrq.basetasks.cleaning.RequeueRetryJobs",

docs/jobs.md

@@ -8,8 +8,9 @@ A **Job** is an instance of the execution of a Task. It must link to a specific
 
 MRQ defines a list of statuses for jobs. A job can only be in one of them at a time.
 
-When everything goes fine, a job will go through 3 statuses:
+When everything goes fine, a job will go through 4 statuses:
 
+* ```delayed```: The Job has been created and it is delayed to be queued later.
 * ```queued```: The Job has been created and it is waiting to be dequeued by a Worker.
 * ```started```: A Worker has dequeued the job and started executing it.
 * ```success```: The job was successfully ran.
@@ -51,6 +52,11 @@ Queues a job. If `queue` is not provided, the default queue for that Task as def
 
 Queues multiple jobs at once. Returns a list of IDs of the jobs.
 
+* `queue_job(main_task_path, params, delay=120, queue=None)`
+
+Create a job with `delayed` status, the job will be queued after of at least `delay` seconds. 
+Remember to add the base delayed job as explained in [Jobs maintenance](jobs-maintenance.md) to have `delayed` jobs actually queued.
+
 * `queue_raw_jobs(queue, params_list, batch_size=1000)`
 
 Queues multiple jobs at once on a [raw queue](queues.md#raw-queues). The queued jobs have no IDs on a raw queue so this function has no return.

mrq/agent.py

@@ -8,7 +8,11 @@
 import traceback
 from collections import defaultdict
 from bson import ObjectId
-from redis.lock import LuaLock
+try:
+ from redis.lock import LuaLock
+except ImportError:
+ # Change name to avoid NameError raised when use of LuaLock at line 147
+ from redis.lock import Lock as LuaLock
 from .processes import Process, ProcessPool
 from .utils import MovingETA, normalize_command
 from .queue import Queue

mrq/basetasks/cleaning.py

@@ -27,14 +27,41 @@ class RequeueRetryJobs(Task):
  max_concurrency = 1
 
  def run(self, params):
- print("IN")
  return run_task("mrq.basetasks.utils.JobAction", {
  "status": "retry",
  "dateretry": {"$lte": datetime.datetime.utcnow()},
  "action": "requeue_retry"
  })
 
 
+class QueueDelayedJobs(Task):
+
+ """ Requeue jobs that were marked as delayed. """
+
+ max_concurrency = 1
+
+ def run(self, params):
+ return run_task("mrq.basetasks.utils.JobAction", {
+ "status": "delayed",
+ "dateretry": {"$lte": datetime.datetime.utcnow()},
+ "action": "requeue"
+ })
+
+
+class DeleteExpiresJobs(Task):
+
+ """ Delete jobs that were dateexpires is less than the current date time. """
+
+ max_concurrency = 1
+
+ def run(self, params):
+ return run_task("mrq.basetasks.utils.JobAction", {
+ "status": ["success","cancel"],
+ "dateexpires": {"$lte": datetime.datetime.utcnow()},
+ "action": "delete"
+ })
+
+
 class RequeueStartedJobs(Task):
 
  """ Requeue jobs that were marked as status=started and never finished.

mrq/basetasks/utils.py

@@ -10,6 +10,7 @@
 from mrq.utils import group_iter
 import datetime
 import ujson as json
+import sys
 
 
 def get_task_cfg(taskpath):
@@ -45,6 +46,7 @@ def build_query(self):
  "status",
  "worker",
  "path",
+ "dateexpires",
  "dateretry",
  "exceptiontype"]:
  if self.params.get(k):
@@ -70,7 +72,8 @@ def perform_action(self, action, query, destination_queue):
 
  stats = {
  "requeued": 0,
- "cancelled": 0
+ "cancelled": 0,
+ "deleted": 0
  }
 
  if action == "cancel":
@@ -151,7 +154,12 @@ def perform_action(self, action, query, destination_queue):
  self.collection.update({
  "_id": {"$in": jobs_by_queue[queue]}
  }, {"$set": updates}, multi=True)
-
- set_queues_size({queue: len(jobs) for queue, jobs in jobs_by_queue.iteritems()})
-
+ if sys.version_info.major > 2:
+ set_queues_size({queue: len(jobs) for queue, jobs in jobs_by_queue.items()})
+ else:
+ set_queues_size({queue: len(jobs) for queue, jobs in jobs_by_queue.iteritems()})
+ elif action == 'delete':
+ amount_delete = self.collection.delete_many(query)
+ stats["deleted"] = amount_delete.deleted_count
  return stats
+

mrq/context.py

@@ -13,6 +13,8 @@
 from .utils import LazyObject, load_class_by_path
 from .config import get_config
 from .subpool import subpool_map, subpool_imap
+from .exceptions import JobRuntimeError
+
 
 # This should be MRQ's only Python object shared by all the jobs in the same process
 _GLOBAL_CONTEXT = {
@@ -113,10 +115,14 @@ def get_current_config():
  return _GLOBAL_CONTEXT["config"]
 
 
-def retry_current_job(delay=None, max_retries=None, queue=None):
+def retry_current_job(delay=None, max_retries=None, queue=None, exception=None):
  current_job = get_current_job()
  if current_job:
  current_job.retry(delay=delay, max_retries=max_retries, queue=queue)
+ elif exception:
+ raise JobRuntimeError("Error during MRQ task execution") from exception
+ else:
+ raise JobRuntimeError("Error during MRQ task execution")
 
 
 def abort_current_job():
@@ -139,22 +145,32 @@ def versiontuple(v):
  if isinstance(config_obj, basestring):
 
  import redis as pyredis
+ import redis.connection as pyredisconnection
 
+ connection_class = pyredisconnection.Connection
  urllib.parse.uses_netloc.append('redis')
  redis_url = urllib.parse.urlparse(config_obj)
-
+ 
  log.info("%s: Connecting to Redis at %s..." %
  (attr, redis_url.hostname))
 
- redis_pool = pyredis.BlockingConnectionPool(
+ redis_params = dict(
  host=redis_url.hostname,
  port=redis_url.port,
  db=int((redis_url.path or "").replace("/", "") or "0"),
- password=redis_url.password,
+ password=redis_url.password if redis_url.password is not None else redis_url.username,
  max_connections=int(config.get("redis_max_connections")),
  timeout=int(config.get("redis_timeout")),
- decode_responses=False
+ decode_responses=False,
+ connection_class=connection_class
  )
+
+ if redis_url.scheme == "rediss":
+ redis_params["connection_class"] = pyredisconnection.SSLConnection
+ redis_params["ssl_cert_reqs"] = "none"
+
+ redis_pool = pyredis.BlockingConnectionPool(**redis_params)
+
  return pyredis.StrictRedis(connection_pool=redis_pool)
 
  # Let's just assume we got a StrictRedis-like object!

mrq/dashboard/app.py

@@ -54,8 +54,8 @@ def root():
 @requires_auth
 def api_task_exceptions():
  stats = list(connections.mongodb_jobs.mrq_jobs.aggregate([
- {"$match": {"status": "failed"}},
- {"$group": {"_id": {"path": "$path", "exceptiontype": "$exceptiontype"},
+ {"$match": {"status": {"$in": ["failed","maxretries"]}}},
+ {"$group": {"_id": {"path": "$path", "exceptiontype": "$exceptiontype","status":"$status"},
  "jobs": {"$sum": 1}}},
  ]))
 

mrq/dashboard/static/js/views/jobs.js

@@ -347,6 +347,9 @@ define(["jquery", "underscore", "views/generic/datatablepage", "models"],functio
  if (source.time) {
  display.push("cputime "+String(source.time).substring(0,6)+"s ("+source.switches+" switches)");
  }
+ if (source.status === 'retry' || source.status === 'delayed') {
+ display.push("requeue "+moment.utc(source.dateretry).fromNow());
+ }
 
  return "<small>" + display.join("<br/>") + "</small>";
 

mrq/dashboard/static/js/views/taskexceptions.js

@@ -32,7 +32,7 @@ define(["jquery", "underscore", "views/generic/datatablepage", "models"],functio
  "sClass": "col-exception",
  "sType":"numeric",
  "mData":function(source, type, val) {
- return "<a href='/#jobs?path="+source._id.path+"&status=failed&exceptiontype="+source._id.exceptiontype+ "'>"+source._id.exceptiontype+"</a>"
+ return "<a href='/#jobs?path=" + source._id.path + "&status=" + source._id.status +"&exceptiontype="+source._id.exceptiontype+ "'>"+source._id.exceptiontype+"</a>"
  }
  },
  {
@@ -43,7 +43,7 @@ define(["jquery", "underscore", "views/generic/datatablepage", "models"],functio
  var cnt = source.jobs || 0;
 
  if (type == "display") {
- return "<a href='/#jobs?path="+source._id.path+"&status=failed&exceptiontype="+source._id.exceptiontype+"'>"+cnt+"</a>"
+ return "<a href='/#jobs?path=" + source._id.path + "&status=" + source._id.status +"&exceptiontype="+source._id.exceptiontype+"'>"+cnt+"</a>"
  + "<br/>"
  + '<span class="inlinesparkline" values="'+self.addToCounter("taskexceptions."+source._id.path+" "+source._id.exceptiontype, cnt, 50).join(",")+'"></span>';
  } else {

mrq/dashboard/templates/index.html

@@ -391,6 +391,7 @@ <h4 class="modal-title"></h4>
  <select class="form-control input-sm js-datatable-filters-status" id="jobs-form-status">
  <option <%= filters.status==""?"selected='selected'":"" %> value="">-statuses-</option>
  <% _.each({
+ "delayed": "delayed",
  "queued": "queued",
  "started": "started",
  "success": "success",

mrq/exceptions.py

@@ -24,22 +24,33 @@ class TimeoutInterrupt(_MrqInterrupt):
 
 
 class AbortInterrupt(_MrqInterrupt):
- pass
+ path = None
+
+ def _get_exception_name(self):
+ return "%s" % (
+ self.path
+ )
 
 
 class RetryInterrupt(_MrqInterrupt):
  delay = None
  queue = None
  retry_count = 0
+ path = None
 
  def _get_exception_name(self):
  return "%s #%s: %s seconds, %s queue" % (
- self.__class__.__name__, self.retry_count, self.delay, self.queue
+ self.path, self.retry_count, self.delay, self.queue
  )
 
 
 class MaxRetriesInterrupt(_MrqInterrupt):
- pass
+ path = None
+
+ def _get_exception_name(self):
+ return "%s" % (
+ self.path
+ )
 
 
 class StopRequested(GreenletExit):
@@ -53,4 +64,13 @@ class JobInterrupt(GreenletExit):
 
 
 class MaxConcurrencyInterrupt(_MrqInterrupt):
+ path = None
+
+ def _get_exception_name(self):
+ return "%s" % (
+ self.path
+ )
+
+
+class JobRuntimeError(_MrqInterrupt):
  pass