Submit metrics for transform tasks

This submits coarse statsd metrics for transform tasks:
* Counter for task dispatches
* Timer for running tasks
* Task error counter.

backdrop/core/errors.py

@@ -1,3 +1,6 @@
+from functools import wraps
+
+
 class BackdropError(StandardError):
     pass
 
@@ -23,3 +26,21 @@ class InvalidOperationError(ValueError):
     """Raised if an invalid collect function is provided, or if an error
     is raised from a collect function"""
     pass
+
+
+def incr_on_error(stats_client, metric_name):
+    """
+    Increments metric_name if func() raises an exception, then
+    re-raises the exception.
+    """
+
+    def decorator(func):
+        @wraps(func)
+        def wrapped(*args, **kwargs):
+            try:
+                func(*args, **kwargs)
+            except:
+                stats_client.incr(metric_name)
+                raise
+        return wrapped
+    return decorator

backdrop/transformers/dispatch.py

@@ -3,7 +3,10 @@
 
 from os import getenv
 
+from statsd import StatsClient
+
 from backdrop.core.log_handler import get_log_file_handler
+from backdrop.core.errors import incr_on_error
 
 from worker import app, config
 
@@ -15,6 +18,9 @@
 logger.addHandler(
     get_log_file_handler("log/{}.log".format(GOVUK_ENV), logging.DEBUG))
 
+stats_client = StatsClient(prefix=getenv("GOVUK_STATSD_PREFIX",
+                                         "pp.apps.backdrop.transformers.worker"))
+
 
 @app.task(ignore_result=True)
 def entrypoint(dataset_id, earliest, latest):
@@ -36,6 +42,7 @@ def entrypoint(dataset_id, earliest, latest):
             'backdrop.transformers.dispatch.run_transform',
             args=(data_set_config, transform, earliest, latest)
         )
+        stats_client.incr('dispatch')
 
 
 def get_query_parameters(transform, earliest, latest):
@@ -93,6 +100,8 @@ def get_or_get_and_create_output_dataset(transform, input_dataset):
 
 
 @app.task(ignore_result=True)
+@stats_client.timer('run_transform')
+@incr_on_error(stats_client, 'run_transform.error')
 def run_transform(data_set_config, transform, earliest, latest):
     data_set = DataSet.from_group_and_type(
         config.BACKDROP_READ_URL,
@@ -113,3 +122,5 @@ def run_transform(data_set_config, transform, earliest, latest):
         transform,
         data_set_config)
     output_data_set.post(transformed_data)
+
+    stats_client.incr('run_transform.success')