Add span compression

lib/elastic_apm.rb

@@ -211,6 +211,7 @@ def with_transaction(
     # @param parent [Transaction,Span] The parent transaction or span.
     #   Relevant when the span is created in another thread.
     # @param sync [Boolean] Whether the span is created synchronously or not.
+    # @param exit_span [Boolean] Whether the span is an exit span.
     # @return [Span]
     def start_span(
       name,
@@ -221,7 +222,8 @@ def start_span(
       include_stacktrace: true,
       trace_context: nil,
       parent: nil,
-      sync: nil
+      sync: nil,
+      exit_span: false
     )
       agent&.start_span(
         name,
@@ -231,7 +233,8 @@ def start_span(
         context: context,
         trace_context: trace_context,
         parent: parent,
-        sync: sync
+        sync: sync,
+        exit_span: exit_span
       ).tap do |span|
         break unless span && include_stacktrace
         break unless agent.config.span_frames_min_duration?
@@ -265,6 +268,7 @@ def end_span(span = nil)
     # @param parent [Transaction,Span] The parent transaction or span.
     #   Relevant when the span is created in another thread.
     # @param sync [Boolean] Whether the span is created synchronously or not.
+    # @param exit_span [Boolean] Whether the span is an exit span.
     # @yield [Span]
     # @return Result of block
     def with_span(
@@ -276,7 +280,8 @@ def with_span(
       include_stacktrace: true,
       trace_context: nil,
       parent: nil,
-      sync: nil
+      sync: nil,
+      exit_span: false
     )
       unless block_given?
         raise ArgumentError,
@@ -296,7 +301,8 @@ def with_span(
             include_stacktrace: include_stacktrace,
             trace_context: trace_context,
             parent: parent,
-            sync: sync
+            sync: sync,
+            exit_span: exit_span
           )
         result = yield span
         span&.outcome ||= Span::Outcome::SUCCESS

lib/elastic_apm/agent.rb

@@ -189,7 +189,8 @@ def start_span(
       context: nil,
       trace_context: nil,
       parent: nil,
-      sync: nil
+      sync: nil,
+      exit_span: nil
     )
       detect_forking!
 
@@ -206,7 +207,8 @@ def start_span(
         context: context,
         trace_context: trace_context,
         parent: parent,
-        sync: sync
+        sync: sync,
+        exit_span: exit_span
       )
     end
     # rubocop:enable Metrics/ParameterLists

lib/elastic_apm/child_durations.rb

@@ -26,11 +26,13 @@ def child_durations
         @child_durations ||= Durations.new
       end
 
-      def child_started
+      def child_started(child)
+        super(child)
         child_durations.start
       end
 
-      def child_stopped
+      def child_stopped(child)
+        super(child)
         child_durations.stop
       end
     end

lib/elastic_apm/compression_buffer.rb

@@ -0,0 +1,128 @@
+# Licensed to Elasticsearch B.V. under one or more contributor
+# license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Elasticsearch B.V. licenses this file to you under
+# the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# frozen_string_literal: true
+
+module ElasticAPM
+  # @api private
+  module CompressionBuffer
+    # @api private
+    class Composite
+      EXACT_MATCH = 'exact_match'
+      SAME_KIND = 'same_kind'
+
+      def initialize(count: 0, sum: 0, compression_strategy: nil)
+        @count = count
+        @sum = sum
+        @compression_strategy = compression_strategy
+      end
+
+      attr_accessor :count, :sum, :compression_strategy
+    end
+
+    def child_stopped(child)
+      super(child)
+
+      if !child.compression_eligible?
+        return # report
+      end
+
+      return buffer(child) if compression_buffer.nil?
+      return if compression_buffer.try_compress(child)
+
+      # child.compression_buffered = false
+      buffer(child)
+    end
+
+    attr_accessor :compression_buffer
+    attr_accessor :compression_buffered
+    alias :compression_buffered? :compression_buffered
+
+    def try_compress(other)
+      can_compress =
+        if composite?
+          try_compress_composite(other)
+        else
+          try_compress_regular(other)
+        end
+
+      return false unless can_compress
+
+      unless composite?
+        self.composite = Composite.new(count: 1, sum: duration)
+      end
+
+      composite.count += 1
+      composite.sum += other.duration
+
+      true
+    end
+
+    def try_compress_regular(other)
+      return false unless same_kind?(other)
+
+      self.composite = Composite.new(count: 1, sum: duration)
+
+      exact_match_duration_us = transaction.span_compression_exact_match_duration * 1_000_000
+
+      if name == other.name
+        if duration <= exact_match_duration_us && other.duration <= exact_match_duration_us
+          self.composite.compression_strategy = Composite::EXACT_MATCH
+          return true
+        end
+
+        return false
+      end
+
+      if duration <= exact_match_duration_us && other.duration <= exact_match_duration_us
+        self.composite.compression_strategy = Composite::SAME_KIND
+        self.name = "Calls to #{context.destination.service.resource}"
+        return true
+      end
+
+      return false
+    end
+
+    def try_compress_composite(other)
+      exact_match_duration_us = transaction.span_compression_exact_match_duration * 1_000_000
+
+      case composite.compression_strategy
+      when Composite::EXACT_MATCH
+        return same_kind?(other) && name == other.name && other.duration <= exact_match_duration_us
+      when Composite::SAME_KIND
+        return same_kind?(other) && other.duration <= exact_match_duration_us
+      else
+        # raise or log?
+      end
+    end
+
+    def same_kind?(other)
+      return false unless type == other.type
+      return false unless subtype == other.subtype
+      return false unless context.destination&.service&.resource == other.context.destination&.service&.resource
+
+      true
+    end
+
+    private
+
+    def buffer(span)
+      self.compression_buffer = span
+      # span.compression_buffered = true
+    end
+  end
+end

lib/elastic_apm/config.rb

@@ -34,69 +34,71 @@ class Config
       %w[password passwd pwd secret *key *token* *session* *credit* *card* authorization set-cookie].freeze
 
     # rubocop:disable Layout/LineLength, Layout/ExtraSpacing
-    option :config_file,                       type: :string, default: 'config/elastic_apm.yml'
-    option :server_url,                        type: :url,    default: 'http://localhost:8200'
-    option :secret_token,                      type: :string
-    option :api_key,                           type: :string
-
-    option :api_buffer_size,                   type: :int,    default: 256
-    option :api_request_size,                  type: :bytes,  default: '750kb', converter: Bytes.new
-    option :api_request_time,                  type: :float,  default: '10s',   converter: Duration.new
-    option :breakdown_metrics,                 type: :bool,   default: true
-    option :capture_body,                      type: :string, default: 'off'
-    option :capture_headers,                   type: :bool,   default: true
-    option :capture_elasticsearch_queries,     type: :bool,   default: false
-    option :capture_env,                       type: :bool,   default: true
-    option :central_config,                    type: :bool,   default: true
-    option :cloud_provider,                    type: :string, default: 'auto'
-    option :current_user_email_method,         type: :string, default: 'email'
-    option :current_user_id_method,            type: :string, default: 'id'
-    option :current_user_username_method,      type: :string, default: 'username'
-    option :custom_key_filters,                type: :list,   default: [],      converter: RegexpList.new
-    option :default_labels,                    type: :dict,   default: {}
-    option :disable_metrics,                   type: :list,   default: [],      converter: WildcardPatternList.new
-    option :disable_send,                      type: :bool,   default: false
-    option :disable_start_message,             type: :bool,   default: false
-    option :disable_instrumentations,          type: :list,   default: %w[json]
-    option :disabled_spies,                    type: :list,   default: []
-    option :enabled,                           type: :bool,   default: true
-    option :environment,                       type: :string, default: ENV['RAILS_ENV'] || ENV['RACK_ENV']
-    option :framework_name,                    type: :string
-    option :framework_version,                 type: :string
-    option :filter_exception_types,            type: :list,   default: []
-    option :global_labels,                     type: :dict
-    option :hostname,                          type: :string
-    option :http_compression,                  type: :bool,   default: true
-    option :ignore_url_patterns,               type: :list,   default: [],      converter: RegexpList.new
-    option :instrument,                        type: :bool,   default: true
-    option :instrumented_rake_tasks,           type: :list,   default: []
-    option :log_ecs_formatting,                type: :string, default: 'off'
-    option :log_level,                         type: :int,    default: Logger::INFO, converter: LogLevelMap.new
-    option :log_path,                          type: :string
-    option :metrics_interval,                  type: :int,    default: '30s',   converter: Duration.new
-    option :pool_size,                         type: :int,    default: 1
-    option :proxy_address,                     type: :string
-    option :proxy_headers,                     type: :dict
-    option :proxy_password,                    type: :string
-    option :proxy_port,                        type: :int
-    option :proxy_username,                    type: :string
-    option :recording,                         type: :bool,   default: true
-    option :sanitize_field_names,              type: :list,   default: SANITIZE_FIELD_NAMES_DEFAULT, converter: WildcardPatternList.new
-    option :server_ca_cert_file,               type: :string
-    option :service_name,                      type: :string
-    option :service_node_name,                 type: :string
-    option :service_version,                   type: :string
-    option :source_lines_error_app_frames,     type: :int,    default: 5
-    option :source_lines_error_library_frames, type: :int,    default: 0
-    option :source_lines_span_app_frames,      type: :int,    default: 5
-    option :source_lines_span_library_frames,  type: :int,    default: 0
-    option :span_frames_min_duration,          type: :float,  default: '5ms',   converter: Duration.new(default_unit: 'ms')
-    option :stack_trace_limit,                 type: :int,    default: 999_999
-    option :transaction_ignore_urls,           type: :list,   default: [],      converter: WildcardPatternList.new
-    option :transaction_max_spans,             type: :int,    default: 500
-    option :transaction_sample_rate,           type: :float,  default: 1.0,     converter: RoundFloat.new
-    option :use_elastic_traceparent_header,    type: :bool,   default: true
-    option :verify_server_cert,                type: :bool,   default: true
+    option :config_file,                           type: :string, default: 'config/elastic_apm.yml'
+    option :server_url,                            type: :url,    default: 'http://localhost:8200'
+    option :secret_token,                          type: :string
+    option :api_key,                               type: :string
+
+    option :api_buffer_size,                         type: :int,    default: 256
+    option :api_request_size,                        type: :bytes,  default: '750kb', converter: Bytes.new
+    option :api_request_time,                        type: :float,  default: '10s', converter: Duration.new
+    option :breakdown_metrics,                       type: :bool,   default: true
+    option :capture_body,                            type: :string, default: 'off'
+    option :capture_headers,                         type: :bool,   default: true
+    option :capture_elasticsearch_queries,           type: :bool,   default: false
+    option :capture_env,                             type: :bool,   default: true
+    option :central_config,                          type: :bool,   default: true
+    option :cloud_provider,                          type: :string, default: 'auto'
+    option :current_user_email_method,               type: :string, default: 'email'
+    option :current_user_id_method,                  type: :string, default: 'id'
+    option :current_user_username_method,            type: :string, default: 'username'
+    option :custom_key_filters,                      type: :list,   default: [], converter: RegexpList.new
+    option :default_labels,                          type: :dict,   default: {}
+    option :disable_metrics,                         type: :list,   default: [], converter: WildcardPatternList.new
+    option :disable_send,                            type: :bool,   default: false
+    option :disable_start_message,                   type: :bool,   default: false
+    option :disable_instrumentations,                type: :list,   default: %w[json]
+    option :disabled_spies,                          type: :list,   default: []
+    option :enabled,                                 type: :bool,   default: true
+    option :environment,                             type: :string, default: ENV['RAILS_ENV'] || ENV['RACK_ENV']
+    option :framework_name,                          type: :string
+    option :framework_version,                       type: :string
+    option :filter_exception_types,                  type: :list,   default: []
+    option :global_labels,                           type: :dict
+    option :hostname,                                type: :string
+    option :http_compression,                        type: :bool,   default: true
+    option :ignore_url_patterns,                     type: :list,   default: [], converter: RegexpList.new
+    option :instrument,                              type: :bool,   default: true
+    option :instrumented_rake_tasks,                 type: :list,   default: []
+    option :log_ecs_formatting,                      type: :string, default: 'off'
+    option :log_level,                               type: :int,    default: Logger::INFO, converter: LogLevelMap.new
+    option :log_path,                                type: :string
+    option :metrics_interval,                        type: :int,    default: '30s', converter: Duration.new
+    option :pool_size,                               type: :int,    default: 1
+    option :proxy_address,                           type: :string
+    option :proxy_headers,                           type: :dict
+    option :proxy_password,                          type: :string
+    option :proxy_port,                              type: :int
+    option :proxy_username,                          type: :string
+    option :recording,                               type: :bool,   default: true
+    option :sanitize_field_names,                    type: :list,   default: SANITIZE_FIELD_NAMES_DEFAULT, converter: WildcardPatternList.new
+    option :server_ca_cert_file,                     type: :string
+    option :service_name,                            type: :string
+    option :service_node_name,                       type: :string
+    option :service_version,                         type: :string
+    option :source_lines_error_app_frames,           type: :int,    default: 5
+    option :source_lines_error_library_frames,       type: :int,    default: 0
+    option :source_lines_span_app_frames,            type: :int,    default: 5
+    option :source_lines_span_library_frames,        type: :int,    default: 0
+    option :span_compression_exact_match_duration,   type: :float,  default: '5ms', converter: Duration.new(default_unit: 'ms')
+    option :span_compression_same_kind_max_duration, type: :float,  default: '5ms', converter: Duration.new(default_unit: 'ms')
+    option :span_frames_min_duration,                type: :float,  default: '5ms', converter: Duration.new(default_unit: 'ms')
+    option :stack_trace_limit,                       type: :int,    default: 999_999
+    option :transaction_ignore_urls,                 type: :list,   default: [], converter: WildcardPatternList.new
+    option :transaction_max_spans,                   type: :int,    default: 500
+    option :transaction_sample_rate,                 type: :float,  default: 1.0, converter: RoundFloat.new
+    option :use_elastic_traceparent_header,          type: :bool,   default: true
+    option :verify_server_cert,                      type: :bool,   default: true
 
     # rubocop:enable Layout/LineLength, Layout/ExtraSpacing
     def initialize(options = {})

lib/elastic_apm/instrumenter.rb

@@ -19,6 +19,7 @@
 
 require 'elastic_apm/trace_context'
 require 'elastic_apm/child_durations'
+require 'elastic_apm/compression_buffer'
 require 'elastic_apm/span'
 require 'elastic_apm/transaction'
 require 'elastic_apm/span_helpers'
@@ -148,9 +149,13 @@ def end_transaction(result = nil)
 
       self.current_transaction = nil
 
-      transaction.done result
+      transaction.done(result)
 
-      enqueue.call transaction
+      if child = transaction.compression_buffer
+        enqueue.call(child)
+      end
+
+      enqueue.call(transaction)
 
       update_transaction_metrics(transaction)
 
@@ -244,7 +249,11 @@ def end_span(span = nil)
 
       span.done
 
-      enqueue.call span
+      enqueue.call(span) unless span.parent.compression_buffer
+
+      if child = span.compression_buffer
+        enqueue.call(child)
+      end
 
       update_span_metrics(span)