Skip to content

Commit dd02868

Browse files
nvh0412nvh0412
committed
Add Karafka integration
1 parent 0be95cb commit dd02868

File tree

19 files changed

+709
-192
lines changed

19 files changed

+709
-192
lines changed

‎appraisal/ruby-2.7.rb

+1
Original file line numberDiff line numberDiff line change
@@ -193,6 +193,7 @@
193193
gem 'concurrent-ruby'
194194
gem 'dalli', '>= 3.0.0'
195195
gem 'grpc'
196+
gem 'karafka'
196197
gem 'mongo', '>= 2.8.0', '< 2.15.0' # TODO: FIX TEST BREAKAGES ON >= 2.15 https://github.com/DataDog/dd-trace-rb/issues/1596
197198
gem 'rack-test' # Dev dependencies for testing rack-based code
198199
gem 'rake', '>= 12.3'

‎appraisal/ruby-3.0.rb

+1
Original file line numberDiff line numberDiff line change
@@ -115,6 +115,7 @@
115115
gem 'concurrent-ruby'
116116
gem 'dalli', '>= 3.0.0'
117117
gem 'grpc', '>= 1.38.0', platform: :ruby # Minimum version with Ruby 3.0 support
118+
gem 'karafka'
118119
gem 'mongo', '>= 2.8.0', '< 2.15.0' # TODO: FIX TEST BREAKAGES ON >= 2.15 https://github.com/DataDog/dd-trace-rb/issues/1596
119120
gem 'rack-test' # Dev dependencies for testing rack-based code
120121
gem 'rake', '>= 12.3'

‎appraisal/ruby-3.1.rb

+1
Original file line numberDiff line numberDiff line change
@@ -115,6 +115,7 @@
115115
gem 'concurrent-ruby'
116116
gem 'dalli', '>= 3.0.0'
117117
gem 'grpc', '>= 1.38.0', platform: :ruby # Minimum version with Ruby 3.0 support
118+
gem 'karafka'
118119
gem 'mongo', '>= 2.8.0', '< 2.15.0' # TODO: FIX TEST BREAKAGES ON >= 2.15 https://github.com/DataDog/dd-trace-rb/issues/1596
119120
gem 'rack-test' # Dev dependencies for testing rack-based code
120121
gem 'rake', '>= 12.3'

‎appraisal/ruby-3.2.rb

+1
Original file line numberDiff line numberDiff line change
@@ -115,6 +115,7 @@
115115
gem 'concurrent-ruby'
116116
gem 'dalli', '>= 3.0.0'
117117
gem 'grpc', '>= 1.38.0', platform: :ruby # Minimum version with Ruby 3.0 support
118+
gem 'karafka'
118119
gem 'mongo', '>= 2.8.0', '< 2.15.0' # TODO: FIX TEST BREAKAGES ON >= 2.15 https://github.com/DataDog/dd-trace-rb/issues/1596
119120
gem 'rack-test' # Dev dependencies for testing rack-based code
120121
gem 'rake', '>= 12.3'

‎appraisal/ruby-3.3.rb

+1
Original file line numberDiff line numberDiff line change
@@ -115,6 +115,7 @@
115115
gem 'concurrent-ruby'
116116
gem 'dalli', '>= 3.0.0'
117117
gem 'grpc', '>= 1.38.0', platform: :ruby # Minimum version with Ruby 3.0 support
118+
gem 'karafka'
118119
gem 'mongo', '>= 2.8.0', '< 2.15.0' # TODO: FIX TEST BREAKAGES ON >= 2.15 https://github.com/DataDog/dd-trace-rb/issues/1596
119120
gem 'rack-test' # Dev dependencies for testing rack-based code
120121
gem 'rake', '>= 12.3'

‎appraisal/ruby-3.4.rb

+1
Original file line numberDiff line numberDiff line change
@@ -122,6 +122,7 @@
122122
gem 'concurrent-ruby'
123123
gem 'dalli', '>= 3.0.0'
124124
gem 'grpc', '>= 1.38.0', platform: :ruby # Minimum version with Ruby 3.0 support
125+
gem 'karafka'
125126
gem 'mongo', '>= 2.8.0', '< 2.15.0' # TODO: FIX TEST BREAKAGES ON >= 2.15 https://github.com/DataDog/dd-trace-rb/issues/1596
126127
gem 'rack-test' # Dev dependencies for testing rack-based code
127128
gem 'rake', '>= 12.3'

‎docs/GettingStarted.md

+214-192
Large diffs are not rendered by default.

‎lib/datadog/tracing/contrib.rb

+1
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,7 @@ module Contrib
5555
require_relative 'contrib/httprb/integration'
5656
require_relative 'contrib/integration'
5757
require_relative 'contrib/kafka/integration'
58+
require_relative 'contrib/karafka'
5859
require_relative 'contrib/lograge/integration'
5960
require_relative 'contrib/mongodb/integration'
6061
require_relative 'contrib/mysql2/integration'

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

+37
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
# frozen_string_literal: true
2+
3+
require_relative 'component'
4+
require_relative 'karafka/integration'
5+
require_relative 'karafka/distributed/propagation'
6+
7+
module Datadog
8+
module Tracing
9+
module Contrib
10+
# `Karafka` integration public API
11+
module Karafka
12+
def self.inject(digest, data)
13+
raise 'Please invoke Datadog.configure at least once before calling this method' unless @propagation
14+
15+
@propagation.inject!(digest, data)
16+
end
17+
18+
def self.extract(data)
19+
raise 'Please invoke Datadog.configure at least once before calling this method' unless @propagation
20+
21+
@propagation.extract(data)
22+
end
23+
24+
Contrib::Component.register('karafka') do |config|
25+
tracing = config.tracing
26+
tracing.propagation_style
27+
28+
@propagation = Sidekiq::Distributed::Propagation.new(
29+
propagation_style_inject: tracing.propagation_style_inject,
30+
propagation_style_extract: tracing.propagation_style_extract,
31+
propagation_extract_first: tracing.propagation_extract_first
32+
)
33+
end
34+
end
35+
end
36+
end
37+
end

‎lib/datadog/tracing/contrib/karafka/configuration/settings.rb

+27
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
# frozen_string_literal: true
2+
3+
require_relative '../../configuration/settings'
4+
require_relative '../ext'
5+
6+
module Datadog
7+
module Tracing
8+
module Contrib
9+
module Karafka
10+
module Configuration
11+
# @public_api
12+
class Settings < Contrib::Configuration::Settings
13+
option :enabled do |o|
14+
o.type :bool
15+
o.env Ext::ENV_ENABLED
16+
o.default true
17+
end
18+
19+
option :service_name
20+
21+
option :distributed_tracing, default: false, type: :bool
22+
end
23+
end
24+
end
25+
end
26+
end
27+
end

‎lib/datadog/tracing/contrib/karafka/distributed/propagation.rb

+47
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
# frozen_string_literal: true
2+
3+
require_relative '../../../distributed/fetcher'
4+
require_relative '../../../distributed/propagation'
5+
require_relative '../../../distributed/b3_multi'
6+
require_relative '../../../distributed/b3_single'
7+
require_relative '../../../distributed/datadog'
8+
require_relative '../../../distributed/none'
9+
require_relative '../../../distributed/trace_context'
10+
require_relative '../../../configuration/ext'
11+
12+
module Datadog
13+
module Tracing
14+
module Contrib
15+
module Karafka
16+
module Distributed
17+
# Extracts and injects propagation through Kafka message headers.
18+
class Propagation < Tracing::Distributed::Propagation
19+
def initialize(
20+
propagation_style_inject:,
21+
propagation_style_extract:,
22+
propagation_extract_first:
23+
)
24+
super(
25+
propagation_styles: {
26+
Tracing::Configuration::Ext::Distributed::PROPAGATION_STYLE_B3_MULTI_HEADER =>
27+
Tracing::Distributed::B3Multi.new(fetcher: Tracing::Distributed::Fetcher),
28+
Tracing::Configuration::Ext::Distributed::PROPAGATION_STYLE_B3_SINGLE_HEADER =>
29+
Tracing::Distributed::B3Single.new(fetcher: Tracing::Distributed::Fetcher),
30+
Tracing::Configuration::Ext::Distributed::PROPAGATION_STYLE_DATADOG =>
31+
Tracing::Distributed::Datadog.new(fetcher: Tracing::Distributed::Fetcher),
32+
Tracing::Configuration::Ext::Distributed::PROPAGATION_STYLE_TRACE_CONTEXT =>
33+
Tracing::Distributed::TraceContext.new(fetcher: Tracing::Distributed::Fetcher),
34+
Tracing::Configuration::Ext::Distributed::PROPAGATION_STYLE_NONE => Tracing::Distributed::None.new
35+
},
36+
propagation_style_inject: propagation_style_inject,
37+
propagation_style_extract: propagation_style_extract,
38+
propagation_extract_first: propagation_extract_first
39+
)
40+
end
41+
end
42+
end
43+
end
44+
end
45+
end
46+
end
47+

‎lib/datadog/tracing/contrib/karafka/event.rb

+28
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
# frozen_string_literal: true
2+
3+
require_relative '../analytics'
4+
require_relative 'ext'
5+
6+
module Datadog
7+
module Tracing
8+
module Contrib
9+
module Karafka
10+
module Event
11+
def self.included(base)
12+
base.extend(ClassMethods)
13+
end
14+
15+
module ClassMethods
16+
def span_options
17+
{ service: configuration[:service_name] }
18+
end
19+
20+
def configuration
21+
Datadog.configuration.tracing[:karafka]
22+
end
23+
end
24+
end
25+
end
26+
end
27+
end
28+
end

‎lib/datadog/tracing/contrib/karafka/events.rb

+28
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
# frozen_string_literal: true
2+
3+
require_relative 'events/worker/process'
4+
5+
module Datadog
6+
module Tracing
7+
module Contrib
8+
module Karafka
9+
# Defines collection of instrumented Kafka events
10+
module Events
11+
ALL = [
12+
Events::Worker::Process,
13+
]
14+
15+
module_function
16+
17+
def all
18+
self::ALL
19+
end
20+
21+
def subscribe!
22+
all.each(&:subscribe!)
23+
end
24+
end
25+
end
26+
end
27+
end
28+
end

‎lib/datadog/tracing/contrib/karafka/events/error/occur.rb

+28
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
# frozen_string_literal: true
2+
3+
require_relative '../../ext'
4+
require_relative '../../event'
5+
6+
require 'byebug'
7+
8+
module Datadog
9+
module Tracing
10+
module Contrib
11+
module Karafka
12+
module Events
13+
module Error
14+
module Occur
15+
include Karafka::Event
16+
17+
def self.subscribe!
18+
::Karafka.monitor.subscribe 'error.consume' do |event|
19+
end
20+
end
21+
end
22+
end
23+
end
24+
end
25+
end
26+
end
27+
end
28+

‎lib/datadog/tracing/contrib/karafka/events/worker/process.rb

+77
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,77 @@
1+
# frozen_string_literal: true
2+
3+
require_relative '../../ext'
4+
require_relative '../../event'
5+
6+
module Datadog
7+
module Tracing
8+
module Contrib
9+
module Karafka
10+
module Events
11+
module Worker
12+
module Process
13+
include Karafka::Event
14+
15+
def self.subscribe!
16+
::Karafka.monitor.subscribe 'worker.process' do |event|
17+
# Start a trace
18+
span = Tracing.trace(Ext::SPAN_WORKER_PROCESS, **span_options)
19+
20+
job = event[:job]
21+
job_type = fetch_job_type(job.class)
22+
consumer = job.executor.topic.consumer
23+
topic = job.executor.topic.name
24+
25+
action = case job_type
26+
when 'Periodic'
27+
'tick'
28+
when 'PeriodicNonBlocking'
29+
'tick'
30+
when 'Shutdown'
31+
'shutdown'
32+
when 'Revoked'
33+
'revoked'
34+
when 'RevokedNonBlocking'
35+
'revoked'
36+
when 'Idle'
37+
'idle'
38+
when 'Eofed'
39+
'eofed'
40+
when 'EofedNonBlocking'
41+
'eofed'
42+
else
43+
'consume'
44+
end
45+
46+
span.resource = "#{consumer}##{action}"
47+
span.set_tag(Ext::TAG_TOPIC, topic) if topic
48+
49+
if action == 'consume'
50+
span.set_tag(Ext::TAG_MESSAGE_COUNT, job.messages.count)
51+
span.set_tag(Ext::TAG_PARTITION, job.executor.partition)
52+
span.set_tag(Ext::TAG_OFFSET, job.messages.first.metadata.offset)
53+
end
54+
55+
span
56+
end
57+
58+
::Karafka.monitor.subscribe 'worker.completed' do |event|
59+
Tracing.active_span&.finish
60+
end
61+
end
62+
63+
def self.span_options
64+
super.merge({ tags: { Tracing::Metadata::Ext::TAG_OPERATION => Ext::TAG_OPERATION_PROCESS_BATCH } })
65+
end
66+
67+
def self.fetch_job_type(job_class)
68+
@job_types_cache ||= {}
69+
@job_types_cache[job_class] ||= job_class.to_s.split('::').last
70+
end
71+
end
72+
end
73+
end
74+
end
75+
end
76+
end
77+
end

‎lib/datadog/tracing/contrib/karafka/ext.rb

+25
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
# frozen_string_literal: true
2+
3+
module Datadog
4+
module Tracing
5+
module Contrib
6+
module Karafka
7+
module Ext
8+
ENV_ENABLED = 'DD_TRACE_KARAFKA_ENABLED'
9+
10+
SPAN_MESSAGE_CONSUME = 'karafka.consume'
11+
SPAN_WORKER_PROCESS = 'worker.process'
12+
13+
TAG_TOPIC = 'kafka.topic'
14+
TAG_PARTITION = 'kafka.partition'
15+
TAG_OFFSET = 'kafka.offset'
16+
TAG_OFFSET_LAG = 'kafka.offset_lag'
17+
TAG_MESSAGE_COUNT = 'kafka.message_count'
18+
TAG_MESSAGE_KEY = 'kafka.message_key'
19+
20+
TAG_OPERATION_PROCESS_BATCH = 'consumer.process_batch'
21+
end
22+
end
23+
end
24+
end
25+
end

0 commit comments

Comments
 (0)