fix: Sidekiq spans crashing when enqueued by Exq

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

@@ -31,13 +31,10 @@ def call(_worker, msg, _queue)
                           end
 
               extracted_context = OpenTelemetry.propagation.extract(msg)
-              created_at = time_from_timestamp(msg['created_at'])
-              enqueued_at = time_from_timestamp(msg['created_at'])
               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)
+                    add_span_timestamps(span, msg)
                     yield
                   end
                 else
@@ -46,8 +43,7 @@ def call(_worker, msg, _queue)
                   links << OpenTelemetry::Trace::Link.new(span_context) if instrumentation_config[:propagation_style] == :link && span_context.valid?
                   span = tracer.start_root_span(span_name, attributes: attributes, links: links, kind: :consumer)
                   OpenTelemetry::Trace.with_span(span) do
-                    span.add_event('created_at', timestamp: created_at)
-                    span.add_event('enqueued_at', timestamp: enqueued_at)
+                    add_span_timestamps(span, msg)
                     yield
                   rescue Exception => e # rubocop:disable Lint/RescueException
                     span.record_exception(e)
@@ -70,6 +66,13 @@ def tracer
               Sidekiq::Instrumentation.instance.tracer
             end
 
+            def add_span_timestamps(span, msg)
+              created_at = msg['created_at']
+              enqueued_at = msg['enqueued_at']
+              span.add_event('created_at', timestamp: time_from_timestamp(created_at)) if created_at
+              span.add_event('enqueued_at', timestamp: time_from_timestamp(enqueued_at)) if enqueued_at
+            end
+
             def time_from_timestamp(timestamp)
               if timestamp.is_a?(Float)
                 # old format, timestamps were stored as fractional seconds since the epoch

instrumentation/sidekiq/test/opentelemetry/instrumentation/sidekiq/middlewares/server/tracer_middleware_test.rb

@@ -16,6 +16,7 @@
   let(:job_span) { spans.last }
   let(:root_span) { spans.find { |s| s.parent_span_id == OpenTelemetry::Trace::INVALID_SPAN_ID } }
   let(:config) { {} }
+  let(:timestamp) { Process.clock_gettime(Process::CLOCK_REALTIME, :millisecond) }
 
   before do
     instrumentation.install(config)
@@ -67,6 +68,65 @@
       _(job_span.attributes['messaging.operation']).must_equal 'process'
     end
 
+    it 'traces when enqueued with only enqueued_at' do
+      payload = { 'queue' => 'default', 'args' => [], 'class' => SimpleJob.to_s, 'enqueued_at' => timestamp, 'jid' => '4' }
+      Sidekiq::Queues.push('default', 'SimpleJob', payload)
+      Sidekiq::Worker.drain_all
+
+      _(job_span.name).must_equal 'default process'
+      _(job_span.kind).must_equal :consumer
+      _(job_span.attributes['messaging.system']).must_equal 'sidekiq'
+      _(job_span.attributes['messaging.sidekiq.job_class']).must_equal 'SimpleJob'
+      _(job_span.attributes['messaging.message_id']).must_equal '4'
+      _(job_span.attributes['messaging.destination']).must_equal 'default'
+      _(job_span.attributes['messaging.destination_kind']).must_equal 'queue'
+      _(job_span.attributes['messaging.operation']).must_equal 'process'
+      _(job_span.attributes['peer.service']).must_be_nil
+      _(job_span.events.size).must_equal(1)
+
+      enqueued_event = job_span.events[0]
+      _(enqueued_event.name).must_equal('enqueued_at')
+      _(enqueued_event.timestamp.digits.count).must_equal(19)
+    end
+
+    it 'traces when enqueued with only created_at' do
+      payload = { 'queue' => 'default', 'args' => [], 'class' => SimpleJob.to_s, 'created_at' => timestamp, 'jid' => '4' }
+      Sidekiq::Queues.push('default', 'SimpleJob', payload)
+      Sidekiq::Worker.drain_all
+
+      _(job_span.name).must_equal 'default process'
+      _(job_span.kind).must_equal :consumer
+      _(job_span.attributes['messaging.system']).must_equal 'sidekiq'
+      _(job_span.attributes['messaging.sidekiq.job_class']).must_equal 'SimpleJob'
+      _(job_span.attributes['messaging.message_id']).must_equal '4'
+      _(job_span.attributes['messaging.destination']).must_equal 'default'
+      _(job_span.attributes['messaging.destination_kind']).must_equal 'queue'
+      _(job_span.attributes['messaging.operation']).must_equal 'process'
+      _(job_span.attributes['peer.service']).must_be_nil
+      _(job_span.events.size).must_equal(1)
+
+      enqueued_event = job_span.events[0]
+      _(enqueued_event.name).must_equal('created_at')
+      _(enqueued_event.timestamp.digits.count).must_equal(19)
+    end
+
+    it 'traces when enqueued with minimal data' do
+      payload = { 'queue' => 'default', 'args' => [], 'class' => SimpleJob.to_s, 'jid' => '4' }
+      Sidekiq::Queues.push('default', 'SimpleJob', payload)
+      Sidekiq::Worker.drain_all
+
+      _(job_span.name).must_equal 'default process'
+      _(job_span.kind).must_equal :consumer
+      _(job_span.attributes['messaging.system']).must_equal 'sidekiq'
+      _(job_span.attributes['messaging.sidekiq.job_class']).must_equal 'SimpleJob'
+      _(job_span.attributes['messaging.message_id']).must_equal '4'
+      _(job_span.attributes['messaging.destination']).must_equal 'default'
+      _(job_span.attributes['messaging.destination_kind']).must_equal 'queue'
+      _(job_span.attributes['messaging.operation']).must_equal 'process'
+      _(job_span.attributes['peer.service']).must_be_nil
+      _(job_span.events).must_be_nil
+    end
+
     it 'defaults to using links to the enqueing span but does not continue the trace' do
       SimpleJob.perform_async
       SimpleJob.drain