Rework `Sidekiq::JobsThreads` into `Monitor`

This makes:
- very shallow `Middleware::Monitor` to only request tracking
  of sidekiq jobs,
- `SidekiqStatus::Monitor` to be responsible to maintain persistent
  connection to receive messages,
- `SidekiqStatus::Monitor` to always use structured logging
  and instance variables

3 files changed, 235 insertions, 83 deletions

diff --git a/spec/lib/gitlab/sidekiq_middleware/jobs_threads_spec.rb b/spec/lib/gitlab/sidekiq_middleware/jobs_threads_spec.rb
deleted file mode 100644
index 58cae3e42e0..00000000000
--- a/spec/lib/gitlab/sidekiq_middleware/jobs_threads_spec.rb
+++ /dev/null
@@ -1,83 +0,0 @@
-require 'spec_helper'
-
-describe Gitlab::SidekiqMiddleware::JobsThreads do
-  subject { described_class.new }
-
-  let(:worker) { double(:worker, class: Chaos::SleepWorker) }
-  let(:jid) { '581f90fbd2f24deabcbde2f9' }
-  let(:job) { { 'jid' => jid } }
-  let(:jid_thread) { '684f90fbd2f24deabcbde2f9' }
-  let(:job_thread) { { 'jid' => jid_thread } }
-  let(:queue) { 'test_queue' }
-  let(:mark_job_as_cancelled) { Sidekiq.redis {|c| c.setex("cancelled-#{jid}", 2, 1) } }
-
-  def run_job
-    subject.call(worker, job, queue) do
-      sleep 2
-      "mock return from yield"
-    end
-  end
-
-  def run_job_thread
-    Thread.new do
-      subject.call(worker, job_thread, queue) do
-        sleep 3
-        "mock return from yield"
-      end
-    end
-  end
-
-  describe '.call' do
-    context 'by default' do
-      it 'return from yield' do
-        expect(run_job).to eq("mock return from yield")
-      end
-    end
-
-    context 'when job is marked as cancelled' do
-      before do
-        mark_job_as_cancelled
-      end
-
-      it 'return directly' do
-        expect(run_job).to be_nil
-      end
-    end
-  end
-
-  describe '.self.interrupt' do
-    before do
-      run_job_thread
-      sleep 1
-    end
-
-    it 'interrupt the job with correct jid' do
-      expect(described_class.jobs[jid_thread]).to receive(:raise).with(Interrupt)
-      expect(described_class.interrupt jid_thread).to eq(described_class.jobs[jid_thread])
-    end
-
-    it 'do nothing with wrong jid' do
-      expect(described_class.jobs[jid_thread]).not_to receive(:raise)
-      expect(described_class.interrupt 'wrong_jid').to be_nil
-    end
-  end
-
-  describe '.self.cancelled?' do
-    it 'return true when job is marked as cancelled' do
-      mark_job_as_cancelled
-      expect(described_class.cancelled? jid).to be true
-    end
-
-    it 'return false when job is not marked as cancelled' do
-      expect(described_class.cancelled? 'non-exists-jid').to be false
-    end
-  end
-
-  describe '.self.mark_job_as_cancelled' do
-    it 'set Redis key' do
-      described_class.mark_job_as_cancelled('jid_123')
-
-      expect(described_class.cancelled? 'jid_123').to be true
-    end
-  end
-end
diff --git a/spec/lib/gitlab/sidekiq_middleware/monitor_spec.rb b/spec/lib/gitlab/sidekiq_middleware/monitor_spec.rb
new file mode 100644
index 00000000000..3ca2ddf3cb1
--- /dev/null
+++ b/spec/lib/gitlab/sidekiq_middleware/monitor_spec.rb
@@ -0,0 +1,29 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe Gitlab::SidekiqMiddleware::Monitor do
+  let(:monitor) { described_class.new }
+
+  describe '#call' do
+    let(:worker) { double }
+    let(:job) { { 'jid' => 'job-id' } }
+    let(:queue) { 'my-queue' }
+
+    it 'calls SidekiqMonitor' do
+      expect(Gitlab::SidekiqMonitor.instance).to receive(:within_job)
+        .with('job-id', 'my-queue')
+        .and_call_original
+
+      expect { |blk| monitor.call(worker, job, queue, &blk) }.to yield_control
+    end
+
+    it 'passthroughs the return value' do
+      result = monitor.call(worker, job, queue) do
+        'value'
+      end
+
+      expect(result).to eq('value')
+    end
+  end
+end
diff --git a/spec/lib/gitlab/sidekiq_monitor_spec.rb b/spec/lib/gitlab/sidekiq_monitor_spec.rb
new file mode 100644
index 00000000000..7c9fc598b02
--- /dev/null
+++ b/spec/lib/gitlab/sidekiq_monitor_spec.rb
@@ -0,0 +1,206 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe Gitlab::SidekiqMonitor do
+  let(:monitor) { described_class.new }
+
+  describe '#within_job' do
+    it 'tracks thread' do
+      blk = proc do
+        expect(monitor.jobs_thread['jid']).not_to be_nil
+
+        "OK"
+      end
+
+      expect(monitor.within_job('jid', 'queue', &blk)).to eq("OK")
+    end
+
+    context 'when job is canceled' do
+      let(:jid) { SecureRandom.hex }
+
+      before do
+        described_class.cancel_job(jid)
+      end
+
+      it 'does not execute a block' do
+        expect do |blk|
+          monitor.within_job(jid, 'queue', &blk)
+        rescue described_class::CancelledError
+        end.not_to yield_control
+      end
+
+      it 'raises exception' do
+        expect { monitor.within_job(jid, 'queue') }.to raise_error(described_class::CancelledError)
+      end
+    end
+  end
+
+  describe '#start_working' do
+    subject { monitor.start_working }
+
+    context 'when structured logging is used' do
+      before do
+        allow_any_instance_of(::Redis).to receive(:subscribe)
+      end
+
+      it 'logs start message' do
+        expect(Sidekiq.logger).to receive(:info)
+          .with(
+            class: described_class,
+            action: 'start',
+            message: 'Starting Monitor Daemon')
+
+        subject
+      end
+
+      it 'logs stop message' do
+        expect(Sidekiq.logger).to receive(:warn)
+          .with(
+            class: described_class,
+            action: 'stop',
+            message: 'Stopping Monitor Daemon')
+
+        subject
+      end
+
+      it 'logs exception message' do
+        expect(Sidekiq.logger).to receive(:warn)
+          .with(
+            class: described_class,
+            action: 'exception',
+            message: 'My Exception')
+
+        expect(::Gitlab::Redis::SharedState).to receive(:with)
+          .and_raise(Exception, 'My Exception')
+
+        expect { subject }.to raise_error(Exception, 'My Exception')
+      end
+    end
+
+    context 'when message is published' do
+      let(:subscribed) { double }
+
+      before do
+        expect_any_instance_of(::Redis).to receive(:subscribe)
+          .and_yield(subscribed)
+
+        expect(subscribed).to receive(:message)
+          .and_yield(
+            described_class::NOTIFICATION_CHANNEL,
+            payload
+          )
+
+        expect(Sidekiq.logger).to receive(:info)
+          .with(
+            class: described_class,
+            action: 'start',
+            message: 'Starting Monitor Daemon')
+
+        expect(Sidekiq.logger).to receive(:info)
+          .with(
+            class: described_class,
+            channel: described_class::NOTIFICATION_CHANNEL,
+            message: 'Received payload on channel',
+            payload: payload
+          )
+      end
+
+      context 'and message is valid' do
+        let(:payload) { '{"action":"cancel","jid":"my-jid"}' }
+
+        it 'processes cancel' do
+          expect(monitor).to receive(:process_job_cancel).with('my-jid')
+
+          subject
+        end
+      end
+
+      context 'and message is not valid json' do
+        let(:payload) { '{"action"}' }
+
+        it 'skips processing' do
+          expect(monitor).not_to receive(:process_job_cancel)
+
+          subject
+        end
+      end
+    end
+  end
+
+  describe '#process_job_cancel' do
+    subject { monitor.send(:process_job_cancel, jid) }
+
+    context 'when jid is missing' do
+      let(:jid) { nil }
+
+      it 'does not run thread' do
+        expect(subject).to be_nil
+      end
+    end
+
+    context 'when jid is provided' do
+      let(:jid) { 'my-jid' }
+
+      context 'when jid is not found' do
+        it 'does not log cancellation message' do
+          expect(Sidekiq.logger).not_to receive(:warn)
+          expect(subject).to be_a(Thread)
+
+          subject.join
+        end
+      end
+
+      context 'when jid is found' do
+        let(:thread) { Thread.new { sleep 1000 } }
+
+        before do
+          monitor.jobs_thread[jid] = thread
+        end
+
+        it 'does log cancellation message' do
+          expect(Sidekiq.logger).to receive(:warn)
+            .with(
+              class: described_class,
+              action: 'cancel',
+              message: 'Canceling thread with CancelledError',
+              jid: 'my-jid',
+              thread_id: thread.object_id)
+
+          expect(subject).to be_a(Thread)
+
+          subject.join
+        end
+
+        it 'does cancel the thread' do
+          expect(subject).to be_a(Thread)
+
+          subject.join
+
+          expect(thread).not_to be_alive
+          expect { thread.value }.to raise_error(described_class::CancelledError)
+        end
+      end
+    end
+  end
+
+  describe '.cancel_job' do
+    subject { described_class.cancel_job('my-jid') }
+
+    it 'sets a redis key' do
+      expect_any_instance_of(::Redis).to receive(:setex)
+        .with('sidekiq:cancel:my-jid', anything, 1)
+
+      subject
+    end
+
+    it 'notifies all workers' do
+      payload = '{"action":"cancel","jid":"my-jid"}'
+
+      expect_any_instance_of(::Redis).to receive(:publish)
+        .with('sidekiq:cancel:notifications', payload)
+
+      subject
+    end
+  end
+end