Support different karafka configurations per topic

Author: Drowze

lib/datadog/tracing/contrib/karafka/integration.rb

@@ -38,6 +38,10 @@ def new_configuration
           def patcher
             Patcher
           end
+
+          def resolver
+            @resolver ||= Contrib::Configuration::Resolvers::PatternResolver.new
+          end
         end
       end
     end

lib/datadog/tracing/contrib/karafka/patcher.rb

@@ -10,14 +10,6 @@ module Contrib
       module Karafka
         # Patch to add tracing to Karafka::Messages::Messages
         module MessagesPatch
-          def configuration
-            Datadog.configuration.tracing[:karafka]
-          end
-
-          def propagation
-            @propagation ||= Contrib::Karafka::Distributed::Propagation.new
-          end
-
           # `each` is the most popular access point to Karafka messages,
           # but not the only one
           #  Other access patterns do not have a straightforward tracing avenue
@@ -28,6 +20,7 @@ def each(&block)
             parent_trace_digest = Datadog::Tracing.active_trace&.to_digest
 
             @messages_array.each do |message|
+              configuration = datadog_configuration(message.topic)
               trace_digest = if configuration[:distributed_tracing]
                 headers = if message.metadata.respond_to?(:raw_headers)
                   message.metadata.raw_headers
@@ -61,6 +54,12 @@ def each(&block)
               end
             end
           end
+
+          private
+
+          def datadog_configuration(topic)
+            Datadog.configuration.tracing[:karafka, topic]
+          end
         end
 
         module AppPatch

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

@@ -17,6 +17,7 @@
   before do
     Datadog.configure do |c|
       c.tracing.instrument :karafka, configuration_options
+      c.tracing.instrument :karafka, describes: /special_/, distributed_tracing: false
     end
   end
 
@@ -31,16 +32,12 @@
     let(:span_name) { Datadog::Tracing::Contrib::Karafka::Ext::SPAN_MESSAGE_CONSUME }
 
     it 'is expected to send a span' do
-      metadata = ::Karafka::Messages::Metadata.new
-      metadata['offset'] = 412
+      metadata = ::Karafka::Messages::Metadata.new(offset: 412, timestamp: Time.now, topic: 'topic_a')
       raw_payload = rand.to_s
 
       message = ::Karafka::Messages::Message.new(raw_payload, metadata)
-      allow(message).to receive(:timestamp).and_return(Time.now)
-      allow(message).to receive(:topic).and_return('topic_a')
-
-      topic = ::Karafka::Routing::Topic.new('topic_a', double(id: 0))
 
+      topic = ::Karafka::Routing::Topic.new(message.topic, double(id: 0))
       messages = ::Karafka::Messages::Builders::Messages.call([message], topic, 0, Time.now)
 
       expect(messages).to all(be_a(::Karafka::Messages::Message))
@@ -55,6 +52,7 @@
     end
 
     context 'when the message has tracing headers' do
+      let(:topic_name) { "topic_a" }
       let(:message) do
         headers = {}
         producer_trace = nil
@@ -64,15 +62,15 @@
           producer_trace = trace
           Datadog::Tracing::Contrib::Karafka.inject(trace.to_digest, headers)
         end
-        metadata = ::Karafka::Messages::Metadata.new
-        metadata['offset'] = 412
-        metadata[headers_accessor] = headers
+        metadata = ::Karafka::Messages::Metadata.new(
+          offset: 412,
+          headers_accessor => headers,
+          topic: topic_name,
+          timestamp: Time.now
+        )
         raw_payload = rand.to_s
 
-        message = ::Karafka::Messages::Message.new(raw_payload, metadata)
-        allow(message).to receive(:timestamp).and_return(Time.now)
-        allow(message).to receive(:topic).and_return('topic_a')
-        message
+        ::Karafka::Messages::Message.new(raw_payload, metadata)
       end
       let(:headers_accessor) do
         ::Karafka::Messages::Metadata.members.include?(:raw_headers) ? 'raw_headers' : 'headers'
@@ -89,7 +87,7 @@
             consumer_span = Datadog::Tracing.active_span
             consumer_trace = Datadog::Tracing.active_trace
 
-            topic = ::Karafka::Routing::Topic.new('topic_a', double(id: 0))
+            topic = ::Karafka::Routing::Topic.new(topic_name, double(id: 0))
             messages = ::Karafka::Messages::Builders::Messages.call([message], topic, 0, Time.now)
             expect(messages).to all(be_a(::Karafka::Messages::Message))
 
@@ -113,6 +111,37 @@
         end
       end
 
+      context "when distributed tracing is disabled for the topic in particular" do
+        let(:topic_name) { "special_topic" }
+
+        it 'does not continue the span that produced the message' do
+          consumer_span = nil
+          consumer_trace = nil
+
+          Datadog::Tracing.trace('consumer') do
+            consumer_span = Datadog::Tracing.active_span
+            consumer_trace = Datadog::Tracing.active_trace
+
+            topic = ::Karafka::Routing::Topic.new(topic_name, double(id: 0))
+            messages = ::Karafka::Messages::Builders::Messages.call([message], topic, 0, Time.now)
+            expect(messages).to all(be_a(::Karafka::Messages::Message))
+
+            # assert that the current trace re-set to the original trace after iterating the messages
+            expect(Datadog::Tracing.active_trace).to eq(consumer_trace)
+            expect(Datadog::Tracing.active_span).to eq(consumer_span)
+          end
+
+          expect(spans).to have(3).items
+
+          # assert that the message span is not continuation of the producer span
+          expect(span.parent_id).to eq(consumer_span.id)
+          expect(span.trace_id).to eq(consumer_trace.id)
+
+          expect(span.links).to be_empty
+          expect(consumer_span.links).to be_empty
+        end
+      end
+
       context 'when distributed tracing is not enabled' do
         let(:configuration_options) { { distributed_tracing: false } }
 
@@ -124,7 +153,7 @@
             consumer_span = Datadog::Tracing.active_span
             consumer_trace = Datadog::Tracing.active_trace
 
-            topic = ::Karafka::Routing::Topic.new('topic_a', double(id: 0))
+            topic = ::Karafka::Routing::Topic.new(topic_name, double(id: 0))
             messages = ::Karafka::Messages::Builders::Messages.call([message], topic, 0, Time.now)
             expect(messages).to all(be_a(::Karafka::Messages::Message))
 
@@ -150,12 +179,11 @@
     let(:span_name) { Datadog::Tracing::Contrib::Karafka::Ext::SPAN_WORKER_PROCESS }
 
     it 'is expected to send a span' do
-      metadata = ::Karafka::Messages::Metadata.new
-      metadata['offset'] = 412
+      metadata = ::Karafka::Messages::Metadata.new(offset: 412, topic: 'topic_a')
       raw_payload = rand.to_s
 
       message = ::Karafka::Messages::Message.new(raw_payload, metadata)
-      job = double(executor: double(topic: double(name: 'topic_a', consumer: 'ABC'), partition: 0), messages: [message])
+      job = double(executor: double(topic: double(name: message.topic, consumer: 'ABC'), partition: 0), messages: [message])
 
       Karafka.monitor.instrument('worker.processed', { job: job }) do
         # Noop