[dogstatsd] Recover gracefully from serialization errors

dogstatsd.py

@@ -34,6 +34,7 @@
 # project
 from aggregator import get_formatter, MetricsBucketAggregator
 from checks.check_status import DogstatsdStatus
+from checks.metric_types import MetricTypes
 from config import get_config, get_version
 from daemon import AgentSupervisor, Daemon
 from util import chunks, get_hostname, get_uuid, plural
@@ -62,8 +63,53 @@
 COMPRESS_THRESHOLD = 1024
 
 
-def serialize_metrics(metrics):
-    serialized = json.dumps({"series": metrics})
+def add_serialization_status_metric(status, hostname):
+    """
+    Add a metric to track the number of metric serializations,
+    tagged by their status.
+    """
+    interval = 10.0
+    value = 1
+    return {
+        'tags': ["status:{0}".format(status)],
+        'metric': 'datadog.dogstatsd.serialization_status',
+        'interval': interval,
+        'device_name': None,
+        'host': hostname,
+        'points': [(time(), value / interval)],
+        'type': MetricTypes.RATE,
+    }
+
+
+def unicode_metrics(metrics):
+    for i, metric in enumerate(metrics):
+        for key, value in metric.items():
+            if isinstance(value, basestring):
+                metric[key] = unicode(value, errors='replace')
+            elif isinstance(value, tuple) or isinstance(value, list):
+                value_list = list(value)
+                for j, value_element in enumerate(value_list):
+                    if isinstance(value_element, basestring):
+                        value_list[j] = unicode(value_element, errors='replace')
+                metric[key] = tuple(value_list)
+        metrics[i] = metric
+    return metrics
+
+
+def serialize_metrics(metrics, hostname):
+    try:
+        metrics.append(add_serialization_status_metric("success", hostname))
+        serialized = json.dumps({"series": metrics})
+    except UnicodeDecodeError as e:
+        log.exception("Unable to serialize payload. Trying to replace bad characters. %s", e)
+        metrics.append(add_serialization_status_metric("failure", hostname))
+        try:
+            log.error(metrics)
+            serialized = json.dumps({"series": unicode_metrics(metrics)})
+        except Exception as e:
+            log.exception("Unable to serialize payload. Giving up. %s", e)
+            serialized = json.dumps({"series": [add_serialization_status_metric("permanent_failure", hostname)]})
+
     if len(serialized) > COMPRESS_THRESHOLD:
         headers = {'Content-Type': 'application/json',
                    'Content-Encoding': 'deflate'}
@@ -91,6 +137,7 @@ def __init__(self, interval, metrics_aggregator, api_host, api_key=None,
         self.metrics_aggregator = metrics_aggregator
         self.flush_count = 0
         self.log_count = 0
+        self.hostname = get_hostname()
 
         self.watchdog = None
         if use_watchdog:
@@ -174,7 +221,7 @@ def flush(self):
                 log.exception("Error flushing metrics")
 
     def submit(self, metrics):
-        body, headers = serialize_metrics(metrics)
+        body, headers = serialize_metrics(metrics, self.hostname)
         params = {}
         if self.api_key:
             params['api_key'] = self.api_key

tests/core/test_bucket_aggregator.py

@@ -118,7 +118,7 @@ def test_tags_gh442(self):
         import dogstatsd
         from aggregator import api_formatter
 
-        serialized = dogstatsd.serialize_metrics([api_formatter("foo", 12, 1, ('tag',), 'host')])
+        serialized = dogstatsd.serialize_metrics([api_formatter("foo", 12, 1, ('tag',), 'host')], "test-host")
         self.assertTrue('"tags": ["tag"]' in serialized[0], serialized)
 
     def test_counter(self):

tests/core/test_dogstatsd.py

@@ -167,7 +167,7 @@ def test_tags_gh442(self):
         import dogstatsd
         from aggregator import api_formatter
 
-        serialized = dogstatsd.serialize_metrics([api_formatter("foo", 12, 1, ('tag',), 'host')])
+        serialized = dogstatsd.serialize_metrics([api_formatter("foo", 12, 1, ('tag',), 'host')], "test-host")
         assert '"tags": ["tag"]' in serialized[0]
 
     def test_counter(self):