Merge pull request #71 from DataDog/benjamin/update-sampler

Update sampler

Author: LotharSee

ddtrace/compat.py

@@ -21,12 +21,9 @@
 
 
 try:
-    import ujson as json
+    import simplejson as json
 except ImportError:
-    try:
-        import simplejson as json
-    except ImportError:
-        import json
+    import json
 
 def iteritems(obj, **kwargs):
     func = getattr(obj, "iteritems", None)

ddtrace/sampler.py

@@ -2,15 +2,17 @@
 
 Any `sampled = False` trace won't be written, and can be ignored by the instrumentation.
 """
-
 import logging
 import array
 import threading
 
-from .span import MAX_TRACE_ID
-
 log = logging.getLogger(__name__)
 
+MAX_TRACE_ID = 2 ** 64
+
+# Has to be the same factor and key as the Agent to allow chained sampling
+KNUTH_FACTOR = 1111111111111111111
+SAMPLE_RATE_METRIC_KEY = "_sample_rate"
 
 class AllSampler(object):
     """Sampler sampling all the traces"""
@@ -42,17 +44,17 @@ def set_sample_rate(self, sample_rate):
         self.sampling_id_threshold = sample_rate * MAX_TRACE_ID
 
     def sample(self, span):
-        span.sampled = span.trace_id <= self.sampling_id_threshold
-        # `weight` is an attribute applied to all spans to help scaling related statistics
-        span.weight = 1 / (self.sample_rate or 1)
-
+        span.sampled = ((span.trace_id * KNUTH_FACTOR) % MAX_TRACE_ID) <= self.sampling_id_threshold
+        span.set_metric(SAMPLE_RATE_METRIC_KEY, self.sample_rate)
 
 class ThroughputSampler(object):
     """ Sampler applying a strict limit over the trace volume.
 
         Stop tracing once reached more than `tps` traces per second.
         Computation is based on a circular buffer over the last
         `BUFFER_DURATION` with a `BUFFER_SIZE` size.
+
+        DEPRECATED: Outdated implementation.
     """
 
     # Reasonable values

ddtrace/span.py

@@ -30,7 +30,6 @@ class Span(object):
         'duration',
         # Sampler attributes
         'sampled',
-        'weight',
         # Internal attributes
         '_tracer',
         '_finished',
@@ -89,7 +88,6 @@ def __init__(
 
         # sampling
         self.sampled = True
-        self.weight = 1
 
         self._tracer = tracer
         self._parent = None
@@ -185,7 +183,6 @@ def to_dict(self):
             'resource' : self.resource,
             'name' : self.name,
             'error': self.error,
-            'weight': self.weight,
         }
 
         if self.start:
@@ -277,8 +274,6 @@ def __repr__(self):
             self.name,
         )
 
-MAX_TRACE_ID = 2 ** 63
 def _new_id():
-    """Generate a random trace_id"""
-    return random.getrandbits(63)
-
+    """Generate a random trace_id or span_id"""
+    return random.getrandbits(64)

docs/index.rst

@@ -96,6 +96,24 @@ API
 
 .. _integrations:
 
+
+Sampling
+--------
+
+It is possible to sample traces with `ddtrace`.
+While the Trace Agent already samples traces to reduce the bandwidth usage, this client sampling
+reduces performance overhead.
+
+`RateSampler` samples a ratio of the traces. Its usage is simple::
+
+    from ddtrace.sampler import RateSampler
+
+    # Sample rate is between 0 (nothing sampled) to 1 (everything sampled).
+    # Sample 50% of the traces.
+    sample_rate = 0.5
+    tracer.sampler = RateSampler(sample_rate)
+
+
 Integrations
 ------------
 

tests/test_sampler.py

@@ -6,47 +6,39 @@
 import threading
 
 from ddtrace.tracer import Tracer
-from ddtrace.sampler import RateSampler, ThroughputSampler
+from ddtrace.sampler import RateSampler, ThroughputSampler, SAMPLE_RATE_METRIC_KEY
 from .test_tracer import DummyWriter
 from .util import patch_time
 
 
 class RateSamplerTest(unittest.TestCase):
 
-    def test_random_sequence(self):
+    def test_sample_rate_deviation(self):
         writer = DummyWriter()
-        tracer = Tracer()
-        tracer.writer = writer
-        tracer.sampler = RateSampler(0.5)
-
-        # Set the seed so that the choice of sampled traces
-        # is deterministic, then write tests accordingly
-        random.seed(4012)
-
-        # First trace, sampled
-        with tracer.trace("foo") as s:
-            assert s.sampled
-            assert s.weight == 2
-        assert writer.pop()
-
-        # Second trace, not sampled
-        with tracer.trace("figh") as s:
-            assert not s.sampled
-            s2 = tracer.trace("what")
-            assert not s2.sampled
-            s2.finish()
-            with tracer.trace("ever") as s3:
-                assert not s3.sampled
-                s4 = tracer.trace("!")
-                assert not s4.sampled
-                s4.finish()
-        spans = writer.pop()
-        assert not spans, spans
-
-        # Third trace, not sampled
-        with tracer.trace("ters") as s:
-            assert s.sampled
-        assert writer.pop()
+
+        for sample_rate in [0.1, 0.25, 0.5, 1]:
+            tracer = Tracer()
+            tracer.writer = writer
+
+            sample_rate = 0.5
+            tracer.sampler = RateSampler(sample_rate)
+
+            random.seed(1234)
+
+            iterations = int(2e4)
+
+            for i in range(iterations):
+                span = tracer.trace(i)
+                span.finish()
+
+            samples = writer.pop()
+
+            # We must have at least 1 sample, check that it has its sample rate properly assigned
+            assert samples[0].get_metric(SAMPLE_RATE_METRIC_KEY) == 0.5
+
+            # Less than 1% deviation when "enough" iterations (arbitrary, just check if it converges)
+            deviation = abs(len(samples) - (iterations * sample_rate)) / (iterations * sample_rate)
+            assert deviation < 0.01, "Deviation too high %f with sample_rate %f" % (deviation, sample_rate)
 
 
 class ThroughputSamplerTest(unittest.TestCase):