Infinite Tracing: batching and compression

CHANGELOG.md

@@ -6,6 +6,14 @@
 
   Following the 3.2.0 release of Ruby, the New Relic Ruby Agent has confirmed compatibility with and now supports the official release of Ruby 3.2.0. [PR#1715](https://github.com/newrelic/newrelic-ruby-agent/pull/1715)
 
+- **Infinite Tracing: Use batching and compression**
+
+  For [Infinite Tracing](https://docs.newrelic.com/docs/distributed-tracing/infinite-tracing/introduction-infinite-tracing/) which Ruby applications can leverage with the `newrelic-infinite_tracing` gem, payloads will now be batched and compressed to signficantly decrease the amount of outbound network traffic. [PR#1723](https://github.com/newrelic/newrelic-ruby-agent/pull/1723)
+
+  | Configuration name                | Default | Behavior                                                                                                                      |
+  | --------------------------------- | ------- | ----------------------------------------------------------------------------------------------------------------------------- |
+  | `infinite_tracing.batching` | true    | If true (the default), data sent to the Trace Observer will be batched instead of each span being sent individually |
+  | `infinite_tracing.compression_level` | high    | Configure the compression level for data sent to the Trace Observer. May be one of [none|low|medium|high]. 'high' is the default. Set the level to 'none' to disable compression. |
 
 ## v8.14.0
 

infinite_tracing/lib/infinite_tracing/channel.rb

@@ -5,8 +5,8 @@
 module NewRelic::Agent
   module InfiniteTracing
     class Channel
-      COMPRESSION_LEVELS = %i[none low medium high].freeze
-      DEFAULT_COMPRESSION_LEVEL = :none
+      SETTINGS_BASE = {'grpc.enable_deadline_checking' => 0}.freeze
+      SETTINGS_COMPRESSION_DISABLED = SETTINGS_BASE.merge({'grpc.minimal_stack' => 1}).freeze
 
       def stub
         NewRelic::Agent.logger.debug("Infinite Tracer Opening Channel to #{host_and_port}")
@@ -19,64 +19,29 @@ def stub
         )
       end
 
-      def channel
-        GRPC::Core::Channel.new(host_and_port, settings, credentials)
-      end
-
-      def channel_args
-        return NewRelic::EMPTY_HASH unless compression_enabled?
-
-        GRPC::Core::CompressionOptions.new(compression_options).to_channel_arg_hash
-      end
-
-      def compression_enabled?
-        compression_level != :none
-      end
-
-      def compression_level
-        @compression_level ||= begin
-          level = if valid_compression_level?(configured_compression_level)
-            configured_compression_level
-          else
-            DEFAULT_COMPRESSION_LEVEL
-          end
-          NewRelic::Agent.logger.debug("Infinite Tracer compression level set to #{level}")
-          level
-        end
-      end
-
-      def compression_options
-        {default_algorithm: :gzip,
-         default_level: compression_level}
-      end
-
-      def configured_compression_level
-        NewRelic::Agent.config[:'infinite_tracing.compression_level']
+      def host_and_port
+        Config.trace_observer_host_and_port
       end
 
       def credentials
-        # Uses system configured certificates by default
-        GRPC::Core::ChannelCredentials.new
+        @credentials ||= GRPC::Core::ChannelCredentials.new
       end
 
-      def host_and_port
-        Config.trace_observer_host_and_port
+      def channel
+        GRPC::Core::Channel.new(host_and_port, settings, credentials)
       end
 
       def settings
-        {
-          'grpc.minimal_stack' => 1,
-          'grpc.enable_deadline_checking' => 0
-        }
-      end
+        return channel_args.merge(SETTINGS_COMPRESSION_DISABLED).freeze unless Config.compression_enabled?
 
-      def valid_compression_level?(level)
-        return true if COMPRESSION_LEVELS.include?(level)
+        channel_args.merge(SETTINGS_BASE).freeze
+      end
 
-        NewRelic::Agent.logger.error("Invalid compression level '#{level}' specified! Must be one of " \
-          "#{COMPRESSION_LEVELS.join('|')}. Using default level of '#{DEFAULT_COMPRESSION_LEVEL}'")
+      def channel_args
+        return NewRelic::EMPTY_HASH unless Config.compression_enabled?
 
-        false
+        GRPC::Core::CompressionOptions.new(default_algorithm: :gzip,
+          default_level: Config.compression_level).to_channel_arg_hash
       end
     end
   end

infinite_tracing/lib/infinite_tracing/config.rb

@@ -7,6 +7,9 @@ module InfiniteTracing
     module Config
       extend self
 
+      COMPRESSION_LEVEL_DISABLED = :none
+      COMPRESSION_LEVEL_DEFAULT = :high
+      COMPRESSION_LEVEL_LIST = %i[none low medium high].freeze
       TRACE_OBSERVER_NOT_CONFIGURED_ERROR = "Trace Observer host not configured!"
 
       # We only want to load the infinite tracing gem's files when
@@ -109,6 +112,29 @@ def test_framework?
       def trace_observer_configured?
         trace_observer_host != NewRelic::EMPTY_STR
       end
+
+      def compression_enabled?
+        compression_level != COMPRESSION_LEVEL_DISABLED
+      end
+
+      def compression_level
+        @compression_level ||= begin
+          level = if COMPRESSION_LEVEL_LIST.include?(configured_compression_level)
+            configured_compression_level
+          else
+            NewRelic::Agent.logger.error("Invalid compression level '#{configured_compression_level}' specified! " \
+                                         "Must be one of #{COMPRESSION_LEVEL_LIST.join('|')}. Using default level " \
+                                         "of '#{COMPRESSION_LEVEL_DEFAULT}'")
+            COMPRESSION_LEVEL_DEFAULT
+          end
+          NewRelic::Agent.logger.debug("Infinite Tracer compression level set to #{level}")
+          level
+        end
+      end
+
+      def configured_compression_level
+        NewRelic::Agent.config[:'infinite_tracing.compression_level']
+      end
     end
   end
 end

infinite_tracing/lib/infinite_tracing/connection.rb

@@ -17,6 +17,12 @@
 module NewRelic::Agent
   module InfiniteTracing
     class Connection
+      GZIP_METADATA = {'grpc-internal-encoding-request' => 'gzip',
+                       'grpc-encoding' => 'gzip',
+                       'grpc-accept-encoding' => ['gzip'],
+                       'content-coding' => 'gzip',
+                       'content-encoding' => 'gzip'}.freeze
+
       # listens for server side configurations added to the agent.  When a new config is
       # added, we have a new agent run token and need to restart the client's RPC stream
       # with the new metadata information.
@@ -100,9 +106,18 @@ def metadata
             "agent_run_token" => agent_id
           }
           @metadata.merge!(request_headers_map)
+          merge_gzip_metadata
         end
       end
 
+      # If (gzip based) compression is enabled, explicitly provide all
+      # relevant metadata
+      def merge_gzip_metadata
+        return @metadata unless Config.compression_enabled?
+
+        @metadata.merge!(GZIP_METADATA)
+      end
+
       # Initializes rpc so we can get a Channel and Stub (connect to gRPC server)
       # Initializes metadata so we use newest values in establishing channel
       # Sets the agent_connected flag and signals the agent started so any

infinite_tracing/test/infinite_tracing/channel_test.rb

@@ -56,40 +56,43 @@ def test_channel_is_really_secure_for_remote_host
           Config.unstub(:test_framework?)
         end
 
-        def test_compression_enabled_returns_true
-          with_config(remote_config.merge('infinite_tracing.compression_level': :high)) do
-            assert_predicate Channel.new, :compression_enabled?
-          end
-        end
-
-        def test_compression_enabled_returns_false
-          with_config(remote_config.merge('infinite_tracing.compression_level': :none)) do
-            refute Channel.new.compression_enabled?
-          end
-        end
-
-        def test_invalid_compression_level
-          channel = Channel.new
-
-          refute channel.valid_compression_level?(:bogus)
-        end
-
         def test_channel_args_are_empty_if_compression_is_disabled
+          reset_compression_level
+
           with_config(remote_config.merge('infinite_tracing.compression_level': :none)) do
-            assert_equal Channel.new.channel_args, NewRelic::EMPTY_HASH
+            assert_equal NewRelic::EMPTY_HASH, Channel.new.channel_args
           end
         end
 
         def test_channel_args_includes_compression_settings_if_compression_is_enabled
+          reset_compression_level
+
           level = :low
           expected_result = {'grpc.default_compression_level' => 1,
                              'grpc.default_compression_algorithm' => 2,
                              'grpc.compression_enabled_algorithms_bitset' => 7}
 
           with_config(remote_config.merge('infinite_tracing.compression_level': level)) do
-            assert_equal Channel.new.channel_args, expected_result
+            assert_equal expected_result, Channel.new.channel_args
+          end
+        end
+
+        def test_settings_apply_minimal_stack_when_compression_is_disabled
+          reset_compression_level
+
+          with_config(remote_config.merge('infinite_tracing.compression_level': :none)) do
+            assert_equal Channel::SETTINGS_COMPRESSION_DISABLED, Channel.new.settings
           end
         end
+
+        def test_settings_do_not_apply_minimal_stack_when_compression_is_enabled
+          reset_compression_level
+          expected = Channel::SETTINGS_BASE.merge({'grpc.default_compression_level' => 3,
+                                                   'grpc.default_compression_algorithm' => 2,
+                                                   'grpc.compression_enabled_algorithms_bitset' => 7})
+
+          assert_equal expected, Channel.new.settings
+        end
       end
     end
   end

infinite_tracing/test/infinite_tracing/config_test.rb

@@ -108,6 +108,36 @@ def test_unset_trace_observer_host_raises_error
           assert_match(/not configured/, error.message)
         end
 
+        def test_compression_enabled_returns_true
+          reset_compression_level
+
+          with_config({'infinite_tracing.compression_level': :high}) do
+            assert_predicate Config, :compression_enabled?
+          end
+        end
+
+        def test_compression_enabled_returns_false
+          reset_compression_level
+
+          with_config({'infinite_tracing.compression_level': :none}) do
+            refute Config.compression_enabled?
+          end
+        end
+
+        def test_invalid_compression_level
+          reset_compression_level
+          logger = MiniTest::Mock.new
+          logger.expect :error, nil, [/Invalid compression level/]
+          logger.expect :debug, nil, [/compression level set to/]
+
+          with_config({'infinite_tracing.compression_level': :bogus}) do
+            NewRelic::Agent.stub :logger, logger do
+              assert_equal Config::COMPRESSION_LEVEL_DEFAULT, Config.compression_level
+              logger.verify
+            end
+          end
+        end
+
         private
 
         def non_test_files

infinite_tracing/test/infinite_tracing/connection_test.rb

@@ -207,6 +207,42 @@ def test_increment_retry_period
           assert_equal 300, next_retry_period
         end
 
+        def test_gzip_related_parameters_exist_in_metadata_when_compression_is_enabled
+          reset_compression_level
+
+          with_serial_lock do
+            with_config(localhost_config) do
+              connection = Connection.instance # instantiate before simulation
+              simulate_connect_to_collector(fiddlesticks_config, 0.01) do |simulator|
+                simulator.join # ensure our simulation happens!
+                metadata = connection.send(:metadata)
+
+                NewRelic::Agent::InfiniteTracing::Connection::GZIP_METADATA.each do |key, value|
+                  assert_equal value, metadata[key]
+                end
+              end
+            end
+          end
+        end
+
+        def test_gzip_related_parameters_are_absent_when_compression_is_disabled
+          reset_compression_level
+
+          with_serial_lock do
+            with_config(localhost_config.merge({'infinite_tracing.compression_level': :none})) do
+              connection = Connection.instance # instantiate before simulation
+              simulate_connect_to_collector(fiddlesticks_config, 0.01) do |simulator|
+                simulator.join # ensure our simulation happens!
+                metadata = connection.send(:metadata)
+
+                NewRelic::Agent::InfiniteTracing::Connection::GZIP_METADATA.each do |key, _value|
+                  refute metadata.key?(key)
+                end
+              end
+            end
+          end
+        end
+
         private
 
         def next_retry_period

infinite_tracing/test/test_helper.rb

@@ -93,3 +93,9 @@ def with_detailed_trace
     yield
   end
 end
+
+def reset_compression_level
+  return unless NewRelic::Agent::InfiniteTracing::Config.instance_variables.include?(:@compression_level)
+
+  NewRelic::Agent::InfiniteTracing::Config.remove_instance_variable(:@compression_level)
+end

lib/new_relic/agent/configuration/default_source.rb

@@ -2317,21 +2317,21 @@ def self.enforce_fallback(allowed_values: nil, fallback: nil)
         },
         :'infinite_tracing.batching' => {
           :default => false,
-          :public => false,
+          :public => true,
           :type => Boolean,
           :allowed_from_server => false,
           :external => :infinite_tracing,
-          :description => "If true, data sent to the Trace Observer will be batched instead of the default of each " \
+          :description => "If true (the default), data sent to the Trace Observer will be batched\ninstead of each " \
                           "span being sent individually"
         },
         :'infinite_tracing.compression_level' => {
-          :default => :none,
-          :public => false,
+          :default => :high,
+          :public => true,
           :type => Symbol,
           :allowed_from_server => false,
           :external => :infinite_tracing,
           :description => "Configure the compression level for data sent to the Trace Observer\nMay be one of " \
-                          "[none|low|medium|high]\nBy default, compression is not used (level = none)"
+          "[none|low|medium|high]\n'high' is the default. Set the level to 'none' to disable compression"
         },
         :js_agent_file => {
           :default => '',

newrelic.yml

@@ -317,12 +317,12 @@ common: &default_settings
 
   # Configure the compression level for data sent to the Trace Observer
   # May be one of [none|low|medium|high]
-  # By default, compression is not used (level = none)
-  # infinite_tracing.compression_level: none
+  # 'high' is the default. Set the level to 'none' to disable compression
+  # infinite_tracing.compression_level: high
 
-  # If true, data sent to the Trace Observer will be batched instead of
-  # the default of each span being sent individually
-  # infinite_tracing.batching: false
+  # If true (the default), data sent to the Trace Observer will be batched
+  # instead of each span being sent individually
+  # infinite_tracing.batching: true
 
   # Controls auto-instrumentation of bunny at start up.
   # May be one of [auto|prepend|chain|disabled].