Merge branch '209831-propagate-integrations-using-batching-and-queues' into 'master'

Propagate integrations using batching and queues

Closes #249564

See merge request gitlab-org/gitlab!42128

app/models/concerns/integration.rb

@@ -16,7 +16,7 @@ module Integration
       Project.where(id: custom_integration_project_ids)
     end
 
-    def ids_without_integration(integration, limit)
+    def without_integration(integration)
       services = Service
         .select('1')
         .where('services.project_id = projects.id')
@@ -26,8 +26,6 @@ module Integration
         .where('NOT EXISTS (?)', services)
         .where(pending_delete: false)
         .where(archived: false)
-        .limit(limit)
-        .pluck(:id)
     end
   end
 end

app/models/data_list.rb

 # frozen_string_literal: true
 
 class DataList
-  def initialize(batch_ids, data_fields_hash, klass)
-    @batch_ids = batch_ids
+  def initialize(batch, data_fields_hash, klass)
+    @batch = batch
     @data_fields_hash = data_fields_hash
     @klass = klass
   end
@@ -13,15 +13,15 @@ class DataList
 
   private
 
-  attr_reader :batch_ids, :data_fields_hash, :klass
+  attr_reader :batch, :data_fields_hash, :klass
 
   def columns
     data_fields_hash.keys << 'service_id'
   end
 
   def values
-    batch_ids.map do |row|
-      data_fields_hash.values << row['id']
+    batch.map do |record|
+      data_fields_hash.values << record['id']
     end
   end
 end

app/models/group.rb

@@ -15,6 +15,7 @@ class Group < Namespace
   include WithUploads
   include Gitlab::Utils::StrongMemoize
   include GroupAPICompatibility
+  include EachBatch
 
   ACCESS_REQUEST_APPROVERS_TO_BE_NOTIFIED_LIMIT = 10
 
@@ -140,6 +141,15 @@ class Group < Namespace
       end
     end
 
+    def without_integration(integration)
+      services = Service
+        .select('1')
+        .where('services.group_id = namespaces.id')
+        .where(type: integration.type)
+
+      where('NOT EXISTS (?)', services)
+    end
+
     private
 
     def public_to_user_arel(user)

app/models/project.rb

@@ -33,6 +33,7 @@ class Project < ApplicationRecord
   include FromUnion
   include IgnorableColumns
   include Integration
+  include EachBatch
   extend Gitlab::Cache::RequestCache
 
   extend Gitlab::ConfigHelper

app/models/service.rb

@@ -63,6 +63,7 @@ class Service < ApplicationRecord
   scope :active, -> { where(active: true) }
   scope :by_type, -> (type) { where(type: type) }
   scope :by_active_flag, -> (flag) { where(active: flag) }
+  scope :inherit_from_id, -> (id) { where(inherit_from_id: id) }
   scope :for_group, -> (group) { where(group_id: group, type: available_services_types) }
   scope :for_template, -> { where(template: true, type: available_services_types) }
   scope :for_instance, -> { where(instance: true, type: available_services_types) }

app/models/service_list.rb

 # frozen_string_literal: true
 
 class ServiceList
-  def initialize(batch_ids, service_hash, association)
-    @batch_ids = batch_ids
+  def initialize(batch, service_hash, association)
+    @batch = batch
     @service_hash = service_hash
     @association = association
   end
@@ -13,15 +13,15 @@ class ServiceList
 
   private
 
-  attr_reader :batch_ids, :service_hash, :association
+  attr_reader :batch, :service_hash, :association
 
   def columns
-    (service_hash.keys << "#{association}_id")
+    service_hash.keys << "#{association}_id"
   end
 
   def values
-    batch_ids.map do |id|
-      (service_hash.values << id)
+    batch.select(:id).map do |record|
+      service_hash.values << record.id
     end
   end
 end

app/services/admin/propagate_integration_service.rb

@@ -14,59 +14,19 @@ module Admin
     private
 
     # rubocop: disable Cop/InBatches
-    # rubocop: disable CodeReuse/ActiveRecord
     def update_inherited_integrations
-      Service.where(type: integration.type, inherit_from_id: integration.id).in_batches(of: BATCH_SIZE) do |batch|
-        bulk_update_from_integration(batch)
+      Service.by_type(integration.type).inherit_from_id(integration.id).in_batches(of: BATCH_SIZE) do |services|
+        min_id, max_id = services.pick("MIN(services.id), MAX(services.id)")
+        PropagateIntegrationInheritWorker.perform_async(integration.id, min_id, max_id)
       end
     end
     # rubocop: enable Cop/InBatches
-    # rubocop: enable CodeReuse/ActiveRecord
-
-    # rubocop: disable CodeReuse/ActiveRecord
-    def bulk_update_from_integration(batch)
-      # Retrieving the IDs instantiates the ActiveRecord relation (batch)
-      # into concrete models, otherwise update_all will clear the relation.
-      # https://stackoverflow.com/q/34811646/462015
-      batch_ids = batch.pluck(:id)
-
-      Service.transaction do
-        batch.update_all(service_hash)
-
-        if data_fields_present?
-          integration.data_fields.class.where(service_id: batch_ids).update_all(data_fields_hash)
-        end
-      end
-    end
-    # rubocop: enable CodeReuse/ActiveRecord
 
     def create_integration_for_groups_without_integration
-      loop do
-        batch = Group.uncached { group_ids_without_integration(integration, BATCH_SIZE) }
-
-        bulk_create_from_integration(batch, 'group') unless batch.empty?
-
-        break if batch.size < BATCH_SIZE
+      Group.without_integration(integration).each_batch(of: BATCH_SIZE) do |groups|
+        min_id, max_id = groups.pick("MIN(namespaces.id), MAX(namespaces.id)")
+        PropagateIntegrationGroupWorker.perform_async(integration.id, min_id, max_id)
       end
     end
-
-    def service_hash
-      @service_hash ||= integration.to_service_hash
-        .tap { |json| json['inherit_from_id'] = integration.id }
-    end
-
-    # rubocop:disable CodeReuse/ActiveRecord
-    def group_ids_without_integration(integration, limit)
-      services = Service
-        .select('1')
-        .where('services.group_id = namespaces.id')
-        .where(type: integration.type)
-
-      Group
-        .where('NOT EXISTS (?)', services)
-        .limit(limit)
-        .pluck(:id)
-    end
-    # rubocop:enable CodeReuse/ActiveRecord
   end
 end

app/services/admin/propagate_service_template.rb

@@ -9,11 +9,5 @@ module Admin
 
       create_integration_for_projects_without_integration
     end
-
-    private
-
-    def service_hash
-      @service_hash ||= integration.to_service_hash
-    end
   end
 end

app/services/bulk_create_integration_service.rb

+# frozen_string_literal: true
+
+class BulkCreateIntegrationService
+  def initialize(integration, batch, association)
+    @integration = integration
+    @batch = batch
+    @association = association
+  end
+
+  def execute
+    service_list = ServiceList.new(batch, service_hash, association).to_array
+
+    Service.transaction do
+      results = bulk_insert(*service_list)
+
+      if integration.data_fields_present?
+        data_list = DataList.new(results, data_fields_hash, integration.data_fields.class).to_array
+
+        bulk_insert(*data_list)
+      end
+
+      run_callbacks(batch) if association == 'project'
+    end
+  end
+
+  private
+
+  attr_reader :integration, :batch, :association
+
+  def bulk_insert(klass, columns, values_array)
+    items_to_insert = values_array.map { |array| Hash[columns.zip(array)] }
+
+    klass.insert_all(items_to_insert, returning: [:id])
+  end
+
+  # rubocop: disable CodeReuse/ActiveRecord
+  def run_callbacks(batch)
+    if integration.issue_tracker?
+      Project.where(id: batch.select(:id)).update_all(has_external_issue_tracker: true)
+    end
+
+    if integration.type == 'ExternalWikiService'
+      Project.where(id: batch.select(:id)).update_all(has_external_wiki: true)
+    end
+  end
+  # rubocop: enable CodeReuse/ActiveRecord
+
+  def service_hash
+    if integration.template?
+      integration.to_service_hash
+    else
+      integration.to_service_hash.tap { |json| json['inherit_from_id'] = integration.id }
+    end
+  end
+
+  def data_fields_hash
+    integration.to_data_fields_hash
+  end
+end

app/services/bulk_update_integration_service.rb

+# frozen_string_literal: true
+
+class BulkUpdateIntegrationService
+  def initialize(integration, batch)
+    @integration = integration
+    @batch = batch
+  end
+
+  # rubocop: disable CodeReuse/ActiveRecord
+  def execute
+    Service.transaction do
+      batch.update_all(service_hash)
+
+      if integration.data_fields_present?
+        integration.data_fields.class.where(service_id: batch.select(:id)).update_all(data_fields_hash)
+      end
+    end
+  end
+  # rubocop: enable CodeReuse/ActiveRecord
+
+  private
+
+  attr_reader :integration, :batch
+
+  def service_hash
+    integration.to_service_hash.tap { |json| json['inherit_from_id'] = integration.id }
+  end
+
+  def data_fields_hash
+    integration.to_data_fields_hash
+  end
+end

app/services/concerns/admin/propagate_service.rb

@@ -4,9 +4,7 @@ module Admin
   module PropagateService
     extend ActiveSupport::Concern
 
-    BATCH_SIZE = 100
-
-    delegate :data_fields_present?, to: :integration
+    BATCH_SIZE = 10_000
 
     class_methods do
       def propagate(integration)
@@ -23,51 +21,10 @@ module Admin
     attr_reader :integration
 
     def create_integration_for_projects_without_integration
-      loop do
-        batch_ids = Project.uncached { Project.ids_without_integration(integration, BATCH_SIZE) }
-
-        bulk_create_from_integration(batch_ids, 'project') unless batch_ids.empty?
-
-        break if batch_ids.size < BATCH_SIZE
-      end
-    end
-
-    def bulk_create_from_integration(batch_ids, association)
-      service_list = ServiceList.new(batch_ids, service_hash, association).to_array
-
-      Service.transaction do
-        results = bulk_insert(*service_list)
-
-        if data_fields_present?
-          data_list = DataList.new(results, data_fields_hash, integration.data_fields.class).to_array
-
-          bulk_insert(*data_list)
-        end
-
-        run_callbacks(batch_ids) if association == 'project'
+      Project.without_integration(integration).each_batch(of: BATCH_SIZE) do |projects|
+        min_id, max_id = projects.pick("MIN(projects.id), MAX(projects.id)")
+        PropagateIntegrationProjectWorker.perform_async(integration.id, min_id, max_id)
       end
     end
-
-    def bulk_insert(klass, columns, values_array)
-      items_to_insert = values_array.map { |array| Hash[columns.zip(array)] }
-
-      klass.insert_all(items_to_insert, returning: [:id])
-    end
-
-    # rubocop: disable CodeReuse/ActiveRecord
-    def run_callbacks(batch_ids)
-      if integration.issue_tracker?
-        Project.where(id: batch_ids).update_all(has_external_issue_tracker: true)
-      end
-
-      if integration.type == 'ExternalWikiService'
-        Project.where(id: batch_ids).update_all(has_external_wiki: true)
-      end
-    end
-    # rubocop: enable CodeReuse/ActiveRecord
-
-    def data_fields_hash
-      @data_fields_hash ||= integration.to_data_fields_hash
-    end
   end
 end

app/workers/all_queues.yml

@@ -1724,6 +1724,30 @@
   :weight: 1
   :idempotent: true
   :tags: []
+- :name: propagate_integration_group
+  :feature_category: :integrations
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
+- :name: propagate_integration_inherit
+  :feature_category: :integrations
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
+- :name: propagate_integration_project
+  :feature_category: :integrations
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
 - :name: propagate_service_template
   :feature_category: :integrations
   :has_external_dependencies:

app/workers/propagate_integration_group_worker.rb

+# frozen_string_literal: true
+
+class PropagateIntegrationGroupWorker
+  include ApplicationWorker
+
+  feature_category :integrations
+  idempotent!
+
+  # rubocop: disable CodeReuse/ActiveRecord
+  def perform(integration_id, min_id, max_id)
+    integration = Service.find_by_id(integration_id)
+    return unless integration
+
+    batch = Group.where(id: min_id..max_id).without_integration(integration)
+
+    BulkCreateIntegrationService.new(integration, batch, 'group').execute
+  end
+  # rubocop: enable CodeReuse/ActiveRecord
+end

app/workers/propagate_integration_inherit_worker.rb

+# frozen_string_literal: true
+
+class PropagateIntegrationInheritWorker
+  include ApplicationWorker
+
+  feature_category :integrations
+  idempotent!
+
+  # rubocop: disable CodeReuse/ActiveRecord
+  def perform(integration_id, min_id, max_id)
+    integration = Service.find_by_id(integration_id)
+    return unless integration
+
+    services = Service.where(id: min_id..max_id).by_type(integration.type).inherit_from_id(integration.id)
+
+    BulkUpdateIntegrationService.new(integration, services).execute
+  end
+  # rubocop: enable CodeReuse/ActiveRecord
+end

app/workers/propagate_integration_project_worker.rb

+# frozen_string_literal: true
+
+class PropagateIntegrationProjectWorker
+  include ApplicationWorker
+
+  feature_category :integrations
+  idempotent!
+
+  # rubocop: disable CodeReuse/ActiveRecord
+  def perform(integration_id, min_id, max_id)
+    integration = Service.find_by_id(integration_id)
+    return unless integration
+
+    batch = Project.where(id: min_id..max_id).without_integration(integration)
+
+    BulkCreateIntegrationService.new(integration, batch, 'project').execute
+  end
+  # rubocop: enable CodeReuse/ActiveRecord
+end

changelogs/unreleased/209831-propagate-integrations-using-batching-and-queues.yml

+---
+title: Update database index on namespaces for type and id
+merge_request: 42128
+author:
+type: other

config/sidekiq_queues.yml

@@ -234,6 +234,12 @@
   - 1
 - - propagate_integration
   - 1
+- - propagate_integration_group
+  - 1
+- - propagate_integration_inherit
+  - 1
+- - propagate_integration_project
+  - 1
 - - propagate_service_template
   - 1
 - - reactive_caching

db/post_migrate/20200917165525_update_index_on_namespaces_for_type_and_id.rb

+# frozen_string_literal: true
+
+class UpdateIndexOnNamespacesForTypeAndId < ActiveRecord::Migration[6.0]
+  include Gitlab::Database::MigrationHelpers
+
+  DOWNTIME = false
+  disable_ddl_transaction!
+
+  OLD_INDEX_NAME = 'index_namespaces_on_type_partial'
+  NEW_INDEX_NAME = 'index_namespaces_on_type_and_id_partial'
+
+  def up
+    add_concurrent_index(:namespaces, [:type, :id], where: 'type IS NOT NULL', name: NEW_INDEX_NAME)
+
+    remove_concurrent_index_by_name(:namespaces, OLD_INDEX_NAME)
+  end
+
+  def down
+    add_concurrent_index(:namespaces, :type, where: 'type IS NOT NULL', name: OLD_INDEX_NAME)
+
+    remove_concurrent_index_by_name(:namespaces, NEW_INDEX_NAME)
+  end
+end

db/schema_migrations/20200917165525

+0080b9192ba5b4ea3853cfd930d58e10b9619f3d9a54016b574712e5ec2084f6
\ No newline at end of file

db/structure.sql

@@ -20600,7 +20600,7 @@ CREATE UNIQUE INDEX index_namespaces_on_runners_token_encrypted ON namespaces US
 
 CREATE INDEX index_namespaces_on_shared_and_extra_runners_minutes_limit ON namespaces USING btree (shared_runners_minutes_limit, extra_shared_runners_minutes_limit);
 
-CREATE INDEX index_namespaces_on_type_partial ON namespaces USING btree (type) WHERE (type IS NOT NULL);
+CREATE INDEX index_namespaces_on_type_and_id_partial ON namespaces USING btree (type, id) WHERE (type IS NOT NULL);
 
 CREATE INDEX index_non_requested_project_members_on_source_id_and_type ON members USING btree (source_id, source_type) WHERE ((requested_at IS NULL) AND ((type)::text = 'ProjectMember'::text));
 

spec/models/group_spec.rb

@@ -224,6 +224,20 @@ RSpec.describe Group do
     end
   end
 
+  describe '.without_integration' do
+    let(:another_group) { create(:group) }
+    let(:instance_integration) { build(:jira_service, :instance) }
+
+    before do
+      create(:jira_service, group: group, project: nil)
+      create(:slack_service, group: another_group, project: nil)
+    end
+
+    it 'returns groups without integration' do
+      expect(Group.without_integration(instance_integration)).to contain_exactly(another_group)
+    end
+  end
+
   describe '.public_or_visible_to_user' do
     let!(:private_group)  { create(:group, :private)  }
     let!(:internal_group) { create(:group, :internal) }

spec/models/integration_spec.rb

@@ -11,18 +11,18 @@ RSpec.describe Integration do
   before do
     create(:jira_service, project: project_1, inherit_from_id: instance_integration.id)
     create(:jira_service, project: project_2, inherit_from_id: nil)
-    create(:slack_service, project: project_1, inherit_from_id: nil)
+    create(:slack_service, project: project_3, inherit_from_id: nil)
   end
 
-  describe '#with_custom_integration_for' do
+  describe '.with_custom_integration_for' do
     it 'returns projects with custom integrations' do
       expect(Project.with_custom_integration_for(instance_integration)).to contain_exactly(project_2)
     end
   end
 
-  describe '#ids_without_integration' do
-    it 'returns projects ids without an integration' do
-      expect(Project.ids_without_integration(instance_integration, 100)).to contain_exactly(project_3.id)
+  describe '.without_integration' do
+    it 'returns projects without integration' do
+      expect(Project.without_integration(instance_integration)).to contain_exactly(project_3)
     end
   end
 end

spec/services/admin/propagate_integration_service_spec.rb

@@ -10,9 +10,7 @@ RSpec.describe Admin::PropagateIntegrationService do
       stub_jira_service_test
     end
 
-    let(:excluded_attributes) { %w[id project_id group_id inherit_from_id instance created_at updated_at default] }
-    let!(:project) { create(:project) }
-    let!(:group) { create(:group) }
+    let_it_be(:project) { create(:project) }
     let!(:instance_integration) do
       JiraService.create!(
         instance: true,
@@ -39,7 +37,7 @@ RSpec.describe Admin::PropagateIntegrationService do
 
     let!(:not_inherited_integration) do
       JiraService.create!(
-        project: create(:project),
+        project: project,
         inherit_from_id: nil,
         instance: false,
         active: true,
@@ -52,7 +50,7 @@ RSpec.describe Admin::PropagateIntegrationService do
 
     let!(:different_type_inherited_integration) do
       BambooService.create!(
-        project: create(:project),
+        project: project,
         inherit_from_id: instance_integration.id,
         instance: false,
         active: true,
@@ -64,75 +62,37 @@ RSpec.describe Admin::PropagateIntegrationService do
       )
     end
 
-    shared_examples 'inherits settings from integration' do
-      it 'updates the inherited integrations' do
-        described_class.propagate(instance_integration)
+    context 'with inherited integration' do
+      let(:integration) { inherited_integration }
 
-        expect(integration.reload.inherit_from_id).to eq(instance_integration.id)
-        expect(integration.attributes.except(*excluded_attributes))
-          .to eq(instance_integration.attributes.except(*excluded_attributes))
-      end
+      it 'calls to PropagateIntegrationProjectWorker' do
+        expect(PropagateIntegrationInheritWorker).to receive(:perform_async)
+          .with(instance_integration.id, inherited_integration.id, inherited_integration.id)
 
-      context 'integration with data fields' do
-        let(:excluded_attributes) { %w[id service_id created_at updated_at] }
-
-        it 'updates the data fields from inherited integrations' do
-          described_class.propagate(instance_integration)
-
-          expect(integration.reload.data_fields.attributes.except(*excluded_attributes))
-            .to eq(instance_integration.data_fields.attributes.except(*excluded_attributes))
-        end
-      end
-    end
-
-    shared_examples 'does not inherit settings from integration' do
-      it 'does not update the not inherited integrations' do
         described_class.propagate(instance_integration)
-
-        expect(integration.reload.attributes.except(*excluded_attributes))
-          .not_to eq(instance_integration.attributes.except(*excluded_attributes))
       end
     end
 
-    context 'update only inherited integrations' do
-      it_behaves_like 'inherits settings from integration' do
-        let(:integration) { inherited_integration }
-      end
-
-      it_behaves_like 'does not inherit settings from integration' do
-        let(:integration) { not_inherited_integration }
-      end
-
-      it_behaves_like 'does not inherit settings from integration' do
-        let(:integration) { different_type_inherited_integration }
-      end
+    context 'with a project without integration' do
+      let!(:another_project) { create(:project) }
 
-      it_behaves_like 'inherits settings from integration' do
-        let(:integration) { project.jira_service }
-      end
+      it 'calls to PropagateIntegrationProjectWorker' do
+        expect(PropagateIntegrationProjectWorker).to receive(:perform_async)
+          .with(instance_integration.id, another_project.id, another_project.id)
 
-      it_behaves_like 'inherits settings from integration' do
-        let(:integration) { Service.find_by(group_id: group.id) }
+        described_class.propagate(instance_integration)
       end
     end
 
-    it 'updates project#has_external_issue_tracker for issue tracker services' do
-      described_class.propagate(instance_integration)
+    context 'with a group without integration' do
+      let!(:group) { create(:group) }
 
-      expect(project.reload.has_external_issue_tracker).to eq(true)
-    end
+      it 'calls to PropagateIntegrationProjectWorker' do
+        expect(PropagateIntegrationGroupWorker).to receive(:perform_async)
+          .with(instance_integration.id, group.id, group.id)
 
-    it 'updates project#has_external_wiki for external wiki services' do
-      instance_integration = ExternalWikiService.create!(
-        instance: true,
-        active: true,
-        push_events: false,
-        external_wiki_url: 'http://external-wiki-url.com'
-      )
-
-      described_class.propagate(instance_integration)
-
-      expect(project.reload.has_external_wiki).to eq(true)
+        described_class.propagate(instance_integration)
+      end
     end
   end
 end

spec/services/admin/propagate_service_template_spec.rb

@@ -4,6 +4,7 @@ require 'spec_helper'
 
 RSpec.describe Admin::PropagateServiceTemplate do
   describe '.propagate' do
+    let_it_be(:project) { create(:project) }
     let!(:service_template) do
       PushoverService.create!(
         template: true,
@@ -19,124 +20,40 @@ RSpec.describe Admin::PropagateServiceTemplate do
       )
     end
 
-    let!(:project) { create(:project) }
-    let(:excluded_attributes) { %w[id project_id template created_at updated_at default] }
-
-    it 'creates services for projects' do
-      expect(project.pushover_service).to be_nil
-
-      described_class.propagate(service_template)
-
-      expect(project.reload.pushover_service).to be_present
-    end
-
-    it 'creates services for a project that has another service' do
-      BambooService.create!(
-        active: true,
-        project: project,
-        properties: {
-          bamboo_url: 'http://gitlab.com',
-          username: 'mic',
-          password: 'password',
-          build_key: 'build'
-        }
-      )
-
-      expect(project.pushover_service).to be_nil
+    it 'calls to PropagateIntegrationProjectWorker' do
+      expect(PropagateIntegrationProjectWorker).to receive(:perform_async)
+        .with(service_template.id, project.id, project.id)
 
       described_class.propagate(service_template)
-
-      expect(project.reload.pushover_service).to be_present
     end
 
-    it 'does not create the service if it exists already' do
-      other_service = BambooService.create!(
-        template: true,
-        active: true,
-        properties: {
-          bamboo_url: 'http://gitlab.com',
-          username: 'mic',
-          password: 'password',
-          build_key: 'build'
-        }
-      )
-
-      Service.build_from_integration(service_template, project_id: project.id).save!
-      Service.build_from_integration(other_service, project_id: project.id).save!
-
-      expect { described_class.propagate(service_template) }
-        .not_to change { Service.count }
-    end
-
-    it 'creates the service containing the template attributes' do
-      described_class.propagate(service_template)
-
-      expect(project.pushover_service.properties).to eq(service_template.properties)
-
-      expect(project.pushover_service.attributes.except(*excluded_attributes))
-        .to eq(service_template.attributes.except(*excluded_attributes))
-    end
-
-    context 'service with data fields' do
-      include JiraServiceHelper
-
-      let(:service_template) do
-        stub_jira_service_test
-
-        JiraService.create!(
-          template: true,
+    context 'with a project that has another service' do
+      before do
+        BambooService.create!(
           active: true,
-          push_events: false,
-          url: 'http://jira.instance.com',
-          username: 'user',
-          password: 'secret'
+          project: project,
+          properties: {
+            bamboo_url: 'http://gitlab.com',
+            username: 'mic',
+            password: 'password',
+            build_key: 'build'
+          }
         )
       end
 
-      it 'creates the service containing the template attributes' do
-        described_class.propagate(service_template)
-
-        expect(project.jira_service.attributes.except(*excluded_attributes))
-          .to eq(service_template.attributes.except(*excluded_attributes))
-
-        excluded_attributes = %w[id service_id created_at updated_at]
-        expect(project.jira_service.data_fields.attributes.except(*excluded_attributes))
-          .to eq(service_template.data_fields.attributes.except(*excluded_attributes))
-      end
-    end
-
-    describe 'bulk update', :use_sql_query_cache do
-      let(:project_total) { 5 }
-
-      before do
-        stub_const('Admin::PropagateServiceTemplate::BATCH_SIZE', 3)
-
-        project_total.times { create(:project) }
+      it 'calls to PropagateIntegrationProjectWorker' do
+        expect(PropagateIntegrationProjectWorker).to receive(:perform_async)
+          .with(service_template.id, project.id, project.id)
 
         described_class.propagate(service_template)
       end
-
-      it 'creates services for all projects' do
-        expect(Service.all.reload.count).to eq(project_total + 2)
-      end
     end
 
-    describe 'external tracker' do
-      it 'updates the project external tracker' do
-        service_template.update!(category: 'issue_tracker')
-
-        expect { described_class.propagate(service_template) }
-          .to change { project.reload.has_external_issue_tracker }.to(true)
-      end
-    end
-
-    describe 'external wiki' do
-      it 'updates the project external tracker' do
-        service_template.update!(type: 'ExternalWikiService')
+    it 'does not create the service if it exists already' do
+      Service.build_from_integration(service_template, project_id: project.id).save!
 
-        expect { described_class.propagate(service_template) }
-          .to change { project.reload.has_external_wiki }.to(true)
-      end
+      expect { described_class.propagate(service_template) }
+        .not_to change { Service.count }
     end
   end
 end

spec/services/bulk_create_integration_service_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkCreateIntegrationService do
+  include JiraServiceHelper
+
+  before do
+    stub_jira_service_test
+  end
+
+  let(:excluded_attributes) { %w[id project_id group_id inherit_from_id instance template created_at updated_at] }
+  let!(:instance_integration) { create(:jira_service, :instance) }
+  let!(:template_integration) { create(:jira_service, :template) }
+
+  shared_examples 'creates integration from batch ids' do
+    it 'updates the inherited integrations' do
+      described_class.new(integration, batch, association).execute
+
+      expect(created_integration.attributes.except(*excluded_attributes))
+        .to eq(integration.attributes.except(*excluded_attributes))
+    end
+
+    context 'integration with data fields' do
+      let(:excluded_attributes) { %w[id service_id created_at updated_at] }
+
+      it 'updates the data fields from inherited integrations' do
+        described_class.new(integration, batch, association).execute
+
+        expect(created_integration.reload.data_fields.attributes.except(*excluded_attributes))
+          .to eq(integration.data_fields.attributes.except(*excluded_attributes))
+      end
+    end
+  end
+
+  shared_examples 'updates inherit_from_id' do
+    it 'updates inherit_from_id attributes' do
+      described_class.new(integration, batch, association).execute
+
+      expect(created_integration.reload.inherit_from_id).to eq(integration.id)
+    end
+  end
+
+  shared_examples 'runs project callbacks' do
+    it 'updates projects#has_external_issue_tracker for issue tracker services' do
+      described_class.new(integration, batch, association).execute
+
+      expect(project.reload.has_external_issue_tracker).to eq(true)
+    end
+
+    context 'with an external wiki integration' do
+      let(:integration) do
+        ExternalWikiService.create!(
+          instance: true,
+          active: true,
+          push_events: false,
+          external_wiki_url: 'http://external-wiki-url.com'
+        )
+      end
+
+      it 'updates projects#has_external_wiki for external wiki services' do
+        described_class.new(integration, batch, association).execute
+
+        expect(project.reload.has_external_wiki).to eq(true)
+      end
+    end
+  end
+
+  context 'with an instance-level integration' do
+    let(:integration) { instance_integration }
+
+    context 'with a project association' do
+      let!(:project) { create(:project) }
+      let(:created_integration) { project.jira_service }
+      let(:batch) { Project.all }
+      let(:association) { 'project' }
+
+      it_behaves_like 'creates integration from batch ids'
+      it_behaves_like 'updates inherit_from_id'
+      it_behaves_like 'runs project callbacks'
+    end
+
+    context 'with a group association' do
+      let!(:group) { create(:group) }
+      let(:created_integration) { Service.find_by(group: group) }
+      let(:batch) { Group.all }
+      let(:association) { 'group' }
+
+      it_behaves_like 'creates integration from batch ids'
+      it_behaves_like 'updates inherit_from_id'
+    end
+  end
+
+  context 'with a template integration' do
+    let(:integration) { template_integration }
+
+    context 'with a project association' do
+      let!(:project) { create(:project) }
+      let(:created_integration) { project.jira_service }
+      let(:batch) { Project.all }
+      let(:association) { 'project' }
+
+      it_behaves_like 'creates integration from batch ids'
+      it_behaves_like 'runs project callbacks'
+    end
+  end
+end

spec/services/bulk_update_integration_service_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkUpdateIntegrationService do
+  include JiraServiceHelper
+
+  before do
+    stub_jira_service_test
+  end
+
+  let(:excluded_attributes) { %w[id project_id group_id inherit_from_id instance template created_at updated_at] }
+  let!(:instance_integration) do
+    JiraService.create!(
+      instance: true,
+      active: true,
+      push_events: true,
+      url: 'http://update-jira.instance.com',
+      username: 'user',
+      password: 'secret'
+    )
+  end
+
+  let!(:integration) do
+    JiraService.create!(
+      project: create(:project),
+      inherit_from_id: instance_integration.id,
+      instance: false,
+      active: true,
+      push_events: false,
+      url: 'http://jira.instance.com',
+      username: 'user',
+      password: 'secret'
+    )
+  end
+
+  context 'with inherited integration' do
+    it 'updates the integration' do
+      described_class.new(instance_integration, Service.inherit_from_id(instance_integration.id)).execute
+
+      expect(integration.reload.inherit_from_id).to eq(instance_integration.id)
+      expect(integration.attributes.except(*excluded_attributes))
+        .to eq(instance_integration.attributes.except(*excluded_attributes))
+    end
+
+    context 'with integration with data fields' do
+      let(:excluded_attributes) { %w[id service_id created_at updated_at] }
+
+      it 'updates the data fields from the integration' do
+        described_class.new(instance_integration, Service.inherit_from_id(instance_integration.id)).execute
+
+        expect(integration.reload.data_fields.attributes.except(*excluded_attributes))
+          .to eq(instance_integration.data_fields.attributes.except(*excluded_attributes))
+      end
+    end
+  end
+end

spec/workers/propagate_integration_group_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe PropagateIntegrationGroupWorker do
+  describe '#perform' do
+    let_it_be(:group1) { create(:group) }
+    let_it_be(:group2) { create(:group) }
+    let_it_be(:integration) { create(:redmine_service, :instance) }
+
+    before do
+      allow(BulkCreateIntegrationService).to receive(:new)
+        .with(integration, match_array([group1, group2]), 'group')
+        .and_return(double(execute: nil))
+    end
+
+    it_behaves_like 'an idempotent worker' do
+      let(:job_args) { [integration.id, group1.id, group2.id] }
+
+      it 'calls to BulkCreateIntegrationService' do
+        expect(BulkCreateIntegrationService).to receive(:new)
+          .with(integration, match_array([group1, group2]), 'group')
+          .and_return(double(execute: nil))
+
+        subject
+      end
+    end
+
+    context 'with an invalid integration id' do
+      it 'returns without failure' do
+        expect(BulkCreateIntegrationService).not_to receive(:new)
+
+        subject.perform(0, group1.id, group2.id)
+      end
+    end
+  end
+end

spec/workers/propagate_integration_inherit_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe PropagateIntegrationInheritWorker do
+  describe '#perform' do
+    let_it_be(:integration) { create(:redmine_service, :instance) }
+    let_it_be(:integration1) { create(:redmine_service, inherit_from_id: integration.id) }
+    let_it_be(:integration2) { create(:bugzilla_service, inherit_from_id: integration.id) }
+    let_it_be(:integration3) { create(:redmine_service) }
+
+    before do
+      allow(BulkUpdateIntegrationService).to receive(:new)
+        .with(integration, match_array(integration1))
+        .and_return(double(execute: nil))
+    end
+
+    it_behaves_like 'an idempotent worker' do
+      let(:job_args) { [integration.id, integration1.id, integration3.id] }
+
+      it 'calls to BulkCreateIntegrationService' do
+        expect(BulkUpdateIntegrationService).to receive(:new)
+          .with(integration, match_array(integration1))
+          .and_return(double(execute: nil))
+
+        subject
+      end
+    end
+
+    context 'with an invalid integration id' do
+      it 'returns without failure' do
+        expect(BulkUpdateIntegrationService).not_to receive(:new)
+
+        subject.perform(0, integration1.id, integration3.id)
+      end
+    end
+  end
+end

spec/workers/propagate_integration_project_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe PropagateIntegrationProjectWorker do
+  describe '#perform' do
+    let_it_be(:project1) { create(:project) }
+    let_it_be(:project2) { create(:project) }
+    let_it_be(:integration) { create(:redmine_service, :instance) }
+
+    before do
+      allow(BulkCreateIntegrationService).to receive(:new)
+        .with(integration, match_array([project1, project2]), 'project')
+        .and_return(double(execute: nil))
+    end
+
+    it_behaves_like 'an idempotent worker' do
+      let(:job_args) { [integration.id, project1.id, project2.id] }
+
+      it 'calls to BulkCreateIntegrationService' do
+        expect(BulkCreateIntegrationService).to receive(:new)
+          .with(integration, match_array([project1, project2]), 'project')
+          .and_return(double(execute: nil))
+
+        subject
+      end
+    end
+
+    context 'with an invalid integration id' do
+      it 'returns without failure' do
+        expect(BulkCreateIntegrationService).not_to receive(:new)
+
+        subject.perform(0, project1.id, project2.id)
+      end
+    end
+  end
+end