Renamed processing pipeline into filters

bmermet · bmermet · commit e42b881af43c · 2017-07-19T13:31:58.000+02:00
diff --git a/ddtrace/constants.py b/ddtrace/constants.py
@@ -1 +1 @@
-PROCESSING_PIPELINE_KEY = 'PROCESSING_PIPELINE'
+FILTERS_KEY = 'FILTERS'
diff --git a/ddtrace/filters.py b/ddtrace/filters.py
@@ -37,10 +37,10 @@ def __init__(self, regexps):
 
     def process_trace(self, trace):
         """
-        process_trace is called by the processing pipeline on each trace
-        before it is sent to the agent, the returned value will be fed to the
-        next step of the pipeline. If process_trace returns None, the whole
-        trace is discarded.
+        When the filter is registered in the tracer, process_trace is called by
+        on each trace before it is sent to the agent, the returned value will
+        be fed to the next filter in the list. If process_trace returns None,
+        the whole trace is discarded.
         """
         for span in trace:
             if span.parent_id is None and span.get_tag(http.URL) is not None:
diff --git a/ddtrace/tracer.py b/ddtrace/tracer.py
@@ -7,7 +7,7 @@
 from .sampler import AllSampler
 from .writer import AgentWriter
 from .span import Span
-from .constants import PROCESSING_PIPELINE_KEY as PP_KEY
+from .constants import FILTERS_KEY
 from . import compat
 from os import getpid
 
@@ -91,15 +91,15 @@ def configure(self, enabled=None, hostname=None, port=None, sampler=None,
         if enabled is not None:
             self.enabled = enabled
 
-        processing_pipeline = None
+        filters = None
         if settings is not None:
-                processing_pipeline = settings.get(PP_KEY)
+                filters = settings.get(FILTERS_KEY)
 
-        if hostname is not None or port is not None or processing_pipeline is not None:
+        if hostname is not None or port is not None or filters is not None:
             self.writer = AgentWriter(
                 hostname or self.DEFAULT_HOSTNAME,
                 port or self.DEFAULT_PORT,
-                processing_pipeline=processing_pipeline
+                filters=filters
             )
 
         if sampler is not None:
diff --git a/ddtrace/writer.py b/ddtrace/writer.py
@@ -22,12 +22,12 @@
 
 class AgentWriter(object):
 
-    def __init__(self, hostname='localhost', port=8126, processing_pipeline=None):
+    def __init__(self, hostname='localhost', port=8126, filters=None):
         self._pid = None
         self._traces = None
         self._services = None
         self._worker = None
-        self._processing_pipeline = processing_pipeline
+        self._filters = filters
         self.api = api.API(hostname, port)
 
     def write(self, spans=None, services=None):
@@ -57,19 +57,19 @@ def _reset_worker(self):
                 self.api,
                 self._traces,
                 self._services,
-                processing_pipeline=self._processing_pipeline,
+                filters=self._filters,
             )
 
 
 class AsyncWorker(object):
 
-    def __init__(self, api, trace_queue, service_queue, shutdown_timeout=DEFAULT_TIMEOUT, processing_pipeline=None):
+    def __init__(self, api, trace_queue, service_queue, shutdown_timeout=DEFAULT_TIMEOUT, filters=None):
         self._trace_queue = trace_queue
         self._service_queue = service_queue
         self._lock = threading.Lock()
         self._thread = None
         self._shutdown_timeout = shutdown_timeout
-        self._processing_pipeline = processing_pipeline
+        self._filters = filters
         self._last_error_ts = 0
         self.api = api
         self.start()
@@ -128,11 +128,11 @@ def _target(self):
             traces = self._trace_queue.pop()
             if traces:
                 # Before sending the traces, make them go through the
-                # processing pipeline
+                # filters
                 try:
-                    traces = self._apply_processing_pipeline(traces)
+                    traces = self._apply_filters(traces)
                 except Exception as err:
-                    log.error("error while processing traces:{0}".format(err))
+                    log.error("error while filtering traces:{0}".format(err))
             if traces:
                 # If we have data, let's try to send it.
                 try:
@@ -169,22 +169,22 @@ def _log_error_status(self, result, result_name):
                       getattr(result, "status", None), getattr(result, "reason", None),
                       getattr(result, "msg", None))
 
-    def _apply_processing_pipeline(self, traces):
+    def _apply_filters(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.
+        Here we make each trace go through the filters 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 = []
+        if self._filters is not None:
+            filtered_traces = []
             for trace in traces:
-                for processor in self._processing_pipeline:
-                    trace = processor.process_trace(trace)
+                for filtr in self._filters:
+                    trace = filtr.process_trace(trace)
                     if trace is None:
                         break
                 if trace is not None:
-                    processed_traces.append(trace)
-            return processed_traces
+                    filtered_traces.append(trace)
+            return filtered_traces
         return traces
 
 
diff --git a/docs/index.rst b/docs/index.rst
@@ -313,45 +313,45 @@ 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
+configuring the tracer with a filters list. For instance, to filter out
 all traces of incoming requests to a specific url::
 
     Tracer.configure(settings={
-        'PROCESSING_PIPELINE': [
+        'FILTERS': [
             FilterRequestsOnUrl(r'http://test\.example\.com'),
         ],
     })
 
-All the processors in the processing pipeline will be evaluated sequentially
+All the filters in the filters list 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.
+discarded depending on the output.
 
-**Use the standard processors**
+**Use the standard filters**
 
-The library comes with a FilterRequestsOnUrl processor that can be used to
+The library comes with a FilterRequestsOnUrl filter that can be used to
 filter out incoming requests to specific urls:
 
-.. autoclass:: ddtrace.processors.FilterRequestsOnUrl
+.. autoclass:: ddtrace.filters.FilterRequestsOnUrl
     :members:
 
-**Write a custom processor**
+**Write a custom filter**
 
-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
+Creating your own filters is as simple as implementing a class with a
+process_trace method and adding it to the filters 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):
+    class FilterExample(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})
+    filters = [FilterExample()]
+    Tracer.configure(settings={'FILTERS': filters})
 
-(see processors.py for other example implementations)
+(see filters.py for other example implementations)
 
 
 API
diff --git a/tests/test_filters.py b/tests/test_filters.py
@@ -1,34 +1,34 @@
 from unittest import TestCase
 
-from ddtrace.processors import FilterRequestsOnUrl
+from ddtrace.filters import FilterRequestsOnUrl
 from ddtrace.span import Span
 from ddtrace.ext.http import URL
 
 class FilterRequestOnUrlTests(TestCase):
     def test_is_match(self):
         span = Span(name='Name', tracer=None)
         span.set_tag(URL, r'http://example.com')
-        processor = FilterRequestsOnUrl('http://examp.*.com')
-        trace = processor.process_trace([span])
+        filtr = FilterRequestsOnUrl('http://examp.*.com')
+        trace = filtr.process_trace([span])
         self.assertIsNone(trace)
 
     def test_is_not_match(self):
         span = Span(name='Name', tracer=None)
         span.set_tag(URL, r'http://anotherexample.com')
-        processor = FilterRequestsOnUrl('http://examp.*.com')
-        trace = processor.process_trace([span])
+        filtr = FilterRequestsOnUrl('http://examp.*.com')
+        trace = filtr.process_trace([span])
         self.assertIsNotNone(trace)
 
     def test_list_match(self):
         span = Span(name='Name', tracer=None)
         span.set_tag(URL, r'http://anotherdomain.example.com')
-        processor = FilterRequestsOnUrl(['http://domain\.example\.com', 'http://anotherdomain\.example\.com'])
-        trace = processor.process_trace([span])
+        filtr = FilterRequestsOnUrl(['http://domain\.example\.com', 'http://anotherdomain\.example\.com'])
+        trace = filtr.process_trace([span])
         self.assertIsNone(trace)
 
     def test_list_no_match(self):
         span = Span(name='Name', tracer=None)
         span.set_tag(URL, r'http://cooldomain.example.com')
-        processor = FilterRequestsOnUrl(['http://domain\.example\.com', 'http://anotherdomain\.example\.com'])
-        trace = processor.process_trace([span])
+        filtr = FilterRequestsOnUrl(['http://domain\.example\.com', 'http://anotherdomain\.example\.com'])
+        trace = filtr.process_trace([span])
         self.assertIsNotNone(trace)
diff --git a/tests/test_integration.py b/tests/test_integration.py
@@ -10,8 +10,8 @@
 
 from ddtrace.api import API
 from ddtrace.ext import http
-from ddtrace.processors import FilterRequestsOnUrl
-from ddtrace.constants import PROCESSING_PIPELINE_KEY as PP_KEY
+from ddtrace.filters import FilterRequestsOnUrl
+from ddtrace.constants import FILTERS_KEY
 from ddtrace.span import Span
 from ddtrace.tracer import Tracer
 from ddtrace.encoding import JSONEncoder, MsgpackEncoder, get_encoder
@@ -204,7 +204,7 @@ def test_worker_http_error_logging(self):
             in logged_errors[0])
 
     def test_worker_filter_request(self):
-        self.tracer.configure(settings={PP_KEY: [FilterRequestsOnUrl(r'http://example\.com/health')]})
+        self.tracer.configure(settings={FILTERS_KEY: [FilterRequestsOnUrl(r'http://example\.com/health')]})
         # spy the send() method
         self.api = self.tracer.writer.api
         self.api._put = mock.Mock(self.api._put, wraps=self.api._put)
diff --git a/tests/test_writer.py b/tests/test_writer.py
@@ -3,29 +3,29 @@
 from ddtrace.span import Span
 from ddtrace.writer import AsyncWorker, Q
 
-class RemoveAllProcessor():
+class RemoveAllFilter():
     def __init__(self):
-        self.processed_traces = 0
+        self.filtered_traces = 0
 
     def process_trace(self, trace):
-        self.processed_traces += 1
+        self.filtered_traces += 1
         return None
 
-class KeepAllProcessor():
+class KeepAllFilter():
     def __init__(self):
-        self.processed_traces = 0
+        self.filtered_traces = 0
 
     def process_trace(self, trace):
-        self.processed_traces += 1
+        self.filtered_traces += 1
         return trace
 
-class AddTagProcessor():
+class AddTagFilter():
     def __init__(self, tag_name):
         self.tag_name = tag_name
-        self.processed_traces = 0
+        self.filtered_traces = 0
 
     def process_trace(self, trace):
-        self.processed_traces += 1
+        self.filtered_traces += 1
         for span in trace:
             span.set_tag(self.tag_name, "A value")
         return trace
@@ -48,42 +48,42 @@ def setUp(self):
         for i in range(N_TRACES):
             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):
-        processor = KeepAllProcessor()
-        processing_pipeline = [processor]
-        worker = AsyncWorker(self.api, self.traces, self.services, processing_pipeline=processing_pipeline)
+    def test_filters_keep_all(self):
+        filtr = KeepAllFilter()
+        filters = [filtr]
+        worker = AsyncWorker(self.api, self.traces, self.services, filters=filters)
         worker.stop()
         worker.join()
         self.assertEqual(len(self.api.traces), N_TRACES)
-        self.assertEqual(processor.processed_traces, N_TRACES)
+        self.assertEqual(filtr.filtered_traces, N_TRACES)
 
-    def test_processing_pipeline_remove_all(self):
-        processor = RemoveAllProcessor()
-        processing_pipeline = [processor]
-        worker = AsyncWorker(self.api, self.traces, self.services, processing_pipeline=processing_pipeline)
+    def test_filters_remove_all(self):
+        filtr = RemoveAllFilter()
+        filters = [filtr]
+        worker = AsyncWorker(self.api, self.traces, self.services, filters=filters)
         worker.stop()
         worker.join()
         self.assertEqual(len(self.api.traces), 0)
-        self.assertEqual(processor.processed_traces, N_TRACES)
+        self.assertEqual(filtr.filtered_traces, N_TRACES)
 
-    def test_processing_pipeline_add_tag(self):
+    def test_filters_add_tag(self):
         tag_name = "Tag"
-        processor = AddTagProcessor(tag_name)
-        processing_pipeline = [processor]
-        worker = AsyncWorker(self.api, self.traces, self.services, processing_pipeline=processing_pipeline)
+        filtr = AddTagFilter(tag_name)
+        filters = [filtr]
+        worker = AsyncWorker(self.api, self.traces, self.services, filters=filters)
         worker.stop()
         worker.join()
         self.assertEqual(len(self.api.traces), N_TRACES)
-        self.assertEqual(processor.processed_traces, N_TRACES)
+        self.assertEqual(filtr.filtered_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)
+    def test_filters_short_circuit(self):
+        filtr = KeepAllFilter()
+        filters = [RemoveAllFilter(), filtr]
+        worker = AsyncWorker(self.api, self.traces, self.services, filters=filters)
         worker.stop()
         worker.join()
         self.assertEqual(len(self.api.traces), 0)
-        self.assertEqual(processor.processed_traces, 0)
+        self.assertEqual(filtr.filtered_traces, 0)

Original file line number	Diff line number	Diff line change
`@@ -1 +1 @@`
`1`		`-PROCESSING_PIPELINE_KEY = 'PROCESSING_PIPELINE'`
	`1`	`+FILTERS_KEY = 'FILTERS'`