Add slow task warnings for TrackedTasks

Summary:
When tasks run for longer than they should, we want to hear about it.
This restricts the timing to the actual body of the task, as that should better restrict
alerts to behaviors that are the fault of that task.

Test Plan: None

Reviewers: jukka

Reviewed By: jukka

Subscribers: changesbot, wwu

Differential Revision: https://tails.corp.dropbox.com/D97190

changes/queue/task.py

@@ -4,8 +4,9 @@
 
 from datetime import datetime, timedelta
 from functools import wraps
-from threading import local, Lock
+from threading import local, Lock, Timer
 from uuid import uuid4
+from contextlib import contextmanager
 
 from changes.config import db, queue, statsreporter
 from changes.constants import Result, Status
@@ -23,6 +24,9 @@
 
 MAX_RETRIES = 10
 
+# How many seconds we let tasks run before warning that they're slow.
+_SLOW_RUN_THRESHOLD = 5 * 60
+
 
 class NotFinished(Exception):
     def __init__(self, message=None, retry_after=None):
@@ -95,7 +99,8 @@ def _run(self, kwargs):
         })
 
         try:
-            self.func(**kwargs)
+            with self._report_slow(_SLOW_RUN_THRESHOLD, self.task_name):
+                self.func(**kwargs)
 
         except NotFinished as e:
             self.logger.info(
@@ -411,6 +416,22 @@ def _report_created(self):
         """Reports to monitoring that a new Task was created."""
         statsreporter.stats().incr('new_task_created_' + self.task_name)
 
+    @contextmanager
+    def _report_slow(self, threshold, msg):
+        """Reports a warning if the wrapped code is taking too long
+        Args:
+            threshold (int): Time to wait before warning, in seconds.
+            msg (str): Message to be included in the report.
+        """
+        def report():
+            self.logger.warning("Task taking too long ({}s so far): {}", threshold, msg)
+        t = Timer(threshold, report)
+        t.start()
+        try:
+            yield
+        finally:
+            t.cancel()
+
 
 # bind to a decorator-like naming scheme
 def tracked_task(func=None, **kwargs):