diff options
Diffstat (limited to 'app/workers')
30 files changed, 445 insertions, 199 deletions
diff --git a/app/workers/all_queues.yml b/app/workers/all_queues.yml index 48bdee4062b..bfb70e0d496 100644 --- a/app/workers/all_queues.yml +++ b/app/workers/all_queues.yml @@ -192,6 +192,15 @@ :weight: 1 :idempotent: true :tags: [] +- :name: cronjob:bulk_imports_stuck_import + :worker_name: BulkImports::StuckImportWorker + :feature_category: :importers + :has_external_dependencies: + :urgency: :low + :resource_boundary: :unknown + :weight: 1 + :idempotent: true + :tags: [] - :name: cronjob:ci_archive_traces_cron :worker_name: Ci::ArchiveTracesCronWorker :feature_category: :continuous_integration @@ -255,6 +264,15 @@ :weight: 1 :idempotent: true :tags: [] +- :name: cronjob:ci_update_locked_unknown_artifacts + :worker_name: Ci::UpdateLockedUnknownArtifactsWorker + :feature_category: :build_artifacts + :has_external_dependencies: + :urgency: :throttled + :resource_boundary: :unknown + :weight: 1 + :idempotent: + :tags: [] - :name: cronjob:clusters_integrations_check_prometheus_health :worker_name: Clusters::Integrations::CheckPrometheusHealthWorker :feature_category: :incident_management @@ -318,6 +336,24 @@ :weight: 1 :idempotent: true :tags: [] +- :name: cronjob:database_ci_namespace_mirrors_consistency_check + :worker_name: Database::CiNamespaceMirrorsConsistencyCheckWorker + :feature_category: :sharding + :has_external_dependencies: + :urgency: :low + :resource_boundary: :unknown + :weight: 1 + :idempotent: true + :tags: [] +- :name: cronjob:database_ci_project_mirrors_consistency_check + :worker_name: Database::CiProjectMirrorsConsistencyCheckWorker + :feature_category: :sharding + :has_external_dependencies: + :urgency: :low + :resource_boundary: :unknown + :weight: 1 + :idempotent: true + :tags: [] - :name: cronjob:database_drop_detached_partitions :worker_name: Database::DropDetachedPartitionsWorker :feature_category: :database @@ -579,15 +615,6 @@ :weight: 1 :idempotent: :tags: [] -- :name: cronjob:quality_test_data_cleanup - :worker_name: Quality::TestDataCleanupWorker - :feature_category: :quality_management - :has_external_dependencies: - :urgency: :low - :resource_boundary: :unknown - :weight: 1 - :idempotent: true - :tags: [] - :name: cronjob:releases_manage_evidence :worker_name: Releases::ManageEvidenceWorker :feature_category: :release_evidence @@ -2578,15 +2605,6 @@ :weight: 1 :idempotent: :tags: [] -- :name: namespaces_invite_team_email - :worker_name: Namespaces::InviteTeamEmailWorker - :feature_category: :experimentation_activation - :has_external_dependencies: - :urgency: :low - :resource_boundary: :unknown - :weight: 1 - :idempotent: - :tags: [] - :name: namespaces_onboarding_issue_created :worker_name: Namespaces::OnboardingIssueCreatedWorker :feature_category: :onboarding @@ -2771,7 +2789,7 @@ :worker_name: ProjectExportWorker :feature_category: :importers :has_external_dependencies: - :urgency: :throttled + :urgency: :low :resource_boundary: :memory :weight: 1 :idempotent: @@ -2812,6 +2830,15 @@ :weight: 1 :idempotent: true :tags: [] +- :name: projects_record_target_platforms + :worker_name: Projects::RecordTargetPlatformsWorker + :feature_category: :experimentation_activation + :has_external_dependencies: + :urgency: :low + :resource_boundary: :unknown + :weight: 1 + :idempotent: true + :tags: [] - :name: projects_refresh_build_artifacts_size_statistics :worker_name: Projects::RefreshBuildArtifactsSizeStatisticsWorker :feature_category: :build_artifacts diff --git a/app/workers/bulk_import_worker.rb b/app/workers/bulk_import_worker.rb index d560ebcc6e6..157586ca397 100644 --- a/app/workers/bulk_import_worker.rb +++ b/app/workers/bulk_import_worker.rb @@ -3,15 +3,12 @@ class BulkImportWorker # rubocop:disable Scalability/IdempotentWorker include ApplicationWorker - data_consistency :always + PERFORM_DELAY = 5.seconds + data_consistency :always feature_category :importers - sidekiq_options retry: false, dead: false - PERFORM_DELAY = 5.seconds - DEFAULT_BATCH_SIZE = 5 - def perform(bulk_import_id) @bulk_import = BulkImport.find_by_id(bulk_import_id) @@ -19,11 +16,10 @@ class BulkImportWorker # rubocop:disable Scalability/IdempotentWorker return if @bulk_import.finished? || @bulk_import.failed? return @bulk_import.fail_op! if all_entities_failed? return @bulk_import.finish! if all_entities_processed? && @bulk_import.started? - return re_enqueue if max_batch_size_exceeded? # Do not start more jobs if max allowed are already running @bulk_import.start! if @bulk_import.created? - created_entities.first(next_batch_size).each do |entity| + created_entities.find_each do |entity| entity.create_pipeline_trackers! BulkImports::ExportRequestWorker.perform_async(entity.id) @@ -45,10 +41,6 @@ class BulkImportWorker # rubocop:disable Scalability/IdempotentWorker @entities ||= @bulk_import.entities end - def started_entities - entities.with_status(:started) - end - def created_entities entities.with_status(:created) end @@ -61,14 +53,6 @@ class BulkImportWorker # rubocop:disable Scalability/IdempotentWorker entities.all? { |entity| entity.failed? } end - def max_batch_size_exceeded? - started_entities.count >= DEFAULT_BATCH_SIZE - end - - def next_batch_size - [DEFAULT_BATCH_SIZE - started_entities.count, 0].max - end - # A new BulkImportWorker job is enqueued to either # - Process the new BulkImports::Entity created during import (e.g. for the subgroups) # - Or to mark the `bulk_import` as finished diff --git a/app/workers/bulk_imports/entity_worker.rb b/app/workers/bulk_imports/entity_worker.rb index 70d6626df91..f6b1c693fe4 100644 --- a/app/workers/bulk_imports/entity_worker.rb +++ b/app/workers/bulk_imports/entity_worker.rb @@ -4,24 +4,32 @@ module BulkImports class EntityWorker # rubocop:disable Scalability/IdempotentWorker include ApplicationWorker + idempotent! + deduplicate :until_executing data_consistency :always - feature_category :importers - sidekiq_options retry: false, dead: false - worker_has_external_dependencies! - idempotent! - deduplicate :until_executed, including_scheduled: true - def perform(entity_id, current_stage = nil) - return if stage_running?(entity_id, current_stage) + if stage_running?(entity_id, current_stage) + logger.info( + structured_payload( + entity_id: entity_id, + current_stage: current_stage, + message: 'Stage running' + ) + ) + + return + end logger.info( - worker: self.class.name, - entity_id: entity_id, - current_stage: current_stage + structured_payload( + entity_id: entity_id, + current_stage: current_stage, + message: 'Stage starting' + ) ) next_pipeline_trackers_for(entity_id).each do |pipeline_tracker| @@ -33,10 +41,11 @@ module BulkImports end rescue StandardError => e logger.error( - worker: self.class.name, - entity_id: entity_id, - current_stage: current_stage, - error_message: e.message + structured_payload( + entity_id: entity_id, + current_stage: current_stage, + message: e.message + ) ) Gitlab::ErrorTracking.track_exception(e, entity_id: entity_id) diff --git a/app/workers/bulk_imports/export_request_worker.rb b/app/workers/bulk_imports/export_request_worker.rb index 21040178cee..0d3e4f013dd 100644 --- a/app/workers/bulk_imports/export_request_worker.rb +++ b/app/workers/bulk_imports/export_request_worker.rb @@ -42,10 +42,12 @@ module BulkImports correlation_id_value: Labkit::Correlation::CorrelationId.current_or_new_id } - Gitlab::Import::Logger.warn( - attributes.merge( - bulk_import_id: entity.bulk_import.id, - bulk_import_entity_type: entity.source_type + Gitlab::Import::Logger.error( + structured_payload( + attributes.merge( + bulk_import_id: entity.bulk_import.id, + bulk_import_entity_type: entity.source_type + ) ) ) diff --git a/app/workers/bulk_imports/pipeline_worker.rb b/app/workers/bulk_imports/pipeline_worker.rb index 03ec2f058ca..1a98705c151 100644 --- a/app/workers/bulk_imports/pipeline_worker.rb +++ b/app/workers/bulk_imports/pipeline_worker.rb @@ -4,14 +4,11 @@ module BulkImports class PipelineWorker # rubocop:disable Scalability/IdempotentWorker include ApplicationWorker - data_consistency :always - - NDJSON_PIPELINE_PERFORM_DELAY = 1.minute + NDJSON_PIPELINE_PERFORM_DELAY = 10.seconds + data_consistency :always feature_category :importers - sidekiq_options retry: false, dead: false - worker_has_external_dependencies! def perform(pipeline_tracker_id, stage, entity_id) @@ -21,18 +18,20 @@ module BulkImports if pipeline_tracker.present? logger.info( - worker: self.class.name, - entity_id: pipeline_tracker.entity.id, - pipeline_name: pipeline_tracker.pipeline_name + structured_payload( + entity_id: pipeline_tracker.entity.id, + pipeline_name: pipeline_tracker.pipeline_name + ) ) run(pipeline_tracker) else logger.error( - worker: self.class.name, - entity_id: entity_id, - pipeline_tracker_id: pipeline_tracker_id, - message: 'Unstarted pipeline not found' + structured_payload( + entity_id: entity_id, + pipeline_tracker_id: pipeline_tracker_id, + message: 'Unstarted pipeline not found' + ) ) end @@ -66,10 +65,11 @@ module BulkImports rescue BulkImports::NetworkError => e if e.retriable?(pipeline_tracker) logger.error( - worker: self.class.name, - entity_id: pipeline_tracker.entity.id, - pipeline_name: pipeline_tracker.pipeline_name, - message: "Retrying error: #{e.message}" + structured_payload( + entity_id: pipeline_tracker.entity.id, + pipeline_name: pipeline_tracker.pipeline_name, + message: "Retrying error: #{e.message}" + ) ) pipeline_tracker.update!(status_event: 'retry', jid: jid) @@ -86,10 +86,11 @@ module BulkImports pipeline_tracker.update!(status_event: 'fail_op', jid: jid) logger.error( - worker: self.class.name, - entity_id: pipeline_tracker.entity.id, - pipeline_name: pipeline_tracker.pipeline_name, - message: exception.message + structured_payload( + entity_id: pipeline_tracker.entity.id, + pipeline_name: pipeline_tracker.pipeline_name, + message: exception.message + ) ) Gitlab::ErrorTracking.track_exception( diff --git a/app/workers/bulk_imports/relation_export_worker.rb b/app/workers/bulk_imports/relation_export_worker.rb index 9324b79cc75..dcac841b3b2 100644 --- a/app/workers/bulk_imports/relation_export_worker.rb +++ b/app/workers/bulk_imports/relation_export_worker.rb @@ -3,12 +3,12 @@ module BulkImports class RelationExportWorker include ApplicationWorker - - data_consistency :always include ExceptionBacktrace idempotent! + deduplicate :until_executed loggable_arguments 2, 3 + data_consistency :always feature_category :importers sidekiq_options status_expiration: StuckExportJobsWorker::EXPORT_JOBS_EXPIRATION diff --git a/app/workers/bulk_imports/stuck_import_worker.rb b/app/workers/bulk_imports/stuck_import_worker.rb new file mode 100644 index 00000000000..3fa4221728b --- /dev/null +++ b/app/workers/bulk_imports/stuck_import_worker.rb @@ -0,0 +1,31 @@ +# frozen_string_literal: true + +module BulkImports + class StuckImportWorker + include ApplicationWorker + + # This worker does not schedule other workers that require context. + include CronjobQueue # rubocop:disable Scalability/CronWorkerContext + + idempotent! + data_consistency :always + + feature_category :importers + + def perform + BulkImport.stale.find_each do |import| + import.cleanup_stale + end + + BulkImports::Entity.includes(:trackers).stale.find_each do |import| # rubocop: disable CodeReuse/ActiveRecord + ApplicationRecord.transaction do + import.cleanup_stale + + import.trackers.find_each do |tracker| + tracker.cleanup_stale + end + end + end + end + end +end diff --git a/app/workers/ci/update_locked_unknown_artifacts_worker.rb b/app/workers/ci/update_locked_unknown_artifacts_worker.rb new file mode 100644 index 00000000000..2d37ebb3c93 --- /dev/null +++ b/app/workers/ci/update_locked_unknown_artifacts_worker.rb @@ -0,0 +1,26 @@ +# frozen_string_literal: true + +module Ci + class UpdateLockedUnknownArtifactsWorker # rubocop:disable Scalability/IdempotentWorker + include ApplicationWorker + + data_consistency :sticky + urgency :throttled + + # rubocop:disable Scalability/CronWorkerContext + # This worker does not perform work scoped to a context + include CronjobQueue + # rubocop:enable Scalability/CronWorkerContext + + feature_category :build_artifacts + + def perform + return unless ::Feature.enabled?(:ci_job_artifacts_backlog_work) + + artifact_counts = Ci::JobArtifacts::UpdateUnknownLockedStatusService.new.execute + + log_extra_metadata_on_done(:removed_count, artifact_counts[:removed]) + log_extra_metadata_on_done(:locked_count, artifact_counts[:locked]) + end + end +end diff --git a/app/workers/concerns/chaos_queue.rb b/app/workers/concerns/chaos_queue.rb index a9c557f0175..23e58b5182b 100644 --- a/app/workers/concerns/chaos_queue.rb +++ b/app/workers/concerns/chaos_queue.rb @@ -5,6 +5,6 @@ module ChaosQueue included do queue_namespace :chaos - feature_category_not_owned! + feature_category :not_owned # rubocop:todo Gitlab/AvoidFeatureCategoryNotOwned end end diff --git a/app/workers/concerns/git_garbage_collect_methods.rb b/app/workers/concerns/git_garbage_collect_methods.rb index 13b7e7b5b1f..308ffacfc6b 100644 --- a/app/workers/concerns/git_garbage_collect_methods.rb +++ b/app/workers/concerns/git_garbage_collect_methods.rb @@ -121,8 +121,12 @@ module GitGarbageCollectMethods end.new(repository) end + # The option to enable/disable bitmaps has been removed in https://gitlab.com/gitlab-org/gitlab/-/issues/353777 + # Now the options is always enabled + # This method and all the deprecated RPCs are going to be removed in + # https://gitlab.com/gitlab-org/gitlab/-/issues/353779 def bitmaps_enabled? - Gitlab::CurrentSettings.housekeeping_bitmaps_enabled + true end def flush_ref_caches(resource) diff --git a/app/workers/concerns/packages/cleanup_artifact_worker.rb b/app/workers/concerns/packages/cleanup_artifact_worker.rb index d4ad023b4a8..a01d7e8abba 100644 --- a/app/workers/concerns/packages/cleanup_artifact_worker.rb +++ b/app/workers/concerns/packages/cleanup_artifact_worker.rb @@ -14,7 +14,9 @@ module Packages artifact.destroy! rescue StandardError - artifact&.error! + unless artifact&.destroyed? + artifact&.update_column(:status, :error) + end end after_destroy @@ -48,7 +50,7 @@ module Packages to_delete = next_item if to_delete - to_delete.processing! + to_delete.update_column(:status, :processing) log_cleanup_item(to_delete) end diff --git a/app/workers/concerns/reactive_cacheable_worker.rb b/app/workers/concerns/reactive_cacheable_worker.rb index 78fcf8087c2..a598b8a9d7d 100644 --- a/app/workers/concerns/reactive_cacheable_worker.rb +++ b/app/workers/concerns/reactive_cacheable_worker.rb @@ -8,7 +8,10 @@ module ReactiveCacheableWorker sidekiq_options retry: 3 - feature_category_not_owned! + # Feature category is different depending on the model that is using the + # reactive cache. Identified by the `related_class` attribute. + feature_category :not_owned # rubocop:todo Gitlab/AvoidFeatureCategoryNotOwned + loggable_arguments 0 def self.context_for_arguments(arguments) diff --git a/app/workers/concerns/worker_attributes.rb b/app/workers/concerns/worker_attributes.rb index 6f91418e38c..8f7a3da5429 100644 --- a/app/workers/concerns/worker_attributes.rb +++ b/app/workers/concerns/worker_attributes.rb @@ -35,17 +35,9 @@ module WorkerAttributes class_methods do def feature_category(value, *extras) - raise "Invalid category. Use `feature_category_not_owned!` to mark a worker as not owned" if value == :not_owned - set_class_attribute(:feature_category, value) end - # Special case: mark this work as not associated with a feature category - # this should be used for cross-cutting concerns, such as mailer workers. - def feature_category_not_owned! - set_class_attribute(:feature_category, :not_owned) - end - # Special case: if a worker is not owned, get the feature category # (if present) from the calling context. def get_feature_category diff --git a/app/workers/container_registry/migration/enqueuer_worker.rb b/app/workers/container_registry/migration/enqueuer_worker.rb index 5feaba870e6..8705deb0cb2 100644 --- a/app/workers/container_registry/migration/enqueuer_worker.rb +++ b/app/workers/container_registry/migration/enqueuer_worker.rb @@ -6,6 +6,9 @@ module ContainerRegistry include ApplicationWorker include CronjobQueue # rubocop:disable Scalability/CronWorkerContext include Gitlab::Utils::StrongMemoize + include ExclusiveLeaseGuard + + DEFAULT_LEASE_TIMEOUT = 30.minutes.to_i.freeze data_consistency :always feature_category :container_registry @@ -14,70 +17,103 @@ module ContainerRegistry idempotent! def perform - return unless migration.enabled? - return unless below_capacity? - return unless waiting_time_passed? + re_enqueue = false + try_obtain_lease do + break unless runnable? - re_enqueue_if_capacity if handle_aborted_migration || handle_next_migration - rescue StandardError => e - Gitlab::ErrorTracking.log_exception( - e, - next_repository_id: next_repository&.id, - next_aborted_repository_id: next_aborted_repository&.id - ) - - next_repository&.abort_import + re_enqueue = handle_aborted_migration || handle_next_migration + end + re_enqueue_if_capacity if re_enqueue end private def handle_aborted_migration - return unless next_aborted_repository&.retry_aborted_migration + return unless next_aborted_repository - log_extra_metadata_on_done(:container_repository_id, next_aborted_repository.id) log_extra_metadata_on_done(:import_type, 'retry') + log_repository(next_aborted_repository) + + next_aborted_repository.retry_aborted_migration + + true + rescue StandardError => e + Gitlab::ErrorTracking.log_exception(e, next_aborted_repository_id: next_aborted_repository&.id) true + ensure + log_repository_migration_state(next_aborted_repository) end def handle_next_migration return unless next_repository + + log_extra_metadata_on_done(:import_type, 'next') + log_repository(next_repository) + # We return true because the repository was successfully processed (migration_state is changed) return true if tag_count_too_high? return unless next_repository.start_pre_import - log_extra_metadata_on_done(:container_repository_id, next_repository.id) - log_extra_metadata_on_done(:import_type, 'next') - true + rescue StandardError => e + Gitlab::ErrorTracking.log_exception(e, next_repository_id: next_repository&.id) + next_repository&.abort_import + + false + ensure + log_repository_migration_state(next_repository) end def tag_count_too_high? return false unless next_repository.tags_count > migration.max_tags_count next_repository.skip_import(reason: :too_many_tags) + log_extra_metadata_on_done(:tags_count_too_high, true) + log_extra_metadata_on_done(:max_tags_count_setting, migration.max_tags_count) true end def below_capacity? - current_capacity <= maximum_capacity + current_capacity < maximum_capacity end def waiting_time_passed? delay = migration.enqueue_waiting_time return true if delay == 0 - return true unless last_step_completed_repository + return true unless last_step_completed_repository&.last_import_step_done_at last_step_completed_repository.last_import_step_done_at < Time.zone.now - delay end - def current_capacity - strong_memoize(:current_capacity) do - ContainerRepository.with_migration_states( - %w[pre_importing pre_import_done importing] - ).count + def runnable? + unless migration.enabled? + log_extra_metadata_on_done(:migration_enabled, false) + return false + end + + unless below_capacity? + log_extra_metadata_on_done(:max_capacity_setting, maximum_capacity) + log_extra_metadata_on_done(:below_capacity, false) + + return false + end + + unless waiting_time_passed? + log_extra_metadata_on_done(:waiting_time_passed, false) + log_extra_metadata_on_done(:current_waiting_time_setting, migration.enqueue_waiting_time) + + return false end + + true + end + + def current_capacity + ContainerRepository.with_migration_states( + %w[pre_importing pre_import_done importing] + ).count end def maximum_capacity @@ -107,10 +143,31 @@ module ContainerRegistry end def re_enqueue_if_capacity - return unless current_capacity < maximum_capacity + return unless below_capacity? self.class.perform_async end + + def log_repository(repository) + log_extra_metadata_on_done(:container_repository_id, repository&.id) + log_extra_metadata_on_done(:container_repository_path, repository&.path) + end + + def log_repository_migration_state(repository) + return unless repository + + log_extra_metadata_on_done(:container_repository_migration_state, repository.migration_state) + end + + # used by ExclusiveLeaseGuard + def lease_key + 'container_registry:migration:enqueuer_worker' + end + + # used by ExclusiveLeaseGuard + def lease_timeout + DEFAULT_LEASE_TIMEOUT + end end end end diff --git a/app/workers/container_registry/migration/guard_worker.rb b/app/workers/container_registry/migration/guard_worker.rb index 77ae111c1cb..bab6b8c2a72 100644 --- a/app/workers/container_registry/migration/guard_worker.rb +++ b/app/workers/container_registry/migration/guard_worker.rb @@ -29,46 +29,45 @@ module ContainerRegistry log_extra_metadata_on_done(:stale_migrations_count, repositories.to_a.size) repositories.each do |repository| - if abortable?(repository) + if actively_importing?(repository) + # if a repository is actively importing but not yet long_running, do nothing + if long_running_migration?(repository) + long_running_migration_ids << repository.id + cancel_long_running_migration(repository) + aborts_count += 1 + end + else repository.abort_import aborts_count += 1 - else - long_running_migration_ids << repository.id if long_running_migration?(repository) end end log_extra_metadata_on_done(:aborted_stale_migrations_count, aborts_count) if long_running_migration_ids.any? - log_extra_metadata_on_done(:long_running_stale_migration_container_repository_ids, long_running_migration_ids) + log_extra_metadata_on_done(:aborted_long_running_migration_ids, long_running_migration_ids) end end private - # This can ping the Container Registry API. - # We loop on a set of repositories to calls this function (see #perform) - # In the worst case scenario, we have a n+1 API calls situation here. - # - # This is reasonable because the maximum amount of repositories looped - # on is `25`. See ::ContainerRegistry::Migration.capacity. - # - # TODO We can remove this n+1 situation by having a Container Registry API - # endpoint that accepts multiple repository paths at once. This is issue + # A repository is actively_importing if it has an importing migration state + # and that state matches the state in the registry + # TODO We can have an API call n+1 situation here. It can be solved when the + # endpoint accepts multiple repository paths at once. This is issue # https://gitlab.com/gitlab-org/container-registry/-/issues/582 - def abortable?(repository) - # early return to save one Container Registry API request - return true unless repository.importing? || repository.pre_importing? - return true unless external_migration_in_progress?(repository) + def actively_importing?(repository) + return false unless repository.importing? || repository.pre_importing? + return false unless external_state_matches_migration_state?(repository) - false + true end def long_running_migration?(repository) migration_start_timestamp(repository).before?(long_running_migration_threshold) end - def external_migration_in_progress?(repository) + def external_state_matches_migration_state?(repository) status = repository.external_import_status (status == 'pre_import_in_progress' && repository.pre_importing?) || @@ -96,6 +95,21 @@ module ContainerRegistry def long_running_migration_threshold @threshold ||= 30.minutes.ago end + + def cancel_long_running_migration(repository) + result = repository.migration_cancel + + case result[:status] + when :ok + repository.skip_import(reason: :migration_canceled) + when :bad_request + repository.reconcile_import_status(result[:state]) do + repository.abort_import + end + else + repository.abort_import + end + end end end end diff --git a/app/workers/database/batched_background_migration/ci_database_worker.rb b/app/workers/database/batched_background_migration/ci_database_worker.rb index 98ec6f98123..13314cf95e2 100644 --- a/app/workers/database/batched_background_migration/ci_database_worker.rb +++ b/app/workers/database/batched_background_migration/ci_database_worker.rb @@ -4,6 +4,10 @@ module Database class CiDatabaseWorker # rubocop:disable Scalability/IdempotentWorker include SingleDatabaseWorker + def self.enabled? + Feature.enabled?(:execute_batched_migrations_on_schedule_ci_database, type: :ops, default_enabled: :yaml) + end + def self.tracking_database @tracking_database ||= Gitlab::Database::CI_DATABASE_NAME end diff --git a/app/workers/database/batched_background_migration/single_database_worker.rb b/app/workers/database/batched_background_migration/single_database_worker.rb index 78c82a6549f..aeadda4b8e1 100644 --- a/app/workers/database/batched_background_migration/single_database_worker.rb +++ b/app/workers/database/batched_background_migration/single_database_worker.rb @@ -23,6 +23,10 @@ module Database def tracking_database raise NotImplementedError, "#{self.name} does not implement #{__method__}" end + + def enabled? + raise NotImplementedError, "#{self.name} does not implement #{__method__}" + end # :nocov: def lease_key @@ -41,7 +45,7 @@ module Database end Gitlab::Database::SharedModel.using_connection(base_model.connection) do - break unless Feature.enabled?(:execute_batched_migrations_on_schedule, type: :ops, default_enabled: :yaml) && active_migration + break unless self.class.enabled? && active_migration with_exclusive_lease(active_migration.interval) do # Now that we have the exclusive lease, reload migration in case another process has changed it. diff --git a/app/workers/database/batched_background_migration_worker.rb b/app/workers/database/batched_background_migration_worker.rb index 29804be832d..6a41fe70915 100644 --- a/app/workers/database/batched_background_migration_worker.rb +++ b/app/workers/database/batched_background_migration_worker.rb @@ -4,6 +4,10 @@ module Database class BatchedBackgroundMigrationWorker # rubocop:disable Scalability/IdempotentWorker include BatchedBackgroundMigration::SingleDatabaseWorker + def self.enabled? + Feature.enabled?(:execute_batched_migrations_on_schedule, type: :ops, default_enabled: :yaml) + end + def self.tracking_database @tracking_database ||= Gitlab::Database::MAIN_DATABASE_NAME.to_sym end diff --git a/app/workers/database/ci_namespace_mirrors_consistency_check_worker.rb b/app/workers/database/ci_namespace_mirrors_consistency_check_worker.rb new file mode 100644 index 00000000000..2b4253947ac --- /dev/null +++ b/app/workers/database/ci_namespace_mirrors_consistency_check_worker.rb @@ -0,0 +1,28 @@ +# frozen_string_literal: true + +module Database + class CiNamespaceMirrorsConsistencyCheckWorker + include ApplicationWorker + include CronjobQueue # rubocop: disable Scalability/CronWorkerContext + + sidekiq_options retry: false + feature_category :sharding + data_consistency :sticky + idempotent! + + version 1 + + def perform + return if Feature.disabled?(:ci_namespace_mirrors_consistency_check, default_enabled: :yaml) + + results = ConsistencyCheckService.new( + source_model: Namespace, + target_model: Ci::NamespaceMirror, + source_columns: %w[id traversal_ids], + target_columns: %w[namespace_id traversal_ids] + ).execute + + log_extra_metadata_on_done(:results, results) + end + end +end diff --git a/app/workers/database/ci_project_mirrors_consistency_check_worker.rb b/app/workers/database/ci_project_mirrors_consistency_check_worker.rb new file mode 100644 index 00000000000..e9413256617 --- /dev/null +++ b/app/workers/database/ci_project_mirrors_consistency_check_worker.rb @@ -0,0 +1,28 @@ +# frozen_string_literal: true + +module Database + class CiProjectMirrorsConsistencyCheckWorker + include ApplicationWorker + include CronjobQueue # rubocop: disable Scalability/CronWorkerContext + + sidekiq_options retry: false + feature_category :sharding + data_consistency :sticky + idempotent! + + version 1 + + def perform + return if Feature.disabled?(:ci_project_mirrors_consistency_check, default_enabled: :yaml) + + results = ConsistencyCheckService.new( + source_model: Project, + target_model: Ci::ProjectMirror, + source_columns: %w[id namespace_id], + target_columns: %w[project_id namespace_id] + ).execute + + log_extra_metadata_on_done(:results, results) + end + end +end diff --git a/app/workers/delete_stored_files_worker.rb b/app/workers/delete_stored_files_worker.rb index d1080c8df64..86167a7fafe 100644 --- a/app/workers/delete_stored_files_worker.rb +++ b/app/workers/delete_stored_files_worker.rb @@ -7,7 +7,7 @@ class DeleteStoredFilesWorker # rubocop:disable Scalability/IdempotentWorker sidekiq_options retry: 3 - feature_category_not_owned! + feature_category :not_owned # rubocop:todo Gitlab/AvoidFeatureCategoryNotOwned loggable_arguments 0 def perform(class_name, keys) diff --git a/app/workers/environments/auto_stop_worker.rb b/app/workers/environments/auto_stop_worker.rb index 672a4f4121e..aee6e977550 100644 --- a/app/workers/environments/auto_stop_worker.rb +++ b/app/workers/environments/auto_stop_worker.rb @@ -10,8 +10,10 @@ module Environments def perform(environment_id, params = {}) Environment.find_by_id(environment_id).try do |environment| - user = environment.stop_action&.user - environment.stop_with_action!(user) + stop_actions = environment.stop_actions + + user = stop_actions.last&.user + environment.stop_with_actions!(user) end end end diff --git a/app/workers/flush_counter_increments_worker.rb b/app/workers/flush_counter_increments_worker.rb index c4a3a5283cc..e21a7ee35e7 100644 --- a/app/workers/flush_counter_increments_worker.rb +++ b/app/workers/flush_counter_increments_worker.rb @@ -12,7 +12,10 @@ class FlushCounterIncrementsWorker sidekiq_options retry: 3 - feature_category_not_owned! + # The increments in `ProjectStatistics` are owned by several teams depending + # on the counter + feature_category :not_owned # rubocop:disable Gitlab/AvoidFeatureCategoryNotOwned + urgency :low deduplicate :until_executing, including_scheduled: true diff --git a/app/workers/namespaces/invite_team_email_worker.rb b/app/workers/namespaces/invite_team_email_worker.rb deleted file mode 100644 index eabf33a7fba..00000000000 --- a/app/workers/namespaces/invite_team_email_worker.rb +++ /dev/null @@ -1,22 +0,0 @@ -# frozen_string_literal: true - -module Namespaces - class InviteTeamEmailWorker # rubocop:disable Scalability/IdempotentWorker - include ApplicationWorker - - data_consistency :always - - feature_category :experimentation_activation - urgency :low - - def perform(group_id, user_id) - # rubocop: disable CodeReuse/ActiveRecord - user = User.find_by(id: user_id) - group = Group.find_by(id: group_id) - # rubocop: enable CodeReuse/ActiveRecord - return unless user && group - - Namespaces::InviteTeamEmailService.send_email(user, group) - end - end -end diff --git a/app/workers/namespaces/root_statistics_worker.rb b/app/workers/namespaces/root_statistics_worker.rb index b97dbca2c1c..e1271dae335 100644 --- a/app/workers/namespaces/root_statistics_worker.rb +++ b/app/workers/namespaces/root_statistics_worker.rb @@ -20,8 +20,17 @@ module Namespaces Namespaces::StatisticsRefresherService.new.execute(namespace) namespace.aggregation_schedule.destroy + + notify_storage_usage(namespace) rescue ::Namespaces::StatisticsRefresherService::RefresherError, ActiveRecord::RecordNotFound => ex Gitlab::ErrorTracking.track_exception(ex, namespace_id: namespace_id, namespace: namespace&.full_path) end + + private + + def notify_storage_usage(namespace) + end end end + +Namespaces::RootStatisticsWorker.prepend_mod_with('Namespaces::RootStatisticsWorker') diff --git a/app/workers/object_storage/background_move_worker.rb b/app/workers/object_storage/background_move_worker.rb index 2204e504702..bb51f0d7e1f 100644 --- a/app/workers/object_storage/background_move_worker.rb +++ b/app/workers/object_storage/background_move_worker.rb @@ -8,7 +8,7 @@ module ObjectStorage include ObjectStorageQueue sidekiq_options retry: 5 - feature_category_not_owned! + feature_category :not_owned # rubocop:todo Gitlab/AvoidFeatureCategoryNotOwned loggable_arguments 0, 1, 2, 3 def perform(uploader_class_name, subject_class_name, file_field, subject_id) diff --git a/app/workers/object_storage/migrate_uploads_worker.rb b/app/workers/object_storage/migrate_uploads_worker.rb index ea4a90cf9d2..b7d938e6b68 100644 --- a/app/workers/object_storage/migrate_uploads_worker.rb +++ b/app/workers/object_storage/migrate_uploads_worker.rb @@ -10,7 +10,7 @@ module ObjectStorage sidekiq_options retry: 3 include ObjectStorageQueue - feature_category_not_owned! + feature_category :not_owned # rubocop:todo Gitlab/AvoidFeatureCategoryNotOwned loggable_arguments 0, 1, 2, 3 SanityCheckError = Class.new(StandardError) diff --git a/app/workers/project_export_worker.rb b/app/workers/project_export_worker.rb index e3f8c4bcd9d..ee892d43313 100644 --- a/app/workers/project_export_worker.rb +++ b/app/workers/project_export_worker.rb @@ -8,7 +8,7 @@ class ProjectExportWorker # rubocop:disable Scalability/IdempotentWorker feature_category :importers worker_resource_boundary :memory - urgency :throttled + urgency :low loggable_arguments 2, 3 sidekiq_options retry: false, dead: false sidekiq_options status_expiration: StuckExportJobsWorker::EXPORT_JOBS_EXPIRATION @@ -21,7 +21,10 @@ class ProjectExportWorker # rubocop:disable Scalability/IdempotentWorker export_job&.start - ::Projects::ImportExport::ExportService.new(project, current_user, params).execute(after_export) + export_service = ::Projects::ImportExport::ExportService.new(project, current_user, params) + export_service.execute(after_export) + + log_exporters_duration(export_service) export_job&.finish rescue ActiveRecord::RecordNotFound => e @@ -46,4 +49,13 @@ class ProjectExportWorker # rubocop:disable Scalability/IdempotentWorker def log_failure(project_id, ex) logger.error("Failed to export project #{project_id}: #{ex.message}") end + + def log_exporters_duration(export_service) + export_service.exporters.each do |exporter| + exporter_key = "#{exporter.class.name.demodulize.underscore}_duration_s".to_sym # e.g. uploads_saver_duration_s + exporter_duration = exporter.duration_s&.round(6) + + log_extra_metadata_on_done(exporter_key, exporter_duration) + end + end end diff --git a/app/workers/projects/record_target_platforms_worker.rb b/app/workers/projects/record_target_platforms_worker.rb new file mode 100644 index 00000000000..5b1f85ecca0 --- /dev/null +++ b/app/workers/projects/record_target_platforms_worker.rb @@ -0,0 +1,55 @@ +# frozen_string_literal: true + +module Projects + class RecordTargetPlatformsWorker + include ApplicationWorker + include ExclusiveLeaseGuard + + LEASE_TIMEOUT = 1.hour.to_i + APPLE_PLATFORM_LANGUAGES = %w(swift objective-c).freeze + + feature_category :experimentation_activation + data_consistency :always + deduplicate :until_executed + urgency :low + idempotent! + + def perform(project_id) + @project = Project.find_by_id(project_id) + + return unless project + return unless uses_apple_platform_languages? + + try_obtain_lease do + @target_platforms = Projects::RecordTargetPlatformsService.new(project).execute + log_target_platforms_metadata + end + end + + private + + attr_reader :target_platforms, :project + + def uses_apple_platform_languages? + project.repository_languages.with_programming_language(*APPLE_PLATFORM_LANGUAGES).present? + end + + def log_target_platforms_metadata + return unless target_platforms.present? + + log_extra_metadata_on_done(:target_platforms, target_platforms) + end + + def lease_key + @lease_key ||= "#{self.class.name.underscore}:#{project.id}" + end + + def lease_timeout + LEASE_TIMEOUT + end + + def lease_release? + false + end + end +end diff --git a/app/workers/quality/test_data_cleanup_worker.rb b/app/workers/quality/test_data_cleanup_worker.rb deleted file mode 100644 index 68b36cacbbf..00000000000 --- a/app/workers/quality/test_data_cleanup_worker.rb +++ /dev/null @@ -1,33 +0,0 @@ -# frozen_string_literal: true - -module Quality - class TestDataCleanupWorker - include ApplicationWorker - - data_consistency :always - feature_category :quality_management - urgency :low - - include CronjobQueue - idempotent! - - KEEP_RECENT_DATA_DAY = 3 - GROUP_PATH_PATTERN = 'test-group-fulfillment' - GROUP_OWNER_EMAIL_PATTERN = %w(test-user- gitlab-qa-user qa-user-).freeze - - # Remove test groups generated in E2E tests on gstg - # rubocop: disable CodeReuse/ActiveRecord - def perform - return unless Gitlab.staging? - - Group.where('path like ?', "#{GROUP_PATH_PATTERN}%").where('created_at < ?', KEEP_RECENT_DATA_DAY.days.ago).each do |group| - next unless GROUP_OWNER_EMAIL_PATTERN.any? { |pattern| group.owners.first.email.include?(pattern) } - - with_context(namespace: group, user: group.owners.first) do - Groups::DestroyService.new(group, group.owners.first).execute - end - end - end - # rubocop: enable CodeReuse/ActiveRecord - end -end |