Reduce memory usage of the HTTP transport

lib/ddtrace/span.rb

@@ -1,3 +1,5 @@
+# frozen_string_literal: true
+
 require 'time'
 require 'thread'
 
@@ -262,7 +264,7 @@ def allocations
 
     # Return the hash representation of the current span.
     def to_hash
-      {
+      h = {
         span_id: @span_id,
         parent_id: @parent_id,
         trace_id: @trace_id,
@@ -274,12 +276,72 @@ def to_hash
         metrics: @metrics,
         allocations: allocations,
         error: @status
-      }.tap do |h|
-        if finished?
-          h[:start] = (start_time.to_f * 1e9).to_i
-          h[:duration] = (duration.to_f * 1e9).to_i
-        end
+      }
+
+      if finished?
+        h[:start] = start_time_nano
+        h[:duration] = duration_nano
+      end
+
+      h
+    end
+
+    # MessagePack serializer interface. Making this object
+    # respond to `#to_msgpack` allows it to be automatically
+    # serialized by MessagePack.
+    #
+    # This is more efficient than doing +MessagePack.pack(span.to_hash)+
+    # as we don't have to create an intermediate Hash.
+    #
+    # @param packer [MessagePack::Packer] serialization buffer, can be +nil+ with JRuby
+    def to_msgpack(packer = nil)
+      # As of 1.3.3, JRuby implementation doesn't pass an existing packer
+      packer ||= MessagePack::Packer.new
+
+      if finished?
+        packer.write_map_header(13) # Set header with how many elements in the map
+
+        packer.write('start')
+        packer.write(start_time_nano)
+
+        packer.write('duration')
+        packer.write(duration_nano)
+      else
+        packer.write_map_header(11) # Set header with how many elements in the map
       end
+
+      # DEV: We use strings as keys here, instead of symbols, as
+      # DEV: MessagePack will ultimately convert them to strings.
+      # DEV: By providing strings directly, we skip this indirection operation.
+      packer.write('span_id')
+      packer.write(@span_id)
+      packer.write('parent_id')
+      packer.write(@parent_id)
+      packer.write('trace_id')
+      packer.write(@trace_id)
+      packer.write('name')
+      packer.write(@name)
+      packer.write('service')
+      packer.write(@service)
+      packer.write('resource')
+      packer.write(@resource)
+      packer.write('type')
+      packer.write(@span_type)
+      packer.write('meta')
+      packer.write(@meta)
+      packer.write('metrics')
+      packer.write(@metrics)
+      packer.write('allocations')
+      packer.write(allocations)
+      packer.write('error')
+      packer.write(@status)
+      packer
+    end
+
+    # JSON serializer interface.
+    # Used by older version of the transport.
+    def to_json(*args)
+      to_hash.to_json(*args)
     end
 
     # Return a human readable version of the span
@@ -327,9 +389,9 @@ def finished?
 
     def duration
       if @duration_end.nil? || @duration_start.nil?
-        (@end_time - @start_time).to_f rescue 0.0
+        @end_time - @start_time
       else
-        (@duration_end - @duration_start).to_f rescue 0.0
+        @duration_end - @duration_start
       end
     end
 
@@ -352,5 +414,17 @@ def now_allocations
         GC.stat(:total_allocated_objects)
       end
     end
+
+    # Used for serialization
+    # @return [Integer] in nanoseconds since Epoch
+    def start_time_nano
+      @start_time.to_i * 1000000000 + @start_time.nsec
+    end
+
+    # Used for serialization
+    # @return [Integer] in nanoseconds since Epoch
+    def duration_nano
+      (duration * 1e9).to_i
+    end
   end
 end

lib/ddtrace/transport/http/adapters/net.rb

@@ -20,8 +20,14 @@ def initialize(hostname, port, options = {})
           end
 
           def open
-            # Open connection
-            ::Net::HTTP.start(hostname, port, open_timeout: timeout, read_timeout: timeout) do |http|
+            # DEV Initializing +Net::HTTP+ directly help us avoid expensive
+            # options processing done in +Net::HTTP.start+:
+            # https://github.com/ruby/ruby/blob/b2d96abb42abbe2e01f010ffc9ac51f0f9a50002/lib/net/http.rb#L614-L618
+            req = ::Net::HTTP.new(hostname, port, nil)
+
+            req.open_timeout = req.read_timeout = timeout
+
+            req.start do |http|
               yield(http)
             end
           end

lib/ddtrace/transport/http/statistics.rb

@@ -18,11 +18,24 @@ def metrics_for_response(response)
               # Add status code tag to api.responses metric
               if metrics.key?(:api_responses)
                 (metrics[:api_responses].options[:tags] ||= []).tap do |tags|
-                  tags << "status_code:#{response.code}"
+                  tags << metrics_tag_value(response.code)
                 end
               end
             end
           end
+
+          private
+
+          # The most common status code on a healthy tracer
+          STATUS_CODE_200 = 'status_code:200'.freeze
+
+          def metrics_tag_value(status_code)
+            if status_code == 200
+              STATUS_CODE_200 # DEV Saves string concatenation/creation for common case
+            else
+              "status_code:#{status_code}"
+            end
+          end
         end
       end
     end

lib/ddtrace/transport/traces.rb

@@ -57,7 +57,12 @@ def initialize(encoder, max_size: DEFAULT_MAX_PAYLOAD_SIZE)
         # @return [Enumerable[Array[Bytes,Integer]]] list of encoded chunks: each containing a byte array and
         #   number of traces
         def encode_in_chunks(traces)
-          encoded_traces = traces.map { |t| encode_one(t) }.reject(&:nil?)
+          encoded_traces = if traces.respond_to?(:filter_map)
+                             # DEV Supported since Ruby 2.7, saves an intermediate object creation
+                             traces.filter_map { |t| encode_one(t) }
+                           else
+                             traces.map { |t| encode_one(t) }.reject(&:nil?)
+                           end
 
           Datadog::Chunker.chunk_by_size(encoded_traces, max_size).map do |chunk|
             [encoder.join(chunk), chunk.size]
@@ -86,7 +91,7 @@ module Encoder
         module_function
 
         def encode_trace(encoder, trace)
-          encoder.encode(trace.map(&:to_hash))
+          encoder.encode(trace)
         end
       end
 

spec/ddtrace/benchmark/microbenchmark_spec.rb

@@ -1,153 +1,8 @@
 require 'spec_helper'
 
-require 'datadog/statsd'
-require 'ddtrace'
-
-require 'benchmark/ips'
-if !PlatformHelpers.jruby? && Gem::Version.new(RUBY_VERSION) >= Gem::Version.new('2.1.0')
-  require 'benchmark/memory'
-  require 'memory_profiler'
-end
-
-require 'fileutils'
-require 'json'
+require_relative 'support/benchmark_helper'
 
 RSpec.describe 'Microbenchmark' do
-  shared_context 'benchmark' do
-    # When applicable, runs the test subject for different input sizes.
-    # Similar to how N in Big O notation works.
-    #
-    # This value is provided to the `subject(i)` method in order for the test
-    # to appropriately execute its run based on input size.
-    let(:steps) { defined?(super) ? super() : [1, 10, 100] }
-
-    # How many times we run our program when testing for memory allocation.
-    # In theory, we should only need to run it once, as memory tests are not
-    # dependent on competing system resources.
-    # But occasionally we do see a few blimps of inconsistency, making the benchmarks skewed.
-    # By running the benchmarked snippet many times, we drown out any one-off anomalies, allowing
-    # the real memory culprits to surface.
-    let(:memory_iterations) { defined?(super) ? super() : 100 }
-
-    # Outputs human readable information to STDERR.
-    # Most of the benchmarks have nicely formatted reports
-    # that are by default printed to terminal.
-    before do |e|
-      @test = e.metadata[:example_group][:full_description]
-      @type = e.description
-
-      STDERR.puts "Test:#{e.metadata[:example_group][:full_description]} #{e.description}"
-
-      # Warm up
-      steps.each do |s|
-        subject(s)
-      end
-    end
-
-    # Report JSON result objects to ./tmp/benchmark/ folder
-    # Theses results can be historically tracked (e.g. plotting) if needed.
-    def write_result(result)
-      STDERR.puts(@test, @type, result)
-
-      path = File.join('tmp', 'benchmark', @test, @type)
-      FileUtils.mkdir_p(File.dirname(path))
-
-      File.write(path, JSON.pretty_generate(result))
-
-      STDERR.puts("Result written to #{path}")
-    end
-
-    # Measure execution time
-    it 'timing' do
-      report = Benchmark.ips do |x|
-        x.config(time: 1, warmup: 0.1)
-
-        steps.each do |s|
-          x.report(s) do
-            subject(s)
-          end
-        end
-
-        x.compare!
-      end
-
-      result = report.entries.each_with_object({}) do |entry, hash|
-        hash[entry.label] = { ips: entry.stats.central_tendency, error: entry.stats.error_percentage / 100 }
-      end.to_h
-
-      write_result(result)
-    end
-
-    # Measure memory usage (object creation and memory size)
-    it 'memory' do
-      if PlatformHelpers.jruby? || Gem::Version.new(RUBY_VERSION) < Gem::Version.new('2.1.0')
-        skip("'benchmark/memory' not supported")
-      end
-
-      report = Benchmark.memory do |x|
-        steps.each do |s|
-          x.report(s) do
-            memory_iterations.times { subject(s) }
-          end
-        end
-
-        x.compare!
-      end
-
-      result = report.entries.map do |entry|
-        row = entry.measurement.map do |metric|
-          { type: metric.type, allocated: metric.allocated, retained: metric.retained }
-        end
-
-        [entry.label, row]
-      end.to_h
-
-      write_result(result)
-    end
-
-    # Measure GC cycles triggered during run
-    it 'gc' do
-      skip if PlatformHelpers.jruby?
-
-      io = StringIO.new
-      GC::Profiler.enable
-
-      memory_iterations.times { subject(steps[0]) }
-
-      GC.disable # Prevent data collection from influencing results
-
-      data = GC::Profiler.raw_data
-      GC::Profiler.report(io)
-      GC::Profiler.disable
-
-      GC.enable
-
-      puts io.string
-
-      result = { count: data.size, time: data.map { |d| d[:GC_TIME] }.inject(0, &:+) }
-      write_result(result)
-    end
-
-    # Reports that generate non-aggregated data.
-    # Useful for debugging.
-    context 'detailed report' do
-      before { skip('Detailed report are too verbose for CI') if ENV.key?('CI') }
-
-      # Memory report with reference to each allocation site
-      it 'memory report' do
-        if PlatformHelpers.jruby? || Gem::Version.new(RUBY_VERSION) < Gem::Version.new('2.1.0')
-          skip("'benchmark/memory' not supported")
-        end
-
-        report = MemoryProfiler.report do
-          memory_iterations.times { subject(steps[0]) }
-        end
-
-        report.pretty_print
-      end
-    end
-  end
-
   # Empty benchmark to assess the overhead benchmarking tools
   # and stability of underlying hardware.
   describe 'baseline' do