Fix potential data loss when using direct ingestion (#76)

* Fix the thread blocking issue

* Add points back to buffer when sending points fails

Author: xshipeng

open_source_license.txt

@@ -1,6 +1,6 @@
 open_source_license.txt
 
-Wavefront by VMware SDK for Python 1.7.5 GA
+Wavefront by VMware SDK for Python 1.7.6 GA
 
 ======================================================================
 
@@ -291,4 +291,4 @@ Source Files is valid for three years from the date you acquired or last used th
 Software product. Alternatively, the Source Files may accompany the
 VMware service.
 
-[WAVEFRONTSDKPYTHON15GANT070819]
+[WAVEFRONTSDKPYTHON176GANT070819]

setup.py

@@ -15,7 +15,7 @@
 
 setuptools.setup(
     name='wavefront-sdk-python',
-    version='1.7.5',
+    version='1.7.6',
     author='Wavefront by VMware',
     author_email='chitimba@wavefront.com',
     url='https://github.com/wavefrontHQ/wavefront-sdk-python',

wavefront_sdk/common/constants.py

@@ -56,3 +56,6 @@
 
 # Distribution name for sdk
 WAVEFRONT_SDK_PYTHON = 'wavefront-sdk-python'
+
+# Default http status code for sending points
+NO_HTTP_RESPONSE = -1

wavefront_sdk/direct.py

@@ -185,6 +185,7 @@ def _report(self, points, data_format, entity_prefix, report_errors):
         @param data_format: Type of data to be sent
         @type data_format: str
         """
+        status_code = constants.NO_HTTP_RESPONSE
         try:
             if data_format == self.WAVEFRONT_EVENT_FORMAT:
                 response = requests.post(self.server + self.EVENT_END_POINT,
@@ -198,23 +199,29 @@ def _report(self, points, data_format, entity_prefix, report_errors):
                                          params=params,
                                          headers=self._headers,
                                          data=compressed_data)
-
-            self._sdk_metrics_registry.new_delta_counter(
-                '{}.report.{}'.format(entity_prefix,
-                                      response.status_code)).inc()
-            response.raise_for_status()
-        except Exception as error:
+            status_code = response.status_code
+            self._sdk_metrics_registry.new_delta_counter('{}.report.{}'.format(
+                entity_prefix, status_code)).inc()
+        except requests.exceptions.RequestException:
             report_errors.inc()
-            raise error
+        return status_code
 
     def _batch_report(self, batch_line_data, data_format, entity_prefix,
-                      report_errors):
+                      report_errors, data_buffer, dropped_point_counter):
         """One api call sending one given list of data.
 
         @param batch_line_data: List of data to be sent
         @type batch_line_data: list
         @param data_format: Type of data to be sent
         @type data_format: str
+        @param entity_prefix: Metric prefix of data type
+        @type entity_prefix: str
+        @param report_errors: Counter of errors
+        @type report_errors: WavefrontSdkCounter
+        @param data_buffer: Data buffer to be flush and sent
+        @type data_buffer: Queue
+        @param dropped_point_counter: Counter of dropped points
+        @type dropped_point_counter: WavefrontSdkCounter
         """
         # Sending events through direct ingestion does not support batching.
         batch_size = int(
@@ -223,31 +230,91 @@ def _batch_report(self, batch_line_data, data_format, entity_prefix,
         # Split data into chunks, each with the size of given batch_size
         for batch in utils.chunks(batch_line_data, batch_size):
             # report once per batch
+            status_code = self._report('\n'.join(batch) + '\n', data_format,
+                                       entity_prefix, report_errors)
+            if 400 <= status_code <= 599 or status_code == -1:
+                if status_code == 401:
+                    logging.error(
+                        'Failed to report %s data points to wavefront '
+                        '(HTTP %d). Please verify that your API Token is '
+                        'correct! All %s data points are discarded. ',
+                        data_format, status_code, data_format)
+                    dropped_point_counter.inc(len(batch))
+                elif status_code == 403:
+                    if data_format == self.WAVEFRONT_METRIC_FORMAT:
+                        logging.error(
+                            'Failed to report %s data points to wavefront '
+                            '(HTTP %d). Please verify that Direct Data '
+                            'Ingestion is enabled for your account! '
+                            'All %s data points are discarded. ', data_format,
+                            status_code, data_format)
+                        dropped_point_counter.inc(len(batch))
+                    else:
+                        logging.error(
+                            'Failed to report %s data points to wavefront '
+                            '(HTTP %d). Please verify that Direct Data '
+                            'Ingestion and %s data points are enabled for '
+                            'your account! All %s data points are discarded. ',
+                            data_format, status_code, data_format, data_format)
+                        dropped_point_counter.inc(len(batch))
+                else:
+                    logging.error(
+                        'Failed to report %s data points to wavefront '
+                        '(HTTP %d). Data will be requeued and resent.',
+                        data_format, status_code)
+                    self._requeue(batch, data_format, data_buffer,
+                                  dropped_point_counter)
+
+    @staticmethod
+    def _requeue(points, data_format, data_buffer, dropped_point_counter):
+        """Add point data back to buffer queue.
+
+        @param points: Point data in line format
+        @type points: List[str]
+        @param data_format: Type of data to be sent
+        @type data_format: str
+        @param data_buffer: Data buffer to be flush and sent
+        @type data_buffer: Queue
+        @param dropped_point_counter: Counter of dropped points
+        @type dropped_point_counter: WavefrontSdkCounter
+        """
+        added_back_to_buffer_count = 0
+        for point in points:
             try:
-                self._report('\n'.join(batch) + '\n', data_format,
-                             entity_prefix, report_errors)
-            # pylint: disable=broad-except,fixme
-            # TODO: Please replace a generic Exception with a specific one.
-            except Exception as error:
+                data_buffer.put_nowait(point)
+            except queue.Full:
+                dropped_point_count = len(points) - added_back_to_buffer_count
+                dropped_point_counter.inc(dropped_point_count)
                 logging.error(
-                    'Failed to report %s data points to wavefront %s',
-                    data_format, error)
+                    'Buffer full, dropping %d %s data points.'
+                    'Consider increasing the batch size of '
+                    'your sender to increase throughput.', dropped_point_count,
+                    data_format)
+                break
+            added_back_to_buffer_count += 1
 
     def _internal_flush(self, data_buffer, data_format, entity_prefix,
-                        report_errors):
+                        report_errors, dropped_point_counter):
         """Get all data from one data buffer to a list, and report that list.
 
         @param data_buffer: Data buffer to be flush and sent
-        @type: Queue
+        @type data_buffer: Queue
         @param data_format: Type of data to be sent
-        @type: str
+        @type data_format: str
+        @param entity_prefix: Metric prefix of data type
+        @type entity_prefix: str
+        @param report_errors: Counter of errors
+        @type report_errors: WavefrontSdkCounter
+        @param dropped_point_counter: Counter of dropped points
+        @type dropped_point_counter: WavefrontSdkCounter
         """
         data = []
         size = data_buffer.qsize()
         while size > 0 and not data_buffer.empty():
             data.append(data_buffer.get())
             size -= 1
-        self._batch_report(data, data_format, entity_prefix, report_errors)
+        self._batch_report(data, data_format, entity_prefix, report_errors,
+                           data_buffer, dropped_point_counter)
 
     def _schedule_timer(self):
         # Flush every 5 secs by default
@@ -270,19 +337,21 @@ def flush_now(self):
         """Flush all the data buffer immediately."""
         self._internal_flush(self._metrics_buffer,
                              self.WAVEFRONT_METRIC_FORMAT, 'points',
-                             self._points_report_errors)
+                             self._points_report_errors, self._points_dropped)
         self._internal_flush(self._histograms_buffer,
                              self.WAVEFRONT_HISTOGRAM_FORMAT, 'histograms',
-                             self._histograms_report_errors)
+                             self._histograms_report_errors,
+                             self._histograms_dropped)
         self._internal_flush(self._tracing_spans_buffer,
                              self.WAVEFRONT_TRACING_SPAN_FORMAT, 'spans',
-                             self._spans_report_errors)
+                             self._spans_report_errors, self._spans_dropped)
         self._internal_flush(self._spans_log_buffer,
                              self.WAVEFRONT_SPAN_LOG_FORMAT, 'span_logs',
-                             self._span_logs_report_errors)
-        self._internal_flush(self._events_buffer,
-                             self.WAVEFRONT_EVENT_FORMAT, 'events',
-                             self._events_report_errors)
+                             self._span_logs_report_errors,
+                             self._span_logs_dropped)
+        self._internal_flush(self._events_buffer, self.WAVEFRONT_EVENT_FORMAT,
+                             'events', self._events_report_errors,
+                             self._events_dropped)
 
     def close(self):
         """Flush all buffer before close the client."""
@@ -320,7 +389,7 @@ def send_metric(self, name, value, timestamp, source, tags):
             self._points_invalid.inc()
             raise error
         try:
-            self._metrics_buffer.put(line_data)
+            self._metrics_buffer.put_nowait(line_data)
         except queue.Full as error:
             self._points_dropped.inc()
             raise error
@@ -335,7 +404,8 @@ def send_metric_now(self, metrics):
         @type metrics: list[str]
         """
         self._batch_report(metrics, self.WAVEFRONT_METRIC_FORMAT, 'points',
-                           self._points_report_errors)
+                           self._points_report_errors, self._metrics_buffer,
+                           self._points_dropped)
 
     def send_distribution(self, name, centroids, histogram_granularities,
                           timestamp, source, tags):
@@ -369,7 +439,7 @@ def send_distribution(self, name, centroids, histogram_granularities,
             self._histograms_invalid.inc()
             raise error
         try:
-            self._histograms_buffer.put(line_data)
+            self._histograms_buffer.put_nowait(line_data)
         except queue.Full as error:
             self._histograms_dropped.inc()
             raise error
@@ -384,7 +454,8 @@ def send_distribution_now(self, distributions):
         @type distributions: list[str]
         """
         self._batch_report(distributions, self.WAVEFRONT_HISTOGRAM_FORMAT,
-                           'histograms', self._histograms_report_errors)
+                           'histograms', self._histograms_report_errors,
+                           self._histograms_buffer, self._histograms_dropped)
 
     # pylint: disable=too-many-arguments
 
@@ -431,7 +502,7 @@ def send_span(self, name, start_millis, duration_millis, source, trace_id,
             self._spans_invalid.inc()
             raise error
         try:
-            self._tracing_spans_buffer.put(line_data)
+            self._tracing_spans_buffer.put_nowait(line_data)
         except queue.Full as error:
             self._spans_dropped.inc()
             raise error
@@ -444,7 +515,7 @@ def send_span(self, name, start_millis, duration_millis, source, trace_id,
                 self._span_logs_invalid.inc()
                 raise error
             try:
-                self._spans_log_buffer.put(line_data)
+                self._spans_log_buffer.put_nowait(line_data)
             except queue.Full as error:
                 self._span_logs_dropped.inc()
                 raise error
@@ -460,7 +531,8 @@ def send_span_now(self, spans):
         @type spans: list[str]
         """
         self._batch_report(spans, self.WAVEFRONT_TRACING_SPAN_FORMAT, 'spans',
-                           self._spans_report_errors)
+                           self._spans_report_errors,
+                           self._tracing_spans_buffer, self._spans_dropped)
 
     def send_span_log_now(self, span_logs):
         """
@@ -473,7 +545,8 @@ def send_span_log_now(self, span_logs):
         @type span_logs: list[str]
         """
         self._batch_report(span_logs, self.WAVEFRONT_SPAN_LOG_FORMAT,
-                           'span_logs', self._span_logs_report_errors)
+                           'span_logs', self._span_logs_report_errors,
+                           self._spans_log_buffer, self._span_logs_dropped)
 
     def send_event(self, name, start_time, end_time, source, tags,
                    annotations):
@@ -510,7 +583,7 @@ def send_event(self, name, start_time, end_time, source, tags,
             self._events_invalid.inc()
             raise error
         try:
-            self._events_buffer.put(line_data)
+            self._events_buffer.put_nowait(line_data)
         except queue.Full as error:
             self._events_dropped.inc()
             raise error
@@ -525,7 +598,8 @@ def send_event_now(self, events):
         @type events: list[str]
         """
         self._batch_report(events, self.WAVEFRONT_EVENT_FORMAT, 'events',
-                           self._events_report_errors)
+                           self._events_report_errors, self._events_buffer,
+                           self._events_dropped)
 
     def get_failure_count(self):
         """Get failure count for one connection."""