Force the push worker to time out stuck tasks
@@ -11,6 +11,7 @@ from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
from celery.utils.log import get_task_logger
from eventlet.green import httplib
from eventlet.green.httplib import BadStatusLine
+from eventlet.timeout import Timeout as EventletTimeout
from socket import error as SocketError
from urlparse import urlparse
from .distributed_tq import TaskName
@@ -26,6 +27,9 @@ from google.appengine.api import datastore_distributed
from google.appengine.ext import db
+# The maximum number of seconds a task is permitted to take.
+MAX_TASK_DURATION = 10 * 60
+
app_id = os.environ['APP_ID']
remote_host = os.environ['HOST']
@@ -91,6 +95,8 @@ def execute_task(task, headers, args):
'Args: {}'.format(args['task_name'], headers, loggable_args))
url = urlparse(args['url'])
+ timeout = EventletTimeout(MAX_TASK_DURATION)
+ try:
redirects_left = 1
while True:
urlpath = url.path
@@ -202,6 +208,15 @@ def execute_task(task, headers, args):
wait=wait_time))
logger.warning(message)
raise task.retry(countdown=wait_time)
+ except EventletTimeout as thrown_timeout:
+ if thrown_timeout != timeout:
+ raise
+
+ logger.exception('Task timed out')
+ # This could probably be calculated, but for now, just retry immediately.
+ raise task.retry(countdown=0)
+ finally:
+ timeout.cancel()
for queue in celery.conf['CELERY_QUEUES']:
