Graphite Standalone container

A new Graphite container is used
Metrics are sent to both Graphite instances
However, it seems that some metrics might be missing
This is however a direct plugin. There are differences in
carbon.conf, storage-schemas.conf and other files as well.
It is suggested to write own Dockerfile instead of using
the image available. We anyway have to do it with respect
ARM Support.

Change-Id: Id34c728f598150caac23ac167c3cce5eaf183a6c
JIRA: STORPERF-142
Signed-off-by: Shrenik <shrenik.jain@research.iiit.ac.in>
Signed-off-by: mbeierl <mark.beierl@dell.com>

2 files changed, 189 insertions, 48 deletions

diff --git a/docker/storperf-master/tests/carbon_tests/emitter_test.py b/docker/storperf-master/tests/carbon_tests/emitter_test.py
index 7f61049..f5a78d1 100644
--- a/docker/storperf-master/tests/carbon_tests/emitter_test.py
+++ b/docker/storperf-master/tests/carbon_tests/emitter_test.py
@@ -7,27 +7,39 @@
 # http://www.apache.org/licenses/LICENSE-2.0
 ##############################################################################
 
-import SocketServer
 import json
-from storperf.carbon import converter
-from storperf.carbon.emitter import CarbonMetricTransmitter
-import threading
-from time import sleep, strptime
+from time import strptime
 import unittest
 
 import mock
 
+from storperf.carbon import converter
+from storperf.carbon.emitter import CarbonMetricTransmitter
+
+
+addresses = []
+data = []
+connect_exception = []
+send_exception = []
+
+
+class MockSocket(object):
 
-class MetricsHandler(SocketServer.BaseRequestHandler):
+    def __init__(self, *args):
+        pass
 
-    def handle(self):
-        # Echo the back to the client
-        CarbonMetricTransmitterTest.response = self.request.recv(1024)
-        return
+    def connect(self, address):
+        if len(connect_exception) != 0:
+            raise connect_exception[0]
+        addresses.append(address)
 
+    def send(self, datum):
+        if len(send_exception) != 0:
+            raise send_exception[0]
+        data.append(datum)
 
-class MetricsServer(SocketServer.ThreadingMixIn, SocketServer.TCPServer):
-    pass
+    def close(self):
+        pass
 
 
 class CarbonMetricTransmitterTest(unittest.TestCase):
@@ -35,17 +47,37 @@ class CarbonMetricTransmitterTest(unittest.TestCase):
     response = None
 
     def setUp(self):
+        del addresses[:]
+        del data[:]
+        del connect_exception[:]
+        del send_exception[:]
+
+    @mock.patch("socket.socket")
+    @mock.patch("time.gmtime")
+    def test_transmit_metrics(self, mock_time, mock_socket):
+
+        mock_socket.side_effect = MockSocket
+
+        mock_time.return_value = strptime("30 Nov 00", "%d %b %y")
+
+        testconv = converter.Converter()
+        json_object = json.loads(
+            """{"timestamp" : "975542400", "key":123.0 }""")
+        result = testconv.convert_json_to_flat(json_object, "host.run-name")
 
-        address = ('localhost', 0)
-        server = MetricsServer(address, MetricsHandler)
-        ip, self.listen_port = server.server_address
+        emitter = CarbonMetricTransmitter()
+        emitter.carbon_port = self.listen_port
+        emitter.transmit_metrics(result, None)
 
-        t = threading.Thread(target=server.serve_forever)
-        t.setDaemon(True)
-        t.start()
+        self.assertEqual("host.run-name.key 123.0 975542400\n",
+                         data[1],
+                         data[1])
 
+    @mock.patch("socket.socket")
     @mock.patch("time.gmtime")
-    def test_transmit_metrics(self, mock_time):
+    def test_skip_non_numeric_metrics(self, mock_time, mock_socket):
+
+        mock_socket.side_effect = MockSocket
 
         mock_time.return_value = strptime("30 Nov 00", "%d %b %y")
 
@@ -56,17 +88,126 @@ class CarbonMetricTransmitterTest(unittest.TestCase):
 
         emitter = CarbonMetricTransmitter()
         emitter.carbon_port = self.listen_port
-        emitter.transmit_metrics(result)
+        emitter.transmit_metrics(result, None)
+
+        self.assertEqual("None.commit-marker 975542400 975542400\n",
+                         data[1],
+                         data[1])
+
+    @mock.patch("socket.socket")
+    def test_connect_fails(self, mock_socket):
+
+        mock_socket.side_effect = MockSocket
+        connect_exception.append(Exception("Mock connection error"))
+
+        testconv = converter.Converter()
+        json_object = json.loads(
+            """{"timestamp" : "975542400", "key":"value" }""")
+        result = testconv.convert_json_to_flat(json_object, "host.run-name")
+
+        emitter = CarbonMetricTransmitter()
+        emitter.carbon_port = self.listen_port
+        emitter.transmit_metrics(result, None)
+
+        self.assertEqual(0,
+                         len(data),
+                         len(data))
 
-        count = 0
+    @mock.patch("socket.socket")
+    def test_send_fails(self, mock_socket):
 
-        while (CarbonMetricTransmitterTest.response is None and count < 10):
-            count += 1
-            sleep(0.1)
+        mock_socket.side_effect = MockSocket
+        send_exception.append(Exception("Mock send error"))
+
+        testconv = converter.Converter()
+        json_object = json.loads(
+            """{"timestamp" : "975542400", "key":"value" }""")
+        result = testconv.convert_json_to_flat(json_object, "host.run-name")
+
+        emitter = CarbonMetricTransmitter()
+        emitter.carbon_port = self.listen_port
+        emitter.transmit_metrics(result, None)
+
+        self.assertEqual(0,
+                         len(data),
+                         len(data))
+
+    @mock.patch("storperf.db.graphite_db.GraphiteDB.fetch_item")
+    def test_confirm_commit(self, mock_graphite_db):
+        graphite_return = json.loads("""[
+          {"target":
+           "rw.queue-depth.2.block-size.2048.10-10-243-154.commit-marker",
+           "datapoints": [[1503078366.0, 1503078370]]}]
+           """)
+        mock_graphite_db.return_value = graphite_return
+
+        commit_marker = "commit-marker"
+
+        emitter = CarbonMetricTransmitter()
+        emitter.commit_markers[commit_marker] = 1503078366
+
+        committed = emitter.confirm_commit(commit_marker)
+        self.assertTrue(committed)
+
+    @mock.patch("storperf.db.graphite_db.GraphiteDB.fetch_item")
+    def test_confirm_multiple_commits(self, mock_graphite_db):
+        graphite_return = json.loads("""[
+          {"target":
+           "rw.queue-depth.2.block-size.2048.10-10-243-154.commit-marker",
+           "datapoints": [
+             [1503078300.0, 1503078350],
+             [1503078366.0, 1503078360]]}]
+           """)
+        mock_graphite_db.return_value = graphite_return
+
+        commit_marker = "commit-marker"
+
+        emitter = CarbonMetricTransmitter()
+        emitter.commit_markers[commit_marker] = 1503078366
+
+        committed = emitter.confirm_commit(commit_marker)
+        self.assertTrue(committed)
+
+    @mock.patch("storperf.db.graphite_db.GraphiteDB.fetch_item")
+    def test_empty_commit(self, mock_graphite_db):
+        graphite_return = json.loads("[]")
+        mock_graphite_db.return_value = graphite_return
+
+        commit_marker = "commit-marker"
+
+        emitter = CarbonMetricTransmitter()
+        emitter.commit_markers[commit_marker] = 1503078366
+
+        committed = emitter.confirm_commit(commit_marker)
+        self.assertFalse(committed)
+
+    @mock.patch("storperf.db.graphite_db.GraphiteDB.fetch_item")
+    def test_badtimestamp_commit(self, mock_graphite_db):
+        graphite_return = json.loads("""[
+          {"target":
+           "rw.queue-depth.2.block-size.2048.10-10-243-154.commit-marker",
+           "datapoints": [[1234, 1503078370]]}]
+           """)
+        mock_graphite_db.return_value = graphite_return
+
+        commit_marker = "commit-marker"
+
+        emitter = CarbonMetricTransmitter()
+        emitter.commit_markers[commit_marker] = 1503078366
+
+        committed = emitter.confirm_commit(commit_marker)
+        self.assertFalse(committed)
+
+    def test_timestamp_parse(self):
+        emitter = CarbonMetricTransmitter()
+        result = json.loads("""[
+          {"target":
+           "rw.queue-depth.2.block-size.2048.10-10-243-154.commit-marker",
+           "datapoints": [[1503078366.0, 1503078370]]}]
+           """)
+        timestamps = emitter.parse_timestamp(result)
+        self.assertEqual(1503078366, timestamps[0], timestamps[0])
 
-        self.assertEqual("host.run-name.key value 975542400\n",
-                         CarbonMetricTransmitterTest.response,
-                         CarbonMetricTransmitterTest.response)
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/docker/storperf-master/tests/utilities_tests/data_handler_test.py b/docker/storperf-master/tests/utilities_tests/data_handler_test.py
index f514ae7..6010bd3 100644
--- a/docker/storperf-master/tests/utilities_tests/data_handler_test.py
+++ b/docker/storperf-master/tests/utilities_tests/data_handler_test.py
@@ -8,11 +8,12 @@
 ##############################################################################
 
 import os
-from storperf.utilities.data_handler import DataHandler
 import unittest
 
 import mock
 
+from storperf.utilities.data_handler import DataHandler
+
 
 class MockGraphiteDB(object):
 
@@ -76,11 +77,11 @@ class DataHandlerTest(unittest.TestCase):
         mock_graphite_db.return_value = expected
         mock_time.return_value = expected[-1][0] + 10
 
-        self.current_workload = ("%s.%s.queue-depth.%s.block-size.%s" %
-                                 ("job_id",
-                                  "rw",
-                                  8,
-                                  8192))
+        self.current_workload = ("%s.%s.queue-depth.%s.block-size.%s"
+                                 % ("job_id",
+                                    "rw",
+                                    8,
+                                    8192))
 
         actual = self.data_handler._lookup_prior_data(self, 'read', 'iops')
         self.assertEqual(expected, actual)
@@ -148,8 +149,9 @@ class DataHandlerTest(unittest.TestCase):
     @mock.patch("time.time")
     @mock.patch("storperf.db.test_results_db.push_results_to_db")
     @mock.patch("storperf.db.graphite_db.GraphiteDB.fetch_series")
-    @mock.patch("storperf.db.graphite_db.JobDB.fetch_workloads")
-    def test_non_terminated_report(self, mock_job_db, mock_graphite_db,
+    @mock.patch("storperf.db.job_db.JobDB.fetch_workloads")
+    def test_non_terminated_report(self, mock_job_db,
+                                   mock_graphite_db,
                                    mock_results_db, mock_time):
         self._terminated = False
         mock_results_db.side_effect = self.push_results_to_db
@@ -169,12 +171,11 @@ class DataHandlerTest(unittest.TestCase):
         expected_range = 17.78
         expected_average = 212.49777777777774
 
-        self.current_workload = ("%s.%s.queue-depth.%s.block-size.%s" %
-                                 ("job_id",
-                                  "rw",
-                                  8,
-                                  8192))
-
+        self.current_workload = ("%s.%s.queue-depth.%s.block-size.%s"
+                                 % ("job_id",
+                                    "rw",
+                                    8,
+                                    8192))
         mock_job_db.return_value = [[self.current_workload, 4804559000, None]]
 
         self.data_handler.data_event(self)
@@ -201,7 +202,7 @@ class DataHandlerTest(unittest.TestCase):
     @mock.patch("time.time")
     @mock.patch("storperf.db.test_results_db.push_results_to_db")
     @mock.patch("storperf.db.graphite_db.GraphiteDB.fetch_series")
-    @mock.patch("storperf.db.graphite_db.JobDB.fetch_workloads")
+    @mock.patch("storperf.db.job_db.JobDB.fetch_workloads")
     def test_report_that_causes_termination(self,
                                             mock_job_db,
                                             mock_graphite_db,
@@ -236,12 +237,11 @@ class DataHandlerTest(unittest.TestCase):
         expected_range = 17.78
         expected_average = 209.2135
 
-        self.current_workload = ("%s.%s.queue-depth.%s.block-size.%s" %
-                                 ("job_id",
-                                  "rw",
-                                  8,
-                                  8192))
-
+        self.current_workload = ("%s.%s.queue-depth.%s.block-size.%s"
+                                 % ("job_id",
+                                    "rw",
+                                    8,
+                                    8192))
         mock_job_db.return_value = [[self.current_workload, 4804559000, None]]
 
         self.data_handler.data_event(self)