implement dropping of fast exit spans

docs/configuration.asciidoc

@@ -734,6 +734,24 @@ Two spans are considered to be of the same kind if the following attributes are
  * span subtype
  * destination resource (e.g. the Database name)
 
+[float]
+[[config-exit-span-min-duration]]
+==== `exit_span_min_duration`
+
+<<dynamic-configuration, image:./images/dynamic-config.svg[] >>
+
+[options="header"]
+|============
+| Environment                            | Django/Flask               | Default
+| `ELASTIC_APM_EXIT_SPAN_MIN_DURATION` | `EXIT_SPAN_MIN_DURATION` | `"1ms"`
+|============
+
+Exit spans are spans that represent a call to an external service, like a database.
+If such calls are very short, they are usually not relevant and can be ignored.
+
+NOTE: if a span propagates distributed tracing IDs, it will not be ignored, even if it is shorter than the configured threshold.
+This is to ensure that no broken traces are recorded.
+
 [float]
 [[config-api-request-size]]
 ==== `api_request_size`
@@ -1222,6 +1240,7 @@ The unit is provided as a suffix directly after the number–without any separat
 
 *Supported units*
 
+ * `us` (microseconds)
  * `ms` (milliseconds)
  * `s` (seconds)
  * `m` (minutes)

elasticapm/conf/__init__.py

@@ -290,7 +290,9 @@ def __call__(self, value, field_name):
         return rounded
 
 
-duration_validator = UnitValidator(r"^((?:-)?\d+)(ms|s|m)$", r"\d+(ms|s|m)", {"ms": 1, "s": 1000, "m": 60000})
+duration_validator = UnitValidator(
+    r"^((?:-)?\d+)(us|ms|s|m)$", r"\d+(us|ms|s|m)", {"us": 0.001, "ms": 1, "s": 1000, "m": 60000}
+)
 size_validator = UnitValidator(
     r"^(\d+)(b|kb|mb|gb)$", r"\d+(b|KB|MB|GB)", {"b": 1, "kb": 1024, "mb": 1024 * 1024, "gb": 1024 * 1024 * 1024}
 )
@@ -590,6 +592,12 @@ class Config(_ConfigBase):
         validators=[duration_validator],
         type=int,
     )
+    exit_span_min_duration = _ConfigValue(
+        "exit_span_min_duration",
+        default=1,
+        validators=[duration_validator],
+        type=float,
+    )
     collect_local_variables = _ConfigValue("COLLECT_LOCAL_VARIABLES", default="errors")
     source_lines_error_app_frames = _ConfigValue("SOURCE_LINES_ERROR_APP_FRAMES", type=int, default=5)
     source_lines_error_library_frames = _ConfigValue("SOURCE_LINES_ERROR_LIBRARY_FRAMES", type=int, default=5)

elasticapm/traces.py

@@ -232,7 +232,6 @@ def end(self, skip_frames: int = 0, duration: Optional[float] = None):
                 self._breakdown.timer("span.self_time", reset_on_collect=True, unit="us", **labels).update(
                     int(val[0] * 1000000), val[1]
                 )
-            labels = {"transaction.name": self.name, "transaction.type": self.transaction_type}
             if self.is_sampled:
                 self._breakdown.timer(
                     "span.self_time",
@@ -413,6 +412,16 @@ def is_sampled(self, is_sampled):
     def tracer(self) -> "Tracer":
         return self._tracer
 
+    def track_dropped_span(self, span: SpanType):
+        with self._span_timers_lock:
+            try:
+                resource = span.context["destination"]["service"]["resource"]
+                stats = self._dropped_span_statistics[(resource, span.outcome)]
+                stats["count"] += 1
+                stats["duration.sum.us"] += span.duration
+            except KeyError:
+                pass
+
 
 class Span(BaseSpan):
     __slots__ = (
@@ -586,6 +595,10 @@ def is_exact_match(self, other_span: SpanType) -> bool:
     def is_compression_eligible(self) -> bool:
         return self.leaf and not self.dist_tracing_propagated and self.outcome in (None, constants.OUTCOME.SUCCESS)
 
+    @property
+    def discardable(self) -> bool:
+        return self.leaf and not self.dist_tracing_propagated and self.outcome == constants.OUTCOME.SUCCESS
+
     def end(self, skip_frames: int = 0, duration: Optional[float] = None):
         """
         End this span and queue it for sending.
@@ -611,7 +624,11 @@ def end(self, skip_frames: int = 0, duration: Optional[float] = None):
         p.child_ended(self)
 
     def report(self) -> None:
-        self.tracer.queue_func(SPAN, self.to_dict())
+        if self.discardable and self.duration < self.tracer.config.exit_span_min_duration:
+            self.transaction.track_dropped_span(self)
+            self.transaction.dropped_spans += 1
+        else:
+            self.tracer.queue_func(SPAN, self.to_dict())
 
     def try_to_compress(self, sibling: SpanType) -> bool:
         compression_strategy = (
@@ -836,7 +853,7 @@ def __init__(
         span_subtype: Optional[str] = None,
         span_action: Optional[str] = None,
         start: Optional[int] = None,
-        duration: Optional[int] = None,
+        duration: Optional[float] = None,
         sync: Optional[bool] = None,
     ):
         self.name = name
@@ -901,17 +918,11 @@ def handle_exit(
             try:
                 outcome = "failure" if exc_val else "success"
                 span = transaction.end_span(self.skip_frames, duration=self.duration, outcome=outcome)
-                should_send = (
+                should_track_dropped = (
                     transaction.tracer._agent.check_server_version(gte=(7, 16)) if transaction.tracer._agent else True
                 )
-                if should_send and isinstance(span, DroppedSpan) and span.context:
-                    try:
-                        resource = span.context["destination"]["service"]["resource"]
-                        stats = transaction._dropped_span_statistics[(resource, span.outcome)]
-                        stats["count"] += 1
-                        stats["duration.sum.us"] += span.duration
-                    except KeyError:
-                        pass
+                if should_track_dropped and isinstance(span, DroppedSpan) and span.context:
+                    transaction.track_dropped_span(span)
                 if exc_val and not isinstance(span, DroppedSpan):
                     try:
                         exc_val._elastic_apm_span_id = span.id

tests/client/dropped_spans_tests.py

@@ -119,3 +119,26 @@ def test_transaction_max_span_dropped_statistics_not_collected_for_incompatible_
     spans = elasticapm_client.events[constants.SPAN]
     assert len(spans) == 1
     assert "dropped_spans_stats" not in transaction
+
+
+@pytest.mark.parametrize("elasticapm_client", [{"exit_span_min_duration": "1ms"}], indirect=True)
+def test_transaction_fast_exit_span(elasticapm_client):
+    elasticapm_client.begin_transaction("test_type")
+    with elasticapm.capture_span(span_type="x", name="x", leaf=True, duration=2):  # not dropped, too long
+        pass
+    with elasticapm.capture_span(span_type="y", name="y", leaf=True, duration=0.1):  # dropped
+        pass
+    with elasticapm.capture_span(span_type="z", name="z", leaf=False, duration=0.1):  # not dropped, not exit
+        pass
+    elasticapm_client.end_transaction("foo", duration=2.2)
+    transaction = elasticapm_client.events[constants.TRANSACTION][0]
+    spans = elasticapm_client.events[constants.SPAN]
+    breakdown = elasticapm_client._metrics.get_metricset("elasticapm.metrics.sets.breakdown.BreakdownMetricSet")
+    metrics = list(breakdown.collect())
+    assert len(spans) == 2
+    assert transaction["span_count"]["started"] == 3
+    assert transaction["span_count"]["dropped"] == 1
+    assert metrics[0]["span"]["type"] == "x"
+    assert metrics[0]["samples"]["span.self_time.sum.us"]["value"] == 2000000
+    assert metrics[1]["span"]["type"] == "y"
+    assert metrics[1]["samples"]["span.self_time.sum.us"]["value"] == 100000

tests/config/tests.py

@@ -227,12 +227,14 @@ class MyConfig(_ConfigBase):
 
 def test_duration_validation():
     class MyConfig(_ConfigBase):
+        microsecond = _ConfigValue("US", type=float, validators=[duration_validator])
         millisecond = _ConfigValue("MS", type=int, validators=[duration_validator])
         second = _ConfigValue("S", type=int, validators=[duration_validator])
         minute = _ConfigValue("M", type=int, validators=[duration_validator])
         wrong_pattern = _ConfigValue("WRONG_PATTERN", type=int, validators=[duration_validator])
 
-    c = MyConfig({"MS": "-10ms", "S": "5s", "M": "17m", "WRONG_PATTERN": "5 ms"})
+    c = MyConfig({"US": "10us", "MS": "-10ms", "S": "5s", "M": "17m", "WRONG_PATTERN": "5 ms"})
+    assert c.microsecond == 0.01
     assert c.millisecond == -10
     assert c.second == 5 * 1000
     assert c.minute == 17 * 1000 * 60

tests/contrib/django/fixtures.py

@@ -88,6 +88,7 @@ def django_sending_elasticapm_client(request, validating_httpserver):
     client_config.setdefault("span_frames_min_duration", -1)
     client_config.setdefault("span_compression_exact_match_max_duration", "0ms")
     client_config.setdefault("span_compression_same_kind_max_duration", "0ms")
+    client_config.setdefault("exit_span_min_duration", "0ms")
     app = apps.get_app_config("elasticapm")
     old_client = app.client
     client = DjangoClient(**client_config)

tests/fixtures.py

@@ -191,6 +191,7 @@ def elasticapm_client(request):
     client_config.setdefault("cloud_provider", False)
     client_config.setdefault("span_compression_exact_match_max_duration", "0ms")
     client_config.setdefault("span_compression_same_kind_max_duration", "0ms")
+    client_config.setdefault("exit_span_min_duration", "0ms")
     client = TempStoreClient(**client_config)
     yield client
     client.close()

tests/instrumentation/transactions_store_tests.py

@@ -45,7 +45,7 @@
 
 
 @pytest.fixture()
-def tracer():
+def tracer(elasticapm_client):
     frames = [
         {
             "function": "something_expensive",
@@ -177,7 +177,7 @@ def tracer():
     def queue(event_type, event, flush=False):
         events[event_type].append(event)
 
-    store = Tracer(lambda: frames, lambda frames: frames, queue, Config(), None)
+    store = Tracer(lambda: frames, lambda frames: frames, queue, elasticapm_client.config, elasticapm_client)
     store.events = events
     return store