Flush on execution end + tests

lib/ddtrace/contrib/delayed_job/plugin.rb

@@ -17,12 +17,20 @@ def self.instrument(job, &block)
             span.set_tag('delayed_job.attempts'.freeze, job.attempts)
             span.span_type = pin.app_type
 
-            block.call(job)
+            yield job
           end
         end
 
+        def self.flush(worker, &block)
+          yield worker
+
+          pin = Pin.get_from(::Delayed::Worker)
+          pin.tracer.shutdown! if pin && pin.tracer
+        end
+
         callbacks do |lifecycle|
           lifecycle.around(:invoke_job, &method(:instrument))
+          lifecycle.around(:execute, &method(:flush))
         end
       end
     end

spec/ddtrace/contrib/delayed_job/plugin_spec.rb

@@ -16,6 +16,25 @@
     Delayed::Worker.delay_jobs = false
   end
 
+  describe 'instrumenting worker execution' do
+    let(:worker) { double(:worker, name: 'worker') }
+    before do
+      allow(tracer).to receive(:shutdown!).and_call_original
+    end
+
+    it 'execution callback yields control' do
+      expect { |b| Delayed::Worker.lifecycle.run_callbacks(:execute, worker, &b) }.to yield_with_args(worker)
+    end
+
+    it 'shutdown happens after yielding' do
+      Delayed::Worker.lifecycle.run_callbacks(:execute, worker) do
+        expect(tracer).not_to have_received(:shutdown!)
+      end
+
+      expect(tracer).to have_received(:shutdown!)
+    end
+  end
+
   describe 'instrumented job invocation' do
     let(:job_params) { {} }
     subject(:job_run) { Delayed::Job.enqueue(SampleJob.new, job_params) }