Merge branch '1129-geo-clear-cache' into 'master'

Geo: Fix cache clearing after repository update on secondary node

When replicating operations originated from `git`, we don't have the same entry point (`git_push_service.rb`) on the secondary node, and because of that we had to manually duplicate some code after the update happens on the secondary (Pull vs Push).

This MR will introduces the following changes for Geo (#76):

* cache backfilling by `ProjectCacheWorker`
   - The call will switch from normal to **Geo** specific cache building code (when in a secondary node). 
* triggers after_* repository hooks when necessary to clear cache after specific events (when in a secondary node).

Fixes #1129 

cc @marin @dewetblomerus @patricio 

See merge request !869

app/services/geo/schedule_repo_update_service.rb

 module Geo
   class ScheduleRepoUpdateService
-    attr_reader :id, :clone_url
+    attr_reader :id, :clone_url, :push_data
 
     def initialize(params)
       @id = params[:project_id]
       @clone_url = params[:project][:git_ssh_url]
+      @push_data = { 'type' => params[:object_kind], 'before' => params[:before],
+                     'after' => params[:newref], 'ref' => params[:ref] }
     end
 
     def execute
-      GeoRepositoryUpdateWorker.perform_async(@id, @clone_url)
+      GeoRepositoryUpdateWorker.perform_async(@id, @clone_url, @push_data)
     end
   end
 end

app/workers/geo_repository_update_worker.rb

@@ -5,16 +5,46 @@ class GeoRepositoryUpdateWorker
 
   attr_accessor :project
 
-  def perform(project_id, clone_url)
+  def perform(project_id, clone_url, push_data = nil)
     @project = Project.find(project_id)
+    @push_data = push_data
 
     fetch_repository(clone_url)
+    process_hooks if push_data # we should be compatible with old unprocessed data
   end
 
   private
 
   def fetch_repository(remote_url)
     @project.create_repository unless @project.repository_exists?
+    @project.repository.after_create if @project.empty_repo?
     @project.repository.fetch_geo_mirror(remote_url)
   end
+
+  def process_hooks
+    if @push_data['type'] == 'push'
+      branch = Gitlab::Git.ref_name(@push_data['ref'])
+      process_push(branch, @push_data['after'])
+    end
+  end
+
+  def process_push(branch, revision)
+    @project.repository.after_push_commit(branch, revision)
+
+    if push_remove_branch?
+      @project.repository.after_remove_branch
+    elsif push_to_new_branch?
+      @project.repository.after_create_branch
+    end
+
+    ProjectCacheWorker.perform_async(@project.id)
+  end
+
+  def push_remove_branch?
+    Gitlab::Git.branch_ref?(@push_data['ref']) && Gitlab::Git.blank_ref?(@push_data['after'])
+  end
+
+  def push_to_new_branch?
+    Gitlab::Git.branch_ref?(@push_data['ref']) && Gitlab::Git.blank_ref?(@push_data['before'])
+  end
 end

app/workers/project_cache_worker.rb

@@ -6,6 +6,7 @@
 class ProjectCacheWorker
   include Sidekiq::Worker
   include DedicatedSidekiqQueue
+  prepend EE::Workers::ProjectCacheWorker
 
   LEASE_TIMEOUT = 15.minutes.to_i
 

changelogs/unreleased-ee/1129-geo-clear-cache.yml

+---
+title: Fixed cache clearing on secondary Geo nodes
+merge_request: 869
+author: 

lib/ee/workers/project_cache_worker.rb

+module EE
+  module Workers
+    # Geo specific code for cache re-generation
+    #
+    # This module is intended to encapsulate EE-specific methods
+    # and be **prepended** in the `ProjectCacheWorker` class.
+    module ProjectCacheWorker
+      def update_caches(project_id)
+        if ::Gitlab::Geo.secondary?
+          update_geo_caches(project_id)
+        else
+          super
+        end
+      end
+
+      private
+
+      # Geo should only update Redis based cache, as data store in the database
+      # will be updated on primary and replicated to the secondaries.
+      def update_geo_caches(project_id)
+        project = Project.find(project_id)
+
+        return unless project.repository.exists?
+
+        if project.repository.root_ref
+          project.repository.build_cache
+        end
+      end
+    end
+  end
+end

spec/services/geo/schedule_repo_update_service_spec.rb

+require 'spec_helper'
+
+describe Geo::ScheduleRepoUpdateService, services: true do
+  include RepoHelpers
+
+  let(:user) { create :user }
+  let(:project) { create :project }
+
+  let(:blankrev) { Gitlab::Git::BLANK_SHA }
+  let(:oldrev) { sample_commit.parent_id }
+  let(:newrev) { sample_commit.id }
+  let(:ref) { 'refs/heads/master' }
+
+  let(:service) { execute_push_service(project, user, oldrev, newrev, ref) }
+
+  before do
+    project.team << [user, :master]
+  end
+
+  subject { described_class.new(service.push_data) }
+
+  context 'parsed push_data' do
+    it 'includes required params' do
+      expect(subject.push_data).to include('type', 'before', 'after', 'ref')
+    end
+  end
+
+  context '#execute' do
+    let(:push_data) { service.push_data }
+    let(:args) do
+      [
+        project.id,
+        push_data[:project][:git_ssh_url],
+        {
+          'type' => push_data[:object_kind],
+          'before' => push_data[:before],
+          'after' => push_data[:newref],
+          'ref' => push_data[:ref]
+        }
+      ]
+    end
+
+    it 'schedule update service' do
+      expect(GeoRepositoryUpdateWorker).to receive(:perform_async).with(*args)
+
+      subject.execute
+    end
+  end
+
+  def execute_push_service(project, user, oldrev, newrev, ref)
+    service = GitPushService.new(project, user, oldrev: oldrev, newrev: newrev, ref: ref)
+    service.execute
+    service
+  end
+end

spec/workers/geo_repository_update_worker_spec.rb

+require 'spec_helper'
+
+describe GeoRepositoryUpdateWorker do
+  include RepoHelpers
+
+  let(:user) { create :user }
+  let(:project) { create :project }
+
+  let(:blankrev) { Gitlab::Git::BLANK_SHA }
+  let(:oldrev) { sample_commit.parent_id }
+  let(:newrev) { sample_commit.id }
+  let(:ref) { 'refs/heads/master' }
+
+  let(:service) { execute_push_service(project, user, oldrev, newrev, ref) }
+  let(:push_data) { service.push_data }
+  let(:parsed_push_data) do
+    {
+      'type' => push_data[:object_kind],
+      'before' => push_data[:before],
+      'after' => push_data[:after],
+      'ref' => push_data[:ref]
+    }
+  end
+
+  let(:clone_url) { push_data[:project][:git_ssh_url] }
+  let(:performed) { subject.perform(project.id, clone_url, parsed_push_data) }
+
+  before do
+    project.team << [user, :master]
+    expect(Project).to receive(:find).at_least(:once).with(project.id) { project }
+  end
+
+  context 'when no repository' do
+    before do
+      allow(project.repository).to receive(:fetch_geo_mirror)
+      allow(project).to receive(:repository_exists?) { false }
+    end
+
+    it 'creates a new repository' do
+      expect(project).to receive(:create_repository)
+
+      performed
+    end
+
+    it 'executes after_create hook' do
+      expect(project.repository).to receive(:after_create)
+
+      performed
+    end
+  end
+
+  context 'when empty repository' do
+    before do
+      allow(project.repository).to receive(:fetch_geo_mirror)
+      allow(project).to receive(:empty_repo?) { true }
+    end
+
+    it 'executes after_create hook' do
+      expect(project.repository).to receive(:after_create).at_least(:once)
+
+      performed
+    end
+  end
+
+  context '#process_hooks' do
+    before { allow(subject).to receive(:fetch_repository) }
+
+    it 'calls if push_data is present' do
+      expect(subject).to receive(:process_hooks)
+
+      performed
+    end
+
+    context 'when no push_data is present' do
+      let(:parsed_push_data) { nil }
+
+      it 'skips process_hooks' do
+        expect(subject).not_to receive(:process_hooks)
+
+        performed
+      end
+    end
+  end
+
+  context '#process_push' do
+    before { allow(subject).to receive(:fetch_repository) }
+
+    it 'executes after_push_commit' do
+      expect(project.repository).to receive(:after_push_commit).at_least(:once).with('master', newrev)
+
+      performed
+    end
+
+    context 'when removing branch' do
+      it 'executes after_remove_branch' do
+        allow(subject).to receive(:push_remove_branch?) { true }
+        expect(project.repository).to receive(:after_remove_branch)
+
+        performed
+      end
+    end
+
+    context 'when updating a new branch' do
+      it 'executes after_create_branch' do
+        allow(subject).to receive(:push_to_new_branch?) { true }
+        expect(project.repository).to receive(:after_create_branch)
+
+        performed
+      end
+    end
+  end
+
+  def execute_push_service(project, user, oldrev, newrev, ref)
+    service = GitPushService.new(project, user, oldrev: oldrev, newrev: newrev, ref: ref)
+    service.execute
+    service
+  end
+end

spec/workers/project_cache_worker_spec.rb

@@ -39,6 +39,8 @@ describe ProjectCacheWorker do
         expect_any_instance_of(Project).to receive(:update_repository_size)
         expect_any_instance_of(Project).to receive(:update_commit_count)
 
+        expect_any_instance_of(Repository).to receive(:build_cache).and_call_original
+
         subject.perform(project.id)
       end
 
@@ -48,6 +50,21 @@ describe ProjectCacheWorker do
 
         subject.perform(project.id)
       end
+
+      context 'when in Geo secondary node' do
+        before do
+          allow(Gitlab::Geo).to receive(:secondary?) { true }
+        end
+
+        it 'updates only non database cache' do
+          expect_any_instance_of(Repository).to receive(:build_cache).and_call_original
+
+          expect_any_instance_of(Project).not_to receive(:update_repository_size)
+          expect_any_instance_of(Project).not_to receive(:update_commit_count)
+
+          subject.perform(project.id)
+        end
+      end
     end
 
     context 'when an exclusive lease can not be obtained' do