If response code from OpenTSDB is 400, don't retry. (#435)

Co-authored-by: Jonathan Creasy <[email protected]>

Author: itamarst

.travis.yml

@@ -7,7 +7,7 @@ python:
   - "3.6"
   - "3.7"
 install:
-  - pip install pylint pylint_runner ordereddict mysqlclient requests feedparser prometheus_client
+  - pip install pylint pylint_runner ordereddict mysqlclient requests feedparser prometheus_client flask
 script:
   - pylint_runner --rcfile .pylintrc
   - ./tests.py

CHANGELOG.md

@@ -8,6 +8,12 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 - A namespace prefix can be added to all metrics.
 - An optional status monitoring API, serving JSON over HTTP
 
+### Bugfixes
+
+- If response code from OpenTSDB is 400, don't retry sending since this means we
+  sent bad data.
+  
+
 ## [1.3.1](https://github.com/OpenTSDB/tcollector/issues?utf8=%E2%9C%93&q=milestone%3A1.3.1+)
 ### Collectors Added
 - docker.py - Pulls metrics from a local Docker instance, tries /var/run/docker.sock, then localhost API

fake_opentsdb.py

@@ -0,0 +1,20 @@
+"""
+A fake OpenTSDB server using Flask.
+
+You can modify the response code using the FAKE_OPENTSDB_RESPONSE environment
+variable.
+"""
+
+import os
+
+from flask import Flask
+app = Flask(__name__)
+
+RESPONSE_CODE = int(os.environ.get("FAKE_OPENTSDB_RESPONSE", 204))
+
+
+@app.route('/api/put', methods=["PUT", "POST"])
+def put_message():
+    # This could be extended to write the messages to disk, for use by
+    # additional tests...
+    return "", RESPONSE_CODE

tcollector.py

@@ -44,11 +44,11 @@
     import importlib
     from queue import Queue, Empty, Full # pylint: disable=import-error
     from urllib.request import Request, urlopen # pylint: disable=maybe-no-member,no-name-in-module,import-error
-    from urllib.error import HTTPError # pylint: disable=maybe-no-member,no-name-in-module,import-error
+    from urllib.error import HTTPError, URLError # pylint: disable=maybe-no-member,no-name-in-module,import-error
     from http.server import HTTPServer, BaseHTTPRequestHandler # pylint: disable=maybe-no-member,no-name-in-module,import-error
 else:
     from Queue import Queue, Empty, Full # pylint: disable=maybe-no-member,no-name-in-module,import-error
-    from urllib2 import Request, urlopen, HTTPError # pylint: disable=maybe-no-member,no-name-in-module,import-error
+    from urllib2 import Request, urlopen, HTTPError, URLError # pylint: disable=maybe-no-member,no-name-in-module,import-error
     from BaseHTTPServer import HTTPServer, BaseHTTPRequestHandler # pylint: disable=maybe-no-member,no-name-in-module,import-error
 
 # global variables.
@@ -846,7 +846,10 @@ def send_data_via_http(self):
                          % base64.b64encode("%s:%s" % (self.http_username, self.http_password)))
         req.add_header("Content-Type", "application/json")
         try:
-            response = urlopen(req, json.dumps(metrics))
+            body = json.dumps(metrics)
+            if not isinstance(body, bytes):
+                body = body.encode("utf-8")
+            response = urlopen(req, body)
             LOG.debug("Received response %s %s", response.getcode(), response.read().rstrip('\n'))
 
             # clear out the sendq
@@ -857,9 +860,16 @@ def send_data_via_http(self):
             #     print line,
             #     print
         except HTTPError as e:
-            LOG.error("Got error %s %s", e, e.read().rstrip('\n'))
-            # for line in http_error:
-            #   print line,
+            if e.code == 400:
+                LOG.error("Some data was bad, so not going to resend it.")
+                # This means one or more of the data points were bad
+                # (http://opentsdb.net/docs/build/html/api_http/put.html#response).
+                # As such, there's no point resending them.
+                self.sendq = []
+
+            LOG.error("Got error %s %s", e, e.read())
+        except URLError as e:
+            LOG.error("Got error URL %s", e)
 
 
 def setup_logging(logfile=DEFAULT_LOG, max_bytes=None, backup_count=None):

tests.py

@@ -14,13 +14,19 @@
 
 import os
 import sys
+import time
 from stat import S_ISDIR, S_ISREG, ST_MODE
 import unittest
+import subprocess
 
 import mocks
 import tcollector
 import json
 import threading
+try:
+    import flask
+except ImportError:
+    flask = None
 
 PY3 = sys.version_info[0] > 2
 
@@ -101,6 +107,54 @@ def test_endtoend(self):
         self.assertEqual(json.loads(r), [c.to_json() for c in collectors.values()])
 
 
+@unittest.skipUnless(flask, "Flask not installed")
+class SenderThreadHTTPTests(unittest.TestCase):
+    """Tests for HTTP sending."""
+
+    def run_fake_opentsdb(self, response_code):
+        env = os.environ.copy()
+        env["FLASK_APP"] = "fake_opentsdb.py"
+        env["FAKE_OPENTSDB_RESPONSE"] = str(response_code)
+        flask = subprocess.Popen(["flask", "run", "--port", "4242"], env=env)
+        time.sleep(1)  # wait for it to start
+        self.addCleanup(flask.terminate)
+
+    def send_query_with_response_code(self, response_code):
+        """
+        Send a HTTP query using SenderThread, with server that returns given
+        response code.
+        """
+        self.run_fake_opentsdb(response_code)
+        reader = tcollector.ReaderThread(1, 10, True)
+        sender = tcollector.SenderThread(
+            reader, False, [("localhost", 4242)], False, {},
+            http=True, http_api_path="api/put"
+        )
+        sender.sendq.append("mymetric 123 12 a=b")
+        sender.send_data()
+        return sender
+
+    def test_normal(self):
+        """If response is OK, sendq is cleared."""
+        sender = self.send_query_with_response_code(204)
+        self.assertEqual(len(sender.sendq), 0)
+
+    def test_error(self):
+        """
+        If response is unexpected, e.g. 500 error, sendq is not cleared so we
+        can retry.
+        """
+        sender = self.send_query_with_response_code(500)
+        self.assertEqual(len(sender.sendq), 1)
+
+    def test_bad_messages(self):
+        """
+        If response is 400, sendq is cleared since there's no point retrying
+        bad messages.
+        """
+        sender = self.send_query_with_response_code(400)
+        self.assertEqual(len(sender.sendq), 0)
+
 class NamespacePrefixTests(unittest.TestCase):
     """Tests for metric namespace prefix."""
 
@@ -114,7 +168,6 @@ def test_prefix_added(self):
         self.assertEqual(collector.lines_received, 1)
         self.assertEqual(collector.lines_invalid, 0)
 
-
 class TSDBlacklistingTests(unittest.TestCase):
     """
     Tests of TSD blacklisting logic
@@ -379,6 +432,8 @@ def test_multi_empty_line_put_cond(self):
         self.assertEqual(stderr, [])
 
 if __name__ == '__main__':
+    import logging
+    logging.basicConfig()
     cdir = os.path.join(os.path.dirname(os.path.realpath(sys.argv[0])),
                         'collectors')
     tcollector.setup_python_path(cdir) # pylint: disable=maybe-no-member