Merge pull request #303 from bmermet/filter-traces

Add a processing pipeline to AsyncWorker

Author: Emanuele Palazzetti

ddtrace/constants.py

@@ -0,0 +1 @@
+FILTERS_KEY = 'FILTERS'

ddtrace/filters.py

@@ -0,0 +1,51 @@
+import re
+
+from .ext import http
+
+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
+    representing the urls to be excluded from tracing. A trace will be excluded
+    if its root span contains a http.url tag and if this tag matches any of
+    the provided regular expression using the standard python regexp match
+    semantic (https://docs.python.org/2/library/re.html#re.match).
+
+    :param list regexps: the  list of regular expressions (as strings) defining
+    the urls that should be filtered out. (a single string is also accepted)
+
+    Examples:
+
+    To filter out http calls to domain api.example.com::
+
+        FilterRequestsOnUrl(r'http://api\.example\.com')
+
+    To filter out http calls to all first level subdomains from example.com::
+
+        FilterRequestOnUrl(r'http://.*+\.example\.com')
+
+    To filter out calls to both http://test.example.com and http://example.com/healthcheck::
+
+        FilterRequestOnUrl([r'http://test\.example\.com', r'http://example\.com/healthcheck'])
+
+
+    """
+    def __init__(self, regexps):
+        if isinstance(regexps, str):
+            regexps = [regexps]
+        self._regexps = [re.compile(regexp) for regexp in regexps]
+
+    def process_trace(self, trace):
+        """
+        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:
+                url = span.get_tag(http.URL)
+                for regexp in self._regexps:
+                    if regexp.match(url):
+                        return None
+        return trace

ddtrace/tracer.py

@@ -7,6 +7,7 @@
 from .sampler import AllSampler
 from .writer import AgentWriter
 from .span import Span
+from .constants import FILTERS_KEY
 from . import compat
 from os import getpid
 
@@ -70,7 +71,7 @@ async def web_handler(request):
         return self._context_provider(*args, **kwargs)
 
     def configure(self, enabled=None, hostname=None, port=None, sampler=None,
-                context_provider=None, wrap_executor=None):
+                context_provider=None, wrap_executor=None, settings=None):
         """
         Configure an existing Tracer the easy way.
         Allow to configure or reconfigure a Tracer instance.
@@ -90,8 +91,16 @@ def configure(self, enabled=None, hostname=None, port=None, sampler=None,
         if enabled is not None:
             self.enabled = enabled
 
-        if hostname is not None or port is not None:
-            self.writer = AgentWriter(hostname or self.DEFAULT_HOSTNAME, port or self.DEFAULT_PORT)
+        filters = None
+        if settings is not None:
+                filters = settings.get(FILTERS_KEY)
+
+        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,
+                filters=filters
+            )
 
         if sampler is not None:
             self.sampler = sampler

ddtrace/writer.py

@@ -22,11 +22,12 @@
 
 class AgentWriter(object):
 
-    def __init__(self, hostname='localhost', port=8126):
+    def __init__(self, hostname='localhost', port=8126, filters=None):
         self._pid = None
         self._traces = None
         self._services = None
         self._worker = None
+        self._filters = filters
         self.api = api.API(hostname, port)
 
     def write(self, spans=None, services=None):
@@ -52,17 +53,23 @@ 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)
+            self._worker = AsyncWorker(
+                self.api,
+                self._traces,
+                self._services,
+                filters=self._filters,
+            )
 
 
 class AsyncWorker(object):
 
-    def __init__(self, api, trace_queue, service_queue, shutdown_timeout=DEFAULT_TIMEOUT):
+    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._filters = filters
         self._last_error_ts = 0
         self.api = api
         self.start()
@@ -119,6 +126,13 @@ def _target(self):
 
         while True:
             traces = self._trace_queue.pop()
+            if traces:
+                # Before sending the traces, make them go through the
+                # filters
+                try:
+                    traces = self._apply_filters(traces)
+                except Exception as err:
+                    log.error("error while filtering traces:{0}".format(err))
             if traces:
                 # If we have data, let's try to send it.
                 try:
@@ -133,7 +147,7 @@ def _target(self):
                 except Exception as err:
                     log.error("cannot send services: {0}".format(err))
 
-            elif self._trace_queue.closed():
+            if self._trace_queue.closed() and self._trace_queue.size() == 0:
                 # no traces and the queue is closed. our work is done
                 return
 
@@ -155,6 +169,24 @@ def _log_error_status(self, result, result_name):
                       getattr(result, "status", None), getattr(result, "reason", None),
                       getattr(result, "msg", None))
 
+    def _apply_filters(self, traces):
+        """
+        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._filters is not None:
+            filtered_traces = []
+            for trace in traces:
+                for filtr in self._filters:
+                    trace = filtr.process_trace(trace)
+                    if trace is None:
+                        break
+                if trace is not None:
+                    filtered_traces.append(trace)
+            return filtered_traces
+        return traces
+
 
 class Q(object):
     """

docs/index.rst

@@ -312,6 +312,51 @@ Users can pass along the parent_trace_id and parent_span_id via whatever method
 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 filters list. For instance, to filter out
+all traces of incoming requests to a specific url::
+
+    Tracer.configure(settings={
+        'FILTERS': [
+            FilterRequestsOnUrl(r'http://test\.example\.com'),
+        ],
+    })
+
+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.
+
+**Use the standard filters**
+
+The library comes with a FilterRequestsOnUrl filter that can be used to
+filter out incoming requests to specific urls:
+
+.. autoclass:: ddtrace.filters.FilterRequestsOnUrl
+    :members:
+
+**Write a custom filter**
+
+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 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
+    filters = [FilterExample()]
+    Tracer.configure(settings={'FILTERS': filters})
+
+(see filters.py for other example implementations)
+
+
 API
 ~~~
 

tests/test_filters.py

@@ -0,0 +1,34 @@
+from unittest import TestCase
+
+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')
+        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')
+        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')
+        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')
+        filtr = FilterRequestsOnUrl(['http://domain\.example\.com', 'http://anotherdomain\.example\.com'])
+        trace = filtr.process_trace([span])
+        self.assertIsNotNone(trace)

tests/test_integration.py

@@ -9,6 +9,9 @@
 from nose.tools import eq_, ok_
 
 from ddtrace.api import API
+from ddtrace.ext import http
+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
@@ -200,6 +203,27 @@ def test_worker_http_error_logging(self):
         ok_('failed_to_send traces to Agent: HTTP error status 400, reason Bad Request, message Content-Type:'
             in logged_errors[0])
 
+    def test_worker_filter_request(self):
+        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)
+
+        span = self.tracer.trace('testing.filteredurl')
+        span.set_tag(http.URL, 'http://example.com/health')
+        span.finish()
+        span = self.tracer.trace('testing.nonfilteredurl')
+        span.set_tag(http.URL, 'http://example.com/api/resource')
+        span.finish()
+        self._wait_thread_flush()
+
+        # Only the second trace should have been sent
+        eq_(self.api._put.call_count, 1)
+        # check and retrieve the right call
+        endpoint, payload = self._get_endpoint_payload(self.api._put.call_args_list, '/v0.3/traces')
+        eq_(endpoint, '/v0.3/traces')
+        eq_(len(payload), 1)
+        eq_(payload[0][0]['name'], 'testing.nonfilteredurl')
 
 @skipUnless(
     os.environ.get('TEST_DATADOG_INTEGRATION', False),

tests/test_writer.py

@@ -0,0 +1,89 @@
+from unittest import TestCase
+
+from ddtrace.span import Span
+from ddtrace.writer import AsyncWorker, Q
+
+class RemoveAllFilter():
+    def __init__(self):
+        self.filtered_traces = 0
+
+    def process_trace(self, trace):
+        self.filtered_traces += 1
+        return None
+
+class KeepAllFilter():
+    def __init__(self):
+        self.filtered_traces = 0
+
+    def process_trace(self, trace):
+        self.filtered_traces += 1
+        return trace
+
+class AddTagFilter():
+    def __init__(self, tag_name):
+        self.tag_name = tag_name
+        self.filtered_traces = 0
+
+    def process_trace(self, trace):
+        self.filtered_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)
+
+N_TRACES = 11
+
+class AsyncWorkerTests(TestCase):
+    def setUp(self):
+        self.api = DummmyAPI()
+        self.traces = Q()
+        self.services = Q()
+        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_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(filtr.filtered_traces, N_TRACES)
+
+    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(filtr.filtered_traces, N_TRACES)
+
+    def test_filters_add_tag(self):
+        tag_name = "Tag"
+        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(filtr.filtered_traces, N_TRACES)
+        for trace in self.api.traces:
+            for span in trace:
+                self.assertIsNotNone(span.get_tag(tag_name))
+
+    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(filtr.filtered_traces, 0)