Add missing index + better worker.stop in tests

Author: sylvinus

mrq/basetasks/indexes.py

@@ -30,7 +30,9 @@ def run(self, params):
         connections.mongodb_jobs.mrq_jobs.ensure_index(
             [("dateretry", 1)], sparse=True, background=True)
         connections.mongodb_jobs.mrq_jobs.ensure_index(
-            [("datequeued", 1)], sparse=True, background=True)
+            [("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)

mrq/config.py

@@ -337,6 +337,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,

mrq/worker.py

@@ -22,7 +22,7 @@
 from .exceptions import (TimeoutInterrupt, StopRequested, JobInterrupt, AbortInterrupt,
                          RetryInterrupt, MaxRetriesInterrupt, MaxConcurrencyInterrupt)
 from .context import (set_current_worker, set_current_job, get_current_job, get_current_config,
-                      connections, enable_greenlet_tracing)
+                      connections, enable_greenlet_tracing, run_task)
 from .queue import Queue
 from .utils import MongoJSONEncoder, MovingAverage
 from .processes import Process
@@ -101,6 +101,9 @@ def __init__(self):
             "total": 0
         }
 
+        if self.config["ensure_indexes"]:
+            run_task("mrq.basetasks.indexes.EnsureIndexes", {})
+
     @property
     def config(self):
         return get_current_config()

tests/conftest.py

@@ -100,6 +100,7 @@ def stop(self, force=False, timeout=None, block=True, sig=15):
         # Call this only one time.
         if self.stopped and not force:
             return
+
         self.stopped = True
         self.started = False
 

tests/test_agent.py

@@ -144,8 +144,6 @@ def test_orchestration_scenarios(worker):
         "worker2": ["MRQ_WORKER_PROFILE=a mrq-worker a"]
     }
 
-    worker.stop()
-
 
 def test_agent_process(worker):
 

tests/test_cli.py

@@ -27,5 +27,3 @@ def test_cli_run_nonblocking(worker):
 
     assert job1.data["status"] == "success"
     assert job1.data["result"] == 42
-
-    worker.stop()

tests/test_interrupts.py

@@ -7,6 +7,7 @@
 import pytest
 from mrq.context import connections
 import json
+import os
 
 
 PROCESS_CONFIGS = [
@@ -32,13 +33,13 @@ def test_interrupt_worker_gracefully(worker, p_flags):
     assert job["status"] == "started"
 
     # Stop the worker gracefully. first job should still finish!
-    worker.stop(block=False, deps=False)
+    os.kill(worker.process.pid, 2)
 
     time.sleep(1)
 
     # Should not be accepting new jobs!
     job_id2 = worker.send_task(
-        "tests.tasks.general.Add", {"a": 42, "b": 1, "sleep": 4}, block=False)
+        "tests.tasks.general.Add", {"a": 42, "b": 1, "sleep": 4}, block=False, start=False)
 
     time.sleep(1)
 
@@ -54,8 +55,6 @@ def test_interrupt_worker_gracefully(worker, p_flags):
     job = Job(job_id2).fetch().data
     assert job.get("status") == "queued"
 
-    worker.stop_deps()
-
 
 @pytest.mark.parametrize(["p_flags"], PROCESS_CONFIGS)
 def test_interrupt_worker_double_sigint(worker, p_flags):
@@ -75,13 +74,13 @@ def test_interrupt_worker_double_sigint(worker, p_flags):
     assert job["status"] == "started"
 
     # Stop the worker gracefully. first job should still finish!
-    worker.stop(block=False, deps=False)
+    os.kill(worker.process.pid, 2)
 
     time.sleep(1)
 
     # Should not be accepting new jobs!
     job_id2 = worker.send_task(
-        "tests.tasks.general.Add", {"a": 42, "b": 1, "sleep": 20}, block=False)
+        "tests.tasks.general.Add", {"a": 42, "b": 1, "sleep": 20}, block=False, start=False)
 
     time.sleep(1)
 
@@ -92,7 +91,7 @@ def test_interrupt_worker_double_sigint(worker, p_flags):
     assert job["status"] == "started"
 
     # Sending a second kill -2 should make it stop
-    worker.stop(block=True, deps=False, force=True)
+    os.kill(worker.process.pid, 2)
 
     while Job(job_id).fetch().data["status"] == "started":
         time.sleep(0.1)
@@ -356,8 +355,10 @@ def test_interrupt_maxconcurrency(worker):
     assert set(job_statuses) == set(["success", "maxconcurrency"])
 
     # the job concurrency key must be equal to 0
-    last_job_id = worker.send_task("tests.tasks.concurrency.LockedAdd",
-        {"a": 1, "b": 1, "sleep": 2}, block=False
+    last_job_id = worker.send_task(
+        "tests.tasks.concurrency.LockedAdd",
+        {"a": 1, "b": 1, "sleep": 2},
+        block=False
     )
 
     last_job = Job(last_job_id).wait(poll_interval=0.01)

tests/test_memoryleaks.py

@@ -29,8 +29,6 @@ def test_max_memory_restart(worker):
     # We must have been restarted at least once.
     assert worker.mongodb_jobs.mrq_workers.find().count() > 1
 
-    worker.stop()
-
 
 def get_diff_after_jobs(worker, n_tasks, leak, sleep=0):
 

tests/test_pause.py

@@ -47,8 +47,6 @@ def test_pause_resume(worker):
 
     assert worker.mongodb_jobs.tests_inserts.count() == 2
 
-    worker.stop()
-
 
 def test_pause_refresh_interval(worker):
 
@@ -74,8 +72,6 @@ def test_pause_refresh_interval(worker):
     assert job1["status"] == "success"
     assert job1["result"] == {"a": 41}
 
-    worker.stop()
-
 
 def test_pause_subqueue(worker):
 
@@ -126,5 +122,3 @@ def test_pause_subqueue(worker):
     assert job2["result"] == {"a": 43}
 
     assert worker.mongodb_jobs.tests_inserts.count() == 2
-
-    worker.stop()

tests/test_performance.py

@@ -21,7 +21,7 @@
 ])
 def test_job_max_latency(worker, p_max_latency, p_min_observed_latency, p_max_observed_latency):
 
-    worker.start(flags=" --greenlets=1 --max_latency=%s" % (p_max_latency), trace=False)
+    worker.start(flags=" --ensure_indexes --greenlets=1 --max_latency=%s" % (p_max_latency), trace=False)
 
     def get_latency():
         t = time.time()
@@ -76,7 +76,7 @@ def test_network_latency(worker, p_latency, p_min, p_max):
 
 def benchmark_task(worker, taskpath, taskparams, tasks=1000, greenlets=50, processes=0, max_seconds=10, profile=False, quiet=True, raw=False, queues="default", config=None):
 
-    worker.start(flags="--processes %s --greenlets %s%s%s%s" % (
+    worker.start(flags="--ensure_indexes --processes %s --greenlets %s%s%s%s" % (
         processes,
         greenlets,
         " --profile" if profile else "",
@@ -321,9 +321,9 @@ def test_performance_queue_cancel_requeue(worker):
 def test_worker_efficiency(worker, p_queue_type, p_greenlets, p_min_efficiency):
 
     if p_queue_type == "regular":
-        worker.start(trace=False, flags="--greenlets %s" % p_greenlets, queues="default")
+        worker.start(trace=False, flags="--ensure_indexes --greenlets %s" % p_greenlets, queues="default")
     elif p_queue_type == "raw":
-        worker.start(trace=False, flags="--greenlets %s --config tests/fixtures/config-raw1.py" % p_greenlets,
+        worker.start(trace=False, flags="--ensure_indexes --greenlets %s --config tests/fixtures/config-raw1.py" % p_greenlets,
                      queues="testperformance_efficiency_raw")
     elif p_queue_type == "raw_nostorage":
         worker.start(trace=False, flags="--greenlets %s --config tests/fixtures/config-raw1.py" % p_greenlets,
@@ -360,7 +360,7 @@ def test_worker_efficiency(worker, p_queue_type, p_greenlets, p_min_efficiency):
     ))
 
     # We can't be perfectly efficient!
-    assert perfect_time < total_time
+    assert (perfect_time - 1) < total_time
 
     # But we should be at least 80% efficient
     assert perfect_time > total_time * p_min_efficiency