Remove defensive coding and make it so DSM contrib libraries require libdatadog_api to be built

Author: ericfirth

Rakefile

@@ -29,6 +29,13 @@ CORE_WITH_LIBDATADOG_API = [
   'spec/datadog/core/ddsketch_spec.rb',
 ].freeze
 
+# Data Streams Monitoring (DSM) requires libdatadog_api for DDSketch
+# Add new instrumentation libraries here as they gain DSM support
+DSM_ENABLED_LIBRARIES = [
+  :kafka,
+  :karafka
+].freeze
+
 # rubocop:disable Metrics/BlockLength
 namespace :test do
   desc 'Run all tests'
@@ -295,6 +302,22 @@ namespace :spec do
     end
   end
 
+  # Ensure DSM-enabled contrib tests compile libdatadog_api before running (MRI Ruby only)
+  # If compilation fails (e.g., new Ruby version without prebuilt extension), tests will skip via DDSketch.supported?
+  unless RUBY_PLATFORM == 'java'
+    task :compile_libdatadog_for_dsm do
+      Rake::Task["compile:libdatadog_api.#{RUBY_VERSION[/\d+.\d+/]}_#{RUBY_PLATFORM}"].invoke
+    rescue => e
+      # Compilation failed (likely unsupported Ruby version) - tests will skip gracefully
+      puts "Warning: libdatadog_api compilation failed: #{e.message}"
+      puts "DSM tests will be skipped for this Ruby version"
+    end
+
+    DSM_ENABLED_LIBRARIES.each do |task_name|
+      Rake::Task["spec:#{task_name}"].enhance([:compile_libdatadog_for_dsm])
+    end
+  end
+
   namespace :appsec do
     task all: [
       :main,

lib/datadog/core/ddsketch.rb

@@ -8,17 +8,13 @@ module Core
     # APIs in this class are implemented as native code.
     class DDSketch
       def self.supported?
-        return @supported if defined?(@supported)
+        return false unless Datadog::Core::LIBDATADOG_API_FAILURE.nil?
 
-        @supported = begin
-          return false unless Datadog::Core::LIBDATADOG_API_FAILURE.nil?
-
-          # Test that DDSketch actually works by trying to instantiate it
-          new
-          true
-        rescue ArgumentError
-          false
-        end
+        # Test that DDSketch actually works by trying to instantiate it
+        new
+        true
+      rescue ArgumentError
+        false
       end
 
       def initialize

lib/datadog/tracing/data_streams/processor.rb

@@ -390,21 +390,15 @@ def record_checkpoint_stats(hash:, parent_hash:, edge_latency_sec:, payload_size
 
             # Get or create stats for this pathway ( aggr_key = (",".join(edge_tags), hash_value, parent_hash))
             aggr_key = [tags.join(','), hash, parent_hash]
-
-            begin
-              stats = bucket[:pathway_stats][aggr_key] ||= create_pathway_stats
-            rescue ArgumentError => e
-              # DDSketch failed (e.g. in forked process where extension isn't available)
-              Datadog.logger.debug("DSM processor disabled: #{e.message}")
-              @enabled = false
-              return nil
-            end
+            stats = bucket[:pathway_stats][aggr_key] ||= create_pathway_stats
 
             # Add latencies to DDSketch ()
             full_pathway_latency_sec = timestamp_sec - @pathway_context.pathway_start_sec
             stats[:edge_latency].add(edge_latency_sec)
             stats[:full_pathway_latency].add(full_pathway_latency_sec)
           end
+
+          true
         end
 
         # Record consumer offset stats for DSM reporting

sig/datadog/tracing/data_streams/processor.rbs

@@ -16,16 +16,14 @@ module Datadog
 
         @stats_mutex: untyped
 
-        @ddsketch_class: untyped
-
         @agent_transport: untyped
 
         include Core::Workers::Polling
         include Core::Workers::IntervalLoop
 
         attr_accessor enabled: untyped
 
-        def initialize: (?ddsketch_class: untyped, ?interval: untyped?) -> void
+        def initialize: (?interval: untyped?) -> void
         def track_kafka_produce: (untyped topic, untyped partition, untyped offset, untyped now_sec) -> (nil | true)
         def track_kafka_commit: (untyped group, untyped topic, untyped partition, untyped offset, untyped now_sec) -> (nil | true)
         def track_kafka_consume: (untyped topic, untyped partition, untyped offset, ?untyped? now_sec) -> (nil | true)

spec/datadog/tracing/contrib/kafka/data_streams_spec.rb

@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 
 require 'datadog/tracing/contrib/support/spec_helper'
+require 'datadog/core/ddsketch'
 require 'ostruct'
 require 'datadog/tracing/contrib/kafka/integration'
 require 'datadog/tracing/contrib/kafka/instrumentation/producer'
@@ -78,6 +79,10 @@ def each_batch(**kwargs)
   end
 
   describe 'pathway context' do
+    before do
+      skip('DDSketch not available') unless Datadog::Core::DDSketch.supported?
+    end
+
     let(:test_producer_class) do
       Class.new do
         attr_accessor :pending_message_queue
@@ -136,6 +141,10 @@ def deliver_messages(**kwargs)
   end
 
   describe 'checkpointing' do
+    before do
+      skip('DDSketch not available') unless Datadog::Core::DDSketch.supported?
+    end
+
     let(:test_producer_class) do
       Class.new do
         attr_accessor :pending_message_queue

spec/datadog/tracing/contrib/karafka/data_streams_spec.rb

@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 
 require 'datadog/tracing/contrib/support/spec_helper'
+require 'datadog/core/ddsketch'
 require 'ostruct'
 
 RSpec.describe 'Karafka Data Streams Integration' do
@@ -73,6 +74,10 @@ def build_karafka_messages(messages_data, topic_name = 'test_topic', partition =
   end
 
   describe 'auto-instrumentation' do
+    before do
+      skip('DDSketch not available') unless Datadog::Core::DDSketch.supported?
+    end
+
     it 'automatically extracts and processes pathway context when consuming messages' do
       processor = Datadog.configuration.tracing.data_streams.processor
 
@@ -139,6 +144,10 @@ def build_karafka_messages(messages_data, topic_name = 'test_topic', partition =
   end
 
   describe 'pathway propagation across services' do
+    before do
+      skip('DDSketch not available') unless Datadog::Core::DDSketch.supported?
+    end
+
     it 'maintains pathway continuity through produce → consume → produce chain' do
       processor = Datadog.configuration.tracing.data_streams.processor