DSM Implementation for Ruby with Karafka instrumented

Rakefile

@@ -27,6 +27,14 @@ CORE_WITH_LIBDATADOG_API = [
   'spec/datadog/core/process_discovery_spec.rb',
   'spec/datadog/core/configuration/stable_config_spec.rb',
   'spec/datadog/core/ddsketch_spec.rb',
+  'spec/datadog/data_streams/**/*_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
@@ -82,8 +90,8 @@ namespace :spec do
   desc '' # "Explicitly hiding from `rake -T`"
   RSpec::Core::RakeTask.new(:main) do |t, args|
     t.pattern = 'spec/**/*_spec.rb'
-    t.exclude_pattern = 'spec/**/{appsec/integration,contrib,benchmark,redis,auto_instrument,opentelemetry,profiling,crashtracking,error_tracking,rubocop}/**/*_spec.rb,' \
-                        ' spec/**/{auto_instrument,opentelemetry,process_discovery,stable_config,ddsketch}_spec.rb,' \
+    t.exclude_pattern = 'spec/**/{appsec/integration,contrib,benchmark,redis,auto_instrument,opentelemetry,profiling,crashtracking,error_tracking,rubocop,data_streams}/**/*_spec.rb,' \
+                        ' spec/**/{auto_instrument,opentelemetry,process_discovery,stable_config,ddsketch}*_spec.rb,' \
                         ' spec/datadog/gem_packaging_spec.rb'
     t.rspec_opts = args.to_a.join(' ')
   end
@@ -295,6 +303,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.class}: #{e}"
+      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,

Steepfile

@@ -137,6 +137,7 @@ target :datadog do
   ignore 'lib/datadog/core/utils/time.rb'
   ignore 'lib/datadog/core/vendor/multipart-post/multipart/post/multipartable.rb'
   ignore 'lib/datadog/core/worker.rb'
+  ignore 'lib/datadog/data_streams/configuration/settings.rb'
   ignore 'lib/datadog/core/workers/async.rb'
   ignore 'lib/datadog/core/workers/interval_loop.rb'
   ignore 'lib/datadog/core/workers/polling.rb'

lib/datadog.rb

@@ -8,6 +8,7 @@
 require_relative 'datadog/profiling'
 require_relative 'datadog/appsec'
 require_relative 'datadog/di'
+require_relative 'datadog/data_streams'
 
 # Line probes will not work on Ruby < 2.6 because of lack of :script_compiled
 # trace point. Activate DI automatically on supported Ruby versions but

lib/datadog/core/configuration/components.rb

@@ -19,6 +19,7 @@
 require_relative '../crashtracking/component'
 require_relative '../environment/agent_info'
 require_relative '../process_discovery'
+require_relative '../../data_streams/processor'
 
 module Datadog
   module Core
@@ -75,6 +76,20 @@ def build_crashtracker(settings, agent_settings, logger:)
 
             Datadog::Core::Crashtracking::Component.build(settings, agent_settings, logger: logger)
           end
+
+          def build_data_streams(settings, agent_settings, logger)
+            return unless settings.data_streams.enabled
+
+            Datadog::DataStreams::Processor.new(
+              interval: settings.data_streams.interval,
+              logger: logger,
+              settings: settings,
+              agent_settings: agent_settings
+            )
+          rescue => e
+            logger.warn("Failed to initialize Data Streams Monitoring: #{e.class}: #{e}")
+            nil
+          end
         end
 
         attr_reader \
@@ -90,7 +105,8 @@ def build_crashtracker(settings, agent_settings, logger:)
           :error_tracking,
           :dynamic_instrumentation,
           :appsec,
-          :agent_info
+          :agent_info,
+          :data_streams
 
         def initialize(settings)
           @settings = settings
@@ -126,6 +142,7 @@ def initialize(settings)
           @appsec = Datadog::AppSec::Component.build_appsec_component(settings, telemetry: telemetry)
           @dynamic_instrumentation = Datadog::DI::Component.build(settings, agent_settings, @logger, telemetry: telemetry)
           @error_tracking = Datadog::ErrorTracking::Component.build(settings, @tracer, @logger)
+          @data_streams = self.class.build_data_streams(settings, agent_settings, @logger)
           @environment_logger_extra[:dynamic_instrumentation_enabled] = !!@dynamic_instrumentation
 
           # Configure non-privileged components.
@@ -195,6 +212,9 @@ def shutdown!(replacement = nil)
           # Shutdown workers
           runtime_metrics.stop(true, close_metrics: false)
 
+          # Shutdown Data Streams Monitoring processor
+          data_streams&.stop(true)
+
           # Shutdown the old metrics, unless they are still being used.
           # (e.g. custom Statsd instances.)
           #

lib/datadog/core/configuration/supported_configurations.rb

@@ -33,6 +33,7 @@ module Configuration
          "DD_APPSEC_WAF_DEBUG" => {version: ["A"]},
          "DD_APPSEC_WAF_TIMEOUT" => {version: ["A"]},
          "DD_CRASHTRACKING_ENABLED" => {version: ["A"]},
+         "DD_DATA_STREAMS_ENABLED" => {version: ["A"]},
          "DD_DBM_PROPAGATION_MODE" => {version: ["A"]},
          "DD_DISABLE_DATADOG_RAILS" => {version: ["A"]},
          "DD_DYNAMIC_INSTRUMENTATION_ENABLED" => {version: ["A"]},

lib/datadog/core/ddsketch.rb

@@ -1,7 +1,5 @@
 # frozen_string_literal: true
 
-require 'datadog/core'
-
 module Datadog
   module Core
     # Used to access ddsketch APIs.

lib/datadog/data_streams.rb

@@ -0,0 +1,100 @@
+# frozen_string_literal: true
+
+require_relative 'data_streams/processor'
+require_relative 'data_streams/pathway_context'
+require_relative 'data_streams/configuration/settings'
+require_relative 'data_streams/extensions'
+require_relative 'core/utils/time'
+
+module Datadog
+  # Datadog Data Streams Monitoring public API.
+  #
+  # The Datadog team ensures that public methods in this module
+  # only receive backwards compatible changes, and breaking changes
+  # will only occur in new major versions releases.
+  # @public_api
+  module DataStreams
+    class << self
+      # Set a produce checkpoint for Data Streams Monitoring
+      #
+      # @param type [String] The type of the checkpoint (e.g., 'kafka', 'kinesis', 'sns')
+      # @param destination [String] The destination (e.g., topic, exchange, stream name)
+      # @param auto_instrumentation [Boolean] Whether this checkpoint was set by auto-instrumentation (default: false)
+      # @param tags [Hash] Additional tags to include
+      # @yield [key, value] Block to inject context into carrier
+      # @return [String, nil] Base64 encoded pathway context or nil if disabled
+      # @public_api
+      def set_produce_checkpoint(type:, destination:, auto_instrumentation: false, tags: {}, &block)
+        processor&.set_produce_checkpoint(
+          type: type,
+          destination: destination,
+          manual_checkpoint: !auto_instrumentation,
+          tags: tags,
+          &block
+        )
+      end
+
+      # Set a consume checkpoint for Data Streams Monitoring
+      #
+      # @param type [String] The type of the checkpoint (e.g., 'kafka', 'kinesis', 'sns')
+      # @param source [String] The source (e.g., topic, exchange, stream name)
+      # @param auto_instrumentation [Boolean] Whether this checkpoint was set by auto-instrumentation (default: false)
+      # @param tags [Hash] Additional tags to include
+      # @yield [key] Block to extract context from carrier
+      # @return [String, nil] Base64 encoded pathway context or nil if disabled
+      # @public_api
+      def set_consume_checkpoint(type:, source:, auto_instrumentation: false, tags: {}, &block)
+        processor&.set_consume_checkpoint(
+          type: type,
+          source: source,
+          manual_checkpoint: !auto_instrumentation,
+          tags: tags,
+          &block
+        )
+      end
+
+      # Track Kafka produce offset for lag monitoring
+      #
+      # @param topic [String] The Kafka topic name
+      # @param partition [Integer] The partition number
+      # @param offset [Integer] The offset of the produced message
+      # @return [Boolean, nil] true if tracking succeeded, nil if disabled
+      # @public_api
+      def track_kafka_produce(topic, partition, offset)
+        processor&.track_kafka_produce(topic, partition, offset, Core::Utils::Time.now)
+      end
+
+      # Track Kafka message consumption for consumer lag monitoring
+      #
+      # @param topic [String] The Kafka topic name
+      # @param partition [Integer] The partition number
+      # @param offset [Integer] The offset of the consumed message
+      # @return [Boolean, nil] true if tracking succeeded, nil if disabled
+      # @public_api
+      def track_kafka_consume(topic, partition, offset)
+        processor&.track_kafka_consume(topic, partition, offset, Core::Utils::Time.now)
+      end
+
+      # Check if Data Streams Monitoring is enabled and available
+      #
+      # @return [Boolean] true if the processor is available
+      # @public_api
+      def enabled?
+        !processor.nil?
+      end
+
+      private
+
+      def processor
+        components.data_streams
+      end
+
+      def components
+        Datadog.send(:components)
+      end
+    end
+
+    # Expose Data Streams to global shared objects
+    Extensions.activate!
+  end
+end

lib/datadog/data_streams/configuration.rb

@@ -0,0 +1,11 @@
+# frozen_string_literal: true
+
+require_relative 'configuration/settings'
+
+module Datadog
+  module DataStreams
+    # Configuration for Data Streams Monitoring
+    module Configuration
+    end
+  end
+end

lib/datadog/data_streams/configuration/settings.rb

@@ -0,0 +1,49 @@
+# frozen_string_literal: true
+
+require_relative '../../core/environment/variable_helpers'
+require_relative '../ext'
+
+module Datadog
+  module DataStreams
+    module Configuration
+      # Configuration settings for Data Streams Monitoring.
+      module Settings
+        def self.extended(base)
+          base = base.singleton_class unless base.is_a?(Class)
+          add_settings!(base)
+        end
+
+        def self.add_settings!(base)
+          base.class_eval do
+            # Data Streams Monitoring configuration
+            # @public_api
+            settings :data_streams do
+              # Whether Data Streams Monitoring is enabled. When enabled, the library will
+              # collect and report data lineage information for messaging systems.
+              #
+              # @default `DD_DATA_STREAMS_ENABLED` environment variable, otherwise `false`.
+              # @return [Boolean]
+              option :enabled do |o|
+                o.type :bool
+                o.env Ext::ENV_ENABLED
+                o.default false
+              end
+
+              # The interval (in seconds) at which Data Streams Monitoring stats are flushed.
+              #
+              # @default 10.0
+              # @env '_DD_TRACE_STATS_WRITER_INTERVAL'
+              # @return [Float]
+              # @!visibility private
+              option :interval do |o|
+                o.type :float
+                o.env '_DD_TRACE_STATS_WRITER_INTERVAL'
+                o.default 10.0
+              end
+            end
+          end
+        end
+      end
+    end
+  end
+end

lib/datadog/data_streams/ext.rb

@@ -0,0 +1,11 @@
+# frozen_string_literal: true
+
+module Datadog
+  module DataStreams
+    # Constants for Data Streams Monitoring configuration
+    # @public_api
+    module Ext
+      ENV_ENABLED = 'DD_DATA_STREAMS_ENABLED'
+    end
+  end
+end

lib/datadog/data_streams/extensions.rb

@@ -0,0 +1,16 @@
+# frozen_string_literal: true
+
+require_relative 'configuration'
+require_relative '../core/configuration'
+
+module Datadog
+  module DataStreams
+    # Extends Datadog with Data Streams Monitoring features
+    module Extensions
+      # Inject Data Streams settings into global configuration.
+      def self.activate!
+        Core::Configuration::Settings.extend(Configuration::Settings)
+      end
+    end
+  end
+end