Implemented span compression algorithm (#1321)

* Implemented span compression algorithm

See elastic/apm#432

* disable span compression by default in tests

* make _try_to_compress_composite and _try_to_compress_regular side effect free

* fix issue with compression buffer not being reported

* fix some type hinting issues/errors

* move traceparent initialization into Transaction constructor

This cleans up a wart where `traceparent` is initialized to
None, and needs to be set by the callee right after, as there
is code that assumes `traceparent` to be set.

* fix breakdown metrics calculation

* fix breakdown metrics test

we no longer only call `child_ended` for breakdown metrics

* update duration of the composite span based on compressed spans

* add docs for compressed span config options

* clean up merge

Co-authored-by: Colton Myers <[email protected]>

docs/configuration.asciidoc

@@ -691,6 +691,49 @@ To disable stack trace collection for spans completely, set the value to `0`.
 Except for the special values `-1` and `0`,
 this setting has to be provided in *<<config-format-duration, duration format>>*.
 
+[float]
+[[config-span-compression-exact-match-max_duration]]
+==== `span_compression_exact_match_max_duration`
+
+<<dynamic-configuration, image:./images/dynamic-config.svg[] >>
+
+[options="header"]
+|============
+| Environment                            | Django/Flask               | Default
+| `ELASTIC_APM_SPAN_COMPRESSION_EXACT_MATCH_MAX_DURATION` | `SPAN_COMPRESSION_EXACT_MATCH_MAX_DURATION` | `"50ms"`
+|============
+
+Consecutive spans that are exact match and that are under this threshold will be compressed into a single composite span.
+This reduces the collection, processing, and storage overhead, and removes clutter from the UI.
+The tradeoff is that the DB statements of all the compressed spans will not be collected.
+
+Two spans are considered exact matches if the following attributes are identical:
+ * span name
+ * span type
+ * span subtype
+ * destination resource (e.g. the Database name)
+
+[float]
+[[config-span-compression-same-kind-max-duration]]
+==== `span_compression_same_kind_max_duration`
+
+<<dynamic-configuration, image:./images/dynamic-config.svg[] >>
+
+[options="header"]
+|============
+| Environment                            | Django/Flask               | Default
+| `ELASTIC_APM_SPAN_COMPRESSION_SAME_KIND_MAX_DURATION` | `SPAN_COMPRESSION_SAME_KIND_MAX_DURATION` | `"5ms"`
+|============
+
+Consecutive spans to the same destination that are under this threshold will be compressed into a single composite span.
+This reduces the collection, processing, and storage overhead, and removes clutter from the UI.
+The tradeoff is that metadata such as database statements of all the compressed spans will not be collected.
+
+Two spans are considered to be of the same kind if the following attributes are identical:
+ * span type
+ * span subtype
+ * destination resource (e.g. the Database name)
+
 [float]
 [[config-api-request-size]]
 ==== `api_request_size`

elasticapm/conf/__init__.py

@@ -579,6 +579,18 @@ class Config(_ConfigBase):
         ],
         type=int,
     )
+    span_compression_exact_match_max_duration = _ConfigValue(
+        "span_compression_exact_match_max_duration",
+        default=5,
+        validators=[duration_validator],
+        type=int,
+    )
+    span_compression_same_kind_max_duration = _ConfigValue(
+        "span_compression_exact_match_max_duration",
+        default=5,
+        validators=[duration_validator],
+        type=int,
+    )
     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/instrumentation/packages/dbapi2.py

@@ -237,6 +237,7 @@ def _trace_sql(self, method, sql, params, action=QUERY_ACTION):
             span_action=action,
             extra={"db": {"type": "sql", "statement": sql_string}, "destination": self._self_destination_info},
             skip_frames=1,
+            leaf=True,
         ) as span:
             if params is None:
                 result = method(sql)

elasticapm/instrumentation/packages/httpcore.py

@@ -96,6 +96,8 @@ def call(self, module, method, wrapped, instance, args, kwargs):
                     span_id=parent_id, trace_options=TracingOptions(recorded=True)
                 )
                 self._set_disttracing_headers(headers, trace_parent, transaction)
+                if leaf_span:
+                    leaf_span.dist_tracing_propagated = True
             response = wrapped(*args, **kwargs)
             if len(response) > 4:
                 # httpcore < 0.11.0

elasticapm/instrumentation/packages/httplib2.py

@@ -93,6 +93,8 @@ def call(self, module, method, wrapped, instance, args, kwargs):
                 span_id=parent_id, trace_options=TracingOptions(recorded=True)
             )
             self._set_disttracing_headers(params["headers"], trace_parent, transaction)
+            if leaf_span:
+                leaf_span.dist_tracing_propagated = True
 
             response, content = wrapped(*args, **kwargs)
             if span.context:

elasticapm/instrumentation/packages/urllib.py

@@ -96,6 +96,8 @@ def call(self, module, method, wrapped, instance, args, kwargs):
                 span_id=parent_id, trace_options=TracingOptions(recorded=True)
             )
             self._set_disttracing_headers(request_object, trace_parent, transaction)
+            if leaf_span:
+                leaf_span.dist_tracing_propagated = True
             response = wrapped(*args, **kwargs)
             if response:
                 status = getattr(response, "status", None) or response.getcode()  # Python 2 compat

elasticapm/instrumentation/packages/urllib3.py

@@ -127,6 +127,8 @@ def call(self, module, method, wrapped, instance, args, kwargs):
                 span_id=parent_id, trace_options=TracingOptions(recorded=True)
             )
             args, kwargs = update_headers(args, kwargs, instance, transaction, trace_parent)
+            if leaf_span:
+                leaf_span.dist_tracing_propagated = True
             response = wrapped(*args, **kwargs)
             if response:
                 if span.context: