diff options
Diffstat (limited to 'app/workers')
30 files changed, 451 insertions, 131 deletions
diff --git a/app/workers/all_queues.yml b/app/workers/all_queues.yml index 8bba5e36b52..9b282340d0a 100644 --- a/app/workers/all_queues.yml +++ b/app/workers/all_queues.yml @@ -786,6 +786,15 @@ :weight: 1 :idempotent: false :tags: [] +- :name: cronjob:users_migrate_records_to_ghost_user_in_batches + :worker_name: Users::MigrateRecordsToGhostUserInBatchesWorker + :feature_category: :users + :has_external_dependencies: false + :urgency: :low + :resource_boundary: :unknown + :weight: 1 + :idempotent: true + :tags: [] - :name: cronjob:x509_issuer_crl_check :worker_name: X509IssuerCrlCheckWorker :feature_category: :source_code_management @@ -1056,6 +1065,15 @@ :weight: 1 :idempotent: false :tags: [] +- :name: github_importer:github_import_import_protected_branch + :worker_name: Gitlab::GithubImport::ImportProtectedBranchWorker + :feature_category: :importers + :has_external_dependencies: true + :urgency: :low + :resource_boundary: :cpu + :weight: 1 + :idempotent: false + :tags: [] - :name: github_importer:github_import_import_pull_request :worker_name: Gitlab::GithubImport::ImportPullRequestWorker :feature_category: :importers @@ -1083,6 +1101,15 @@ :weight: 1 :idempotent: false :tags: [] +- :name: github_importer:github_import_import_release_attachments + :worker_name: Gitlab::GithubImport::ImportReleaseAttachmentsWorker + :feature_category: :importers + :has_external_dependencies: true + :urgency: :low + :resource_boundary: :unknown + :weight: 1 + :idempotent: false + :tags: [] - :name: github_importer:github_import_refresh_import_jid :worker_name: Gitlab::GithubImport::RefreshImportJidWorker :feature_category: :importers @@ -1101,6 +1128,15 @@ :weight: 1 :idempotent: false :tags: [] +- :name: github_importer:github_import_stage_import_attachments + :worker_name: Gitlab::GithubImport::Stage::ImportAttachmentsWorker + :feature_category: :importers + :has_external_dependencies: false + :urgency: :low + :resource_boundary: :unknown + :weight: 1 + :idempotent: false + :tags: [] - :name: github_importer:github_import_stage_import_base_data :worker_name: Gitlab::GithubImport::Stage::ImportBaseDataWorker :feature_category: :importers @@ -1146,6 +1182,15 @@ :weight: 1 :idempotent: false :tags: [] +- :name: github_importer:github_import_stage_import_protected_branches + :worker_name: Gitlab::GithubImport::Stage::ImportProtectedBranchesWorker + :feature_category: :importers + :has_external_dependencies: false + :urgency: :low + :resource_boundary: :unknown + :weight: 1 + :idempotent: false + :tags: [] - :name: github_importer:github_import_stage_import_pull_requests :worker_name: Gitlab::GithubImport::Stage::ImportPullRequestsWorker :feature_category: :importers @@ -1578,6 +1623,15 @@ :weight: 1 :idempotent: true :tags: [] +- :name: pipeline_background:ci_job_artifacts_track_artifact_report + :worker_name: Ci::JobArtifacts::TrackArtifactReportWorker + :feature_category: :code_testing + :has_external_dependencies: false + :urgency: :low + :resource_boundary: :unknown + :weight: 1 + :idempotent: true + :tags: [] - :name: pipeline_background:ci_pending_builds_update_group :worker_name: Ci::PendingBuilds::UpdateGroupWorker :feature_category: :continuous_integration @@ -2379,6 +2433,15 @@ :weight: 1 :idempotent: true :tags: [] +- :name: google_cloud_fetch_google_ip_list + :worker_name: GoogleCloud::FetchGoogleIpListWorker + :feature_category: :build_artifacts + :has_external_dependencies: false + :urgency: :low + :resource_boundary: :unknown + :weight: 1 + :idempotent: true + :tags: [] - :name: group_destroy :worker_name: GroupDestroyWorker :feature_category: :subgroups @@ -2415,6 +2478,15 @@ :weight: 1 :idempotent: true :tags: [] +- :name: groups_update_two_factor_requirement_for_members + :worker_name: Groups::UpdateTwoFactorRequirementForMembersWorker + :feature_category: :authentication_and_authorization + :has_external_dependencies: false + :urgency: :low + :resource_boundary: :unknown + :weight: 1 + :idempotent: true + :tags: [] - :name: import_issues_csv :worker_name: ImportIssuesCsvWorker :feature_category: :team_planning @@ -2496,6 +2568,15 @@ :weight: 1 :idempotent: true :tags: [] +- :name: issues_close + :worker_name: Issues::CloseWorker + :feature_category: :source_code_management + :has_external_dependencies: false + :urgency: :high + :resource_boundary: :unknown + :weight: 2 + :idempotent: true + :tags: [] - :name: issues_placement :worker_name: Issues::PlacementWorker :feature_category: :team_planning diff --git a/app/workers/analytics/usage_trends/counter_job_worker.rb b/app/workers/analytics/usage_trends/counter_job_worker.rb index b3a8f7dd3c2..e6de623f784 100644 --- a/app/workers/analytics/usage_trends/counter_job_worker.rb +++ b/app/workers/analytics/usage_trends/counter_job_worker.rb @@ -3,6 +3,8 @@ module Analytics module UsageTrends class CounterJobWorker + TIMEOUT = 250.seconds + extend ::Gitlab::Utils::Override include ApplicationWorker @@ -15,24 +17,27 @@ module Analytics idempotent! - def perform(measurement_identifier, min_id, max_id, recorded_at) + def perform(measurement_identifier, min_id, max_id, recorded_at, partial_results = nil) query_scope = ::Analytics::UsageTrends::Measurement.identifier_query_mapping[measurement_identifier].call - count = if min_id.nil? || max_id.nil? # table is empty - 0 - else - counter(query_scope, min_id, max_id) - end + result = counter(query_scope, min_id, max_id, partial_results) + + # If the batch counter timed out, schedule a job to continue counting later + if result[:status] == :timeout + return self.class.perform_async(measurement_identifier, result[:continue_from], max_id, recorded_at, result[:partial_results]) + end - return if count == Gitlab::Database::BatchCounter::FALLBACK + return if result[:status] != :completed - UsageTrends::Measurement.insert_all([{ recorded_at: recorded_at, count: count, identifier: measurement_identifier }]) + UsageTrends::Measurement.insert_all([{ recorded_at: recorded_at, count: result[:count], identifier: measurement_identifier }]) end private - def counter(query_scope, min_id, max_id) - Gitlab::Database::BatchCount.batch_count(query_scope, start: min_id, finish: max_id) + def counter(query_scope, min_id, max_id, partial_results) + return { status: :completed, count: 0 } if min_id.nil? || max_id.nil? # table is empty + + Gitlab::Database::BatchCount.batch_count_with_timeout(query_scope, start: min_id, finish: max_id, timeout: TIMEOUT, partial_results: partial_results) end end end diff --git a/app/workers/ci/build_finished_worker.rb b/app/workers/ci/build_finished_worker.rb index 36a50735fed..7503ea3d800 100644 --- a/app/workers/ci/build_finished_worker.rb +++ b/app/workers/ci/build_finished_worker.rb @@ -36,10 +36,10 @@ module Ci build.update_coverage Ci::BuildReportResultService.new.execute(build) - build.feature_flagged_execute_hooks + build.execute_hooks ChatNotificationWorker.perform_async(build.id) if build.pipeline.chat? build.track_deployment_usage - build.track_verify_usage + build.track_verify_environment_usage if build.failed? && !build.auto_retry_expected? ::Ci::MergeRequests::AddTodoWhenBuildFailsWorker.perform_async(build.id) diff --git a/app/workers/ci/job_artifacts/track_artifact_report_worker.rb b/app/workers/ci/job_artifacts/track_artifact_report_worker.rb new file mode 100644 index 00000000000..3df8c284ab3 --- /dev/null +++ b/app/workers/ci/job_artifacts/track_artifact_report_worker.rb @@ -0,0 +1,23 @@ +# frozen_string_literal: true + +module Ci + module JobArtifacts + class TrackArtifactReportWorker + include ApplicationWorker + + data_consistency :delayed + + include PipelineBackgroundQueue + + feature_category :code_testing + + idempotent! + + def perform(pipeline_id) + Ci::Pipeline.find_by_id(pipeline_id).try do |pipeline| + Ci::JobArtifacts::TrackArtifactReportService.new.execute(pipeline) + end + end + end + end +end diff --git a/app/workers/ci/pipeline_artifacts/coverage_report_worker.rb b/app/workers/ci/pipeline_artifacts/coverage_report_worker.rb index 127eb3b6f44..53bed0fa9da 100644 --- a/app/workers/ci/pipeline_artifacts/coverage_report_worker.rb +++ b/app/workers/ci/pipeline_artifacts/coverage_report_worker.rb @@ -20,7 +20,7 @@ module Ci return unless pipeline pipeline.root_ancestor.try do |root_ancestor_pipeline| - next unless root_ancestor_pipeline.self_and_descendants_complete? + next unless root_ancestor_pipeline.self_and_project_descendants_complete? Ci::PipelineArtifacts::CoverageReportService.new(root_ancestor_pipeline).execute end diff --git a/app/workers/cleanup_container_repository_worker.rb b/app/workers/cleanup_container_repository_worker.rb index 73501315575..3a506470743 100644 --- a/app/workers/cleanup_container_repository_worker.rb +++ b/app/workers/cleanup_container_repository_worker.rb @@ -24,7 +24,7 @@ class CleanupContainerRepositoryWorker return unless valid? Projects::ContainerRepository::CleanupTagsService - .new(container_repository, current_user, params) + .new(container_repository: container_repository, current_user: current_user, params: params) .execute end diff --git a/app/workers/flush_counter_increments_worker.rb b/app/workers/flush_counter_increments_worker.rb index e21a7ee35e7..8c7e17587d0 100644 --- a/app/workers/flush_counter_increments_worker.rb +++ b/app/workers/flush_counter_increments_worker.rb @@ -11,6 +11,7 @@ class FlushCounterIncrementsWorker data_consistency :always sidekiq_options retry: 3 + loggable_arguments 0, 2 # The increments in `ProjectStatistics` are owned by several teams depending # on the counter diff --git a/app/workers/gitlab/github_import/advance_stage_worker.rb b/app/workers/gitlab/github_import/advance_stage_worker.rb index 70d18d8004c..fdf4ec6f396 100644 --- a/app/workers/gitlab/github_import/advance_stage_worker.rb +++ b/app/workers/gitlab/github_import/advance_stage_worker.rb @@ -25,6 +25,8 @@ module Gitlab issues_and_diff_notes: Stage::ImportIssuesAndDiffNotesWorker, issue_events: Stage::ImportIssueEventsWorker, notes: Stage::ImportNotesWorker, + attachments: Stage::ImportAttachmentsWorker, + protected_branches: Stage::ImportProtectedBranchesWorker, lfs_objects: Stage::ImportLfsObjectsWorker, finish: Stage::FinishImportWorker }.freeze diff --git a/app/workers/gitlab/github_import/import_protected_branch_worker.rb b/app/workers/gitlab/github_import/import_protected_branch_worker.rb new file mode 100644 index 00000000000..c083d8ee867 --- /dev/null +++ b/app/workers/gitlab/github_import/import_protected_branch_worker.rb @@ -0,0 +1,23 @@ +# frozen_string_literal: true + +module Gitlab + module GithubImport + class ImportProtectedBranchWorker # rubocop:disable Scalability/IdempotentWorker + include ObjectImporter + + worker_resource_boundary :cpu + + def representation_class + Gitlab::GithubImport::Representation::ProtectedBranch + end + + def importer_class + Importer::ProtectedBranchImporter + end + + def object_type + :protected_branch + end + end + end +end diff --git a/app/workers/gitlab/github_import/import_release_attachments_worker.rb b/app/workers/gitlab/github_import/import_release_attachments_worker.rb new file mode 100644 index 00000000000..c6f45ec1d7c --- /dev/null +++ b/app/workers/gitlab/github_import/import_release_attachments_worker.rb @@ -0,0 +1,21 @@ +# frozen_string_literal: true + +module Gitlab + module GithubImport + class ImportReleaseAttachmentsWorker # rubocop:disable Scalability/IdempotentWorker + include ObjectImporter + + def representation_class + Representation::ReleaseAttachments + end + + def importer_class + Importer::ReleaseAttachmentsImporter + end + + def object_type + :release_attachment + end + end + end +end diff --git a/app/workers/gitlab/github_import/stage/import_attachments_worker.rb b/app/workers/gitlab/github_import/stage/import_attachments_worker.rb new file mode 100644 index 00000000000..e9086dca503 --- /dev/null +++ b/app/workers/gitlab/github_import/stage/import_attachments_worker.rb @@ -0,0 +1,58 @@ +# frozen_string_literal: true + +module Gitlab + module GithubImport + module Stage + class ImportAttachmentsWorker # rubocop:disable Scalability/IdempotentWorker + include ApplicationWorker + + data_consistency :always + + sidekiq_options retry: 5 + include GithubImport::Queue + include StageMethods + + # client - An instance of Gitlab::GithubImport::Client. + # project - An instance of Project. + def import(client, project) + return skip_to_next_stage(project) if feature_disabled?(project) + + waiters = importers.each_with_object({}) do |importer, hash| + waiter = start_importer(project, importer, client) + hash[waiter.key] = waiter.jobs_remaining + end + move_to_next_stage(project, waiters) + end + + private + + # For future issue/mr/milestone/etc attachments importers + def importers + [::Gitlab::GithubImport::Importer::ReleasesAttachmentsImporter] + end + + def start_importer(project, importer, client) + info(project.id, message: "starting importer", importer: importer.name) + importer.new(project, client).execute + end + + def skip_to_next_stage(project) + info(project.id, message: "skipping importer", importer: 'Attachments') + move_to_next_stage(project) + end + + def move_to_next_stage(project, waiters = {}) + AdvanceStageWorker.perform_async( + project.id, + waiters, + :protected_branches + ) + end + + def feature_disabled?(project) + Feature.disabled?(:github_importer_attachments_import, project.group, type: :ops) + end + end + end + end +end diff --git a/app/workers/gitlab/github_import/stage/import_notes_worker.rb b/app/workers/gitlab/github_import/stage/import_notes_worker.rb index 167b3e147a3..b53e31ce40e 100644 --- a/app/workers/gitlab/github_import/stage/import_notes_worker.rb +++ b/app/workers/gitlab/github_import/stage/import_notes_worker.rb @@ -21,11 +21,7 @@ module Gitlab hash[waiter.key] = waiter.jobs_remaining end - AdvanceStageWorker.perform_async( - project.id, - waiters, - :lfs_objects - ) + AdvanceStageWorker.perform_async(project.id, waiters, :attachments) end def importers(project) diff --git a/app/workers/gitlab/github_import/stage/import_protected_branches_worker.rb b/app/workers/gitlab/github_import/stage/import_protected_branches_worker.rb new file mode 100644 index 00000000000..6d6dea10e64 --- /dev/null +++ b/app/workers/gitlab/github_import/stage/import_protected_branches_worker.rb @@ -0,0 +1,45 @@ +# frozen_string_literal: true + +module Gitlab + module GithubImport + module Stage + class ImportProtectedBranchesWorker # rubocop:disable Scalability/IdempotentWorker + include ApplicationWorker + + data_consistency :always + + sidekiq_options retry: 3 + include GithubImport::Queue + include StageMethods + + # client - An instance of Gitlab::GithubImport::Client. + # project - An instance of Project. + def import(client, project) + info(project.id, + message: "starting importer", + importer: 'Importer::ProtectedBranchesImporter') + waiter = Importer::ProtectedBranchesImporter + .new(project, client) + .execute + + project.import_state.refresh_jid_expiration + + AdvanceStageWorker.perform_async( + project.id, + { waiter.key => waiter.jobs_remaining }, + :lfs_objects + ) + rescue StandardError => e + Gitlab::Import::ImportFailureService.track( + project_id: project.id, + error_source: self.class.name, + exception: e, + metrics: true + ) + + raise(e) + end + end + end + end +end diff --git a/app/workers/gitlab/jira_import/import_issue_worker.rb b/app/workers/gitlab/jira_import/import_issue_worker.rb index 3824cc1f3ef..eabe988dfc2 100644 --- a/app/workers/gitlab/jira_import/import_issue_worker.rb +++ b/app/workers/gitlab/jira_import/import_issue_worker.rb @@ -15,8 +15,7 @@ module Gitlab loggable_arguments 3 def perform(project_id, jira_issue_id, issue_attributes, waiter_key) - issue_id = create_issue(issue_attributes, project_id) - JiraImport.cache_issue_mapping(issue_id, jira_issue_id, project_id) + create_issue(issue_attributes, project_id) rescue StandardError => ex # Todo: Record jira issue id(or better jira issue key), # so that we can report the list of failed to import issues to the user diff --git a/app/workers/gitlab_service_ping_worker.rb b/app/workers/gitlab_service_ping_worker.rb index a974667e5e0..b02e7318585 100644 --- a/app/workers/gitlab_service_ping_worker.rb +++ b/app/workers/gitlab_service_ping_worker.rb @@ -15,17 +15,24 @@ class GitlabServicePingWorker # rubocop:disable Scalability/IdempotentWorker sidekiq_options retry: 3, dead: false sidekiq_retry_in { |count| (count + 1) * 8.hours.to_i } - def perform - # Disable service ping for GitLab.com + def perform(options = {}) + # Sidekiq does not support keyword arguments, so the args need to be + # passed the old pre-Ruby 2.0 way. + # + # See https://github.com/mperham/sidekiq/issues/2372 + triggered_from_cron = options.fetch('triggered_from_cron', true) + skip_db_write = options.fetch('skip_db_write', false) + + # Disable service ping for GitLab.com unless called manually # See https://gitlab.com/gitlab-org/gitlab/-/issues/292929 for details - return if Gitlab.com? + return if Gitlab.com? && triggered_from_cron # Multiple Sidekiq workers could run this. We should only do this at most once a day. in_lock(LEASE_KEY, ttl: LEASE_TIMEOUT) do # Splay the request over a minute to avoid thundering herd problems. sleep(rand(0.0..60.0).round(3)) - ServicePing::SubmitService.new(payload: usage_data).execute + ServicePing::SubmitService.new(payload: usage_data, skip_db_write: skip_db_write).execute end end diff --git a/app/workers/google_cloud/create_cloudsql_instance_worker.rb b/app/workers/google_cloud/create_cloudsql_instance_worker.rb index 3c15c59b8d9..8c4f4c83339 100644 --- a/app/workers/google_cloud/create_cloudsql_instance_worker.rb +++ b/app/workers/google_cloud/create_cloudsql_instance_worker.rb @@ -8,30 +8,15 @@ module GoogleCloud feature_category :not_owned # rubocop:disable Gitlab/AvoidFeatureCategoryNotOwned idempotent! - def perform(user_id, project_id, options = {}) + def perform(user_id, project_id, params = {}) user = User.find(user_id) project = Project.find(project_id) + params = params.with_indifferent_access - google_oauth2_token = options[:google_oauth2_token] - gcp_project_id = options[:gcp_project_id] - instance_name = options[:instance_name] - database_version = options[:database_version] - environment_name = options[:environment_name] - is_protected = options[:is_protected] - - params = { - google_oauth2_token: google_oauth2_token, - gcp_project_id: gcp_project_id, - instance_name: instance_name, - database_version: database_version, - environment_name: environment_name, - is_protected: is_protected - } - - response = GoogleCloud::SetupCloudsqlInstanceService.new(project, user, params).execute + response = ::GoogleCloud::SetupCloudsqlInstanceService.new(project, user, params).execute if response[:status] == :error - raise response[:message] + raise "Error SetupCloudsqlInstanceService: #{response.to_json}" end end end diff --git a/app/workers/google_cloud/fetch_google_ip_list_worker.rb b/app/workers/google_cloud/fetch_google_ip_list_worker.rb new file mode 100644 index 00000000000..b14b4e735dc --- /dev/null +++ b/app/workers/google_cloud/fetch_google_ip_list_worker.rb @@ -0,0 +1,17 @@ +# frozen_string_literal: true + +module GoogleCloud + class FetchGoogleIpListWorker + include ApplicationWorker + + data_consistency :delayed + feature_category :build_artifacts + urgency :low + deduplicate :until_executing + idempotent! + + def perform + GoogleCloud::FetchGoogleIpListService.new.execute + end + end +end diff --git a/app/workers/groups/update_two_factor_requirement_for_members_worker.rb b/app/workers/groups/update_two_factor_requirement_for_members_worker.rb new file mode 100644 index 00000000000..ac1d3589516 --- /dev/null +++ b/app/workers/groups/update_two_factor_requirement_for_members_worker.rb @@ -0,0 +1,22 @@ +# frozen_string_literal: true + +# Worker for updating two factor requirement for all group members +module Groups + class UpdateTwoFactorRequirementForMembersWorker + include ApplicationWorker + + data_consistency :always + + idempotent! + + feature_category :authentication_and_authorization + + def perform(group_id) + group = Group.find_by_id(group_id) + + return unless group + + group.update_two_factor_requirement_for_members + end + end +end diff --git a/app/workers/issues/close_worker.rb b/app/workers/issues/close_worker.rb new file mode 100644 index 00000000000..0d540ee8c4f --- /dev/null +++ b/app/workers/issues/close_worker.rb @@ -0,0 +1,50 @@ +# frozen_string_literal: true + +module Issues + class CloseWorker + include ApplicationWorker + + data_consistency :always + + sidekiq_options retry: 3 + + idempotent! + deduplicate :until_executed, including_scheduled: true + feature_category :source_code_management + urgency :high + weight 2 + + def perform(project_id, issue_id, issue_type, params = {}) + project = Project.find_by_id(project_id) + + unless project + logger.info(structured_payload(message: "Project not found.", project_id: project_id)) + return + end + + issue = case issue_type + when "ExternalIssue" + ExternalIssue.new(issue_id, project) + else + Issue.find_by_id(issue_id) + end + + unless issue + logger.info(structured_payload(message: "Issue not found.", issue_id: issue_id)) + return + end + + author = User.find_by_id(params["closed_by"]) + + unless author + logger.info(structured_payload(message: "User not found.", user_id: params["closed_by"])) + return + end + + commit = Commit.build_from_sidekiq_hash(project, params["commit_hash"]) + service = Issues::CloseService.new(project: project, current_user: author) + + service.execute(issue, commit: commit) + end + end +end diff --git a/app/workers/namespaces/onboarding_issue_created_worker.rb b/app/workers/namespaces/onboarding_issue_created_worker.rb index aab5767e0f1..4f0cc71cd91 100644 --- a/app/workers/namespaces/onboarding_issue_created_worker.rb +++ b/app/workers/namespaces/onboarding_issue_created_worker.rb @@ -18,7 +18,7 @@ module Namespaces namespace = Namespace.find_by_id(namespace_id) return unless namespace - OnboardingProgressService.new(namespace).execute(action: :issue_created) + Onboarding::ProgressService.new(namespace).execute(action: :issue_created) end end end diff --git a/app/workers/namespaces/onboarding_pipeline_created_worker.rb b/app/workers/namespaces/onboarding_pipeline_created_worker.rb index 4172e286474..c3850880df0 100644 --- a/app/workers/namespaces/onboarding_pipeline_created_worker.rb +++ b/app/workers/namespaces/onboarding_pipeline_created_worker.rb @@ -18,7 +18,7 @@ module Namespaces namespace = Namespace.find_by_id(namespace_id) return unless namespace - OnboardingProgressService.new(namespace).execute(action: :pipeline_created) + Onboarding::ProgressService.new(namespace).execute(action: :pipeline_created) end end end diff --git a/app/workers/namespaces/onboarding_progress_worker.rb b/app/workers/namespaces/onboarding_progress_worker.rb index 77a31d85a9a..49629428459 100644 --- a/app/workers/namespaces/onboarding_progress_worker.rb +++ b/app/workers/namespaces/onboarding_progress_worker.rb @@ -19,7 +19,7 @@ module Namespaces namespace = Namespace.find_by_id(namespace_id) return unless namespace && action - OnboardingProgressService.new(namespace).execute(action: action.to_sym) + Onboarding::ProgressService.new(namespace).execute(action: action.to_sym) end end end diff --git a/app/workers/namespaces/onboarding_user_added_worker.rb b/app/workers/namespaces/onboarding_user_added_worker.rb index 4d17cf9a6e2..a1b349eedd3 100644 --- a/app/workers/namespaces/onboarding_user_added_worker.rb +++ b/app/workers/namespaces/onboarding_user_added_worker.rb @@ -15,7 +15,7 @@ module Namespaces def perform(namespace_id) namespace = Namespace.find(namespace_id) - OnboardingProgressService.new(namespace).execute(action: :user_added) + Onboarding::ProgressService.new(namespace).execute(action: :user_added) end end end diff --git a/app/workers/namespaces/process_sync_events_worker.rb b/app/workers/namespaces/process_sync_events_worker.rb index 2bf2a4a6ef8..d0124c69781 100644 --- a/app/workers/namespaces/process_sync_events_worker.rb +++ b/app/workers/namespaces/process_sync_events_worker.rb @@ -13,7 +13,7 @@ module Namespaces urgency :high idempotent! - deduplicate :until_executing + deduplicate :until_executed, if_deduplicated: :reschedule_once def perform results = ::Ci::ProcessSyncEventsService.new( diff --git a/app/workers/object_storage/migrate_uploads_worker.rb b/app/workers/object_storage/migrate_uploads_worker.rb index b7d938e6b68..3e681c3f111 100644 --- a/app/workers/object_storage/migrate_uploads_worker.rb +++ b/app/workers/object_storage/migrate_uploads_worker.rb @@ -11,7 +11,7 @@ module ObjectStorage include ObjectStorageQueue feature_category :not_owned # rubocop:todo Gitlab/AvoidFeatureCategoryNotOwned - loggable_arguments 0, 1, 2, 3 + loggable_arguments 0 SanityCheckError = Class.new(StandardError) @@ -67,41 +67,19 @@ module ObjectStorage include Report # rubocop: disable CodeReuse/ActiveRecord - def self.enqueue!(uploads, model_class, mounted_as, to_store) - sanity_check!(uploads, model_class, mounted_as) - - perform_async(uploads.ids, model_class.to_s, mounted_as, to_store) + def self.enqueue!(uploads, to_store) + perform_async(uploads.ids, to_store) end # rubocop: enable CodeReuse/ActiveRecord - # We need to be sure all the uploads are for the same uploader and model type - # and that the mount point exists if provided. - # - def self.sanity_check!(uploads, model_class, mounted_as) - upload = uploads.first - uploader_class = upload.uploader.constantize - uploader_types = uploads.map(&:uploader).uniq - model_types = uploads.map(&:model_type).uniq - model_has_mount = mounted_as.nil? || model_class.uploaders[mounted_as] == uploader_class - - raise(SanityCheckError, _("Multiple uploaders found: %{uploader_types}") % { uploader_types: uploader_types }) unless uploader_types.count == 1 - raise(SanityCheckError, _("Multiple model types found: %{model_types}") % { model_types: model_types }) unless model_types.count == 1 - raise(SanityCheckError, _("Mount point %{mounted_as} not found in %{model_class}.") % { mounted_as: mounted_as, model_class: model_class }) unless model_has_mount - end - # rubocop: disable CodeReuse/ActiveRecord def perform(*args) - args_check!(args) - - (ids, model_type, mounted_as, to_store) = args + ids, to_store = retrieve_applicable_args!(args) - @model_class = model_type.constantize - @mounted_as = mounted_as&.to_sym @to_store = to_store uploads = Upload.preload(:model).where(id: ids) - sanity_check!(uploads) results = migrate(uploads) report!(results) @@ -111,31 +89,22 @@ module ObjectStorage end # rubocop: enable CodeReuse/ActiveRecord - def sanity_check!(uploads) - self.class.sanity_check!(uploads, @model_class, @mounted_as) - end - - def args_check!(args) - return if args.count == 4 + private - case args.count - when 3 then raise SanityCheckError, _("Job is missing the `model_type` argument.") - else - raise SanityCheckError, _("Job has wrong arguments format.") - end - end + def retrieve_applicable_args!(args) + return args if args.count == 2 + return args.values_at(0, 3) if args.count == 4 - def build_uploaders(uploads) - uploads.map { |upload| upload.retrieve_uploader(@mounted_as) } + raise SanityCheckError, _("Job has wrong arguments format.") end def migrate(uploads) - build_uploaders(uploads).map(&method(:process_uploader)) + uploads.map(&method(:process_upload)) end - def process_uploader(uploader) - MigrationResult.new(uploader.upload).tap do |result| - uploader.migrate!(@to_store) + def process_upload(upload) + MigrationResult.new(upload).tap do |result| + upload.retrieve_uploader.migrate!(@to_store) rescue StandardError => e result.error = e end diff --git a/app/workers/process_commit_worker.rb b/app/workers/process_commit_worker.rb index a4dfe11c394..cd6ce6eb28b 100644 --- a/app/workers/process_commit_worker.rb +++ b/app/workers/process_commit_worker.rb @@ -34,7 +34,7 @@ class ProcessCommitWorker return unless user - commit = build_commit(project, commit_hash) + commit = Commit.build_from_sidekiq_hash(project, commit_hash) author = commit.author || user process_commit_message(project, commit, user, author, default) @@ -51,12 +51,22 @@ class ProcessCommitWorker end def close_issues(project, user, author, commit, issues) - # We don't want to run permission related queries for every single issue, - # therefore we use IssueCollection here and skip the authorization check in - # Issues::CloseService#execute. - IssueCollection.new(issues).updatable_by_user(user).each do |issue| - Issues::CloseService.new(project: project, current_user: author) - .close_issue(issue, closed_via: commit) + if Feature.enabled?(:process_issue_closure_in_background, project) + Issues::CloseWorker.bulk_perform_async_with_contexts( + issues, + arguments_proc: -> (issue) { + [project.id, issue.id, issue.class.to_s, { closed_by: author.id, commit_hash: commit.to_hash }] + }, + context_proc: -> (issue) { { project: project } } + ) + else + # We don't want to run permission related queries for every single issue, + # therefore we use IssueCollection here and skip the authorization check in + # Issues::CloseService#execute. + IssueCollection.new(issues).updatable_by_user(user).each do |issue| + Issues::CloseService.new(project: project, current_user: author) + .close_issue(issue, closed_via: commit) + end end end @@ -75,19 +85,4 @@ class ProcessCommitWorker .with_first_mention_not_earlier_than(commit.committed_date) .update_all(first_mentioned_in_commit_at: commit.committed_date) end - - def build_commit(project, hash) - date_suffix = '_date' - - # When processing Sidekiq payloads various timestamps are stored as Strings. - # Commit in turn expects Time-like instances upon input, so we have to - # manually parse these values. - hash.each do |key, value| - if key.to_s.end_with?(date_suffix) && value.is_a?(String) - hash[key] = Time.zone.parse(value) - end - end - - Commit.from_hash(hash, project) - end end diff --git a/app/workers/projects/inactive_projects_deletion_cron_worker.rb b/app/workers/projects/inactive_projects_deletion_cron_worker.rb index a280c9203d6..ba6d44ec4a5 100644 --- a/app/workers/projects/inactive_projects_deletion_cron_worker.rb +++ b/app/workers/projects/inactive_projects_deletion_cron_worker.rb @@ -39,8 +39,6 @@ module Projects raise TimeoutError end - next unless Feature.enabled?(:inactive_projects_deletion, project.root_namespace) - with_context(project: project, user: admin_user) do deletion_warning_email_sent_on = notified_inactive_projects["project:#{project.id}"] diff --git a/app/workers/projects/process_sync_events_worker.rb b/app/workers/projects/process_sync_events_worker.rb index 57f3e3dee5e..4bbe1b65e5a 100644 --- a/app/workers/projects/process_sync_events_worker.rb +++ b/app/workers/projects/process_sync_events_worker.rb @@ -13,7 +13,7 @@ module Projects urgency :high idempotent! - deduplicate :until_executing + deduplicate :until_executed, if_deduplicated: :reschedule_once def perform results = ::Ci::ProcessSyncEventsService.new( diff --git a/app/workers/ssh_keys/expired_notification_worker.rb b/app/workers/ssh_keys/expired_notification_worker.rb index dc1efce51ce..768579214c6 100644 --- a/app/workers/ssh_keys/expired_notification_worker.rb +++ b/app/workers/ssh_keys/expired_notification_worker.rb @@ -15,19 +15,20 @@ module SshKeys # rubocop: disable CodeReuse/ActiveRecord def perform - order = Gitlab::Pagination::Keyset::Order.build([ - Gitlab::Pagination::Keyset::ColumnOrderDefinition.new( - attribute_name: 'expires_at_utc', - order_expression: Arel.sql("date(expires_at AT TIME ZONE 'UTC')").asc, - nullable: :not_nullable, - distinct: false, - add_to_projections: true - ), - Gitlab::Pagination::Keyset::ColumnOrderDefinition.new( - attribute_name: 'id', - order_expression: Key.arel_table[:id].asc - ) - ]) + order = Gitlab::Pagination::Keyset::Order.build( + [ + Gitlab::Pagination::Keyset::ColumnOrderDefinition.new( + attribute_name: 'expires_at_utc', + order_expression: Arel.sql("date(expires_at AT TIME ZONE 'UTC')").asc, + nullable: :not_nullable, + distinct: false, + add_to_projections: true + ), + Gitlab::Pagination::Keyset::ColumnOrderDefinition.new( + attribute_name: 'id', + order_expression: Key.arel_table[:id].asc + ) + ]) scope = Key.expired_today_and_not_notified.order(order) diff --git a/app/workers/users/migrate_records_to_ghost_user_in_batches_worker.rb b/app/workers/users/migrate_records_to_ghost_user_in_batches_worker.rb new file mode 100644 index 00000000000..ddddfc106ae --- /dev/null +++ b/app/workers/users/migrate_records_to_ghost_user_in_batches_worker.rb @@ -0,0 +1,22 @@ +# frozen_string_literal: true + +module Users + class MigrateRecordsToGhostUserInBatchesWorker + include ApplicationWorker + include Gitlab::ExclusiveLeaseHelpers + include CronjobQueue # rubocop: disable Scalability/CronWorkerContext + + sidekiq_options retry: false + feature_category :users + data_consistency :always + idempotent! + + def perform + return unless Feature.enabled?(:user_destroy_with_limited_execution_time_worker) + + in_lock(self.class.name.underscore, ttl: Gitlab::Utils::ExecutionTracker::MAX_RUNTIME, retries: 0) do + Users::MigrateRecordsToGhostUserInBatchesService.new.execute + end + end + end +end |