Module: Gitlab::GithubImport::ParallelScheduling
- Includes:
- JobDelayCalculator
- Included in:
- Importer::Attachments::BaseImporter, Importer::CollaboratorsImporter, Importer::DiffNotesImporter, Importer::IssueEventsImporter, Importer::IssuesImporter, Importer::LfsObjectsImporter, Importer::NotesImporter, Importer::ProtectedBranchesImporter, Importer::PullRequests::AllMergedByImporter, Importer::PullRequests::ReviewRequestsImporter, Importer::PullRequests::ReviewsImporter, Importer::PullRequestsImporter, Importer::SingleEndpointDiffNotesImporter, Importer::SingleEndpointIssueEventsImporter, Importer::SingleEndpointIssueNotesImporter, Importer::SingleEndpointMergeRequestNotesImporter
- Defined in:
- lib/gitlab/github_import/parallel_scheduling.rb
Constant Summary collapse
- ALREADY_IMPORTED_CACHE_KEY =
The base cache key to use for tracking already imported objects.
'github-importer/already-imported/%{project}/%{collection}'
- JOB_WAITER_CACHE_KEY =
The base cache key to use for storing job waiter key
'github-importer/job-waiter/%{project}/%{collection}'
- JOB_WAITER_REMAINING_CACHE_KEY =
The base cache key to use for storing job waiter remaining jobs
'github-importer/job-waiter-remaining/%{project}/%{collection}'
Instance Attribute Summary collapse
-
#already_imported_cache_key ⇒ Object
readonly
Returns the value of attribute already_imported_cache_key.
-
#client ⇒ Object
readonly
Returns the value of attribute client.
-
#job_waiter_cache_key ⇒ Object
readonly
Returns the value of attribute job_waiter_cache_key.
-
#job_waiter_remaining_cache_key ⇒ Object
readonly
Returns the value of attribute job_waiter_remaining_cache_key.
-
#page_counter ⇒ Object
readonly
Returns the value of attribute page_counter.
-
#project ⇒ Object
readonly
Returns the value of attribute project.
Instance Method Summary collapse
- #abort_on_failure ⇒ Object
-
#already_imported?(object) ⇒ Boolean
Returns true if the given object has already been imported, false otherwise.
-
#collection_method ⇒ Object
The name of the method to call to retrieve the data to import.
-
#collection_options ⇒ Object
Any options to be passed to the method used for retrieving the data to import.
-
#each_object_to_import ⇒ Object
The method that will be called for traversing through all the objects to import, yielding them to the supplied block.
- #execute ⇒ Object
-
#id_for_already_imported_cache(object) ⇒ Object
Returns the ID to use for the cache used for checking if an object has already been imported or not.
-
#importer_class ⇒ Object
The class to use for importing objects when importing them sequentially.
- #increment_object_counter?(_object) ⇒ Boolean
-
#initialize(project, client, parallel: true) ⇒ Object
project - An instance of ‘Project`.
-
#mark_as_imported(object) ⇒ Object
Marks the given object as “already imported”.
- #object_type ⇒ Object
- #parallel? ⇒ Boolean
-
#parallel_import ⇒ Object
Imports all objects in parallel by scheduling a Sidekiq job for every individual object.
-
#representation_class ⇒ Object
The class used for converting API responses to Hashes when performing the import.
-
#sequential_import ⇒ Object
Imports all the objects in sequence in the current thread.
-
#sidekiq_worker_class ⇒ Object
The Sidekiq worker class used for scheduling the importing of objects in parallel.
- #spread_parallel_import ⇒ Object
Methods included from JobDelayCalculator
Instance Attribute Details
#already_imported_cache_key ⇒ Object (readonly)
Returns the value of attribute already_imported_cache_key.
8 9 10 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 8 def already_imported_cache_key @already_imported_cache_key end |
#client ⇒ Object (readonly)
Returns the value of attribute client.
8 9 10 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 8 def client @client end |
#job_waiter_cache_key ⇒ Object (readonly)
Returns the value of attribute job_waiter_cache_key.
8 9 10 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 8 def job_waiter_cache_key @job_waiter_cache_key end |
#job_waiter_remaining_cache_key ⇒ Object (readonly)
Returns the value of attribute job_waiter_remaining_cache_key.
8 9 10 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 8 def job_waiter_remaining_cache_key @job_waiter_remaining_cache_key end |
#page_counter ⇒ Object (readonly)
Returns the value of attribute page_counter.
8 9 10 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 8 def page_counter @page_counter end |
#project ⇒ Object (readonly)
Returns the value of attribute project.
8 9 10 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 8 def project @project end |
Instance Method Details
#abort_on_failure ⇒ Object
202 203 204 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 202 def abort_on_failure false end |
#already_imported?(object) ⇒ Boolean
Returns true if the given object has already been imported, false otherwise.
object - The object to check.
155 156 157 158 159 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 155 def already_imported?(object) id = id_for_already_imported_cache(object) Gitlab::Cache::Import::Caching.set_includes?(already_imported_cache_key, id) end |
#collection_method ⇒ Object
The name of the method to call to retrieve the data to import.
198 199 200 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 198 def collection_method raise NotImplementedError end |
#collection_options ⇒ Object
Any options to be passed to the method used for retrieving the data to import.
208 209 210 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 208 def {} end |
#each_object_to_import ⇒ Object
The method that will be called for traversing through all the objects to import, yielding them to the supplied block.
111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 111 def each_object_to_import repo = project.import_source # We inject the page number here to make sure that all importers always # start where they left off. Simply starting over wouldn't work for # repositories with a lot of data (e.g. tens of thousands of comments). = .merge(page: page_counter.current) client.each_page(collection_method, repo, ) do |page| # Technically it's possible that the same work is performed multiple # times, as Sidekiq doesn't guarantee there will ever only be one # instance of a job. In such a scenario it's possible for one job to # have a lower page number (e.g. 5) compared to another (e.g. 10). In # this case we skip over all the objects until we have caught up, # reducing the number of duplicate jobs scheduled by the provided # block. next unless page_counter.set(page.number) page.objects.each do |object| object = object.to_h next if already_imported?(object) if increment_object_counter?(object) Gitlab::GithubImport::ObjectCounter.increment(project, object_type, :fetched) end yield object # We mark the object as imported immediately so we don't end up # scheduling it multiple times. mark_as_imported(object) end end end |
#execute ⇒ Object
41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 41 def execute info(project.id, message: "starting importer") retval = if parallel? parallel_import else sequential_import end # Once we have completed all work we can remove our "already exists" # cache so we don't put too much pressure on Redis. # # We don't immediately remove it since it's technically possible for # other instances of this job to still run, instead we set the # expiration time to a lower value. This prevents the other jobs from # still scheduling duplicates while. Since all work has already been # completed those jobs will just cycle through any remaining pages while # not scheduling anything. Gitlab::Cache::Import::Caching.expire(already_imported_cache_key, Gitlab::Cache::Import::Caching::SHORTER_TIMEOUT) info(project.id, message: "importer finished") retval rescue StandardError => e Gitlab::Import::ImportFailureService.track( project_id: project.id, error_source: self.class.name, exception: e, fail_import: abort_on_failure, metrics: true ) raise(e) end |
#id_for_already_imported_cache(object) ⇒ Object
Returns the ID to use for the cache used for checking if an object has already been imported or not.
object - The object we may want to import.
176 177 178 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 176 def id_for_already_imported_cache(object) raise NotImplementedError end |
#importer_class ⇒ Object
The class to use for importing objects when importing them sequentially.
187 188 189 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 187 def importer_class raise NotImplementedError end |
#increment_object_counter?(_object) ⇒ Boolean
147 148 149 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 147 def increment_object_counter?(_object) true end |
#initialize(project, client, parallel: true) ⇒ Object
project - An instance of ‘Project`. client - An instance of `Gitlab::GithubImport::Client`. parallel - When set to true the objects will be imported in parallel.
24 25 26 27 28 29 30 31 32 33 34 35 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 24 def initialize(project, client, parallel: true) @project = project @client = client @parallel = parallel @page_counter = PageCounter.new(project, collection_method) @already_imported_cache_key = ALREADY_IMPORTED_CACHE_KEY % { project: project.id, collection: collection_method } @job_waiter_cache_key = JOB_WAITER_CACHE_KEY % { project: project.id, collection: collection_method } @job_waiter_remaining_cache_key = JOB_WAITER_REMAINING_CACHE_KEY % { project: project.id, collection: collection_method } end |
#mark_as_imported(object) ⇒ Object
Marks the given object as “already imported”.
162 163 164 165 166 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 162 def mark_as_imported(object) id = id_for_already_imported_cache(object) Gitlab::Cache::Import::Caching.set_add(already_imported_cache_key, id) end |
#object_type ⇒ Object
168 169 170 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 168 def object_type raise NotImplementedError end |
#parallel? ⇒ Boolean
37 38 39 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 37 def parallel? @parallel end |
#parallel_import ⇒ Object
Imports all objects in parallel by scheduling a Sidekiq job for every individual object.
87 88 89 90 91 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 87 def parallel_import raise 'Batch settings must be defined for parallel import' if parallel_import_batch.blank? spread_parallel_import end |
#representation_class ⇒ Object
The class used for converting API responses to Hashes when performing the import.
182 183 184 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 182 def representation_class raise NotImplementedError end |
#sequential_import ⇒ Object
Imports all the objects in sequence in the current thread.
77 78 79 80 81 82 83 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 77 def sequential_import each_object_to_import do |object| repr = object_representation(object) importer_class.new(repr, project, client).execute end end |
#sidekiq_worker_class ⇒ Object
The Sidekiq worker class used for scheduling the importing of objects in parallel.
193 194 195 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 193 def sidekiq_worker_class raise NotImplementedError end |
#spread_parallel_import ⇒ Object
93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 |
# File 'lib/gitlab/github_import/parallel_scheduling.rb', line 93 def spread_parallel_import enqueued_job_counter = 0 each_object_to_import do |object| repr = object_representation(object) job_delay = calculate_job_delay(enqueued_job_counter) sidekiq_worker_class.perform_in(job_delay, project.id, repr.to_hash, job_waiter.key) enqueued_job_counter += 1 job_waiter.jobs_remaining = Gitlab::Cache::Import::Caching.increment(job_waiter_remaining_cache_key) end job_waiter end |