Merge pull request #922 from mozilla-services/fix/firehose-spinning

fix: avoid firehose logger spinning the CPU

autopush/logging.py

@@ -266,9 +266,9 @@ def _worker(self):
         self._last_send = time.time()
         while True:
             time_since_sent = time.time() - self._last_send
+            remaining_wait = self.MAX_INTERVAL - time_since_sent
             try:
-                record = self._records.get(
-                    timeout=max(self.MAX_INTERVAL-time_since_sent, 0))
+                record = self._records.get(timeout=remaining_wait)
             except Queue.Empty:
                 # Send the records
                 self._send_record_batch()
@@ -294,6 +294,7 @@ def _worker(self):
         self._send_record_batch()
 
     def _send_record_batch(self):
+        self._last_send = time.time()
         if not self._prepped:
             return
 
@@ -312,4 +313,3 @@ def _send_record_batch(self):
 
         self._prepped = []
         self._total_size = 0
-        self._last_send = time.time()

autopush/tests/test_logging.py

@@ -1,5 +1,6 @@
 import json
 import os
+import Queue
 import sys
 import StringIO
 
@@ -210,10 +211,13 @@ def test_message_max_batch(self):
     def test_queue_timeout(self):
         proc = FirehoseProcessor("test")
         proc.MAX_INTERVAL = 0
+        proc._records.get = mock_get = Mock()
+        proc._send_record_batch = mock_send = Mock()
+        mock_get.side_effect = (Queue.Empty, None)
 
         proc.start()
         proc.stop()
-        eq_(len(self.mock_boto.mock_calls), 1)
+        mock_send.assert_called()
 
     def test_batch_send_failure(self):
         proc = FirehoseProcessor("test")