Merge pull request #116 from DataDog/stats-metric-host

Allow to override stats metrics host

src/dogapi/stats/dog_stats_api.py

@@ -96,7 +96,7 @@ def stop(self):
             return True
 
 
-    def gauge(self, metric_name, value, timestamp=None, tags=None, sample_rate=1):
+    def gauge(self, metric_name, value, timestamp=None, tags=None, sample_rate=1, host=None):
         """
         Record the current *value* of a metric. They most recent value in
         a given flush interval will be recorded. Optionally, specify a set of
@@ -108,9 +108,10 @@ def gauge(self, metric_name, value, timestamp=None, tags=None, sample_rate=1):
         >>> dog_stats_api.gauge('cache.bytes.free', cache.get_free_bytes(), tags=['version:1.0'])
         """
         if not self._disabled:
-            self._aggregator.add_point(metric_name, tags, timestamp or time(), value, Gauge, sample_rate)
+            self._aggregator.add_point(metric_name, tags, timestamp or time(), value, Gauge,
+                sample_rate=sample_rate, host=host)
 
-    def increment(self, metric_name, value=1, timestamp=None, tags=None, sample_rate=1):
+    def increment(self, metric_name, value=1, timestamp=None, tags=None, sample_rate=1, host=None):
         """
         Increment the counter by the given *value*. Optionally, specify a list of
         *tags* to associate with the metric. This is useful for counting things
@@ -120,9 +121,10 @@ def increment(self, metric_name, value=1, timestamp=None, tags=None, sample_rate
         >>> dog_stats_api.increment('bytes.processed', file.size())
         """
         if not self._disabled:
-            self._aggregator.add_point(metric_name, tags, timestamp or time(), value, Counter, sample_rate)
+            self._aggregator.add_point(metric_name, tags, timestamp or time(), value, Counter,
+                sample_rate=sample_rate, host=host)
 
-    def histogram(self, metric_name, value, timestamp=None, tags=None, sample_rate=1):
+    def histogram(self, metric_name, value, timestamp=None, tags=None, sample_rate=1, host=None):
         """
         Sample a histogram value. Histograms will produce metrics that
         describe the distribution of the recorded values, namely the minimum,
@@ -132,10 +134,11 @@ def histogram(self, metric_name, value, timestamp=None, tags=None, sample_rate=1
         >>> dog_stats_api.histogram('uploaded_file.size', uploaded_file.size())
         """
         if not self._disabled:
-            self._aggregator.add_point(metric_name, tags, timestamp or time(), value, Histogram, sample_rate)
+            self._aggregator.add_point(metric_name, tags, timestamp or time(), value, Histogram,
+                sample_rate=sample_rate, host=host)
 
     @contextmanager
-    def timer(self, metric_name, sample_rate=1, tags=None):
+    def timer(self, metric_name, sample_rate=1, tags=None, host=None):
         """
         A context manager that will track the distribution of the contained code's run time.
         Optionally specify a list of tags to associate with the metric.
@@ -160,9 +163,10 @@ def get_user(user_id):
             yield
         finally:
             end = time()
-            self.histogram(metric_name, end - start, end, tags=tags, sample_rate=sample_rate)
+            self.histogram(metric_name, end - start, end, tags=tags,
+                sample_rate=sample_rate, host=host)
 
-    def timed(self, metric_name, sample_rate=1, tags=None):
+    def timed(self, metric_name, sample_rate=1, tags=None, host=None):
         """
         A decorator that will track the distribution of a function's run time.
         Optionally specify a list of tags to associate with the metric.
@@ -183,7 +187,7 @@ def get_user(user_id):
         def wrapper(func):
             @wraps(func)
             def wrapped(*args, **kwargs):
-                with self.timer(metric_name, sample_rate, tags):
+                with self.timer(metric_name, sample_rate, tags, host):
                     result = func(*args, **kwargs)
                     return result
             return wrapped
@@ -228,12 +232,14 @@ def _get_aggregate_metrics(self, flush_time=None):
 
         # FIXME: emit a dictionary from the aggregator
         metrics = []
-        for timestamp, value, name, tags in rolled_up_metrics:
+        for timestamp, value, name, tags, host in rolled_up_metrics:
+            if host is None:
+                host = self.host
             metric = {
                 'metric' : name,
                 'points' : [[timestamp, value]],
                 'type':    MetricType.Gauge,
-                'host':    self.host,
+                'host':    host,
                 'device':  self.device,
                 'tags'  :  tags
             }

src/dogapi/stats/metrics.py

@@ -28,42 +28,45 @@ class Gauge(Metric):
 
     stats_tag = 'g'
 
-    def __init__(self, name, tags):
+    def __init__(self, name, tags, host):
         self.name = name
         self.tags = tags
+        self.host = host
         self.value = None
 
     def add_point(self, value):
         self.value = value
 
     def flush(self, timestamp):
-        return [(timestamp, self.value, self.name, self.tags)]
+        return [(timestamp, self.value, self.name, self.tags, self.host)]
 
 class Counter(Metric):
     """ A counter metric. """
 
     stats_tag = 'c'
 
-    def __init__(self, name, tags):
+    def __init__(self, name, tags, host):
         self.name = name
         self.tags = tags
+        self.host = host
         self.count = 0
 
     def add_point(self, value):
         self.count += value
 
     def flush(self, timestamp):
-        return [(timestamp, self.count, self.name, self.tags)]
+        return [(timestamp, self.count, self.name, self.tags, self.host)]
 
 
 class Histogram(Metric):
     """ A histogram metric. """
 
     stats_tag = 'h'
 
-    def __init__(self, name, tags):
+    def __init__(self, name, tags, host):
         self.name = name
         self.tags = tags
+        self.host = host
         self.max = float("-inf")
         self.min = float("inf")
         self.sum = 0
@@ -86,17 +89,17 @@ def flush(self, timestamp):
         if not self.count:
             return []
         metrics = [
-            (timestamp, self.min,       '%s.min'   % self.name, self.tags),
-            (timestamp, self.max,       '%s.max'   % self.name, self.tags),
-            (timestamp, self.count,     '%s.count' % self.name, self.tags),
-            (timestamp, self.average(), '%s.avg'   % self.name, self.tags)
+            (timestamp, self.min,       '%s.min'   % self.name, self.tags, self.host),
+            (timestamp, self.max,       '%s.max'   % self.name, self.tags, self.host),
+            (timestamp, self.count,     '%s.count' % self.name, self.tags, self.host),
+            (timestamp, self.average(), '%s.avg'   % self.name, self.tags, self.host)
         ]
         length = len(self.samples)
         self.samples.sort()
         for p in self.percentiles:
             val = self.samples[int(round(p * length - 1))]
             name = '%s.%spercentile' % (self.name, int(p * 100))
-            metrics.append((timestamp, val, name, self.tags))
+            metrics.append((timestamp, val, name, self.tags, self.host))
         return metrics
 
     def average(self):
@@ -112,12 +115,12 @@ def __init__(self, roll_up_interval=10):
         self._metrics = defaultdict(lambda: {})
         self._roll_up_interval = roll_up_interval
 
-    def add_point(self, metric, tags, timestamp, value, metric_class, sample_rate=1):
+    def add_point(self, metric, tags, timestamp, value, metric_class, sample_rate=1, host=None):
         # The sample rate is currently ignored for in process stuff
         interval = timestamp - timestamp % self._roll_up_interval
-        key = (metric, tuple(sorted(tags)) if tags else tags)
+        key = (metric, host, tuple(sorted(tags)) if tags else tags)
         if key not in self._metrics[interval]:
-            self._metrics[interval][key] = metric_class(metric, tags)
+            self._metrics[interval][key] = metric_class(metric, tags, host)
         self._metrics[interval][key].add_point(value)
 
     def flush(self, timestamp):

src/dogapi/stats/statsd.py

@@ -18,9 +18,13 @@ def __init__(self, host='localhost', port=8125):
         self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         self.socket_sendto = self.socket.sendto
 
-    def add_point(self, metric, tags, timestamp, value, metric_class, sample_rate=1):
+    def add_point(self, metric, tags, timestamp, value, metric_class, sample_rate=1, host=None):
         if sample_rate == 1 or random() < sample_rate:
             payload = '%s:%s|%s' % (metric, value, metric_class.stats_tag)
+            if host is not None:
+                if not tags:
+                    tags = []
+                tags.append('host:%s' % host)
             if sample_rate != 1:
                 payload += '|@%s' % sample_rate
             if tags:

tests/unit/test_stats_api.py

@@ -43,10 +43,9 @@ class TestUnitDogStatsAPI(object):
     def sort_metrics(self, metrics):
         """ Sort metrics by timestamp of first point and then name """
         def sort(metric):
-            if metric['tags'] is None:
-                return (metric['points'][0][0], metric['metric'], [])
-            else:
-                return (metric['points'][0][0], metric['metric'], metric['tags'])
+            tags = metric['tags'] or []
+            host = metric['host'] or ''
+            return (metric['points'][0][0], metric['metric'], tags, host)
         return sorted(metrics, key=sort)
 
     def test_timed_decorator(self):
@@ -304,6 +303,60 @@ def test():
         for metric in reporter.metrics:
             assert metric['tags'] # this is enough
 
+    def test_host(self):
+        dog = DogStatsApi()
+        dog.start(roll_up_interval=10, flush_in_thread=False, host='default')
+        reporter = dog.reporter = MemoryReporter()
+
+        # Post the same metric with different tags.
+        dog.gauge('gauge', 12, timestamp=100.0, host='') # unset the host
+        dog.gauge('gauge', 10, timestamp=100.0)
+        dog.gauge('gauge', 15, timestamp=100.0, host='test')
+        dog.gauge('gauge', 15, timestamp=100.0, host='test')
+
+        dog.increment('counter', timestamp=100.0)
+        dog.increment('counter', timestamp=100.0)
+        dog.increment('counter', timestamp=100.0, host='test')
+        dog.increment('counter', timestamp=100.0, host='test', tags=['tag'])
+        dog.increment('counter', timestamp=100.0, host='test', tags=['tag'])
+
+        dog.flush(200.0)
+
+        metrics = self.sort_metrics(reporter.metrics)
+        nt.assert_equal(len(metrics), 6)
+
+        [c1, c2, c3, g1, g2, g3] = metrics
+        (nt.assert_equal(c['metric'], 'counter') for c in [c1, c2, c3])
+        nt.assert_equal(c1['host'], 'default')
+        nt.assert_equal(c1['tags'], None)
+        nt.assert_equal(c1['points'][0][1], 2)
+        nt.assert_equal(c2['host'], 'test')
+        nt.assert_equal(c2['tags'], None)
+        nt.assert_equal(c2['points'][0][1], 1)
+        nt.assert_equal(c3['host'], 'test')
+        nt.assert_equal(c3['tags'], ['tag'])
+        nt.assert_equal(c3['points'][0][1], 2)
+
+        (nt.assert_equal(g['metric'], 'gauge')   for g in [g1, g2, g3])
+        nt.assert_equal(g1['host'], '')
+        nt.assert_equal(g1['points'][0][1], 12)
+        nt.assert_equal(g2['host'], 'default')
+        nt.assert_equal(g2['points'][0][1], 10)
+        nt.assert_equal(g3['host'], 'test')
+        nt.assert_equal(g3['points'][0][1], 15)
+
+
+        # Ensure histograms work as well.
+        @dog.timed('timed', host='test')
+        def test():
+            pass
+        test()
+        dog.histogram('timed', 20, timestamp=300.0, host='test')
+        reporter.metrics = []
+        dog.flush(400)
+        for metric in reporter.metrics:
+            assert metric['host'] == 'test'
+
     def test_disabled_mode(self):
         dog = DogStatsApi()
         reporter = dog.reporter = MemoryReporter()