NH-132825: oboe_sampler refactor and use pq for transaction name expire management

.rubocop.yml

@@ -57,3 +57,5 @@ Style/StringLiterals:
     - 'Gemfile'
 Naming/PredicateMethod:
   Enabled: false
+Style/OneClassPerFile:
+  Enabled: false

lib/solarwinds_apm/opentelemetry/otlp_processor.rb

@@ -30,7 +30,6 @@ def initialize(txn_manager)
         @txn_manager = txn_manager
         @metrics     = init_response_time_metrics
         @is_lambda   = SolarWindsAPM::Utils.determine_lambda
-        @transaction_name = nil
       end
 
       # @param [Span] span the (mutable) {Span} that just started.
@@ -52,8 +51,8 @@ def on_finishing(span)
         SolarWindsAPM.logger.debug { "[#{self.class}/#{__method__}] processor on_finishing span attributes: #{span.attributes}" }
         return if non_entry_span(span: span)
 
-        @transaction_name = calculate_transaction_names(span)
-        span.set_attribute(SW_TRANSACTION_NAME, @transaction_name)
+        transaction_name = calculate_transaction_names(span)
+        span.set_attribute(SW_TRANSACTION_NAME, transaction_name)
         @txn_manager.delete_root_context_h(span.context.hex_trace_id)
         SolarWindsAPM.logger.debug { "[#{self.class}/#{__method__}] processor on_finishing end" }
       end
@@ -108,7 +107,7 @@ def init_response_time_metrics
       def meter_attributes(span)
         meter_attrs = {
           SW_IS_ERROR => error?(span) == 1,
-          SW_TRANSACTION_NAME => @transaction_name
+          SW_TRANSACTION_NAME => span.attributes[SW_TRANSACTION_NAME]
         }
 
         is_http_span = span_http?(span)

lib/solarwinds_apm/otel_config.rb

@@ -17,10 +17,12 @@ module OTelConfig
     @@config           = {}
     @@config_map       = {}
     @@agent_enabled    = false
+    @@mutex            = ::Mutex.new
 
     RESOURCE_ATTRIBUTES = 'RESOURCE_ATTRIBUTES'
 
     def self.initialize
+      @@mutex.synchronize { return if @@agent_enabled }
       return unless defined?(::OpenTelemetry::SDK::Configurator)
 
       is_lambda = SolarWindsAPM::Utils.determine_lambda
@@ -87,7 +89,7 @@ def self.initialize
       txn_manager = TxnNameManager.new
       otlp_processor = SolarWindsAPM::OpenTelemetry::OTLPProcessor.new(txn_manager)
 
-      @@config[:metrics_processor] = otlp_processor
+      @@mutex.synchronize { @@config[:metrics_processor] = otlp_processor }
       ::OpenTelemetry.tracer_provider.add_span_processor(otlp_processor)
 
       # collector, service and headers are used for http sampler get settings
@@ -113,17 +115,17 @@ def self.initialize
         remote_parent_not_sampled: sampler
       )
 
-      @@agent_enabled = true
+      @@mutex.synchronize { @@agent_enabled = true }
 
       nil
     end
 
     def self.[](key)
-      @@config[key.to_sym]
+      @@mutex.synchronize { @@config[key.to_sym] }
     end
 
     def self.agent_enabled
-      @@agent_enabled
+      @@mutex.synchronize { @@agent_enabled }
     end
 
     def self.resolve_response_propagator

lib/solarwinds_apm/sampling/oboe_sampler.rb

@@ -51,15 +51,7 @@ def should_sample?(params)
       @logger.debug { "[#{self.class}/#{__method__}] should_sample? params: #{params.inspect}; span type is #{type}" }
 
       # For local spans, we always trust the parent
-      if type == SolarWindsAPM::SpanType::LOCAL
-        if parent_span.context.trace_flags.sampled?
-          return OTEL_SAMPLING_RESULT.new(decision: OTEL_SAMPLING_DECISION::RECORD_AND_SAMPLE,
-                                          tracestate: DEFAULT_TRACESTATE)
-        else
-          return OTEL_SAMPLING_RESULT.new(decision: OTEL_SAMPLING_DECISION::DROP,
-                                          tracestate: DEFAULT_TRACESTATE)
-        end
-      end
+      return handle_local_span(parent_span) if type == SolarWindsAPM::SpanType::LOCAL
 
       sample_state = SampleState.new(OTEL_SAMPLING_DECISION::DROP,
                                      attributes || {},
@@ -73,58 +65,90 @@ def should_sample?(params)
 
       @counters[:request_count].add(1)
 
-      # adding trigger trace attributes to sample_state attribute as part of decision
-      if sample_state.headers['X-Trace-Options']
+      # Parse and validate trace options; may return early on invalid signature
+      early_result = apply_trace_options(sample_state, parent_span)
+      return early_result if early_result
 
-        # TraceOptions.parse_trace_options return TriggerTraceOptions
-        sample_state.trace_options = ::SolarWindsAPM::TraceOptions.parse_trace_options(sample_state.headers['X-Trace-Options'], @logger)
+      # Check settings availability
+      early_result = check_settings_available(sample_state, parent_span)
+      return early_result if early_result
 
-        @logger.debug { "[#{self.class}/#{__method__}] sample_state.trace_options: #{sample_state.trace_options.inspect}" }
+      resolve_sampling_algorithm(sample_state)
 
-        if sample_state.headers['X-Trace-Options-Signature']
+      xtracestate = generate_new_tracestate(parent_span, sample_state)
+      @logger.debug { "[#{self.class}/#{__method__}] final sampling state: #{sample_state.inspect}" }
+
+      OTEL_SAMPLING_RESULT.new(decision: sample_state.decision,
+                               tracestate: xtracestate,
+                               attributes: sample_state.attributes)
+    end
 
-          # this validate_signature is the function from trace_options file
-          sample_state.trace_options.response.auth = TraceOptions.validate_signature(
-            sample_state.headers['X-Trace-Options'],
-            sample_state.headers['X-Trace-Options-Signature'],
-            sample_state.settings[:signature_key],
-            sample_state.trace_options.timestamp
-          )
+    private
 
-          # If the request has an invalid signature, drop the trace
-          if sample_state.trace_options.response.auth != Auth::OK
-            @logger.debug { "[#{self.class}/#{__method__}] signature invalid; tracing disabled (auth=#{sample_state.trace_options.response.auth})" }
+    def handle_local_span(parent_span)
+      if parent_span.context.trace_flags.sampled?
+        OTEL_SAMPLING_RESULT.new(decision: OTEL_SAMPLING_DECISION::RECORD_AND_SAMPLE,
+                                 tracestate: DEFAULT_TRACESTATE)
+      else
+        OTEL_SAMPLING_RESULT.new(decision: OTEL_SAMPLING_DECISION::DROP,
+                                 tracestate: DEFAULT_TRACESTATE)
+      end
+    end
 
-            xtracestate = generate_new_tracestate(parent_span, sample_state)
-            return OTEL_SAMPLING_RESULT.new(decision: OTEL_SAMPLING_DECISION::DROP,
-                                            tracestate: xtracestate,
-                                            attributes: sample_state.attributes)
-          end
+    # Parses X-Trace-Options, validates signature, and applies trace option
+    # attributes. Returns an early OTEL_SAMPLING_RESULT on invalid signature,
+    # or nil to continue normal processing.
+    def apply_trace_options(sample_state, parent_span)
+      return unless sample_state.headers['X-Trace-Options']
+
+      sample_state.trace_options = ::SolarWindsAPM::TraceOptions.parse_trace_options(sample_state.headers['X-Trace-Options'], @logger)
+      @logger.debug { "[#{self.class}/#{__method__}] sample_state.trace_options: #{sample_state.trace_options.inspect}" }
+
+      if sample_state.headers['X-Trace-Options-Signature']
+        sample_state.trace_options.response.auth = TraceOptions.validate_signature(
+          sample_state.headers['X-Trace-Options'],
+          sample_state.headers['X-Trace-Options-Signature'],
+          sample_state.settings[:signature_key],
+          sample_state.trace_options.timestamp
+        )
+
+        if sample_state.trace_options.response.auth != Auth::OK
+          @logger.debug { "[#{self.class}/#{__method__}] signature invalid; tracing disabled (auth=#{sample_state.trace_options.response.auth})" }
+          xtracestate = generate_new_tracestate(parent_span, sample_state)
+          return OTEL_SAMPLING_RESULT.new(decision: OTEL_SAMPLING_DECISION::DROP,
+                                          tracestate: xtracestate,
+                                          attributes: sample_state.attributes)
         end
-
-        # Apply trace options to span attributes and list ignored keys in response
-        sample_state.trace_options.response.trigger_trace = TriggerTrace::NOT_REQUESTED unless sample_state.trace_options.trigger_trace
-        sample_state.attributes[SW_KEYS_ATTRIBUTE] = sample_state.trace_options[:sw_keys] if sample_state.trace_options[:sw_keys]
-        sample_state.trace_options.custom.each { |k, v| sample_state.attributes[k] = v }
-        sample_state.trace_options.response.ignored = sample_state.trace_options[:ignored].map { |k, _| k } if sample_state.trace_options[:ignored].any?
       end
 
-      unless sample_state.settings
-        @logger.debug { "[#{self.class}/#{__method__}] settings unavailable; sampling disabled" }
+      # Apply trace options to span attributes and list ignored keys in response
+      sample_state.trace_options.response.trigger_trace = TriggerTrace::NOT_REQUESTED unless sample_state.trace_options.trigger_trace
+      sample_state.attributes[SW_KEYS_ATTRIBUTE] = sample_state.trace_options[:sw_keys] if sample_state.trace_options[:sw_keys]
+      sample_state.trace_options.custom.each { |k, v| sample_state.attributes[k] = v }
+      sample_state.trace_options.response.ignored = sample_state.trace_options[:ignored].map { |k, _| k } if sample_state.trace_options[:ignored].any?
 
-        sample_state.trace_options.response.trigger_trace = TriggerTrace::SETTINGS_NOT_AVAILABLE if sample_state.trace_options&.trigger_trace
+      nil
+    end
 
-        xtracestate = generate_new_tracestate(parent_span, sample_state)
+    # Returns an early OTEL_SAMPLING_RESULT when settings are unavailable,
+    # or nil to continue normal processing.
+    def check_settings_available(sample_state, parent_span)
+      return if sample_state.settings
 
-        return OTEL_SAMPLING_RESULT.new(decision: OTEL_SAMPLING_DECISION::DROP,
-                                        tracestate: xtracestate,
-                                        attributes: sample_state.attributes)
-      end
+      @logger.debug { "[#{self.class}/#{__method__}] settings unavailable; sampling disabled" }
+      sample_state.trace_options.response.trigger_trace = TriggerTrace::SETTINGS_NOT_AVAILABLE if sample_state.trace_options&.trigger_trace
+      xtracestate = generate_new_tracestate(parent_span, sample_state)
+
+      OTEL_SAMPLING_RESULT.new(decision: OTEL_SAMPLING_DECISION::DROP,
+                               tracestate: xtracestate,
+                               attributes: sample_state.attributes)
+    end
 
+    # Decide which sampling algo to use based on tracestate and flags
+    # https://swicloud.atlassian.net/wiki/spaces/NIT/pages/3815473156/Tracing+Decision+Tree
+    def resolve_sampling_algorithm(sample_state)
       @logger.debug { "[#{self.class}/#{__method__}] sample_state before deciding sampling algo: #{sample_state.inspect}" }
 
-      # Decide which sampling algo to use and add sampling attribute to decision attributes
-      # https://swicloud.atlassian.net/wiki/spaces/NIT/pages/3815473156/Tracing+Decision+Tree
       if sample_state.trace_state && TRACESTATE_REGEXP.match?(sample_state.trace_state)
         parent_based_algo(sample_state)
       elsif sample_state.settings[:flags].anybits?(Flags::SAMPLE_START)
@@ -136,15 +160,10 @@ def should_sample?(params)
       else
         disabled_algo(sample_state)
       end
-
-      xtracestate = generate_new_tracestate(parent_span, sample_state)
-      @logger.debug { "[#{self.class}/#{__method__}] final sampling state: #{sample_state.inspect}" }
-
-      OTEL_SAMPLING_RESULT.new(decision: sample_state.decision,
-                               tracestate: xtracestate,
-                               attributes: sample_state.attributes)
     end
 
+    public
+
     def parent_based_algo(sample_state)
       @logger.debug { "[#{self.class}/#{__method__}] parent_based_algo start" }