We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Allow strings to be elements in dogstatsd sets
Right now it only counts numbers.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `aggregator.py`
Content:
```
1 import logging
2 from time import time
3 
4 
5 logger = logging.getLogger(__name__)
6 
7 
8 class Infinity(Exception): pass
9 class UnknownValue(Exception): pass
10 
11 class Metric(object):
12     """
13     A base metric class that accepts points, slices them into time intervals
14     and performs roll-ups within those intervals.
15     """
16 
17     def sample(self, value, sample_rate):
18         """ Add a point to the given metric. """
19         raise NotImplementedError()
20 
21     def flush(self, timestamp, interval):
22         """ Flush all metrics up to the given timestamp. """
23         raise NotImplementedError()
24 
25 
26 class Gauge(Metric):
27     """ A metric that tracks a value at particular points in time. """
28 
29     def __init__(self, formatter, name, tags, hostname, device_name):
30         self.formatter = formatter
31         self.name = name
32         self.value = None
33         self.tags = tags
34         self.hostname = hostname
35         self.device_name = device_name
36         self.last_sample_time = None
37 
38     def sample(self, value, sample_rate):
39         self.value = value
40         self.last_sample_time = time()
41 
42     def flush(self, timestamp, interval):
43         if self.value is not None:
44             res = [self.formatter(
45                 metric=self.name,
46                 timestamp=timestamp,
47                 value=self.value,
48                 tags=self.tags,
49                 hostname=self.hostname,
50                 device_name=self.device_name
51             )]
52             self.value = None
53             return res
54 
55         return []
56 
57 
58 class Counter(Metric):
59     """ A metric that tracks a counter value. """
60 
61     def __init__(self, formatter, name, tags, hostname, device_name):
62         self.formatter = formatter
63         self.name = name
64         self.value = 0
65         self.tags = tags
66         self.hostname = hostname
67         self.device_name = device_name
68 
69     def sample(self, value, sample_rate):
70         self.value += value * int(1 / sample_rate)
71         self.last_sample_time = time()
72 
73     def flush(self, timestamp, interval):
74         try:
75             value = self.value / interval
76             return [self.formatter(
77                 metric=self.name,
78                 value=value,
79                 timestamp=timestamp,
80                 tags=self.tags,
81                 hostname=self.hostname,
82                 device_name=self.device_name
83             )]
84         finally:
85             self.value = 0
86 
87 
88 class Histogram(Metric):
89     """ A metric to track the distribution of a set of values. """
90 
91     def __init__(self, formatter, name, tags, hostname, device_name):
92         self.formatter = formatter
93         self.name = name
94         self.count = 0
95         self.samples = []
96         self.percentiles = [0.95]
97         self.tags = tags
98         self.hostname = hostname
99         self.device_name = device_name
100 
101     def sample(self, value, sample_rate):
102         self.count += int(1 / sample_rate)
103         self.samples.append(value)
104         self.last_sample_time = time()
105 
106     def flush(self, ts, interval):
107         if not self.count:
108             return []
109 
110         self.samples.sort()
111         length = len(self.samples)
112 
113         max_ = self.samples[-1]
114         med = self.samples[int(round(length/2 - 1))]
115         avg = sum(self.samples) / float(length)
116 
117         metric_aggrs = [
118             ('max', max_),
119             ('median', med),
120             ('avg', avg),
121             ('count', self.count/interval)
122         ]
123 
124         metrics = [self.formatter(
125                 hostname=self.hostname,
126                 device_name=self.device_name,
127                 tags=self.tags,
128                 metric='%s.%s' % (self.name, suffix),
129                 value=value,
130                 timestamp=ts
131             ) for suffix, value in metric_aggrs
132         ]
133 
134         for p in self.percentiles:
135             val = self.samples[int(round(p * length - 1))]
136             name = '%s.%spercentile' % (self.name, int(p * 100))
137             metrics.append(self.formatter(
138                 hostname=self.hostname,
139                 tags=self.tags,
140                 metric=name,
141                 value=val,
142                 timestamp=ts
143             ))
144 
145         # Reset our state.
146         self.samples = []
147         self.count = 0
148 
149         return metrics
150 
151 
152 class Set(Metric):
153     """ A metric to track the number of unique elements in a set. """
154 
155     def __init__(self, formatter, name, tags, hostname, device_name):
156         self.formatter = formatter
157         self.name = name
158         self.tags = tags
159         self.hostname = hostname
160         self.device_name = device_name
161         self.values = set()
162 
163     def sample(self, value, sample_rate):
164         self.values.add(value)
165         self.last_sample_time = time()
166 
167     def flush(self, timestamp, interval):
168         if not self.values:
169             return []
170         try:
171             return [self.formatter(
172                 hostname=self.hostname,
173                 device_name=self.device_name,
174                 tags=self.tags,
175                 metric=self.name,
176                 value=len(self.values),
177                 timestamp=timestamp
178             )]
179         finally:
180             self.values = set()
181 
182 
183 class Rate(Metric):
184     """ Track the rate of metrics over each flush interval """
185 
186     def __init__(self, formatter, name, tags, hostname, device_name):
187         self.formatter = formatter
188         self.name = name
189         self.tags = tags
190         self.hostname = hostname
191         self.device_name = device_name
192         self.samples = []
193 
194     def sample(self, value, sample_rate):
195         ts = time()
196         self.samples.append((int(ts), value))
197         self.last_sample_time = ts
198 
199     def _rate(self, sample1, sample2):
200         interval = sample2[0] - sample1[0]
201         if interval == 0:
202             logger.warn('Metric %s has an interval of 0. Not flushing.' % self.name)
203             raise Infinity()
204 
205         delta = sample2[1] - sample1[1]
206         if delta < 0:
207             logger.warn('Metric %s has a rate < 0. Not flushing.' % self.name)
208             raise UnknownValue()
209 
210         return (delta / interval)
211 
212     def flush(self, timestamp, interval):
213         if len(self.samples) < 2:
214             return []
215         try:
216             try:
217                 val = self._rate(self.samples[-2], self.samples[-1])
218             except:
219                 return []
220 
221             return [self.formatter(
222                 hostname=self.hostname,
223                 device_name=self.device_name,
224                 tags=self.tags,
225                 metric=self.name,
226                 value=val,
227                 timestamp=timestamp
228             )]
229         finally:
230             self.samples = self.samples[-1:]
231 
232 
233 
234 class MetricsAggregator(object):
235     """
236     A metric aggregator class.
237     """
238 
239     def __init__(self, hostname, interval=1.0, expiry_seconds=300, formatter=None):
240         self.metrics = {}
241         self.total_count = 0
242         self.count = 0
243         self.metric_type_to_class = {
244             'g': Gauge,
245             'c': Counter,
246             'h': Histogram,
247             'ms' : Histogram,
248             's'  : Set,
249             '_dd-r': Rate,
250         }
251         self.hostname = hostname
252         self.expiry_seconds = expiry_seconds
253         self.formatter = formatter or self.api_formatter
254         self.interval = float(interval)
255 
256     def packets_per_second(self, interval):
257         return round(float(self.count)/interval, 2)
258 
259     def submit_packets(self, packets):
260 
261         for packet in packets.split("\n"):
262             self.count += 1
263             # We can have colons in tags, so split once.
264             name_and_metadata = packet.split(':', 1)
265 
266             if not packet.strip():
267                 continue
268 
269             if len(name_and_metadata) != 2:
270                 raise Exception('Unparseable packet: %s' % packet)
271 
272             name = name_and_metadata[0]
273             metadata = name_and_metadata[1].split('|')
274 
275             if len(metadata) < 2:
276                 raise Exception('Unparseable packet: %s' % packet)
277 
278             # Try to cast as an int first to avoid precision issues, then as a
279             # float.
280             try:
281                 value = int(metadata[0])
282             except ValueError:
283                 try:
284                     value = float(metadata[0])
285                 except ValueError:
286                     raise Exception('Metric value must be a number: %s, %s' % name, metadata[0])
287 
288             # Parse the optional values - sample rate & tags.
289             sample_rate = 1
290             tags = None
291             for m in metadata[2:]:
292                 # Parse the sample rate
293                 if m[0] == '@':
294                     sample_rate = float(m[1:])
295                     assert 0 <= sample_rate <= 1
296                 elif m[0] == '#':
297                     tags = tuple(sorted(m[1:].split(',')))
298 
299             # Submit the metric
300             mtype = metadata[1]
301             self.submit_metric(name, value, mtype, tags=tags, sample_rate=sample_rate)
302 
303     def submit_metric(self, name, value, mtype, tags=None, hostname=None,
304                                 device_name=None, timestamp=None, sample_rate=1):
305         # Avoid calling extra functions to dedupe tags if there are none
306         if tags is None:
307             context = (name, tuple(), hostname, device_name)
308         else:
309             context = (name, tuple(sorted(set(tags))), hostname, device_name)
310         if context not in self.metrics:
311             metric_class = self.metric_type_to_class[mtype]
312             self.metrics[context] = metric_class(self.formatter, name, tags,
313                 hostname or self.hostname, device_name)
314         self.metrics[context].sample(value, sample_rate)
315 
316     def gauge(self, name, value, tags=None, hostname=None, device_name=None, timestamp=None):
317         self.submit_metric(name, value, 'g', tags, hostname, device_name, timestamp)
318 
319     def increment(self, name, value=1, tags=None, hostname=None, device_name=None):
320         self.submit_metric(name, value, 'c', tags, hostname, device_name)
321 
322     def decrement(self, name, value=-1, tags=None, hostname=None, device_name=None):
323         self.submit_metric(name, value, 'c', tags, hostname, device_name)
324 
325     def rate(self, name, value, tags=None, hostname=None, device_name=None):
326         self.submit_metric(name, value, '_dd-r', tags, hostname, device_name)
327 
328     def histogram(self, name, value, tags=None, hostname=None, device_name=None):
329         self.submit_metric(name, value, 'h', tags, hostname, device_name)
330 
331     def set(self, name, value, tags=None, hostname=None, device_name=None):
332         self.submit_metric(name, value, 's', tags, hostname, device_name)
333 
334     def flush(self):
335         timestamp = time()
336         expiry_timestamp = timestamp - self.expiry_seconds
337 
338         # Flush points and remove expired metrics. We mutate this dictionary
339         # while iterating so don't use an iterator.
340         metrics = []
341         for context, metric in self.metrics.items():
342             if metric.last_sample_time < expiry_timestamp:
343                 logger.debug("%s hasn't been submitted in %ss. Expiring." % (context, self.expiry_seconds))
344                 del self.metrics[context]
345             else:
346                 metrics += metric.flush(timestamp, self.interval)
347 
348         # Save some stats.
349         logger.debug("received %s payloads since last flush" % self.count)
350         self.total_count += self.count
351         self.count = 0
352         return metrics
353 
354     def send_packet_count(self, metric_name):
355         self.submit_metric(metric_name, self.count, 'g')
356 
357     def api_formatter(self, metric, value, timestamp, tags, hostname, device_name=None):
358         return {
359             'metric' : metric,
360             'points' : [(timestamp, value)],
361             'tags' : tags,
362             'host' : hostname,
363             'device_name': device_name
364         }
365
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/aggregator.py b/aggregator.py
--- a/aggregator.py
+++ b/aggregator.py
@@ -236,6 +236,9 @@
     A metric aggregator class.
     """
 
+    # Types of metrics that allow strings
+    ALLOW_STRINGS = ['s', ]
+
     def __init__(self, hostname, interval=1.0, expiry_seconds=300, formatter=None):
         self.metrics = {}
         self.total_count = 0
@@ -283,7 +286,13 @@
                 try:
                     value = float(metadata[0])
                 except ValueError:
-                    raise Exception('Metric value must be a number: %s, %s' % name, metadata[0])
+
+                    # If the data type is Set, we will allow strings
+                    if metadata[1] in self.ALLOW_STRINGS:
+                        value = metadata[0]
+                    else:
+                        # Otherwise, raise an error saying it must be a number
+                        raise Exception('Metric value must be a number: %s, %s' % (name, metadata[0]))
 
             # Parse the optional values - sample rate & tags.
             sample_rate = 1

{"golden_diff": "diff --git a/aggregator.py b/aggregator.py\n--- a/aggregator.py\n+++ b/aggregator.py\n@@ -236,6 +236,9 @@\n     A metric aggregator class.\n     \"\"\"\n \n+    # Types of metrics that allow strings\n+    ALLOW_STRINGS = ['s', ]\n+\n     def __init__(self, hostname, interval=1.0, expiry_seconds=300, formatter=None):\n         self.metrics = {}\n         self.total_count = 0\n@@ -283,7 +286,13 @@\n                 try:\n                     value = float(metadata[0])\n                 except ValueError:\n-                    raise Exception('Metric value must be a number: %s, %s' % name, metadata[0])\n+\n+                    # If the data type is Set, we will allow strings\n+                    if metadata[1] in self.ALLOW_STRINGS:\n+                        value = metadata[0]\n+                    else:\n+                        # Otherwise, raise an error saying it must be a number\n+                        raise Exception('Metric value must be a number: %s, %s' % (name, metadata[0]))\n \n             # Parse the optional values - sample rate & tags.\n             sample_rate = 1\n", "issue": "Allow strings to be elements in dogstatsd sets\nRight now it only counts numbers.\n\n", "before_files": [{"content": "import logging\nfrom time import time\n\n\nlogger = logging.getLogger(__name__)\n\n\nclass Infinity(Exception): pass\nclass UnknownValue(Exception): pass\n\nclass Metric(object):\n    \"\"\"\n    A base metric class that accepts points, slices them into time intervals\n    and performs roll-ups within those intervals.\n    \"\"\"\n\n    def sample(self, value, sample_rate):\n        \"\"\" Add a point to the given metric. \"\"\"\n        raise NotImplementedError()\n\n    def flush(self, timestamp, interval):\n        \"\"\" Flush all metrics up to the given timestamp. \"\"\"\n        raise NotImplementedError()\n\n\nclass Gauge(Metric):\n    \"\"\" A metric that tracks a value at particular points in time. \"\"\"\n\n    def __init__(self, formatter, name, tags, hostname, device_name):\n        self.formatter = formatter\n        self.name = name\n        self.value = None\n        self.tags = tags\n        self.hostname = hostname\n        self.device_name = device_name\n        self.last_sample_time = None\n\n    def sample(self, value, sample_rate):\n        self.value = value\n        self.last_sample_time = time()\n\n    def flush(self, timestamp, interval):\n        if self.value is not None:\n            res = [self.formatter(\n                metric=self.name,\n                timestamp=timestamp,\n                value=self.value,\n                tags=self.tags,\n                hostname=self.hostname,\n                device_name=self.device_name\n            )]\n            self.value = None\n            return res\n\n        return []\n\n\nclass Counter(Metric):\n    \"\"\" A metric that tracks a counter value. \"\"\"\n\n    def __init__(self, formatter, name, tags, hostname, device_name):\n        self.formatter = formatter\n        self.name = name\n        self.value = 0\n        self.tags = tags\n        self.hostname = hostname\n        self.device_name = device_name\n\n    def sample(self, value, sample_rate):\n        self.value += value * int(1 / sample_rate)\n        self.last_sample_time = time()\n\n    def flush(self, timestamp, interval):\n        try:\n            value = self.value / interval\n            return [self.formatter(\n                metric=self.name,\n                value=value,\n                timestamp=timestamp,\n                tags=self.tags,\n                hostname=self.hostname,\n                device_name=self.device_name\n            )]\n        finally:\n            self.value = 0\n\n\nclass Histogram(Metric):\n    \"\"\" A metric to track the distribution of a set of values. \"\"\"\n\n    def __init__(self, formatter, name, tags, hostname, device_name):\n        self.formatter = formatter\n        self.name = name\n        self.count = 0\n        self.samples = []\n        self.percentiles = [0.95]\n        self.tags = tags\n        self.hostname = hostname\n        self.device_name = device_name\n\n    def sample(self, value, sample_rate):\n        self.count += int(1 / sample_rate)\n        self.samples.append(value)\n        self.last_sample_time = time()\n\n    def flush(self, ts, interval):\n        if not self.count:\n            return []\n\n        self.samples.sort()\n        length = len(self.samples)\n\n        max_ = self.samples[-1]\n        med = self.samples[int(round(length/2 - 1))]\n        avg = sum(self.samples) / float(length)\n\n        metric_aggrs = [\n            ('max', max_),\n            ('median', med),\n            ('avg', avg),\n            ('count', self.count/interval)\n        ]\n\n        metrics = [self.formatter(\n                hostname=self.hostname,\n                device_name=self.device_name,\n                tags=self.tags,\n                metric='%s.%s' % (self.name, suffix),\n                value=value,\n                timestamp=ts\n            ) for suffix, value in metric_aggrs\n        ]\n\n        for p in self.percentiles:\n            val = self.samples[int(round(p * length - 1))]\n            name = '%s.%spercentile' % (self.name, int(p * 100))\n            metrics.append(self.formatter(\n                hostname=self.hostname,\n                tags=self.tags,\n                metric=name,\n                value=val,\n                timestamp=ts\n            ))\n\n        # Reset our state.\n        self.samples = []\n        self.count = 0\n\n        return metrics\n\n\nclass Set(Metric):\n    \"\"\" A metric to track the number of unique elements in a set. \"\"\"\n\n    def __init__(self, formatter, name, tags, hostname, device_name):\n        self.formatter = formatter\n        self.name = name\n        self.tags = tags\n        self.hostname = hostname\n        self.device_name = device_name\n        self.values = set()\n\n    def sample(self, value, sample_rate):\n        self.values.add(value)\n        self.last_sample_time = time()\n\n    def flush(self, timestamp, interval):\n        if not self.values:\n            return []\n        try:\n            return [self.formatter(\n                hostname=self.hostname,\n                device_name=self.device_name,\n                tags=self.tags,\n                metric=self.name,\n                value=len(self.values),\n                timestamp=timestamp\n            )]\n        finally:\n            self.values = set()\n\n\nclass Rate(Metric):\n    \"\"\" Track the rate of metrics over each flush interval \"\"\"\n\n    def __init__(self, formatter, name, tags, hostname, device_name):\n        self.formatter = formatter\n        self.name = name\n        self.tags = tags\n        self.hostname = hostname\n        self.device_name = device_name\n        self.samples = []\n\n    def sample(self, value, sample_rate):\n        ts = time()\n        self.samples.append((int(ts), value))\n        self.last_sample_time = ts\n\n    def _rate(self, sample1, sample2):\n        interval = sample2[0] - sample1[0]\n        if interval == 0:\n            logger.warn('Metric %s has an interval of 0. Not flushing.' % self.name)\n            raise Infinity()\n\n        delta = sample2[1] - sample1[1]\n        if delta < 0:\n            logger.warn('Metric %s has a rate < 0. Not flushing.' % self.name)\n            raise UnknownValue()\n\n        return (delta / interval)\n\n    def flush(self, timestamp, interval):\n        if len(self.samples) < 2:\n            return []\n        try:\n            try:\n                val = self._rate(self.samples[-2], self.samples[-1])\n            except:\n                return []\n\n            return [self.formatter(\n                hostname=self.hostname,\n                device_name=self.device_name,\n                tags=self.tags,\n                metric=self.name,\n                value=val,\n                timestamp=timestamp\n            )]\n        finally:\n            self.samples = self.samples[-1:]\n\n\n\nclass MetricsAggregator(object):\n    \"\"\"\n    A metric aggregator class.\n    \"\"\"\n\n    def __init__(self, hostname, interval=1.0, expiry_seconds=300, formatter=None):\n        self.metrics = {}\n        self.total_count = 0\n        self.count = 0\n        self.metric_type_to_class = {\n            'g': Gauge,\n            'c': Counter,\n            'h': Histogram,\n            'ms' : Histogram,\n            's'  : Set,\n            '_dd-r': Rate,\n        }\n        self.hostname = hostname\n        self.expiry_seconds = expiry_seconds\n        self.formatter = formatter or self.api_formatter\n        self.interval = float(interval)\n\n    def packets_per_second(self, interval):\n        return round(float(self.count)/interval, 2)\n\n    def submit_packets(self, packets):\n\n        for packet in packets.split(\"\\n\"):\n            self.count += 1\n            # We can have colons in tags, so split once.\n            name_and_metadata = packet.split(':', 1)\n\n            if not packet.strip():\n                continue\n\n            if len(name_and_metadata) != 2:\n                raise Exception('Unparseable packet: %s' % packet)\n\n            name = name_and_metadata[0]\n            metadata = name_and_metadata[1].split('|')\n\n            if len(metadata) < 2:\n                raise Exception('Unparseable packet: %s' % packet)\n\n            # Try to cast as an int first to avoid precision issues, then as a\n            # float.\n            try:\n                value = int(metadata[0])\n            except ValueError:\n                try:\n                    value = float(metadata[0])\n                except ValueError:\n                    raise Exception('Metric value must be a number: %s, %s' % name, metadata[0])\n\n            # Parse the optional values - sample rate & tags.\n            sample_rate = 1\n            tags = None\n            for m in metadata[2:]:\n                # Parse the sample rate\n                if m[0] == '@':\n                    sample_rate = float(m[1:])\n                    assert 0 <= sample_rate <= 1\n                elif m[0] == '#':\n                    tags = tuple(sorted(m[1:].split(',')))\n\n            # Submit the metric\n            mtype = metadata[1]\n            self.submit_metric(name, value, mtype, tags=tags, sample_rate=sample_rate)\n\n    def submit_metric(self, name, value, mtype, tags=None, hostname=None,\n                                device_name=None, timestamp=None, sample_rate=1):\n        # Avoid calling extra functions to dedupe tags if there are none\n        if tags is None:\n            context = (name, tuple(), hostname, device_name)\n        else:\n            context = (name, tuple(sorted(set(tags))), hostname, device_name)\n        if context not in self.metrics:\n            metric_class = self.metric_type_to_class[mtype]\n            self.metrics[context] = metric_class(self.formatter, name, tags,\n                hostname or self.hostname, device_name)\n        self.metrics[context].sample(value, sample_rate)\n\n    def gauge(self, name, value, tags=None, hostname=None, device_name=None, timestamp=None):\n        self.submit_metric(name, value, 'g', tags, hostname, device_name, timestamp)\n\n    def increment(self, name, value=1, tags=None, hostname=None, device_name=None):\n        self.submit_metric(name, value, 'c', tags, hostname, device_name)\n\n    def decrement(self, name, value=-1, tags=None, hostname=None, device_name=None):\n        self.submit_metric(name, value, 'c', tags, hostname, device_name)\n\n    def rate(self, name, value, tags=None, hostname=None, device_name=None):\n        self.submit_metric(name, value, '_dd-r', tags, hostname, device_name)\n\n    def histogram(self, name, value, tags=None, hostname=None, device_name=None):\n        self.submit_metric(name, value, 'h', tags, hostname, device_name)\n\n    def set(self, name, value, tags=None, hostname=None, device_name=None):\n        self.submit_metric(name, value, 's', tags, hostname, device_name)\n\n    def flush(self):\n        timestamp = time()\n        expiry_timestamp = timestamp - self.expiry_seconds\n\n        # Flush points and remove expired metrics. We mutate this dictionary\n        # while iterating so don't use an iterator.\n        metrics = []\n        for context, metric in self.metrics.items():\n            if metric.last_sample_time < expiry_timestamp:\n                logger.debug(\"%s hasn't been submitted in %ss. Expiring.\" % (context, self.expiry_seconds))\n                del self.metrics[context]\n            else:\n                metrics += metric.flush(timestamp, self.interval)\n\n        # Save some stats.\n        logger.debug(\"received %s payloads since last flush\" % self.count)\n        self.total_count += self.count\n        self.count = 0\n        return metrics\n\n    def send_packet_count(self, metric_name):\n        self.submit_metric(metric_name, self.count, 'g')\n\n    def api_formatter(self, metric, value, timestamp, tags, hostname, device_name=None):\n        return {\n            'metric' : metric,\n            'points' : [(timestamp, value)],\n            'tags' : tags,\n            'host' : hostname,\n            'device_name': device_name\n        }\n", "path": "aggregator.py"}], "after_files": [{"content": "import logging\nfrom time import time\n\n\nlogger = logging.getLogger(__name__)\n\n\nclass Infinity(Exception): pass\nclass UnknownValue(Exception): pass\n\nclass Metric(object):\n    \"\"\"\n    A base metric class that accepts points, slices them into time intervals\n    and performs roll-ups within those intervals.\n    \"\"\"\n\n    def sample(self, value, sample_rate):\n        \"\"\" Add a point to the given metric. \"\"\"\n        raise NotImplementedError()\n\n    def flush(self, timestamp, interval):\n        \"\"\" Flush all metrics up to the given timestamp. \"\"\"\n        raise NotImplementedError()\n\n\nclass Gauge(Metric):\n    \"\"\" A metric that tracks a value at particular points in time. \"\"\"\n\n    def __init__(self, formatter, name, tags, hostname, device_name):\n        self.formatter = formatter\n        self.name = name\n        self.value = None\n        self.tags = tags\n        self.hostname = hostname\n        self.device_name = device_name\n        self.last_sample_time = None\n\n    def sample(self, value, sample_rate):\n        self.value = value\n        self.last_sample_time = time()\n\n    def flush(self, timestamp, interval):\n        if self.value is not None:\n            res = [self.formatter(\n                metric=self.name,\n                timestamp=timestamp,\n                value=self.value,\n                tags=self.tags,\n                hostname=self.hostname,\n                device_name=self.device_name\n            )]\n            self.value = None\n            return res\n\n        return []\n\n\nclass Counter(Metric):\n    \"\"\" A metric that tracks a counter value. \"\"\"\n\n    def __init__(self, formatter, name, tags, hostname, device_name):\n        self.formatter = formatter\n        self.name = name\n        self.value = 0\n        self.tags = tags\n        self.hostname = hostname\n        self.device_name = device_name\n\n    def sample(self, value, sample_rate):\n        self.value += value * int(1 / sample_rate)\n        self.last_sample_time = time()\n\n    def flush(self, timestamp, interval):\n        try:\n            value = self.value / interval\n            return [self.formatter(\n                metric=self.name,\n                value=value,\n                timestamp=timestamp,\n                tags=self.tags,\n                hostname=self.hostname,\n                device_name=self.device_name\n            )]\n        finally:\n            self.value = 0\n\n\nclass Histogram(Metric):\n    \"\"\" A metric to track the distribution of a set of values. \"\"\"\n\n    def __init__(self, formatter, name, tags, hostname, device_name):\n        self.formatter = formatter\n        self.name = name\n        self.count = 0\n        self.samples = []\n        self.percentiles = [0.95]\n        self.tags = tags\n        self.hostname = hostname\n        self.device_name = device_name\n\n    def sample(self, value, sample_rate):\n        self.count += int(1 / sample_rate)\n        self.samples.append(value)\n        self.last_sample_time = time()\n\n    def flush(self, ts, interval):\n        if not self.count:\n            return []\n\n        self.samples.sort()\n        length = len(self.samples)\n\n        max_ = self.samples[-1]\n        med = self.samples[int(round(length/2 - 1))]\n        avg = sum(self.samples) / float(length)\n\n        metric_aggrs = [\n            ('max', max_),\n            ('median', med),\n            ('avg', avg),\n            ('count', self.count/interval)\n        ]\n\n        metrics = [self.formatter(\n                hostname=self.hostname,\n                device_name=self.device_name,\n                tags=self.tags,\n                metric='%s.%s' % (self.name, suffix),\n                value=value,\n                timestamp=ts\n            ) for suffix, value in metric_aggrs\n        ]\n\n        for p in self.percentiles:\n            val = self.samples[int(round(p * length - 1))]\n            name = '%s.%spercentile' % (self.name, int(p * 100))\n            metrics.append(self.formatter(\n                hostname=self.hostname,\n                tags=self.tags,\n                metric=name,\n                value=val,\n                timestamp=ts\n            ))\n\n        # Reset our state.\n        self.samples = []\n        self.count = 0\n\n        return metrics\n\n\nclass Set(Metric):\n    \"\"\" A metric to track the number of unique elements in a set. \"\"\"\n\n    def __init__(self, formatter, name, tags, hostname, device_name):\n        self.formatter = formatter\n        self.name = name\n        self.tags = tags\n        self.hostname = hostname\n        self.device_name = device_name\n        self.values = set()\n\n    def sample(self, value, sample_rate):\n        self.values.add(value)\n        self.last_sample_time = time()\n\n    def flush(self, timestamp, interval):\n        if not self.values:\n            return []\n        try:\n            return [self.formatter(\n                hostname=self.hostname,\n                device_name=self.device_name,\n                tags=self.tags,\n                metric=self.name,\n                value=len(self.values),\n                timestamp=timestamp\n            )]\n        finally:\n            self.values = set()\n\n\nclass Rate(Metric):\n    \"\"\" Track the rate of metrics over each flush interval \"\"\"\n\n    def __init__(self, formatter, name, tags, hostname, device_name):\n        self.formatter = formatter\n        self.name = name\n        self.tags = tags\n        self.hostname = hostname\n        self.device_name = device_name\n        self.samples = []\n\n    def sample(self, value, sample_rate):\n        ts = time()\n        self.samples.append((int(ts), value))\n        self.last_sample_time = ts\n\n    def _rate(self, sample1, sample2):\n        interval = sample2[0] - sample1[0]\n        if interval == 0:\n            logger.warn('Metric %s has an interval of 0. Not flushing.' % self.name)\n            raise Infinity()\n\n        delta = sample2[1] - sample1[1]\n        if delta < 0:\n            logger.warn('Metric %s has a rate < 0. Not flushing.' % self.name)\n            raise UnknownValue()\n\n        return (delta / interval)\n\n    def flush(self, timestamp, interval):\n        if len(self.samples) < 2:\n            return []\n        try:\n            try:\n                val = self._rate(self.samples[-2], self.samples[-1])\n            except:\n                return []\n\n            return [self.formatter(\n                hostname=self.hostname,\n                device_name=self.device_name,\n                tags=self.tags,\n                metric=self.name,\n                value=val,\n                timestamp=timestamp\n            )]\n        finally:\n            self.samples = self.samples[-1:]\n\n\n\nclass MetricsAggregator(object):\n    \"\"\"\n    A metric aggregator class.\n    \"\"\"\n\n    # Types of metrics that allow strings\n    ALLOW_STRINGS = ['s', ]\n\n    def __init__(self, hostname, interval=1.0, expiry_seconds=300, formatter=None):\n        self.metrics = {}\n        self.total_count = 0\n        self.count = 0\n        self.metric_type_to_class = {\n            'g': Gauge,\n            'c': Counter,\n            'h': Histogram,\n            'ms' : Histogram,\n            's'  : Set,\n            '_dd-r': Rate,\n        }\n        self.hostname = hostname\n        self.expiry_seconds = expiry_seconds\n        self.formatter = formatter or self.api_formatter\n        self.interval = float(interval)\n\n    def packets_per_second(self, interval):\n        return round(float(self.count)/interval, 2)\n\n    def submit_packets(self, packets):\n\n        for packet in packets.split(\"\\n\"):\n            self.count += 1\n            # We can have colons in tags, so split once.\n            name_and_metadata = packet.split(':', 1)\n\n            if not packet.strip():\n                continue\n\n            if len(name_and_metadata) != 2:\n                raise Exception('Unparseable packet: %s' % packet)\n\n            name = name_and_metadata[0]\n            metadata = name_and_metadata[1].split('|')\n\n            if len(metadata) < 2:\n                raise Exception('Unparseable packet: %s' % packet)\n\n            # Try to cast as an int first to avoid precision issues, then as a\n            # float.\n            try:\n                value = int(metadata[0])\n            except ValueError:\n                try:\n                    value = float(metadata[0])\n                except ValueError:\n\n                    # If the data type is Set, we will allow strings\n                    if metadata[1] in self.ALLOW_STRINGS:\n                        value = metadata[0]\n                    else:\n                        # Otherwise, raise an error saying it must be a number\n                        raise Exception('Metric value must be a number: %s, %s' % (name, metadata[0]))\n\n            # Parse the optional values - sample rate & tags.\n            sample_rate = 1\n            tags = None\n            for m in metadata[2:]:\n                # Parse the sample rate\n                if m[0] == '@':\n                    sample_rate = float(m[1:])\n                    assert 0 <= sample_rate <= 1\n                elif m[0] == '#':\n                    tags = tuple(sorted(m[1:].split(',')))\n\n            # Submit the metric\n            mtype = metadata[1]\n            self.submit_metric(name, value, mtype, tags=tags, sample_rate=sample_rate)\n\n    def submit_metric(self, name, value, mtype, tags=None, hostname=None,\n                                device_name=None, timestamp=None, sample_rate=1):\n        # If the value is not NaN, +infinity or -infinity\n        if value == value and value != float('inf') and value != float('-inf'):\n            # Avoid calling extra functions to dedupe tags if there are none\n            if tags is None:\n                context = (name, tuple(), hostname, device_name)\n            else:\n                context = (name, tuple(sorted(set(tags))), hostname, device_name)\n            if context not in self.metrics:\n                metric_class = self.metric_type_to_class[mtype]\n                self.metrics[context] = metric_class(self.formatter, name, tags,\n                    hostname or self.hostname, device_name)\n            self.metrics[context].sample(value, sample_rate)\n\n        elif value != value:\n            logger.warning(\"Trying to send a NaN value for metric %s\" % name)\n        elif value == float('inf'):\n            logger.warning(\"Trying to send an Infinity value for metric %s\" % name)\n        elif value == float('-inf'):\n            logger.warning(\"Trying to send an -Infinity value for metric %s\" % name)\n\n    def gauge(self, name, value, tags=None, hostname=None, device_name=None, timestamp=None):\n        self.submit_metric(name, value, 'g', tags, hostname, device_name, timestamp)\n\n    def increment(self, name, value=1, tags=None, hostname=None, device_name=None):\n        self.submit_metric(name, value, 'c', tags, hostname, device_name)\n\n    def decrement(self, name, value=-1, tags=None, hostname=None, device_name=None):\n        self.submit_metric(name, value, 'c', tags, hostname, device_name)\n\n    def rate(self, name, value, tags=None, hostname=None, device_name=None):\n        self.submit_metric(name, value, '_dd-r', tags, hostname, device_name)\n\n    def histogram(self, name, value, tags=None, hostname=None, device_name=None):\n        self.submit_metric(name, value, 'h', tags, hostname, device_name)\n\n    def set(self, name, value, tags=None, hostname=None, device_name=None):\n        self.submit_metric(name, value, 's', tags, hostname, device_name)\n\n    def flush(self):\n        timestamp = time()\n        expiry_timestamp = timestamp - self.expiry_seconds\n\n        # Flush points and remove expired metrics. We mutate this dictionary\n        # while iterating so don't use an iterator.\n        metrics = []\n        for context, metric in self.metrics.items():\n            if metric.last_sample_time < expiry_timestamp:\n                logger.debug(\"%s hasn't been submitted in %ss. Expiring.\" % (context, self.expiry_seconds))\n                del self.metrics[context]\n            else:\n                metrics += metric.flush(timestamp, self.interval)\n\n        # Save some stats.\n        logger.debug(\"received %s payloads since last flush\" % self.count)\n        self.total_count += self.count\n        self.count = 0\n        return metrics\n\n    def send_packet_count(self, metric_name):\n        self.submit_metric(metric_name, self.count, 'g')\n\n    def api_formatter(self, metric, value, timestamp, tags, hostname, device_name=None):\n        return {\n            'metric' : metric,\n            'points' : [(timestamp, value)],\n            'tags' : tags,\n            'host' : hostname,\n            'device_name': device_name\n        }\n", "path": "aggregator.py"}]}