Sha256: d0188b9f82cc1672fb243bcebd7984f8bceb70d2e0147bf55c890cbb5842535d
Contents?: true
Size: 1.61 KB
Versions: 17
Compression:
Stored size: 1.61 KB
Contents
# frozen_string_literal: true
module OpenTracing
module Instrumentation
module Sidekiq
# Sidekiq producer middleware
class ClientMiddleware
extend Forwardable
DEFAULT_SPAN_NAME = 'sidekiq_enqueue'
DEFAULT_SPAN_KIND = 'producer'
attr_reader :tracer
attr_reader :tagger
attr_reader :error_writter
attr_reader :span_kind
attr_reader :span_name
def initialize(
tracer: OpenTracing.global_tracer,
tagger: JobTagger.new,
error_writter: Common::ErrorWriter.new,
span_kind: DEFAULT_SPAN_KIND,
span_name: DEFAULT_SPAN_NAME
)
@tracer = tracer
@tagger = tagger
@error_writter = error_writter
@span_kind = span_kind
@span_name = span_name
end
def call(_worker_class, job, _queue, _redis_pool)
scope = tracer.start_active_span(span_name, **build_span_args(job))
inject(scope.span.context, job)
log(scope.span, job) do
yield
end
ensure
scope.close
end
private
def log(span, job)
tagger.write_args_log(span, job['jid'], job['args'])
yield
rescue StandardError => e
error_writter.write_error(span, e)
raise
end
def build_span_args(job)
{
tags: tagger.build_tags(job, span_kind),
}
end
def inject(span_context, carrier)
tracer.inject(span_context, OpenTracing::FORMAT_TEXT_MAP, carrier)
end
end
end
end
end
Version data entries
17 entries across 17 versions & 1 rubygems