added transaction_max_spans setting to limit the amount of spans th…

…at are recorded per transaction (#127)

closes #127

CHANGELOG.asciidoc

@@ -19,6 +19,7 @@ https://github.com/elastic/apm-agent-python/compare/v1.0.0\...master[Check the H
  * added `transaction.id` to errors to better correlate errors with transactions ({pull}122[#122])
  * added `transaction_sample_rate` to define a rate with which transactions are sampled ({pull}116[#116])
  * added `error.handled` to indicate if an exception was handled or not ({pull}124[#124]).
+ * added `transaction_max_spans` setting to limit the amount of spans that are recorded per transaction ({pull}127[#127])
  * BREAKING: Several settings and APIs have been renamed ({pull}111[#111], {pull}119[#119]):
  ** The decorator for custom instrumentation, `elasticapm.trace`, is now `elasticapm.capture_span`
  ** The setting `traces_send_frequency` has been renamed to `transaction_send_frequency`.

docs/configuration.asciidoc

@@ -282,11 +282,11 @@ If you notice a performance impact when running Elastic APM, changing
 this setting to `errors` can help.
 
 [float]
-[[config-traces-send-frequency]]
+[[config-transaction-send-frequency]]
 ==== `transaction_send_frequency`
 
 |============
-| Environment                    | Django/Flask        | Default
+| Environment                         | Django/Flask            | Default
 | `ELASTIC_APM_TRANSACTION_SEND_FREQ` | `TRANSACTION_SEND_FREQ` | `60`
 |============
 
@@ -296,6 +296,20 @@ while a higher value can increase the memory pressure of your app.
 A higher value also impacts the time until transactions are indexed and searchable in Elasticsearch.
 
 
+[float]
+[[config-transaction-max-spans]]
+==== `transaction_max_spans`
+
+|============
+| Environment                         | Django/Flask            | Default
+| `ELASTIC_APM_TRANSACTION_MAX_SPANS` | `TRANSACTION_MAX_SPANS` | `500`
+|============
+
+Limits the amount of spans that are recorded per transaction.
+This is helpful in cases where a transaction creates a very high amount of spans (e.g. thousands of SQL queries).
+Setting an upper limit will prevent overloading the agent and the APM server with too much work for such edge cases.
+
+
 [float]
 [[config-max-event-queue-length]]
 ==== `max_event_queue_length`

elasticapm/base.py

@@ -135,6 +135,7 @@ def __init__(self, config=None, **defaults):
             ),
             collect_frequency=self.config.transaction_send_frequency,
             sample_rate=self.config.transaction_sample_rate,
+            max_spans=self.config.transaction_max_spans,
             max_queue_length=self.config.max_event_queue_length,
             ignore_patterns=self.config.transactions_ignore_patterns,
         )

elasticapm/conf/__init__.py

@@ -158,6 +158,7 @@ class Config(_ConfigBase):
     ])
     transaction_send_frequency = _ConfigValue('TRACES_SEND_FREQ', type=int, default=60)
     transaction_sample_rate = _ConfigValue('TRANSACTION_SAMPLE_RATE', type=float, default=1.0)
+    transaction_max_spans = _ConfigValue('TRANSACTION_MAX_SPANS', type=int, default=500)
     max_event_queue_length = _ConfigValue('MAX_EVENT_QUEUE_LENGTH', type=int, default=500)
     collect_local_variables = _ConfigValue('COLLECT_LOCAL_VARIABLES', default='errors')
     collect_source = _ConfigValue('COLLECT_SOURCE', default='all')

elasticapm/traces.py

@@ -23,6 +23,10 @@
 TAG_RE = re.compile('^[^.*\"]+$')
 
 
+DROPPED_SPAN = object()
+IGNORED_SPAN = object()
+
+
 def get_transaction(clear=False):
     """
     Get the transaction registered for the current thread.
@@ -37,7 +41,7 @@ def get_transaction(clear=False):
 
 
 class Transaction(object):
-    def __init__(self, frames_collector_func, transaction_type="custom", is_sampled=True):
+    def __init__(self, frames_collector_func, transaction_type="custom", is_sampled=True, max_spans=None):
         self.id = str(uuid.uuid4())
         self.timestamp = datetime.datetime.utcnow()
         self.start_time = _time_func()
@@ -49,6 +53,8 @@ def __init__(self, frames_collector_func, transaction_type="custom", is_sampled=
 
         self.spans = []
         self.span_stack = []
+        self.max_spans = max_spans
+        self.dropped_spans = 0
         self.ignore_subtree = False
         self._context = {}
         self._tags = {}
@@ -63,25 +69,34 @@ def begin_span(self, name, span_type, context=None, leaf=False):
         # If we were already called with `leaf=True`, we'll just push
         # a placeholder on the stack.
         if self.ignore_subtree:
-            self.span_stack.append(None)
+            self.span_stack.append(IGNORED_SPAN)
             return None
 
         if leaf:
             self.ignore_subtree = True
 
-        start = _time_func() - self.start_time
-        span = Span(self._span_counter, name, span_type, start, context)
         self._span_counter += 1
+
+        if self.max_spans and self._span_counter > self.max_spans:
+            self.dropped_spans += 1
+            self.span_stack.append(DROPPED_SPAN)
+            return None
+
+        start = _time_func() - self.start_time
+        span = Span(self._span_counter - 1, name, span_type, start, context)
         self.span_stack.append(span)
         return span
 
     def end_span(self, skip_frames):
         span = self.span_stack.pop()
-        if span is None:
+        if span is IGNORED_SPAN:
             return None
 
         self.ignore_subtree = False
 
+        if span is DROPPED_SPAN:
+            return
+
         span.duration = _time_func() - span.start_time - self.start_time
 
         if self.span_stack:
@@ -106,6 +121,8 @@ def to_dict(self):
         }
         if self.is_sampled:
             result['spans'] = [span_obj.to_dict() for span_obj in self.spans]
+        if self.dropped_spans:
+            result['span_count'] = {'dropped': {'total': self.dropped_spans}}
         return result
 
 
@@ -151,11 +168,12 @@ def to_dict(self):
 
 
 class TransactionsStore(object):
-    def __init__(self, frames_collector_func, collect_frequency, sample_rate=1.0, max_queue_length=None,
+    def __init__(self, frames_collector_func, collect_frequency, sample_rate=1.0, max_spans=0, max_queue_length=None,
                  ignore_patterns=None):
         self.cond = threading.Condition()
         self.collect_frequency = collect_frequency
         self.max_queue_length = max_queue_length
+        self.max_spans = max_spans
         self._frames_collector_func = frames_collector_func
         self._transactions = []
         self._last_collect = _time_func()
@@ -191,7 +209,8 @@ def begin_transaction(self, transaction_type):
         :returns the Transaction object
         """
         is_sampled = self._sample_rate == 1.0 or self._sample_rate > random.random()
-        transaction = Transaction(self._frames_collector_func, transaction_type, is_sampled=is_sampled)
+        transaction = Transaction(self._frames_collector_func, transaction_type, max_spans=self.max_spans,
+                                  is_sampled=is_sampled)
         thread_local.transaction = transaction
         return transaction
 
@@ -236,8 +255,7 @@ def decorated(*args, **kwds):
     def __enter__(self):
         transaction = get_transaction()
         if transaction and transaction.is_sampled:
-            transaction.begin_span(self.name, self.type, self.extra,
-                                   self.leaf)
+            transaction.begin_span(self.name, self.type, context=self.extra, leaf=self.leaf)
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         transaction = get_transaction()

tests/client/client_tests.py

@@ -570,3 +570,56 @@ def test_transaction_sampling(should_collect, elasticapm_client, not_so_random):
     assert len([t for t in transactions if t['sampled']]) == 5
     for transaction in transactions:
         assert transaction['sampled'] or not 'spans' in transaction
+
+
+@pytest.mark.parametrize('elasticapm_client', [{'transaction_max_spans': 5}], indirect=True)
+@mock.patch('elasticapm.base.TransactionsStore.should_collect')
+def test_transaction_max_spans(should_collect, elasticapm_client):
+    should_collect.return_value = False
+    elasticapm_client.begin_transaction('test_type')
+    for i in range(5):
+        with elasticapm.capture_span('nodrop'):
+            pass
+    for i in range(10):
+        with elasticapm.capture_span('drop'):
+            pass
+    transaction_obj = elasticapm_client.end_transaction('test')
+
+    transaction = elasticapm_client.instrumentation_store.get_all()[0]
+
+    assert transaction_obj.max_spans == 5
+    assert transaction_obj.dropped_spans == 10
+    assert len(transaction['spans']) == 5
+    for span in transaction['spans']:
+        assert span['name'] == 'nodrop'
+    assert transaction['span_count'] == {'dropped': {'total': 10}}
+
+
+@pytest.mark.parametrize('elasticapm_client', [{'transaction_max_spans': 3}], indirect=True)
+@mock.patch('elasticapm.base.TransactionsStore.should_collect')
+def test_transaction_max_span_nested(should_collect, elasticapm_client):
+    should_collect.return_value = False
+    elasticapm_client.begin_transaction('test_type')
+    with elasticapm.capture_span('1'):
+        with elasticapm.capture_span('2'):
+            with elasticapm.capture_span('3'):
+                with elasticapm.capture_span('4'):
+                    with elasticapm.capture_span('5'):
+                        pass
+                with elasticapm.capture_span('6'):
+                    pass
+            with elasticapm.capture_span('7'):
+                pass
+        with elasticapm.capture_span('8'):
+            pass
+    with elasticapm.capture_span('9'):
+        pass
+    transaction_obj = elasticapm_client.end_transaction('test')
+
+    transaction = elasticapm_client.instrumentation_store.get_all()[0]
+
+    assert transaction_obj.dropped_spans == 6
+    assert len(transaction['spans']) == 3
+    for span in transaction['spans']:
+        assert span['name'] in ('1', '2', '3')
+    assert transaction['span_count'] == {'dropped': {'total': 6}}