Add limit to concurrent batch exports

Allows administrators to configure a maximum number of simultaneous
exports (default 25) to prevent exhausting resources.

Changelog: added

app/helpers/application_settings_helper.rb

@@ -502,6 +502,7 @@ def visible_attributes
       :invitation_flow_enforcement,
       :can_create_group,
       :bulk_import_concurrent_pipeline_batch_limit,
+      :concurrent_relation_batch_export_limit,
       :bulk_import_enabled,
       :bulk_import_max_download_file_size,
       :silent_admin_exports_enabled,

app/models/application_setting.rb

@@ -530,6 +530,7 @@ def self.kroki_formats_attributes
       :concurrent_bitbucket_import_jobs_limit,
       :concurrent_bitbucket_server_import_jobs_limit,
       :concurrent_github_import_jobs_limit,
+      :concurrent_relation_batch_export_limit,
       :container_registry_token_expire_delay,
       :housekeeping_optimize_repository_period,
       :inactive_projects_delete_after_months,
@@ -623,6 +624,7 @@ def self.kroki_formats_attributes
     concurrent_bitbucket_import_jobs_limit: [:integer, { default: 100 }],
     concurrent_bitbucket_server_import_jobs_limit: [:integer, { default: 100 }],
     concurrent_github_import_jobs_limit: [:integer, { default: 1000 }],
+    concurrent_relation_batch_export_limit: [:integer, { default: 8 }],
     downstream_pipeline_trigger_limit_per_project_user_sha: [:integer, { default: 0 }],
     group_api_limit: [:integer, { default: 400 }],
     group_invited_groups_api_limit: [:integer, { default: 60 }],

app/models/bulk_imports/export_batch.rb

@@ -5,14 +5,21 @@ class ExportBatch < ApplicationRecord
     self.table_name = 'bulk_import_export_batches'
 
     BATCH_SIZE = 1000
+    TIMEOUT_AFTER_START = 1.hour
+    IN_PROGRESS_STATES = %i[created started].freeze
+
+    scope :started_and_not_timed_out, -> { with_status(:started).where(updated_at: TIMEOUT_AFTER_START.ago...) }
 
     belongs_to :export, class_name: 'BulkImports::Export'
     has_one :upload, class_name: 'BulkImports::ExportUpload', foreign_key: :batch_id, inverse_of: :batch
 
     validates :batch_number, presence: true, uniqueness: { scope: :export_id }
 
-    state_machine :status, initial: :started do
-      state :started, value: 0
+    scope :in_progress, -> { with_status(IN_PROGRESS_STATES) }
+
+    state_machine :status, initial: :created do
+      state :created, value: 0
+      state :started, value: 2
       state :finished, value: 1
       state :failed, value: -1
 

app/services/bulk_imports/relation_batch_export_service.rb

@@ -17,6 +17,7 @@ def execute
       ensure_export_file_exists!
       compress_exported_relation
       upload_compressed_file
+      export.touch
 
       finish_batch!
     ensure

app/validators/json_schemas/application_setting_rate_limits.json

@@ -19,6 +19,11 @@
       "minimum": 1,
       "description": "Maximum number of simultaneous import jobs for GitHub importer"
     },
+    "concurrent_relation_batch_export_limit": {
+      "type": "integer",
+      "minimum": 1,
+      "description": "Maximum number of simultaneous batch export jobs to process"
+    },
     "create_organization_api_limit": {
       "type": "integer",
       "minimum": 0,

app/workers/bulk_imports/finish_batched_relation_export_worker.rb

@@ -42,7 +42,7 @@ def export_timeout?
     end
 
     def export_in_progress?
-      export.batches.any?(&:started?)
+      export.batches.in_progress.any?
     end
 
     def finish_export!

app/workers/bulk_imports/relation_batch_export_worker.rb

@@ -3,6 +3,9 @@
 module BulkImports
   class RelationBatchExportWorker
     include ApplicationWorker
+    include Gitlab::Utils::StrongMemoize
+
+    PERFORM_DELAY = 1.minute
 
     idempotent!
     data_consistency :always
@@ -23,10 +26,35 @@ def perform(user_id, batch_id)
       @user = User.find(user_id)
       @batch = BulkImports::ExportBatch.find(batch_id)
 
+      return re_enqueue_job(@user, @batch) if max_exports_already_running?
+
       log_extra_metadata_on_done(:relation, @batch.export.relation)
-      log_extra_metadata_on_done(:objects_count, @batch.objects_count)
+      log_extra_metadata_on_done(:batch_number, @batch.batch_number)
 
       RelationBatchExportService.new(@user, @batch).execute
+
+      log_extra_metadata_on_done(:objects_count, @batch.objects_count)
+    end
+
+    def max_exports_already_running?
+      BulkImports::ExportBatch.started_and_not_timed_out.limit(max_exports).count == max_exports
+    end
+
+    strong_memoize_attr def max_exports
+      ::Gitlab::CurrentSettings.concurrent_relation_batch_export_limit
+    end
+
+    def re_enqueue_job(user, batch)
+      reset_cache_timeout(batch)
+      log_extra_metadata_on_done(:re_enqueue, true)
+
+      self.class.perform_in(PERFORM_DELAY, user.id, batch.id)
+    end
+
+    def reset_cache_timeout(batch)
+      cache_service = BulkImports::BatchedRelationExportService
+      cache_key = cache_service.cache_key(batch.export_id, batch.id)
+      Gitlab::Cache::Import::Caching.expire(cache_key, cache_service::CACHE_DURATION.to_i)
     end
   end
 end

db/migrate/20241022175028_increase_gitlab_com_concurrent_relation_batch_export_limit.rb

+# frozen_string_literal: true
+
+class IncreaseGitlabComConcurrentRelationBatchExportLimit < Gitlab::Database::Migration[2.2]
+  milestone '17.6'
+
+  restrict_gitlab_migration gitlab_schema: :gitlab_main
+
+  class ApplicationSetting < MigrationRecord; end
+
+  def up
+    return unless Gitlab.com?
+
+    application_setting = ApplicationSetting.last
+    return if application_setting.nil?
+
+    application_setting.rate_limits['concurrent_relation_batch_export_limit'] = 10_000
+    application_setting.save!
+  end
+
+  def down
+    return unless Gitlab.com?
+
+    application_setting = ApplicationSetting.last
+    return if application_setting.nil?
+
+    application_setting.rate_limits.delete('concurrent_relation_batch_export_limit')
+    application_setting.save!
+  end
+end

db/schema_migrations/20241022175028

+bf1cc7f9d4dbc3f83919c59db50e1f8a2e01eaa9e8d52511e8f8302ba5374e38
\ No newline at end of file

doc/administration/settings/import_and_export_settings.md

@@ -213,6 +213,24 @@ To modify this setting:
 1. Expand **Import and export settings**.
 1. Set another value for **Maximum number of simultaneous import jobs** for the desired importer.
 
+## Maximum number of simultaneous batch export jobs
+
+> - [Introduced](https://gitlab.com/gitlab-org/gitlab/-/merge_requests/169122) in GitLab 17.6.
+
+Direct transfer exports can consume a significant amount of resources.
+To prevent using up the database or Sidekiq processes,
+administrators can configure the `concurrent_relation_batch_export_limit` setting.
+
+The default value is `8` jobs, which corresponds to a
+[reference architecture for up to 40 RPS or 2,000 users](../../administration/reference_architectures/2k_users.md).
+If you encounter `PG::QueryCanceled: ERROR: canceling statement due to statement timeout` errors
+or jobs getting interrupted due to Sidekiq memory limits, you might want to reduce this number.
+If you have enough resources, you can increase this number to process more concurrent export jobs.
+
+To modify this setting, send an API request to `/api/v4/application/settings`
+with `concurrent_relation_batch_export_limit`.
+For more information, see [application settings API](../../api/settings.md).
+
 ## Troubleshooting
 
 ## Error: `Help page documentation base url is blocked: execution expired`

doc/api/settings.md

@@ -147,6 +147,7 @@ Example response:
   "project_jobs_api_rate_limit": 600,
   "security_txt_content": null,
   "bulk_import_concurrent_pipeline_batch_limit": 25,
+  "concurrent_relation_batch_export_limit": 25,
   "concurrent_github_import_jobs_limit": 1000,
   "concurrent_bitbucket_import_jobs_limit": 100,
   "concurrent_bitbucket_server_import_jobs_limit": 100,
@@ -322,6 +323,7 @@ Example response:
   "project_jobs_api_rate_limit": 600,
   "security_txt_content": null,
   "bulk_import_concurrent_pipeline_batch_limit": 25,
+  "concurrent_relation_batch_export_limit": 25,
   "downstream_pipeline_trigger_limit_per_project_user_sha": 0,
   "concurrent_github_import_jobs_limit": 1000,
   "concurrent_bitbucket_import_jobs_limit": 100,
@@ -699,7 +701,8 @@ listed in the descriptions of the relevant settings.
 | `valid_runner_registrars`                | array of strings | no                                   | List of types which are allowed to register a GitLab Runner. Can be `[]`, `['group']`, `['project']` or `['group', 'project']`. |
 | `whats_new_variant`                      | string           | no                                   | What's new variant, possible values: `all_tiers`, `current_tier`, and `disabled`. |
 | `wiki_page_max_content_bytes`            | integer          | no                                   | Maximum wiki page content size in **bytes**. Default: 52428800 Bytes (50 MB). The minimum value is 1024 bytes. |
-| `bulk_import_concurrent_pipeline_batch_limit` | integer     | no                                   | Maximum simultaneous Direct Transfer batches to process. |
+| `bulk_import_concurrent_pipeline_batch_limit` | integer     | no                                   | Maximum simultaneous direct transfer batch exports to process. |
+| `concurrent_relation_batch_export_limit` | integer          | no                                   | Maximum number of simultaneous batch export jobs to process. [Introduced](https://gitlab.com/gitlab-org/gitlab/-/merge_requests/169122) in GitLab 17.6. |
 | `asciidoc_max_includes`                  | integer          | no                                   | Maximum limit of AsciiDoc include directives being processed in any one document. Default: 32. Maximum: 64. |
 | `duo_features_enabled`                   | boolean          | no                                   | Indicates whether GitLab Duo features are enabled for this instance. [Introduced](https://gitlab.com/gitlab-org/gitlab/-/merge_requests/144931) in GitLab 16.10. Self-managed, Premium and Ultimate only. |
 | `lock_duo_features_enabled`              | boolean          | no                                   | Indicates whether the GitLab Duo features enabled setting is enforced for all subgroups. [Introduced](https://gitlab.com/gitlab-org/gitlab/-/merge_requests/144931) in GitLab 16.10. Self-managed, Premium and Ultimate only. |

lib/api/settings.rb

@@ -214,7 +214,8 @@ def filter_attributes_using_license(attrs)
       optional :jira_connect_application_key, type: String, desc: "ID of the OAuth application used to authenticate with the GitLab for Jira Cloud app."
       optional :jira_connect_public_key_storage_enabled, type: Boolean, desc: 'Enable public key storage for the GitLab for Jira Cloud app.'
       optional :jira_connect_proxy_url, type: String, desc: "URL of the GitLab instance used as a proxy for the GitLab for Jira Cloud app."
-      optional :bulk_import_concurrent_pipeline_batch_limit, type: Integer, desc: 'Maximum simultaneous Direct Transfer pipeline batches to process'
+      optional :bulk_import_concurrent_pipeline_batch_limit, type: Integer, desc: 'Maximum simultaneous direct transfer batch exports to process.'
+      optional :concurrent_relation_batch_export_limit, type: Integer, desc: 'Maximum number of simultaneous batch export jobs to process.'
       optional :bulk_import_enabled, type: Boolean, desc: 'Enable migrating GitLab groups and projects by direct transfer'
       optional :bulk_import_max_download_file, type: Integer, desc: 'Maximum download file size in MB when importing from source GitLab instances by direct transfer'
       optional :concurrent_github_import_jobs_limit, type: Integer, desc: 'Github Importer maximum number of simultaneous import jobs'

spec/factories/bulk_import/export_batches.rb

@@ -9,7 +9,7 @@
     status { 0 }
     batch_number { 1 }
 
-    trait :started do
+    trait :created do
       status { 0 }
     end
 
@@ -17,6 +17,10 @@
       status { 1 }
     end
 
+    trait :started do
+      status { 2 }
+    end
+
     trait :failed do
       status { -1 }
     end

spec/migrations/increase_gitlab_com_concurrent_relation_batch_export_limit_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+require_migration!
+
+RSpec.describe IncreaseGitlabComConcurrentRelationBatchExportLimit, feature_category: :database do
+  let(:application_settings_table) { table(:application_settings) }
+
+  let!(:application_settings) { application_settings_table.create! }
+
+  context 'when Gitlab.com? is false' do
+    before do
+      allow(Gitlab).to receive(:com?).and_return(false)
+    end
+
+    it 'does not change the rate limit concurrent_relation_batch_export_limit' do
+      disable_migrations_output do
+        reversible_migration do |migration|
+          migration.before -> {
+            expect(application_settings.reload.rate_limits).not_to have_key('concurrent_relation_batch_export_limit')
+          }
+
+          migration.after -> {
+            expect(application_settings.reload.rate_limits).not_to have_key('concurrent_relation_batch_export_limit')
+          }
+        end
+      end
+    end
+  end
+
+  context 'when Gitlab.com? is true' do
+    before do
+      allow(Gitlab).to receive(:com?).and_return(true)
+    end
+
+    it 'sets the rate_limits concurrent_relation_batch_export_limit to 10k' do
+      disable_migrations_output do
+        reversible_migration do |migration|
+          migration.before -> {
+            expect(application_settings.reload.rate_limits).not_to have_key('concurrent_relation_batch_export_limit')
+          }
+
+          migration.after -> {
+            expect(application_settings.reload.rate_limits['concurrent_relation_batch_export_limit']).to eq(10_000)
+          }
+        end
+      end
+    end
+
+    context 'when there is no application setting' do
+      let!(:application_settings) { nil }
+
+      it 'does not fail' do
+        disable_migrations_output do
+          expect { migrate! }.not_to raise_exception
+        end
+      end
+    end
+  end
+end

spec/models/bulk_imports/export_batch_spec.rb

@@ -14,4 +14,31 @@
     it { is_expected.to validate_presence_of(:batch_number) }
     it { is_expected.to validate_uniqueness_of(:batch_number).scoped_to(:export_id) }
   end
+
+  describe 'scopes' do
+    describe '.in_progress' do
+      it 'returns only batches that are in progress' do
+        created = create(:bulk_import_export_batch, :created)
+        started = create(:bulk_import_export_batch, :started)
+        create(:bulk_import_export_batch, :finished)
+        create(:bulk_import_export_batch, :failed)
+
+        expect(described_class.in_progress).to contain_exactly(created, started)
+      end
+    end
+  end
+
+  describe '.started_and_not_timed_out' do
+    subject(:started_and_not_timed_out) { described_class.started_and_not_timed_out }
+
+    let_it_be(:recently_started_export_batch) { create(:bulk_import_export_batch, :started, updated_at: 1.minute.ago) }
+    let_it_be(:old_started_export_batch) { create(:bulk_import_export_batch, :started, updated_at: 2.hours.ago) }
+    let_it_be(:recently_finished_export_batch) do
+      create(:bulk_import_export_batch, :finished, updated_at: 1.minute.ago)
+    end
+
+    it 'returns records with status started, which were last updated less that 1 hour ago' do
+      is_expected.to contain_exactly(recently_started_export_batch)
+    end
+  end
 end

spec/services/bulk_imports/relation_batch_export_service_spec.rb

@@ -42,6 +42,10 @@
       service.execute
     end
 
+    it 'updates export updated_at so the timeout resets' do
+      expect { service.execute }.to change { export.reload.updated_at }
+    end
+
     context 'when relation is empty and there is nothing to export' do
       let_it_be(:export) { create(:bulk_import_export, :batched, project: project, relation: 'milestones') }
       let_it_be(:batch) { create(:bulk_import_export_batch, export: export) }

spec/workers/bulk_imports/finish_batched_relation_export_worker_spec.rb

@@ -40,10 +40,8 @@
         end
       end
 
-      context 'when export is in progress' do
+      shared_examples 'reenqueues itself' do
         it 'reenqueues itself' do
-          create(:bulk_import_export_batch, :started, export: export)
-
           expect(described_class).to receive(:perform_in).twice.with(described_class::REENQUEUE_DELAY, export.id)
 
           perform_multiple(job_args)
@@ -52,6 +50,22 @@
         end
       end
 
+      context 'when export has started' do
+        before do
+          create(:bulk_import_export_batch, :started, export: export)
+        end
+
+        it_behaves_like 'reenqueues itself'
+      end
+
+      context 'when export has been created' do
+        before do
+          create(:bulk_import_export_batch, :created, export: export)
+        end
+
+        it_behaves_like 'reenqueues itself'
+      end
+
       context 'when export timed out' do
         it 'marks export as failed' do
           expect(export.reload.failed?).to eq(false)

spec/workers/bulk_imports/relation_batch_export_worker_spec.rb

@@ -9,6 +9,8 @@
   let(:job_args) { [user.id, batch.id] }
 
   describe '#perform' do
+    subject(:perform) { described_class.new.perform(user.id, batch.id) }
+
     include_examples 'an idempotent worker' do
       it 'executes RelationBatchExportService' do
         service = instance_double(BulkImports::RelationBatchExportService)
@@ -22,6 +24,51 @@
         perform_multiple(job_args)
       end
     end
+
+    context 'when the max number of exports have already started' do
+      let_it_be(:existing_export) { create(:bulk_import_export_batch, :started) }
+
+      before do
+        stub_application_setting(concurrent_relation_batch_export_limit: 1)
+      end
+
+      it 'does not start the export and schedules it for later' do
+        expect(described_class).to receive(:perform_in).with(described_class::PERFORM_DELAY, user.id, batch.id)
+
+        expect(BulkImports::RelationBatchExportService).not_to receive(:new)
+
+        perform
+      end
+
+      it 'resets the expiration date for the cache key' do
+        cache_key = BulkImports::BatchedRelationExportService.cache_key(batch.export_id, batch.id)
+        Gitlab::Cache::Import::Caching.write(cache_key, "test", timeout: 1.hour.to_i)
+
+        perform
+
+        expires_in_seconds = Gitlab::Cache::Import::Caching.with_redis do |redis|
+          redis.ttl(Gitlab::Cache::Import::Caching.cache_key_for(cache_key))
+        end
+
+        expect(expires_in_seconds).to be_within(10).of(BulkImports::BatchedRelationExportService::CACHE_DURATION.to_i)
+      end
+
+      context 'when the export batch started longer ago than the timeout time' do
+        before do
+          existing_export.update!(updated_at: (BulkImports::ExportBatch::TIMEOUT_AFTER_START + 1.minute).ago)
+        end
+
+        it 'starts the export and does not schedule it for later' do
+          expect(described_class).not_to receive(:perform_in).with(described_class::PERFORM_DELAY, user.id, batch.id)
+
+          expect_next_instance_of(BulkImports::RelationBatchExportService) do |instance|
+            expect(instance).to receive(:execute)
+          end
+
+          perform
+        end
+      end
+    end
   end
 
   describe '.sidekiq_retries_exhausted' do