Allow to have a custom request #3977
Conversation
mvaled
force-pushed
the
custom-request
branch
4 times, most recently
from
18e621e
to
26a7d86
Compare
|
This PR is related to a question in the mailing list: https://groups.google.com/forum/#!topic/celery-users/0QhWtY7na4M Without this PR the current solution I have found is to monkey-patch the This is a fragment of the monkey patching I'm doing right now: from celery.worker.request import Request as BaseRequest
class Request(BaseRequest):
def on_timeout(self, soft, timeout):
"""Handler called if the task times out."""
super(Request, self).on_timeout(soft, timeout)
try:
if not soft:
_report_failure.delay(self.id) # Yes! This is another task
except:
pass
if not getattr(BaseTask, 'Request', None):
# So this is a celery that has not accepted our patch
# (https://github.com/celery/celery/pull/3977). Let's proceed to
# monkey-patch the Request.
from celery.worker import request
_super_create_request_cls = request.create_request_cls
def create_request_cls(base, task, pool, hostname, eventer,
ref=request.ref,
revoked_tasks=request.revoked_tasks,
task_ready=request.task_ready,
trace=request.trace_task_ret):
if base is BaseRequest:
Base = Request
else:
class Base(base, Request):
pass
class PatchedRequest(Base):
pass
return _super_create_request_cls(
PatchedRequest,
task,
pool,
hostname,
eventer,
ref=ref,
revoked_tasks=revoked_tasks,
task_ready=task_ready,
trace=trace
)
request.create_request_cls = create_request_cls |
|
Hi @thedrow, I'll be happy to provide those. In fact, after this PR I extended my own usage to also override the Which document do you think I should touch? The Request is not documented in any of the narrative documents. In fact, at first, I was confused about if As for the tests, I will include them in the next push. |
|
Actually it seems that some of the hooks are exposed through the Task class. I think that this is where we should be documenting this http://docs.celeryproject.org/en/latest/userguide/tasks.html#custom-task-classes The Task class already exposes an on_failure hook. I'm not sure we want to expose them through the Request API. |
|
It seems that When I run something like: $ ps a | grep PoolWorke[r] | awk '{print $1}' | xargs kill -9 in the middle of a task the That's why I overloaded the I could kill the PoolWorker several times; with this in place the system can, at least, inform about the failure. In this case, since I use late acks the task should run again (I'm not seeing that happening, though.) |
|
Hum. I can't manage to get the tests running locally: It remains there doing nothing. This how I'm running tests: TEST_BROKER=redis:// tox -re2.7-unit,2.7-integration-redis |
|
@mvaled do you have the same issue with the test failures with the |
|
Hi @georgepsarakis, Yes, I do get the same errors. I think this is to be expected since the change I'm introducing should not mess with the current set of tests. However, Travis does pass all tests, so I think it should be something local in my machine. So far I can run and pass all sets of unit tests: $ tox -e{2.7,pypy,3.4,3.5,3.6}-unitIt's integration tests which are failing. |
|
I am seeing a similar error (not the same though) in another Pull Request, so if you happen to find out what may be wrong, please share. |
|
The issue is quite slippery. Just now I was able to run all integration tests for Python 2.7, PyPy, 3.4 and 3.5. Same code, no changes. Before, I changed the default |
|
I was thinking that maybe SoftTimeLimitExceeded (in billiard) should inherit from BaseException instead. |
|
@mvaled Are you still interested in working on this? |
|
@thedrow I was a little off of it, because I was using my work-around happily. But I can give it another try to make it part of the celery stable API. That would be better. Basically, what you requested was to include the tests and documentation. I think I can get this done by the end of the week-end. Are you planning to do a freeze of new features? |
|
I could run the test in master: |
Allowing a custom Request eases the task of handling timeouts (even hard
timeouts).
Rationale
Some (poorly written) bits of code catch exceptions quite broadly:
try:
...
except:
...
This hurts tasks when a SoftTimeLimitError is raised inside such blocks of
code. Rewriting those smelly bits of code can take a lot of effort, and
sometimes, the code belongs to a third-party library which makes the task even
harder.
Using a custom request allows to catch hard time limits.
Your app can be customized like:
from celery import Task as BaseTask
from celery.worker.request import Request as BaseRequest
class Request(BaseRequest):
def on_timeout(self, soft, timeout):
super(Request, self).on_timeout(soft, timeout)
if not soft:
print('Something hard hit me!')
class MyTask(BaseTask):
Request = Request
@app.task(base=MyTask, bind=True)
def sometask(self):
pass
|
One question about where to put the documentation: As I argued before, |
|
Somewhere in the task customization section of Tasks should be good. |
|
I created a minimal documentation. It's a bit weird where it is, I think. Since what really changed is the |
docs/userguide/tasks.rst
Outdated
| worker process. An application may leverage such facility to detect failures | ||
| which are not detected using `celery.app.task.Task.on_failure`:meth:. | ||
|
|
||
|
|
There was a problem hiding this comment.
Please provide code examples for how to override the request.
|
Thanks! Awesome work. |
* Allow custom Request, aka custom `on_timeout`.
Allowing a custom Request eases the task of handling timeouts (even hard
timeouts).
Rationale
Some (poorly written) bits of code catch exceptions quite broadly:
try:
...
except:
...
This hurts tasks when a SoftTimeLimitError is raised inside such blocks of
code. Rewriting those smelly bits of code can take a lot of effort, and
sometimes, the code belongs to a third-party library which makes the task even
harder.
Using a custom request allows to catch hard time limits.
Your app can be customized like:
from celery import Task as BaseTask
from celery.worker.request import Request as BaseRequest
class Request(BaseRequest):
def on_timeout(self, soft, timeout):
super(Request, self).on_timeout(soft, timeout)
if not soft:
print('Something hard hit me!')
class MyTask(BaseTask):
Request = Request
@app.task(base=MyTask, bind=True)
def sometask(self):
pass
* Check signatures' types have a default Request.
* Test Request is customizable per Task class.
* Document custom requests.
* Exemplify the usage of the custom requests.
Thanks! Maybe with my next contribution ;) |
|
We have just hit this problem as well. Great to see that it is merged! Any ETA on a new release? |
|
Soon. We have a few more things to merge first. |
|
Its worth noting that a custom |
|
@cjh1 good catch. Please open an issue about it so we won't forget. |
Allowing a custom Request eases the task of handling timeouts (even hard timeouts).
Rationale
Some (poorly written) bits of code catch exceptions quite broadly:
This hurts tasks when a SoftTimeLimitError is raised inside such blocks of
code. Rewriting those smelly bits of code can take a lot of effort, and
sometimes, the code belongs to a third-party library which makes the task even
harder.
Using custom a request allows to catch hard time limits.
Your app can be customized like: