[core] Submitting metrics to the api endpoint

emitter.py

@@ -72,13 +72,10 @@ def sanitize_payload(item, log, sanitize_func):
 
     return item
 
-def http_emitter(message, log, agentConfig, endpoint):
-    "Send payload"
-    url = agentConfig['dd_url']
+def post_payload(url, message, agentConfig, log):
 
     log.debug('http_emitter: attempting postback to ' + url)
 
-    # Post back the data
     try:
         try:
             payload = json.dumps(message)
@@ -107,14 +104,8 @@ def http_emitter(message, log, agentConfig, endpoint):
     log.debug("payload_size=%d, compressed_size=%d, compression_ratio=%.3f"
               % (len(payload), len(zipped), float(len(payload))/float(len(zipped))))
 
-    apiKey = message.get('apiKey', None)
-    if not apiKey:
-        raise Exception("The http emitter requires an api key")
-
-    url = "{0}/intake/{1}?api_key={2}".format(url, endpoint, apiKey)
-
     try:
-        headers = post_headers(agentConfig, zipped)
+        headers = get_post_headers(agentConfig, zipped)
         r = requests.post(url, data=zipped, timeout=5, headers=headers)
 
         r.raise_for_status()
@@ -124,13 +115,57 @@ def http_emitter(message, log, agentConfig, endpoint):
 
     except Exception:
         log.exception("Unable to post payload.")
-        try:
-            log.error("Received status code: {0}".format(r.status_code))
-        except Exception:
-            pass
 
 
-def post_headers(agentConfig, payload):
+def split_payload(legacy_payload):
+    metrics = list(legacy_payload['metrics'])
+    del legacy_payload['metrics']
+
+    metrics_payload = {"series": []}
+
+    # See https://github.com/DataDog/dd-agent/blob/5.11.1/checks/__init__.py#L905-L926 for format
+    for ts in metrics:
+        sample = {
+            "metric": ts[0],
+            "points": [[ts[1], ts[2]]]
+        }
+
+        if len(ts) >= 4:
+            if ts[3].get('type'):
+                sample['type'] = ts[3]['type']
+            if ts[3].get('hostname'):
+                sample['host'] = ts[3]['hostname']
+            if ts[3].get('tags'):
+                sample['tags'] = ts[3]['tags']
+            if ts[3].get('device_name'):
+                sample['device'] = ts[3]['device_name']
+
+        metrics_payload["series"].append(sample)
+
+    return legacy_payload, metrics_payload
+
+def http_emitter(message, log, agentConfig, endpoint):
+    api_key = message.get('apiKey')
+
+    if not api_key:
+        raise Exception("The http emitter requires an api key")
+
+    # For perf reason. We now want to send the metrics to the api endpoint. So we are extracting them
+    # from the payload here, transform them into the expected format and send them (via the forwarder)
+
+    legacy_url = "{0}/intake/{1}?api_key={2}".format(agentConfig['dd_url'], endpoint, api_key)
+    metrics_endpoint = "{0}/api/v1/series?api_key={1}".format(agentConfig['dd_url'], api_key)
+
+    legacy_payload, metrics_payload = split_payload(message)
+
+    # Post legacy payload
+    post_payload(legacy_url, legacy_payload, agentConfig, log)
+
+    # Post metrics payload
+    post_payload(metrics_endpoint, metrics_payload, agentConfig, log)
+
+
+def get_post_headers(agentConfig, payload):
     return {
         'User-Agent': 'Datadog Agent/%s' % agentConfig['version'],
         'Content-Type': 'application/json',

tests/core/test_emitter.py

@@ -2,17 +2,48 @@
 # 3p
 from mock import Mock
 import unittest
+import simplejson as json
 
 # project
 from emitter import (
     remove_control_chars,
     remove_undecodable_chars,
     sanitize_payload,
+    split_payload
 )
 
+import os
+
+FIXTURE_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'fixtures', 'payloads')
 
 class TestEmitter(unittest.TestCase):
 
+
+    def test_payload_splitter(self):
+        with open(FIXTURE_PATH + '/legacy_payload.json') as f:
+            legacy_payload = json.load(f)
+
+        legacy_payload_split, metrics_payload = split_payload(legacy_payload)
+        series = metrics_payload['series']
+        legacy_payload_split['metrics'] = []
+
+        for s in series:
+            attributes = {}
+
+            if s.get('type'):
+                attributes['type'] = s['type']
+            if s.get('host'):
+                attributes['hostname'] = s['host']
+            if s.get('tags'):
+                attributes['tags'] = s['tags']
+            if s.get('device'):
+                attributes['device_name'] = s['device']
+
+            formatted_sample = [s['metric'], s['points'][0][0], s['points'][0][1], attributes]
+            legacy_payload_split['metrics'].append(formatted_sample)
+
+        self.assertEqual(legacy_payload, legacy_payload_split)
+
     def test_remove_control_chars(self):
         messages = [
             (u'#és9df\x7fELF\x02\x01\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x00>\x00\x01\x00\x00\x00\x06@\x00\x00\x00\x00\x00@\x00\x00\x00\x00\x00\x00´wer0sf®ré', u'#és9dfELF>@@´wer0sf®ré'),