Adressing Manu's comments

Author: bmermet

ddtrace/constants.py

@@ -0,0 +1 @@
+PROCESSING_PIPELINE_KEY = 'PROCESSING_PIPELINE'

ddtrace/processors.py

@@ -2,7 +2,7 @@
 
 from .ext import http
 
-class FilterRequestsOnUrl():
+class FilterRequestsOnUrl(object):
     """Filter out traces from incoming http requests based on the request's url
 
     This class takes as argument a list of regular expression patterns

ddtrace/settings.py

@@ -7,7 +7,7 @@
 from .sampler import AllSampler
 from .writer import AgentWriter
 from .span import Span
-from .settings import PP_KEY
+from .constants import PROCESSING_PIPELINE_KEY as PP_KEY
 from . import compat
 from os import getpid
 
@@ -92,15 +92,15 @@ def configure(self, enabled=None, hostname=None, port=None, sampler=None,
             self.enabled = enabled
 
         processing_pipeline = None
-        if settings is not None and PP_KEY in settings:
-                processing_pipeline = settings[PP_KEY]
+        if settings is not None:
+                processing_pipeline = settings.get([PP_KEY])
 
         if hostname is not None or port is not None or processing_pipeline is not None:
             self.writer = AgentWriter(
-                        hostname or self.DEFAULT_HOSTNAME,
-                        port or self.DEFAULT_PORT,
-                        processing_pipeline=processing_pipeline
-                    )
+                hostname or self.DEFAULT_HOSTNAME,
+                port or self.DEFAULT_PORT,
+                processing_pipeline=processing_pipeline
+            )
 
         if sampler is not None:
             self.sampler = sampler

ddtrace/tracer.py

@@ -54,10 +54,11 @@ def _reset_worker(self):
         # ensure we have an active thread working on this queue
         if not self._worker or not self._worker.is_alive():
             self._worker = AsyncWorker(
-                        self.api,
-                        self._traces,
-                        self._services, processing_pipeline=self._processing_pipeline
-                    )
+                self.api,
+                self._traces,
+                self._services,
+                processing_pipeline=self._processing_pipeline,
+            )
 
 
 class AsyncWorker(object):
@@ -169,6 +170,11 @@ def _log_error_status(self, result, result_name):
                       getattr(result, "msg", None))
 
     def _apply_processing_pipeline(self, traces):
+        """
+        Here we make each trace go through the processing pipeline configured
+        in the tracer. There is no need for a lock since the traces are owned
+        by the AsyncWorker at that point.
+        """
         if self._processing_pipeline is not None:
             processed_traces = []
             for trace in traces:

ddtrace/writer.py

@@ -312,23 +312,43 @@ Advanced Usage
 Trace Filtering
 ~~~~~~~~~~~~~~~
 
-It is possible to filter or modify traces before they are sent to the agent by configuring the tracer with a processing pipeline. For instance to filter out all traces of incoming requests to a specific url::
+It is possible to filter or modify traces before they are sent to the agent by
+configuring the tracer with a processing pipeline. For instance to filter out
+all traces of incoming requests to a specific url::
 
     processing_pipeline = [FilterRequestsOnUrl(r'http://test\.example\.com')]
     Tracer.configure(settings={'PROCESSING_PIPELINE': processing_pipeline})
 
-All the processors in the processing pipeline will be evaluated sequentially for each trace and the resulting trace will either be sent to the agent or discarded depending on the output of the pipeline.
+All the processors in the processing pipeline will be evaluated sequentially
+for each trace and the resulting trace will either be sent to the agent or
+discarded depending on the output of the pipeline.
 
 **Use the standard processors**
 
-The library comes with a FilterRequestsOnUrl processor that can be used to filter out incoming requests to specific urls:
+The library comes with a FilterRequestsOnUrl processor that can be used to
+filter out incoming requests to specific urls:
 
 .. autoclass:: ddtrace.processors.FilterRequestsOnUrl
     :members:
 
 **Write a custom processor**
 
-Creating your own processors is as simple as implementing a class with a process_trace method and adding it to the processing pipeline parameter of Tracer.configure. process_trace should either return a trace to be fed to the next step of the pipeline or None if the trace should be discarded. (see processors.py for example implementations)
+Creating your own processors is as simple as implementing a class with a
+process_trace method and adding it to the processing pipeline parameter of
+Tracer.configure. process_trace should either return a trace to be fed to the
+next step of the pipeline or None if the trace should be discarded::
+
+    class ProcessorExample(object):
+        def process_trace(self, trace):
+            # write here your logic to return the `trace` or None;
+            # `trace` instance is owned by the thread and you can alter
+            # each single span or the whole trace if needed
+
+    # And then instantiate it with
+    processing_pipeline = [ProcessorExample()]
+    Tracer.configure(settings={'PROCESSING_PIPELINE': processing_pipeline})
+
+(see processors.py for other example implementations)
 
 
 API

docs/index.rst

@@ -4,24 +4,36 @@
 from ddtrace.writer import AsyncWorker, Q
 
 class RemoveAllProcessor():
+    def __init__(self):
+        self.processed_traces = 0
+
     def process_trace(self, trace):
+        self.processed_traces += 1
         return None
 
 class KeepAllProcessor():
+    def __init__(self):
+        self.processed_traces = 0
+
     def process_trace(self, trace):
+        self.processed_traces += 1
         return trace
 
 class AddTagProcessor():
     def __init__(self, tag_name):
         self.tag_name = tag_name
+        self.processed_traces = 0
+
     def process_trace(self, trace):
+        self.processed_traces += 1
         for span in trace:
             span.set_tag(self.tag_name, "A value")
         return trace
 
 class DummmyAPI():
     def __init__(self):
         self.traces = []
+
     def send_traces(self, traces):
         for trace in traces:
             self.traces.append(trace)
@@ -37,26 +49,41 @@ def setUp(self):
             self.traces.add([Span(tracer=None, name="name", trace_id=i, span_id=j, parent_id=j-1 or None) for j in range(7)])
 
     def test_processing_pipeline_keep_all(self):
-        processing_pipeline = [KeepAllProcessor()]
+        processor = KeepAllProcessor()
+        processing_pipeline = [processor]
         worker = AsyncWorker(self.api, self.traces, self.services, processing_pipeline=processing_pipeline)
         worker.stop()
         worker.join()
         self.assertEqual(len(self.api.traces), N_TRACES)
+        self.assertEqual(processor.processed_traces, N_TRACES)
 
     def test_processing_pipeline_remove_all(self):
-        processing_pipeline = [RemoveAllProcessor()]
+        processor = RemoveAllProcessor()
+        processing_pipeline = [processor]
         worker = AsyncWorker(self.api, self.traces, self.services, processing_pipeline=processing_pipeline)
         worker.stop()
         worker.join()
         self.assertEqual(len(self.api.traces), 0)
+        self.assertEqual(processor.processed_traces, N_TRACES)
 
     def test_processing_pipeline_add_tag(self):
         tag_name = "Tag"
-        processing_pipeline = [AddTagProcessor(tag_name)]
+        processor = AddTagProcessor(tag_name)
+        processing_pipeline = [processor]
         worker = AsyncWorker(self.api, self.traces, self.services, processing_pipeline=processing_pipeline)
         worker.stop()
         worker.join()
         self.assertEqual(len(self.api.traces), N_TRACES)
+        self.assertEqual(processor.processed_traces, N_TRACES)
         for trace in self.api.traces:
             for span in trace:
                 self.assertIsNotNone(span.get_tag(tag_name))
+
+    def test_processing_pipeline_short_circuit(self):
+        processor = KeepAllProcessor()
+        processing_pipeline = [RemoveAllProcessor(), processor]
+        worker = AsyncWorker(self.api, self.traces, self.services, processing_pipeline=processing_pipeline)
+        worker.stop()
+        worker.join()
+        self.assertEqual(len(self.api.traces), 0)
+        self.assertEqual(processor.processed_traces, 0)