feat(sidekiq): scheduled jobs improvements - scheduled operation and add links

When a Job is pushed via `perform_in` or friends, it goes through the following life-cycle:
1. First the Job is pushed with `at` attribute in Redis from `Sidekiq::Client` - this will now create `<...> scheduled` span (before it was `<...> publish`
2. Then the Job is handled from `Sidekiq::Scheduled#enqueue` and pushed without `at` attribute in Redis from `Sidekiq::Client` - this will keep creating `<...> publish` span
3. Then the Job is handled by the Worker as usual

Additionally when using `:propagation_style = :link` and `:trace_poller_enqueue = true` a handy set of links are added between the 3 actors above.

Author: smoke

instrumentation/sidekiq/lib/opentelemetry/instrumentation/sidekiq/middlewares/client/tracer_middleware.rb

@@ -24,14 +24,25 @@ def call(_worker_class, job, _queue, _redis_pool)
               }
               attributes[SemanticConventions::Trace::PEER_SERVICE] = instrumentation_config[:peer_service] if instrumentation_config[:peer_service]
 
+              scheduled_at = job['at']
+              op = scheduled_at.nil? ? "publish" : "scheduled"
+
               span_name = case instrumentation_config[:span_naming]
-                          when :job_class then "#{job['wrapped']&.to_s || job['class']} publish"
-                          else "#{job['queue']} publish"
+                          when :job_class then "#{job['wrapped']&.to_s || job['class']} #{op}"
+                          else "#{job['queue']} #{op}"
                           end
 
-              tracer.in_span(span_name, attributes: attributes, kind: :producer) do |span|
+              # In case this is Scheduled job, there is already context injected, so link to that context
+              # NOTE: :propagation_style = :child is not supported as it is quite tricky when :trace_poller_enqueue = true
+              extracted_context = OpenTelemetry.propagation.extract(job, OpenTelemetry::Trace::SpanContext.INVALID)
+              links = []
+              span_context = OpenTelemetry::Trace.current_span(extracted_context).context
+              links << OpenTelemetry::Trace::Link.new(span_context) if instrumentation_config[:propagation_style] == :link && span_context.valid?
+
+              tracer.in_span(span_name, attributes: attributes, links: links, kind: :producer) do |span|
                 OpenTelemetry.propagation.inject(job)
                 span.add_event('created_at', timestamp: time_from_timestamp(job['created_at']))
+                span.add_event('scheduled_at', timestamp: time_from_timestamp(scheduled_at)) unless scheduled_at.nil?
                 yield
               end
             end

instrumentation/sidekiq/lib/opentelemetry/instrumentation/sidekiq/middlewares/server/tracer_middleware.rb

@@ -33,11 +33,13 @@ def call(_worker, msg, _queue)
               extracted_context = OpenTelemetry.propagation.extract(msg)
               created_at = time_from_timestamp(msg['created_at'])
               enqueued_at = time_from_timestamp(msg['created_at'])
+              scheduled_at = time_from_timestamp(msg['at']) unless msg['at'].nil?
               OpenTelemetry::Context.with_current(extracted_context) do
                 if instrumentation_config[:propagation_style] == :child
                   tracer.in_span(span_name, attributes: attributes, kind: :consumer) do |span|
                     span.add_event('created_at', timestamp: created_at)
                     span.add_event('enqueued_at', timestamp: enqueued_at)
+                    span.add_event('scheduled_at', timestamp: scheduled_at) unless scheduled_at.nil?
                     yield
                   end
                 else
@@ -48,6 +50,7 @@ def call(_worker, msg, _queue)
                   OpenTelemetry::Trace.with_span(span) do
                     span.add_event('created_at', timestamp: created_at)
                     span.add_event('enqueued_at', timestamp: enqueued_at)
+                    span.add_event('scheduled_at', timestamp: scheduled_at) unless scheduled_at.nil?
                     yield
                   rescue Exception => e # rubocop:disable Lint/RescueException
                     span.record_exception(e)