Skip to content

Commit 311c492

Browse files
sylvinussylvinus
committed
Merge branch 'master' into 0.9.x
* master: test API routes fix typo preventing traceback retrieval avoid filters being overriden Fix missing commits in 0.2.1 MRQ 0.2.0 requirement update Max time (#152) more info on workers (#151) Save abort traceback (#149)
2 parents 0fd9f29 + 9d64c41 commit 311c492

14 files changed

Lines changed: 303 additions & 23 deletions

File tree

‎docs/configuration.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,7 @@ USE_LARGE_JOB_IDS = False #Do not use compacted job IDs in Redis. For compatibil
5555
"""
5656
QUEUES = ("default",) # The queues to listen on.Defaults to default , which will listen on all queues.
5757
MAX_JOBS = 0 #Gevent:max number of jobs to do before quitting. Workaround for memory leaks in your tasks. Defaults to 0
58+
MAX_TIME = 0 # max number of seconds a worker runs before quitting
5859
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
5960
GRENLETS = 1 #Max number of greenlets to use.Defaults to 1.
6061
PROCESSES = 0 #Number of processes to launch with supervisord.Defaults to 0.

‎docs/workers.md‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,9 @@ It is started with a list of queues to listen to, in a specific order.
66

77
It can be started with concurrency options (multiple processes and / or multiple greenlets). We call this whole group a single 'worker' even though it is able to dequeue multiple jobs in parallel.
88

9+
If a worker is started with concurrency options, it will poll for waiting jobs and dispatch them to its related processes/greenlets.
10+
For example, if we decide to use the greenlets option, under the hood, the worker will be one python process that has a pool of greenlets which will be in charge of actually running tasks.
11+
912

1013
## Statuses
1114

‎mrq/config.py‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -299,6 +299,13 @@ def add_parser_args(parser, config_type):
299299
help='Gevent: max number of jobs to do before quitting.' +
300300
' Temp workaround for memory leaks')
301301

302+
parser.add_argument(
303+
'--max_time',
304+
default=0.0,
305+
type=float,
306+
action='store',
307+
help='Max time a worker should run before quitting.')
308+
302309
parser.add_argument(
303310
'--max_memory',
304311
default=0,

‎mrq/dashboard/app.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -319,7 +319,7 @@ def api_job_result(job_id):
319319
@app.route('/api/job/<job_id>/traceback')
320320
@requires_auth
321321
def api_job_traceback(job_id):
322-
collection = connections.mongodb_jobs.mrq_jobss
322+
collection = connections.mongodb_jobs.mrq_jobs
323323
if get_current_config().get("save_traceback_history"):
324324

325325
field_sent = "traceback_history"

‎mrq/dashboard/templates/index.html‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@
1616
</head>
1717
<body>
1818

19-
<div class="navbar navbar-default navbar-fixed-top">
19+
<div class="navbar navbar-default navbar-top">
2020
<div class="container">
2121
<div class="navbar-header">
2222
<a href="/#" class="navbar-brand">MRQ Dashboard</a>
@@ -96,7 +96,7 @@
9696
</div>
9797

9898

99-
<div class="container" style="margin-top:52px;">
99+
<div class="container">
100100

101101
<div id="app-container">
102102

‎mrq/job.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -431,7 +431,7 @@ def save_abort(self):
431431
"dateexpires": dateexpires
432432
}
433433

434-
self._save_status("abort", updates)
434+
self._save_status("abort", updates, exception=True)
435435

436436
def _save_status(self, status, updates=None, exception=False, w=None, j=None):
437437

‎mrq/worker.py‎

Lines changed: 14 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,6 @@
1414
import sys
1515
import json as json_stdlib
1616
import ujson as json
17-
import http.server
1817
from bson import ObjectId
1918
from redis.lock import LuaLock
2019
from collections import defaultdict
@@ -53,6 +52,7 @@ def __init__(self):
5352

5453
self.done_jobs = 0
5554
self.max_jobs = self.config["max_jobs"]
55+
self.max_time = datetime.timedelta(seconds=self.config["max_time"]) or None
5656

5757
self.connected = False # MongoDB + Redis
5858

@@ -393,14 +393,13 @@ def wait_for_idle(self):
393393
if outcome is "wait" and dequeue_jobs == 0:
394394
break
395395

396-
397396
def work(self):
398397
"""Starts the work loop.
399398
400399
"""
401400
self.work_init()
402401

403-
self.work_loop(max_jobs=self.max_jobs)
402+
self.work_loop(max_jobs=self.max_jobs, max_time=self.max_time)
404403

405404
self.work_stop()
406405

@@ -436,14 +435,16 @@ def work_init(self):
436435

437436
self.install_signal_handlers()
438437

439-
def work_loop(self, max_jobs=None):
438+
def work_loop(self, max_jobs=None, max_time=None):
440439

441440
self.done_jobs = 0
442441
self.datestarted_work_loop = datetime.datetime.utcnow()
443442
self.queue_offset = 0
444443

445444
try:
446445

446+
max_time_reached = False
447+
447448
while True:
448449

449450
if self.graceful_stop:
@@ -456,6 +457,12 @@ def work_loop(self, max_jobs=None):
456457

457458
while True:
458459

460+
# we put this here to make sure we have a strict limit on max_time
461+
if max_time and datetime.datetime.utcnow() - self.datestarted >= max_time:
462+
self.log.info("Reached max_time=%s" % max_time.seconds)
463+
max_time_reached = True
464+
break
465+
459466
free_pool_slots = self.gevent_pool.free_count()
460467

461468
if max_jobs:
@@ -465,12 +472,14 @@ def work_loop(self, max_jobs=None):
465472
break
466473

467474
if free_pool_slots > 0:
468-
469475
break
470476

471477
self.status = "full"
472478
self.gevent_pool.wait_available(timeout=60)
473479

480+
if max_time_reached:
481+
break
482+
474483
self.status = "spawn"
475484
with self.work_lock:
476485
outcome, dequeue_jobs = self.work_once(free_pool_slots=free_pool_slots, max_jobs=max_jobs)

‎requirements-base.txt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,4 +9,4 @@ objgraph>=1.8.1
99
termcolor>=1.1.0
1010
subprocess32>=3.2.7; python_version < '3.2'
1111
future>=0.15.2
12-
importlib>=1.0.3; python_version < '2.7'
12+
importlib>=1.0.3; python_version < '2.7'

‎setup.py‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ def get_version():
4343
author_email="contact@pricingassistant.com",
4444
url="http://github.com/pricingassistant/mrq",
4545
# download_url="http://chardet.feedparser.org/download/python3-chardet-1.0.1.tgz",
46-
keywords=["worker", "task", "distributed", "queue", "asynchronous", "redis", "mongodb", "job", "processing"],
46+
keywords=["worker", "task", "distributed", "queue", "asynchronous", "redis", "mongodb", "job", "processing", "gevent"],
4747
platforms='any',
4848
entry_points={
4949
'console_scripts': [
@@ -63,11 +63,12 @@ def get_version():
6363
"Programming Language :: Python :: 2.7",
6464
"Programming Language :: Python :: 3.4",
6565
"Programming Language :: Python :: 3.5",
66+
"Programming Language :: Python :: 3.6",
6667
#'Development Status :: 1 - Planning',
6768
#'Development Status :: 2 - Pre-Alpha',
6869
#'Development Status :: 3 - Alpha',
69-
'Development Status :: 4 - Beta',
70-
#'Development Status :: 5 - Production/Stable',
70+
#'Development Status :: 4 - Beta',
71+
'Development Status :: 5 - Production/Stable',
7172
#'Development Status :: 6 - Mature',
7273
#'Development Status :: 7 - Inactive',
7374
"Environment :: Other Environment",

‎tests/conftest.py‎

Lines changed: 82 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,10 @@
3131

3232
os.system("rm -rf dump.rdb")
3333

34+
PYTHON_BIN = "python"
35+
if os.environ.get("PYTHON_BIN"):
36+
PYTHON_BIN = os.environ["PYTHON_BIN"]
37+
3438

3539
class ProcessFixture(object):
3640

@@ -41,6 +45,7 @@ def __init__(self, request, cmdline=None, wait_port=None, quiet=False):
4145
self.wait_port = wait_port
4246
self.quiet = quiet
4347
self.stopped = False
48+
self.started = False
4449

4550
self.request.addfinalizer(self.stop)
4651

@@ -88,12 +93,15 @@ def start(self, cmdline=None, env=None, expected_children=0):
8893
if self.wait_port:
8994
wait_for_net_service("127.0.0.1", int(self.wait_port), poll_interval=0.01)
9095

96+
self.started = True
97+
9198
def stop(self, force=False, timeout=None, block=True, sig=15):
9299

93100
# Call this only one time.
94101
if self.stopped and not force:
95102
return
96103
self.stopped = True
104+
self.started = False
97105

98106
if self.process is not None and self.process.returncode is None:
99107

@@ -133,13 +141,9 @@ def start(self, flush=True, deps=True, trace=True, agent=False, **kwargs):
133141

134142
processes = 0
135143

136-
python_bin = "python"
137-
if os.environ.get("PYTHON_BIN"):
138-
python_bin = os.environ["PYTHON_BIN"]
139-
140144
if agent:
141145

142-
cmdline = "%s mrq/bin/mrq_agent.py %s" % (python_bin, kwargs.get("flags", ""))
146+
cmdline = "%s mrq/bin/mrq_agent.py %s" % (PYTHON_BIN, kwargs.get("flags", ""))
143147

144148
else:
145149

@@ -148,7 +152,7 @@ def start(self, flush=True, deps=True, trace=True, agent=False, **kwargs):
148152
processes = int(m.group(1))
149153

150154
cmdline = "%s mrq/bin/mrq_worker.py --mongodb_logs_size 0 %s %s %s %s" % (
151-
python_bin,
155+
PYTHON_BIN,
152156
"--admin_port %s" % self.admin_port if (processes <= 1) else "",
153157
"--trace_io --trace_greenlets" if trace else "",
154158
kwargs.get("flags", ""),
@@ -237,7 +241,7 @@ def send_task(self, path, params, **kwargs):
237241

238242
def send_task_cli(self, path, params, queue=None, **kwargs):
239243

240-
cli = ["python", "mrq/bin/mrq_run.py", "--quiet"]
244+
cli = [PYTHON_BIN, "mrq/bin/mrq_run.py", "--quiet"]
241245
if queue:
242246
cli += ["--queue", queue]
243247
cli += [path, json.dumps(params)]
@@ -293,6 +297,71 @@ def flush(self):
293297
mongodb.drop_collection(c)
294298

295299

300+
class ApiFixture(ProcessFixture):
301+
302+
def __init__(self, *args, **kwargs):
303+
304+
ProcessFixture.__init__(self, *args, **kwargs)
305+
306+
self.wait_port = 5555
307+
self.host = "127.0.0.1"
308+
self.url = "http://%s:%s" % (self.host, self.wait_port)
309+
self.cmdline = "%s mrq/dashboard/app.py" % PYTHON_BIN
310+
311+
def start(self, *args, **kwargs):
312+
313+
kwargs["env"] = os.environ
314+
315+
ProcessFixture.start(self, *args, **kwargs)
316+
317+
def _login(self, user):
318+
import requests
319+
320+
res = requests.post(self.url + "/api/auth/login", data={
321+
"userEmail": user["email"],
322+
"userPassword": user["password"]
323+
})
324+
325+
assert res.status_code == 200
326+
327+
return {
328+
"cookies": res.cookies
329+
}
330+
331+
def _request(self, method, path, key="local", user=None, assert_200=True, **kwargs):
332+
333+
if not self.started:
334+
self.start()
335+
336+
import requests
337+
338+
if user:
339+
auth = self._login(user)
340+
kwargs["cookies"] = auth["cookies"]
341+
342+
res = getattr(requests, method)(self.url + path, **kwargs)
343+
344+
if assert_200 and res.status_code != 200:
345+
raise Exception(res.status_code)
346+
347+
try:
348+
js = json.loads(res.text)
349+
except:
350+
print("Couldn't json parse", repr(res.text)[0:100])
351+
js = None
352+
353+
return js, res
354+
355+
def GET(self, *args, **kwargs):
356+
return self._request("get", *args, **kwargs)
357+
358+
def POST(self, *args, **kwargs):
359+
return self._request("post", *args, **kwargs)
360+
361+
def DELETE(self, *args, **kwargs):
362+
return self._request("delete", *args, **kwargs)
363+
364+
296365
@pytest.fixture(scope="function")
297366
def httpstatic(request):
298367
return ProcessFixture(request, "/usr/sbin/nginx -c /app/tests/fixtures/httpstatic/nginx.conf", wait_port=8081)
@@ -336,3 +405,9 @@ def worker2(request, mongodb, redis):
336405
def worker_mongodb_with_journal(request, mongodb_with_journal, redis):
337406

338407
return WorkerFixture(request, mongodb=mongodb_with_journal, redis=redis, admin_port=20022)
408+
409+
410+
@pytest.fixture(scope="function")
411+
def api(request, mongodb, redis):
412+
413+
return ApiFixture(request)

0 commit comments

Comments
 (0)