Merge branch '2384-geo-event-log' into 'master'

Add push events to Geo event log

Closes #2384

See merge request !1976

app/models/concerns/geo/model.rb

+module Geo
+  module Model
+    extend ActiveSupport::Concern
+
+    included do
+      def self.table_name_prefix
+        "geo_"
+      end
+    end
+  end
+end

app/models/geo/event_log.rb

+module Geo
+  class EventLog < ActiveRecord::Base
+    include Geo::Model
+
+    belongs_to :repository_updated_event,
+      class_name: 'Geo::RepositoryUpdatedEvent',
+      foreign_key: :repository_updated_event_id
+  end
+end

app/models/geo/repository_updated_event.rb

+module Geo
+  class RepositoryUpdatedEvent < ActiveRecord::Base
+    include Geo::Model
+
+    REPOSITORY = 0
+    WIKI       = 1
+
+    belongs_to :project
+
+    enum source: { repository: REPOSITORY, wiki: WIKI }
+
+    validates :project, presence: true
+  end
+end

app/services/ee/wiki_pages/base_service.rb

+module EE
+  module WikiPages
+    # BaseService EE mixin
+    #
+    # This module is intended to encapsulate EE-specific service logic
+    # and be included in the `WikiPages::BaseService` service
+    module BaseService
+      extend ActiveSupport::Concern
+
+      private
+
+      def execute_hooks(page, action = 'create')
+        super
+        process_wiki_repository_update
+      end
+
+      def process_wiki_repository_update
+        if ::Gitlab::Geo.primary?
+          # Create wiki repository updated event on Geo event log
+          ::Geo::RepositoryUpdatedEventStore.new(project, source: Geo::RepositoryUpdatedEvent::WIKI).create
+
+          # Triggers repository update on secondary nodes
+          ::Gitlab::Geo.notify_wiki_update(project)
+        end
+      end
+    end
+  end
+end

app/services/geo/repository_updated_event_store.rb

+module Geo
+  class RepositoryUpdatedEventStore
+    attr_reader :project, :source, :refs, :changes
+
+    def initialize(project, refs: [], changes: [], source: Geo::RepositoryUpdatedEvent::REPOSITORY)
+      @project = project
+      @refs    = refs
+      @changes = changes
+      @source  = source
+    end
+
+    def create
+      return unless Gitlab::Geo.primary?
+
+      Geo::EventLog.transaction do
+        event_log = Geo::EventLog.new
+        event_log.repository_updated_event = build_event
+        event_log.save!
+      end
+    rescue ActiveRecord::RecordInvalid
+      log("#{Geo::PushEvent.sources.key(source).humanize} updated event could not be created")
+    end
+
+    private
+
+    def build_event
+      Geo::RepositoryUpdatedEvent.new(
+        project: project,
+        source: source,
+        ref: ref,
+        branches_affected: branches_affected,
+        tags_affected: tags_affected,
+        new_branch: push_to_new_branch?,
+        remove_branch: push_remove_branch?
+      )
+    end
+
+    def ref
+      refs.first if refs.length == 1
+    end
+
+    def branches_affected
+      refs.count { |ref| Gitlab::Git.branch_ref?(ref) }
+    end
+
+    def tags_affected
+      refs.count { |ref| Gitlab::Git.tag_ref?(ref) }
+    end
+
+    def push_to_new_branch?
+      changes.any? { |change| Gitlab::Git.branch_ref?(change[:ref]) && Gitlab::Git.blank_ref?(change[:before]) }
+    end
+
+    def push_remove_branch?
+      changes.any? { |change| Gitlab::Git.branch_ref?(change[:ref]) && Gitlab::Git.blank_ref?(change[:after]) }
+    end
+
+    def log(message)
+      Rails.logger.info("#{self.class.name}: #{message} for project #{project.path_with_namespace} (#{project.id})")
+    end
+  end
+end

app/services/wiki_pages/base_service.rb

 module WikiPages
   class BaseService < ::BaseService
+    prepend EE::WikiPages::BaseService
+
     def hook_data(page, action)
       hook_data = {
         object_kind: page.class.name.underscore,

app/services/wiki_pages/destroy_service.rb

@@ -3,9 +3,6 @@ class DestroyService < WikiPages::BaseService
     def execute(page)
       if page&.delete
         execute_hooks(page, 'delete')
-
-        # Triggers repository update on secondary nodes when Geo is enabled
-        Gitlab::Geo.notify_wiki_update(project) if Gitlab::Geo.primary?
       end
       page

app/workers/ee/post_receive.rb

+module EE
+  # PostReceive EE mixin
+  #
+  # This module is intended to encapsulate EE-specific model logic
+  # and be prepended in the `PostReceive` worker
+  module PostReceive
+    extend ActiveSupport::Concern
+    extend ::Gitlab::CurrentSettings
+
+    private
+
+    def after_project_changes_hooks(post_received, user, refs, changes)
+      super
+
+      # Generate repository updated event on Geo event log when Geo is enabled
+      ::Geo::RepositoryUpdatedEventStore.new(post_received.project, refs: refs, changes: changes).create
+    end
+
+    def process_wiki_changes(post_received)
+      super
+
+      update_wiki_es_indexes(post_received)
+
+      if ::Gitlab::Geo.enabled?
+        # Create wiki repository updated event on Geo event log
+        ::Geo::RepositoryUpdatedEventStore.new(post_received.project, source: Geo::RepositoryUpdatedEvent::WIKI).create
+
+        # Triggers repository update on secondary nodes
+        ::Gitlab::Geo.notify_wiki_update(post_received.project)
+      end
+    end
+
+    def update_wiki_es_indexes(post_received)
+      return unless current_application_settings.elasticsearch_indexing?
+
+      post_received.project.wiki.index_blobs
+    end
+  end
+end

app/workers/post_receive.rb

 class PostReceive
   include Sidekiq::Worker
   include DedicatedSidekiqQueue
-  extend Gitlab::CurrentSettings
+  prepend EE::PostReceive
   def perform(project_identifier, identifier, changes)
     project, is_wiki = parse_project_identifier(project_identifier)
@@ -18,37 +18,18 @@ def perform(project_identifier, identifier, changes)
     post_received = Gitlab::GitPostReceive.new(project, identifier, changes)
     if is_wiki
-      update_wiki_es_indexes(post_received)
-
-      # Triggers repository update on secondary nodes when Geo is enabled
-      Gitlab::Geo.notify_wiki_update(post_received.project) if Gitlab::Geo.enabled?
+      process_wiki_changes(post_received)
     else
       process_project_changes(post_received)
-      process_repository_update(post_received)
     end
   end
-  def process_repository_update(post_received)
+  private
+
+  def process_project_changes(post_received)
     changes = []
     refs = Set.new
-    post_received.changes_refs do |oldrev, newrev, ref|
-      @user ||= post_received.identify(newrev)
-
-      unless @user
-        log("Triggered hook for non-existing user \"#{post_received.identifier}\"")
-        return false
-      end
-
-      changes << Gitlab::DataBuilder::Repository.single_change(oldrev, newrev, ref)
-      refs << ref
-    end
-
-    hook_data = Gitlab::DataBuilder::Repository.update(post_received.project, @user, changes, refs.to_a)
-    SystemHooksService.new.execute_hooks(hook_data, :repository_update_hooks)
-  end
-
-  def process_project_changes(post_received)
     post_received.changes_refs do |oldrev, newrev, ref|
       @user ||= post_received.identify(newrev)
@@ -62,16 +43,22 @@ def process_project_changes(post_received)
       elsif Gitlab::Git.branch_ref?(ref)
         GitPushService.new(post_received.project, @user, oldrev: oldrev, newrev: newrev, ref: ref).execute
       end
+
+      changes << Gitlab::DataBuilder::Repository.single_change(oldrev, newrev, ref)
+      refs << ref
     end
-  end
-  def update_wiki_es_indexes(post_received)
-    return unless current_application_settings.elasticsearch_indexing?
-    post_received.project.wiki.index_blobs
+    after_project_changes_hooks(post_received, @user, refs.to_a, changes)
+  end
+  def after_project_changes_hooks(post_received, user, refs, changes)
+    hook_data = Gitlab::DataBuilder::Repository.update(post_received.project, user, changes, refs)
+    SystemHooksService.new.execute_hooks(hook_data, :repository_update_hooks)
   end
-  private
+  def process_wiki_changes(post_received)
+    # Nothing defined here yet.
+  end
   # To maintain backwards compatibility, we accept both gl_repository or
   # repository paths as project identifiers. Our plan is to migrate to

changelogs/unreleased-ee/2384-geo-event-log.yml

+---
+title: Add push events to Geo event log
+merge_request:
+author:

config/initializers/0_inflections.rb

@@ -10,6 +10,6 @@
 # end
 #
 ActiveSupport::Inflector.inflections do |inflect|
-  inflect.uncountable %w(award_emoji project_statistics project_registry file_registry system_note_metadata)
+  inflect.uncountable %w(award_emoji project_statistics system_note_metadata event_log project_registry file_registry)
   inflect.acronym 'EE'
 end

config/initializers/ar_native_database_types.rb

+require 'active_record/connection_adapters/abstract_mysql_adapter'
+
+module ActiveRecord
+  module ConnectionAdapters
+    class AbstractMysqlAdapter
+      NATIVE_DATABASE_TYPES.merge!(
+        bigserial:  { name: 'bigint(20) auto_increment PRIMARY KEY' }
+      )
+    end
+  end
+end

db/migrate/20170522194603_create_geo_repository_updated_events.rb

+class CreateGeoRepositoryUpdatedEvents < ActiveRecord::Migration
+  DOWNTIME = false
+
+  def change
+    create_table :geo_repository_updated_events, id: :bigserial do |t|
+      t.datetime :created_at, null: false
+      t.integer :branches_affected, null: false
+      t.integer :tags_affected, null: false
+      t.references :project, index: true, foreign_key: { on_delete: :cascade }, null: false
+      t.integer :source, limit: 2, index: true, null: false
+      t.boolean :new_branch, default: false, null: false
+      t.boolean :remove_branch, default: false, null: false
+      t.text :ref
+    end
+  end
+end

db/migrate/20170522201800_create_geo_event_log.rb

+class CreateGeoEventLog < ActiveRecord::Migration
+  DOWNTIME = false
+
+  def change
+    create_table :geo_event_log, id: :bigserial do |t|
+      t.datetime :created_at, null: false
+      t.integer :repository_updated_event_id, limit: 8, index: true
+
+      t.foreign_key :geo_repository_updated_events,
+        column: :repository_updated_event_id, on_delete: :cascade
+    end
+  end
+end

db/schema.rb

@@ -535,6 +535,13 @@
   add_index "forked_project_links", ["forked_to_project_id"], name: "index_forked_project_links_on_forked_to_project_id", unique: true, using: :btree
+  create_table "geo_event_log", id: :bigserial, force: :cascade do |t|
+    t.datetime "created_at", null: false
+    t.integer "repository_updated_event_id", limit: 8
+  end
+
+  add_index "geo_event_log", ["repository_updated_event_id"], name: "index_geo_event_log_on_repository_updated_event_id", using: :btree
+
   create_table "geo_nodes", force: :cascade do |t|
     t.string "schema"
     t.string "host"
@@ -555,6 +562,20 @@
   add_index "geo_nodes", ["host"], name: "index_geo_nodes_on_host", using: :btree
   add_index "geo_nodes", ["primary"], name: "index_geo_nodes_on_primary", using: :btree
+  create_table "geo_repository_updated_events", id: :bigserial, force: :cascade do |t|
+    t.datetime "created_at", null: false
+    t.integer "branches_affected", null: false
+    t.integer "tags_affected", null: false
+    t.integer "project_id", null: false
+    t.integer "source", limit: 2, null: false
+    t.boolean "new_branch", default: false, null: false
+    t.boolean "remove_branch", default: false, null: false
+    t.text "ref"
+  end
+
+  add_index "geo_repository_updated_events", ["project_id"], name: "index_geo_repository_updated_events_on_project_id", using: :btree
+  add_index "geo_repository_updated_events", ["source"], name: "index_geo_repository_updated_events_on_source", using: :btree
+
   create_table "historical_data", force: :cascade do |t|
     t.date "date", null: false
     t.integer "active_user_count"
@@ -1718,6 +1739,8 @@
   add_foreign_key "ci_triggers", "users", column: "owner_id", name: "fk_e8e10d1964", on_delete: :cascade
   add_foreign_key "ci_variables", "projects", name: "fk_ada5eb64b3", on_delete: :cascade
   add_foreign_key "container_repositories", "projects"
+  add_foreign_key "geo_event_log", "geo_repository_updated_events", column: "repository_updated_event_id", on_delete: :cascade
+  add_foreign_key "geo_repository_updated_events", "projects", on_delete: :cascade
   add_foreign_key "issue_assignees", "issues", name: "fk_b7d881734a", on_delete: :cascade
   add_foreign_key "issue_assignees", "users", name: "fk_5e0c8d9154", on_delete: :cascade
   add_foreign_key "issue_links", "issues", column: "source_id", name: "fk_c900194ff2", on_delete: :cascade

spec/models/geo/event_log_spec.rb

+require 'spec_helper'
+
+RSpec.describe Geo::EventLog, type: :model do
+  describe 'relationships' do
+    it { is_expected.to belong_to(:repository_updated_event).class_name('Geo::RepositoryUpdatedEvent').with_foreign_key('repository_updated_event_id') }
+  end
+end

spec/models/geo/repository_updated_event_spec.rb

+require 'spec_helper'
+
+RSpec.describe Geo::RepositoryUpdatedEvent, type: :model do
+  describe 'relationships' do
+    it { is_expected.to belong_to(:project) }
+  end
+
+  describe 'validations' do
+    it { is_expected.to validate_presence_of(:project) }
+  end
+
+  describe '#source' do
+    it { is_expected.to define_enum_for(:source).with([:repository, :wiki]) }
+  end
+end

spec/services/ee/wiki_pages/create_service_spec.rb

+require 'spec_helper'
+
+describe WikiPages::CreateService, services: true do
+  let(:project) { create(:empty_project) }
+  let(:user)    { create(:user) }
+
+  let(:opts) do
+    {
+      title: 'Title',
+      content: 'Content for wiki page',
+      format: 'markdown'
+    }
+  end
+
+  subject(:service) { described_class.new(project, user, opts) }
+
+  before do
+    project.add_master(user)
+  end
+
+  describe '#execute' do
+    context 'when running on a Geo primary node' do
+      before do
+        allow(Gitlab::Geo).to receive(:primary?) { true }
+      end
+
+      it 'triggers Geo::RepositoryUpdatedEventStore when Geo is enabled' do
+        expect(Geo::RepositoryUpdatedEventStore).to receive(:new).with(instance_of(Project), source: Geo::RepositoryUpdatedEvent::WIKI).and_call_original
+        expect_any_instance_of(Geo::RepositoryUpdatedEventStore).to receive(:create)
+
+        service.execute
+      end
+
+      it 'triggers wiki update on secondary nodes' do
+        expect(Gitlab::Geo).to receive(:notify_wiki_update).with(instance_of(Project))
+
+        service.execute
+      end
+    end
+  end
+end

spec/services/ee/wiki_pages/destroy_service_spec.rb

+require 'spec_helper'
+
+describe WikiPages::DestroyService, services: true do
+  let(:project) { create(:empty_project) }
+  let(:user)    { create(:user) }
+  let(:page)    { create(:wiki_page) }
+
+  subject(:service) { described_class.new(project, user) }
+
+  before do
+    project.add_master(user)
+  end
+
+  describe '#execute' do
+    context 'when running on a Geo primary node' do
+      before do
+        allow(Gitlab::Geo).to receive(:primary?) { true }
+      end
+
+      it 'triggers Geo::RepositoryUpdatedEventStore when Geo is enabled' do
+        expect(Geo::RepositoryUpdatedEventStore).to receive(:new).with(instance_of(Project), source: Geo::RepositoryUpdatedEvent::WIKI).and_call_original
+        expect_any_instance_of(Geo::RepositoryUpdatedEventStore).to receive(:create)
+
+        service.execute(page)
+      end
+
+      it 'triggers wiki update on secondary nodes' do
+        expect(Gitlab::Geo).to receive(:notify_wiki_update).with(instance_of(Project))
+
+        service.execute(page)
+      end
+    end
+  end
+end

spec/services/ee/wiki_pages/update_service_spec.rb

+require 'spec_helper'
+
+describe WikiPages::UpdateService, services: true do
+  let(:project) { create(:empty_project) }
+  let(:user)    { create(:user) }
+  let(:page)    { create(:wiki_page) }
+
+  let(:opts) do
+    {
+      content: 'New content for wiki page',
+      format: 'markdown',
+      message: 'New wiki message'
+    }
+  end
+
+  subject(:service) { described_class.new(project, user, opts) }
+
+  before do
+    project.add_master(user)
+  end
+
+  describe '#execute' do
+    context 'when running on a Geo primary node' do
+      before do
+        allow(Gitlab::Geo).to receive(:primary?) { true }
+      end
+
+      it 'triggers Geo::RepositoryUpdatedEventStore when Geo is enabled' do
+        expect(Geo::RepositoryUpdatedEventStore).to receive(:new).with(instance_of(Project), source: Geo::RepositoryUpdatedEvent::WIKI).and_call_original
+        expect_any_instance_of(Geo::RepositoryUpdatedEventStore).to receive(:create)
+
+        service.execute(page)
+      end
+
+      it 'triggers wiki update on secondary nodes' do
+        expect(Gitlab::Geo).to receive(:notify_wiki_update).with(instance_of(Project))
+
+        service.execute(page)
+      end
+    end
+  end
+end