Add client side aggregation

datadog/__init__.py

@@ -37,7 +37,9 @@ def initialize(
     api_host=None,  # type: Optional[str]
     statsd_host=None,  # type: Optional[str]
     statsd_port=None,  # type: Optional[int]
+    statsd_disable_aggregator=True,  # type: bool
     statsd_disable_buffering=True,  # type: bool
+    statsd_aggregation_flush_interval=2,  # type: float
     statsd_use_default_route=False,  # type: bool
     statsd_socket_path=None,  # type: Optional[str]
     statsd_namespace=None,  # type: Optional[str]
@@ -76,6 +78,14 @@ def initialize(
                                      (default: True).
     :type statsd_disable_buffering: boolean
 
+    :param statsd_disable_aggregator: Enable/disable statsd client aggregation support
+                                     (default: True).
+    :type statsd_disable_aggregator: boolean
+
+    :param statsd_aggregation_flush_interval: Sets the flush interval for aggregation
+                                     (default: 2 seconds)
+    :type statsd_aggregation_flush_interval: float
+
     :param statsd_use_default_route: Dynamically set the statsd host to the default route
                                      (Useful when running the client in a container)
     :type statsd_use_default_route: boolean
@@ -128,8 +138,11 @@ def initialize(
     if statsd_constant_tags:
         statsd.constant_tags += statsd_constant_tags
 
+    if statsd_disable_aggregator:
+        statsd.disable_aggregation()
+    else:
+        statsd.enable_aggregation(statsd_aggregation_flush_interval)
     statsd.disable_buffering = statsd_disable_buffering
-
     api._return_raw_response = return_raw_response
 
     # HTTP client and API options

datadog/dogstatsd/aggregator.py

@@ -0,0 +1,62 @@
+import threading
+from datadog.dogstatsd.metrics import (
+    CountMetric,
+    GaugeMetric,
+    SetMetric,
+)
+from datadog.dogstatsd.metric_types import MetricType
+
+
+class Aggregator(object):
+    def __init__(self):
+        self.metrics_map = {
+            MetricType.COUNT: {},
+            MetricType.GAUGE: {},
+            MetricType.SET: {},
+        }
+        self._locks = {
+            MetricType.COUNT: threading.RLock(),
+            MetricType.GAUGE: threading.RLock(),
+            MetricType.SET: threading.RLock(),
+        }
+
+    def flush_aggregated_metrics(self):
+        metrics = []
+        for metric_type in self.metrics_map.keys():
+            with self._locks[metric_type]:
+                current_metrics = self.metrics_map[metric_type]
+                self.metrics_map[metric_type] = {}
+            for metric in current_metrics.values():
+                metrics.extend(metric.get_data() if isinstance(metric, SetMetric) else [metric])
+        return metrics
+
+    def get_context(self, name, tags):
+        tags_str = ",".join(tags) if tags is not None else ""
+        return "{}:{}".format(name, tags_str)
+
+    def count(self, name, value, tags, rate, timestamp=0):
+        return self.add_metric(
+            MetricType.COUNT, CountMetric, name, value, tags, rate, timestamp
+        )
+
+    def gauge(self, name, value, tags, rate, timestamp=0):
+        return self.add_metric(
+            MetricType.GAUGE, GaugeMetric, name, value, tags, rate, timestamp
+        )
+
+    def set(self, name, value, tags, rate, timestamp=0):
+        return self.add_metric(
+            MetricType.SET, SetMetric, name, value, tags, rate, timestamp
+        )
+
+    def add_metric(
+        self, metric_type, metric_class, name, value, tags, rate, timestamp=0
+    ):
+        context = self.get_context(name, tags)
+        with self._locks[metric_type]:
+            if context in self.metrics_map[metric_type]:
+                self.metrics_map[metric_type][context].aggregate(value)
+            else:
+                self.metrics_map[metric_type][context] = metric_class(
+                    name, value, tags, rate, timestamp
+                )