diff options
Diffstat (limited to 'app/workers/gitlab')
4 files changed, 208 insertions, 0 deletions
diff --git a/app/workers/gitlab/export/prune_project_export_jobs_worker.rb b/app/workers/gitlab/export/prune_project_export_jobs_worker.rb new file mode 100644 index 00000000000..9a3c0c80f85 --- /dev/null +++ b/app/workers/gitlab/export/prune_project_export_jobs_worker.rb @@ -0,0 +1,23 @@ +# frozen_string_literal: true + +module Gitlab + module Export + class PruneProjectExportJobsWorker + include ApplicationWorker + + # rubocop:disable Scalability/CronWorkerContext + # This worker updates several import states inline and does not schedule + # other jobs. So no context needed + include CronjobQueue + # rubocop:enable Scalability/CronWorkerContext + + feature_category :importers + data_consistency :always + idempotent! + + def perform + ProjectExportJob.prune_expired_jobs + end + end + end +end diff --git a/app/workers/gitlab/github_gists_import/finish_import_worker.rb b/app/workers/gitlab/github_gists_import/finish_import_worker.rb new file mode 100644 index 00000000000..1989b6314ea --- /dev/null +++ b/app/workers/gitlab/github_gists_import/finish_import_worker.rb @@ -0,0 +1,46 @@ +# frozen_string_literal: true + +module Gitlab + module GithubGistsImport + class FinishImportWorker + include ApplicationWorker + + data_consistency :always + queue_namespace :github_gists_importer + feature_category :importers + idempotent! + + sidekiq_options dead: false, retry: 5 + + sidekiq_retries_exhausted do |msg, _| + Gitlab::GithubGistsImport::Status.new(msg['args'][0]).fail! + end + + INTERVAL = 30.seconds.to_i + BLOCKING_WAIT_TIME = 5 + + def perform(user_id, waiter_key, remaining) + waiter = wait_for_jobs(waiter_key, remaining) + + if waiter.nil? + Gitlab::GithubGistsImport::Status.new(user_id).finish! + + Gitlab::GithubImport::Logger.info(user_id: user_id, message: 'GitHub Gists import finished') + else + self.class.perform_in(INTERVAL, user_id, waiter.key, waiter.jobs_remaining) + end + end + + private + + def wait_for_jobs(key, remaining) + waiter = JobWaiter.new(remaining, key) + waiter.wait(BLOCKING_WAIT_TIME) + + return if waiter.jobs_remaining == 0 + + waiter + end + end + end +end diff --git a/app/workers/gitlab/github_gists_import/import_gist_worker.rb b/app/workers/gitlab/github_gists_import/import_gist_worker.rb new file mode 100644 index 00000000000..7e2b3709597 --- /dev/null +++ b/app/workers/gitlab/github_gists_import/import_gist_worker.rb @@ -0,0 +1,75 @@ +# frozen_string_literal: true + +module Gitlab + module GithubGistsImport + class ImportGistWorker # rubocop:disable Scalability/IdempotentWorker + include ApplicationWorker + include Gitlab::NotifyUponDeath + + data_consistency :always + queue_namespace :github_gists_importer + feature_category :importers + + sidekiq_options dead: false, retry: 5 + + def perform(user_id, gist_hash, notify_key) + gist = ::Gitlab::GithubGistsImport::Representation::Gist.from_json_hash(gist_hash) + + with_logging(user_id, gist.github_identifiers) do + result = importer_class.new(gist, user_id).execute + error(user_id, result.errors, gist.github_identifiers) unless result.success? + + JobWaiter.notify(notify_key, jid) + end + rescue StandardError => e + log_and_track_error(user_id, e, gist.github_identifiers) + + raise + end + + private + + def importer_class + ::Gitlab::GithubGistsImport::Importer::GistImporter + end + + def with_logging(user_id, gist_id) + info(user_id, 'start importer', gist_id) + + yield + + info(user_id, 'importer finished', gist_id) + end + + def log_and_track_error(user_id, exception, gist_id) + error(user_id, exception.message, gist_id) + + Gitlab::ErrorTracking.track_exception(exception, + import_type: :github_gists, + user_id: user_id + ) + end + + def error(user_id, error_message, gist_id) + attributes = { + user_id: user_id, + github_identifiers: gist_id, + message: 'importer failed', + 'error.message': error_message + } + + Gitlab::GithubImport::Logger.error(structured_payload(attributes)) + end + + def info(user_id, message, gist_id) + attributes = { + user_id: user_id, + message: message, + github_identifiers: gist_id + } + + Gitlab::GithubImport::Logger.info(structured_payload(attributes)) + end + end + end +end diff --git a/app/workers/gitlab/github_gists_import/start_import_worker.rb b/app/workers/gitlab/github_gists_import/start_import_worker.rb new file mode 100644 index 00000000000..33c91611719 --- /dev/null +++ b/app/workers/gitlab/github_gists_import/start_import_worker.rb @@ -0,0 +1,64 @@ +# frozen_string_literal: true + +module Gitlab + module GithubGistsImport + class StartImportWorker # rubocop:disable Scalability/IdempotentWorker + include ApplicationWorker + + data_consistency :always + queue_namespace :github_gists_importer + feature_category :importers + + sidekiq_options dead: false, retry: 5 + + worker_has_external_dependencies! + + sidekiq_retries_exhausted do |msg, _| + Gitlab::GithubGistsImport::Status.new(msg['args'][0]).fail! + + user = User.find(msg['args'][0]) + Gitlab::GithubImport::PageCounter.new(user, :gists, 'github-gists-importer').expire! + end + + def perform(user_id, encrypted_token) + logger.info(structured_payload(user_id: user_id, message: 'starting importer')) + + user = User.find(user_id) + decrypted_token = Gitlab::CryptoHelper.aes256_gcm_decrypt(encrypted_token) + result = Gitlab::GithubGistsImport::Importer::GistsImporter.new(user, decrypted_token).execute + + if result.success? + schedule_finish_worker(user_id, result.waiter) + elsif result.next_attempt_in + schedule_next_attempt(result.next_attempt_in, user_id, encrypted_token) + else + log_error_and_raise!(user_id, result.error) + end + end + + private + + def schedule_finish_worker(user_id, waiter) + logger.info(structured_payload(user_id: user_id, message: 'importer finished')) + + Gitlab::GithubGistsImport::FinishImportWorker.perform_async(user_id, waiter.key, waiter.jobs_remaining) + end + + def schedule_next_attempt(next_attempt_in, user_id, encrypted_token) + logger.info(structured_payload(user_id: user_id, message: 'rate limit reached')) + + self.class.perform_in(next_attempt_in, user_id, encrypted_token) + end + + def log_error_and_raise!(user_id, error) + logger.error(structured_payload(user_id: user_id, message: 'import failed', 'error.message': error.message)) + + raise(error) + end + + def logger + Gitlab::GithubImport::Logger + end + end + end +end |