Refactor and integrate Grafana annotations

* Introduce observer-like pattern.
* Refactor ReindexAction.keep_track_of and split

lib/gitlab/database/reindexing.rb

@@ -8,9 +8,9 @@ module Gitlab
 
       # candidate_indexes: Array of Gitlab::Database::PostgresIndex
       def self.perform(candidate_indexes, how_many: DEFAULT_INDEXES_PER_INVOCATION)
-        indexes = IndexSelection.new(candidate_indexes).take(how_many)
-
-        Coordinator.new(indexes).perform
+        IndexSelection.new(candidate_indexes).take(how_many).each do |index|
+          Coordinator.new(index).perform
+        end
       end
 
       def self.candidate_indexes

lib/gitlab/database/reindexing/coordinator.rb

@@ -12,26 +12,44 @@ module Gitlab
         # statement timeouts).
         TIMEOUT_PER_ACTION = 1.day
 
-        attr_reader :indexes
+        attr_reader :index, :notifier
 
-        def initialize(indexes)
-          @indexes = indexes
+        def initialize(index, notifier = GrafanaNotifier.new)
+          @index = index
+          @notifier = notifier
         end
 
         def perform
-          indexes.each do |index|
           # This obtains a global lease such that there's
           # only one live reindexing process at a time.
           try_obtain_lease do
-              ReindexAction.keep_track_of(index) do
-                ConcurrentReindex.new(index).perform
-              end
+            action = ReindexAction.create_for(index)
+
+            with_notifications(action) do
+              perform_for(index, action)
             end
           end
         end
 
         private
 
+        def with_notifications(action)
+          notifier.notify_start(action)
+          yield
+        ensure
+          notifier.notify_end(action)
+        end
+
+        def perform_for(index, action)
+          ConcurrentReindex.new(index).perform
+        rescue
+          action.state = :failed
+
+          raise
+        ensure
+          action.finish
+        end
+
         def lease_timeout
           TIMEOUT_PER_ACTION
         end

lib/gitlab/database/reindexing/grafana_notifier.rb

+# frozen_string_literal: true
+
+module Gitlab
+  module Database
+    module Reindexing
+      # This can be used to send annotations for reindexing to a Grafana API
+      class GrafanaNotifier
+        def initialize(api_key = ENV['GITLAB_GRAFANA_API_KEY'], api_url = ENV['GITLAB_GRAFANA_API_URL'])
+          @api_key = api_key
+          @api_url = api_url
+        end
+
+        def notify_start(action)
+          return unless enabled?
+
+          payload = base_payload(action).merge(
+            text: "Started reindexing of #{action.index.name} on #{action.index.tablename}"
+          )
+
+          annotate(payload)
+        end
+
+        def notify_end(action)
+          return unless enabled?
+
+          payload = base_payload(action).merge(
+            text: "Finished reindexing of #{action.index.name} on #{action.index.tablename} (#{action.state})",
+            timeEnd: (action.action_end.utc.to_f * 1000).to_i,
+            isRegion: true
+          )
+
+          annotate(payload)
+        end
+
+        private
+
+        def base_payload(action)
+          {
+            time: (action.action_start.utc.to_f * 1000).to_i,
+            tags: ['reindex', action.index.tablename, action.index.name]
+          }
+        end
+
+        def annotate(payload)
+          headers = {
+            "Content-Type": "application/json",
+            "Authorization": "Bearer #{@api_key}"
+          }
+
+          success = Gitlab::HTTP.post("#{@api_url}/api/annotations", body: payload.to_json, headers: headers, allow_local_requests: true).success?
+
+          log_error("Response code #{response.code}") unless success
+
+          success
+        rescue => err
+          log_error(err)
+
+          false
+        end
+
+        def log_error(err)
+          Gitlab::AppLogger.warn("Unable to notify Grafana from #{self.class}: #{err}")
+        end
+
+        def enabled?
+          !(@api_url.blank? || @api_key.blank?)
+        end
+      end
+    end
+  end
+end

lib/gitlab/database/reindexing/reindex_action.rb

@@ -14,27 +14,23 @@ module Gitlab
 
         scope :recent, -> { where(state: :finished).where('action_end > ?', Time.zone.now - RECENT_THRESHOLD) }
 
-        def self.keep_track_of(index, &block)
-          action = create!(
+        def self.create_for(index)
+          create!(
             index_identifier: index.identifier,
             action_start: Time.zone.now,
             ondisk_size_bytes_start: index.ondisk_size_bytes,
             bloat_estimate_bytes_start: index.bloat_size
           )
+        end
 
-          yield
-
-          action.state = :finished
-        rescue
-          action.state = :failed
-          raise
-        ensure
+        def finish
           index.reload # rubocop:disable Cop/ActiveRecordAssociationReload
 
-          action.action_end = Time.zone.now
-          action.ondisk_size_bytes_end = index.ondisk_size_bytes
+          self.state = :finished unless failed?
+          self.action_end = Time.zone.now
+          self.ondisk_size_bytes_end = index.ondisk_size_bytes
 
-          action.save!
+          save!
         end
       end
     end

spec/lib/gitlab/database/reindexing/coordinator_spec.rb

@@ -3,66 +3,80 @@
 require 'spec_helper'
 
 RSpec.describe Gitlab::Database::Reindexing::Coordinator do
+  include Database::DatabaseHelpers
   include ExclusiveLeaseHelpers
 
   describe '.perform' do
-    subject { described_class.new(indexes).perform }
+    subject { described_class.new(index, notifier).perform }
 
-    let(:indexes) { [instance_double(Gitlab::Database::PostgresIndex), instance_double(Gitlab::Database::PostgresIndex)] }
-    let(:reindexers) { [instance_double(Gitlab::Database::Reindexing::ConcurrentReindex), instance_double(Gitlab::Database::Reindexing::ConcurrentReindex)] }
+    before do
+      swapout_view_for_table(:postgres_indexes)
+
+      allow(Gitlab::Database::Reindexing::ConcurrentReindex).to receive(:new).with(index).and_return(reindexer)
+      allow(Gitlab::Database::Reindexing::ReindexAction).to receive(:create_for).with(index).and_return(action)
+    end
+
+    let(:index) { create(:postgres_index) }
+    let(:notifier) { instance_double(Gitlab::Database::Reindexing::GrafanaNotifier, notify_start: nil, notify_end: nil) }
+    let(:reindexer) { instance_double(Gitlab::Database::Reindexing::ConcurrentReindex, perform: nil) }
+    let(:action) { create(:reindex_action, index: index) }
 
     let!(:lease) { stub_exclusive_lease(lease_key, uuid, timeout: lease_timeout) }
     let(:lease_key) { 'gitlab/database/reindexing/coordinator' }
     let(:lease_timeout) { 1.day }
     let(:uuid) { 'uuid' }
 
-    before do
-      allow(Gitlab::Database::Reindexing::ReindexAction).to receive(:keep_track_of).and_yield
+    context 'locking' do
+      it 'acquires a lock while reindexing' do
+        expect(lease).to receive(:try_obtain).ordered.and_return(uuid)
 
-      indexes.zip(reindexers).each do |index, reindexer|
-        allow(Gitlab::Database::Reindexing::ConcurrentReindex).to receive(:new).with(index).and_return(reindexer)
-        allow(reindexer).to receive(:perform)
-      end
-    end
+        expect(reindexer).to receive(:perform).ordered
 
-    it 'performs concurrent reindexing for each index' do
-      indexes.zip(reindexers).each do |index, reindexer|
-        expect(Gitlab::Database::Reindexing::ConcurrentReindex).to receive(:new).with(index).ordered.and_return(reindexer)
-        expect(reindexer).to receive(:perform)
-      end
+        expect(Gitlab::ExclusiveLease).to receive(:cancel).ordered.with(lease_key, uuid)
 
         subject
       end
 
-    it 'keeps track of actions and creates ReindexAction records' do
-      indexes.each do |index|
-        expect(Gitlab::Database::Reindexing::ReindexAction).to receive(:keep_track_of).with(index).and_yield
-      end
+      it 'does not perform reindexing actions if lease is not granted' do
+        expect(lease).to receive(:try_obtain).ordered.and_return(false)
+        expect(Gitlab::Database::Reindexing::ConcurrentReindex).not_to receive(:new)
 
         subject
       end
-
-    context 'locking' do
-      it 'acquires a lock while reindexing' do
-        indexes.each do |index|
-          expect(lease).to receive(:try_obtain).ordered.and_return(uuid)
-          action = instance_double(Gitlab::Database::Reindexing::ConcurrentReindex)
-          expect(Gitlab::Database::Reindexing::ConcurrentReindex).to receive(:new).ordered.with(index).and_return(action)
-          expect(action).to receive(:perform).ordered
-          expect(Gitlab::ExclusiveLease).to receive(:cancel).ordered.with(lease_key, uuid)
     end
 
+    context 'notifications' do
+      it 'sends #notify_start before reindexing' do
+        expect(notifier).to receive(:notify_start).with(action).ordered
+        expect(reindexer).to receive(:perform).ordered
+
         subject
       end
 
-      it 'does does not perform reindexing actions if lease is not granted' do
-        indexes.each do |index|
-          expect(lease).to receive(:try_obtain).ordered.and_return(false)
-          expect(Gitlab::Database::Reindexing::ConcurrentReindex).not_to receive(:new)
+      it 'sends #notify_end after reindexing and updating the action is done' do
+        expect(action).to receive(:finish).ordered
+        expect(notifier).to receive(:notify_end).with(action).ordered
+
+        subject
       end
+    end
+
+    context 'action tracking' do
+      it 'calls #finish on the action' do
+        expect(reindexer).to receive(:perform).ordered
+        expect(action).to receive(:finish).ordered
 
         subject
       end
+
+      it 'upon error, it still calls finish and raises the error' do
+        expect(reindexer).to receive(:perform).ordered.and_raise('something went wrong')
+        expect(action).to receive(:finish).ordered
+
+        expect { subject }.to raise_error(/something went wrong/)
+
+        expect(action).to be_failed
+      end
     end
   end
 end

spec/lib/gitlab/database/reindexing/grafana_notifier_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Database::Reindexing::GrafanaNotifier do
+  include Database::DatabaseHelpers
+
+  let(:api_key) { "foo" }
+  let(:api_url) { "http://bar"}
+
+  let(:action) { create(:reindex_action) }
+
+  before do
+    swapout_view_for_table(:postgres_indexes)
+  end
+
+  let(:headers) do
+    {
+      'Content-Type': 'application/json',
+      'Authorization': "Bearer #{api_key}"
+    }
+  end
+
+  let(:response) { double('response', success?: true) }
+
+  def expect_api_call(payload)
+    expect(Gitlab::HTTP).to receive(:post).with("#{api_url}/api/annotations", body: payload.to_json, headers: headers, allow_local_requests: true).and_return(response)
+  end
+
+  shared_examples_for 'interacting with Grafana annotations API' do
+    it 'POSTs a JSON payload' do
+      expect_api_call(payload)
+
+      expect(subject).to be_truthy
+    end
+
+    context 'on error' do
+      it 'does not raise the error and returns false' do
+        allow(Gitlab::HTTP).to receive(:post).and_raise('something went wrong')
+
+        expect(subject).to be_falsey
+      end
+
+      context 'when request was not successful' do
+        it 'returns false' do
+          expect_api_call(payload)
+          allow(response).to receive(:success?).and_return(false)
+
+          expect(subject).to be_falsey
+        end
+      end
+    end
+
+    context 'without api_key' do
+      let(:api_key) { '' }
+
+      it 'does not post anything' do
+        expect(Gitlab::HTTP).not_to receive(:post)
+
+        expect(subject).to be_falsey
+      end
+    end
+
+    context 'without api_url' do
+      let(:api_url) { '' }
+
+      it 'does not post anything' do
+        expect(Gitlab::HTTP).not_to receive(:post)
+
+        expect(subject).to be_falsey
+      end
+    end
+  end
+
+  describe '#notify_start' do
+    subject { described_class.new(api_key, api_url).notify_start(action) }
+
+    let(:payload) do
+      {
+        time: (action.action_start.utc.to_f * 1000).to_i,
+        tags: ['reindex', action.index.tablename, action.index.name],
+        text: "Started reindexing of #{action.index.name} on #{action.index.tablename}"
+      }
+    end
+
+    it_behaves_like 'interacting with Grafana annotations API'
+  end
+
+  describe '#notify_end' do
+    subject { described_class.new(api_key, api_url).notify_end(action) }
+
+    let(:payload) do
+      {
+        time: (action.action_start.utc.to_f * 1000).to_i,
+        tags: ['reindex', action.index.tablename, action.index.name],
+        text: "Finished reindexing of #{action.index.name} on #{action.index.tablename} (#{action.state})",
+        timeEnd: (action.action_end.utc.to_f * 1000).to_i,
+        isRegion: true
+      }
+    end
+
+    it_behaves_like 'interacting with Grafana annotations API'
+  end
+end

spec/lib/gitlab/database/reindexing/reindex_action_spec.rb

@@ -2,91 +2,83 @@
 
 require 'spec_helper'
 
-RSpec.describe Gitlab::Database::Reindexing::ReindexAction, '.keep_track_of' do
-  let(:index) { double('index', identifier: 'public.something', ondisk_size_bytes: 10240, reload: nil, bloat_size: 42) }
-  let(:size_after) { 512 }
+RSpec.describe Gitlab::Database::Reindexing::ReindexAction do
+  include Database::DatabaseHelpers
 
-  it 'yields to the caller' do
-    expect { |b| described_class.keep_track_of(index, &b) }.to yield_control
-  end
+  let(:index) { create(:postgres_index) }
 
-  def find_record
-    described_class.find_by(index_identifier: index.identifier)
+  before_all do
+    swapout_view_for_table(:postgres_indexes)
   end
 
-  it 'creates the record with a start time and updates its end time' do
+  describe '.create_for' do
+    subject { described_class.create_for(index) }
+
+    it 'creates a new record for the given index' do
       freeze_time do
-      described_class.keep_track_of(index) do
-        expect(find_record.action_start).to be_within(1.second).of(Time.zone.now)
+        record = subject
+
+        expect(record.index_identifier).to eq(index.identifier)
+        expect(record.action_start).to eq(Time.zone.now)
+        expect(record.ondisk_size_bytes_start).to eq(index.ondisk_size_bytes)
+        expect(subject.bloat_estimate_bytes_start).to eq(index.bloat_size)
 
-        travel(10.seconds)
+        expect(record).to be_persisted
+      end
+    end
   end
 
-      duration = find_record.action_end - find_record.action_start
+  describe '#finish' do
+    subject { action.finish }
 
-      expect(duration).to be_within(1.second).of(10.seconds)
+    let(:action) { build(:reindex_action, index: index) }
+
+    it 'sets #action_end' do
+      freeze_time do
+        subject
+
+        expect(action.action_end).to eq(Time.zone.now)
       end
     end
 
-  it 'creates the record with its status set to :started and updates its state to :finished' do
-    described_class.keep_track_of(index) do
-      expect(find_record).to be_started
-    end
+    it 'sets #ondisk_size_bytes_end after reloading the index record' do
+      new_size = 4711
+      expect(action.index).to receive(:reload).ordered
+      expect(action.index).to receive(:ondisk_size_bytes).and_return(new_size).ordered
 
-    expect(find_record).to be_finished
+      subject
+
+      expect(action.ondisk_size_bytes_end).to eq(new_size)
     end
 
-  it 'creates the record with the indexes start size and updates its end size' do
-    described_class.keep_track_of(index) do
-      expect(find_record.ondisk_size_bytes_start).to eq(index.ondisk_size_bytes)
+    context 'setting #state' do
+      it 'sets #state to finished if not given' do
+        action.state = nil
 
-      expect(index).to receive(:reload).once
-      allow(index).to receive(:ondisk_size_bytes).and_return(size_after)
-    end
+        subject
 
-    expect(find_record.ondisk_size_bytes_end).to eq(size_after)
+        expect(action).to be_finished
       end
 
-  it 'creates the record with the indexes bloat estimate' do
-    described_class.keep_track_of(index) do
-      expect(find_record.bloat_estimate_bytes_start).to eq(index.bloat_size)
-    end
-  end
+      it 'sets #state to finished if not set to started' do
+        action.state = :started
 
-  context 'in case of errors' do
-    it 'sets the state to failed' do
-      expect do
-        described_class.keep_track_of(index) do
-          raise 'something went wrong'
-        end
-      end.to raise_error(/something went wrong/)
+        subject
 
-      expect(find_record).to be_failed
+        expect(action).to be_finished
       end
 
-    it 'records the end time' do
-      freeze_time do
-        expect do
-          described_class.keep_track_of(index) do
-            raise 'something went wrong'
-          end
-        end.to raise_error(/something went wrong/)
+      it 'does not change state if set to failed' do
+        action.state = :failed
 
-        expect(find_record.action_end).to be_within(1.second).of(Time.zone.now)
+        expect { subject }.not_to change { action.state }
       end
     end
 
-    it 'records the resulting index size' do
-      expect(index).to receive(:reload).once
-      allow(index).to receive(:ondisk_size_bytes).and_return(size_after)
-
-      expect do
-        described_class.keep_track_of(index) do
-          raise 'something went wrong'
-        end
-      end.to raise_error(/something went wrong/)
+    it 'saves the record' do
+      expect(action).to receive(:save!)
 
-      expect(find_record.ondisk_size_bytes_end).to eq(size_after)
+      subject
     end
   end
 end

spec/lib/gitlab/database/reindexing_spec.rb

@@ -11,13 +11,16 @@ RSpec.describe Gitlab::Database::Reindexing do
     let(:coordinator) { instance_double(Gitlab::Database::Reindexing::Coordinator) }
     let(:index_selection) { instance_double(Gitlab::Database::Reindexing::IndexSelection) }
     let(:candidate_indexes) { double }
-    let(:indexes) { double }
+    let(:indexes) { [double, double] }
 
     it 'delegates to Coordinator' do
       expect(Gitlab::Database::Reindexing::IndexSelection).to receive(:new).with(candidate_indexes).and_return(index_selection)
       expect(index_selection).to receive(:take).with(2).and_return(indexes)
-      expect(Gitlab::Database::Reindexing::Coordinator).to receive(:new).with(indexes).and_return(coordinator)
+
+      indexes.each do |index|
+        expect(Gitlab::Database::Reindexing::Coordinator).to receive(:new).with(index).and_return(coordinator)
         expect(coordinator).to receive(:perform)
+      end
 
       subject
     end