Class: SharedBroker::Middlewares::OpenTelemetryPropagation

Inherits:
Object
  • Object
show all
Defined in:
lib/shared_broker/middlewares/open_telemetry_propagation.rb

Instance Method Summary collapse

Instance Method Details

#call(topic, message, metadata) ⇒ Object 



8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/shared_broker/middlewares/open_telemetry_propagation.rb', line 8

def call(topic, message, metadata)
  if metadata[:operation] == :publish
    carrier = {}
    ::OpenTelemetry.propagation.inject(carrier)
    carrier.each do |key, value|
      message["_#{key}".to_sym] = value
    end
    yield
  elsif metadata[:operation] == :subscribe
    carrier = {}
    message.each do |key, value|
      if key.to_s.start_with?("_trace")
        carrier[key.to_s.sub(/^_/, "")] = value
      end
    end

    parent_context = ::OpenTelemetry.propagation.extract(carrier)
    ::OpenTelemetry::Context.with_current(parent_context) do
      tracer = ::OpenTelemetry.tracer_provider.tracer("shared_broker")
      tracer.in_span("#{topic} process", kind: :consumer) do |span|
        span.set_attribute("messaging.system", "shared_broker")
        span.set_attribute("messaging.destination", topic)
        yield
      end
    end
  else
    yield
  end
end