Group Migration - Concurrent model
# Problem to Solve
With the complexity of Group Migration increasing with more relations/associations being imported, a concurrent model becomes required to
* Reduce the memory consumption on smaller processes
* Provide better user, and developer, experience by having more granular status about the process
* Improve the Migration process resilience, since retrying a small parts of the migration is more feasible than the whole migration
* Provide an easy way to handle Network Rate Limits (related to retry smaller parts of the migration)
# Proposed Solution
In the spike (https://gitlab.com/gitlab-org/gitlab/-/merge_requests/54970), it was discussed how to approach the concurrent model to be adopt in the Gitlab Group Migration and achieve the goals above.
## ETL
The chosen base architecture, [ETL](https://www.ibm.com/cloud/learn/etl), already provided us with a logical way to split the migration process in Pipelines. Each pipeline is responsible for migrating a single _Relation_[1] from one GitLab instance to another. There is a hierarchical dependency among the Pipelines, for instance Epics cannot be migrated before its Group. But, some Pipelines have the same dependency, which enable them to run in parallel.
1. A _Relation_ in this context can be a Database Table, a Data association among tables or some other data structure. Examples of Relations:
- Group: represents a database table
- EpicEvents: represents a database table but in relation with the already migrated Epic
- Members: Does not represents a table, instead we fetch Group member information to locate existing users in the target GitLab instance and create the membership with the migrated Group.
## Stages
With this in mind, we decided to organize the Migration process in Stages, with the following characteristics:
- The stages runs sequentially in order
- The pipelines of the same stage runs in parallel
- The next stage only starts when all the current stage pipelines are either finished or failed.
This way we can have something like
```mermaid
flowchart LR
subgraph stage0
StartS0["Start"] --> Group --> FinishS0["Finished"]
S0Note["All the following <br> Stages depends <br> on Group"]
S0Note:::note
end
subgraph stage1
StartS1["Start"] --> Subgroup --> FinishSubgroup["Finished"]
StartS1 --> Members --> FinishMembers["Finished"]
StartS1 --> Labels --> FinishLabels["Finished"]
end
subgraph stage2
StartS2["Start"] --> Epic --> FinishEpic["Finished"]
S2Note["Epics depends on<br>Labels and Members<br>That's why it cannot<br>run in the stage1"]
S2Note:::note
end
subgraph stage3
StartS3["Start"] --> EpicAwardEmoji --> FinishEpicAwardEmoji["Finish"]
StartS3 --> EpicEvents --> FinishEpicEvents["Finish"]
S3Note["Epics subrelations<br>Depends on Epics"]
S3Note:::note
end
Start --> stage0
stage0 --> stage1
stage1 --> stage2
stage2 --> stage3
classDef note fill:#ffd,stroke-width:0;
```
### The spike implementation
<details>
<summary>Sequence Diagram</summary>
```mermaid
sequenceDiagram
User ->>+ Controller: Import Group1 and Group 2
Controller ->>+ BulkImportService: Import Group1 and Group 2
Note over BulkImportService, BulkImportService: Creates the BulkImport <br> one Entity for each top level Group <br> and the Trackers for each Entity
BulkImportService -->> BulkImportWorker: perform_async
BulkImportService ->>- User: Import Scheduled
par
BulkImportWorker -->> BulkImportEntityWorker: perform_async (Group1 Entity)
and
BulkImportWorker -->> BulkImportEntityWorker: perform_async (Group2 Entity)
end
BulkImportWorker -->> BulkImportWorker: perform_async
Note over BulkImportWorker, BulkImportWorker: Check if there's more <br> Entities to process <br> (subgroups)
par For each pipeline from stage
BulkImportEntityWorker -->>+ BulkImportsPipelineWorker: perform_async(pipeline, entity)
and
BulkImportEntityWorker -->>+ BulkImportsPipelineWorker: perform_async(pipeline, entity)
end
BulkImportsPipelineWorker ->>+ Pipeline: run(tracker)
Pipeline ->>- BulkImportsPipelineWorker: finished
BulkImportsPipelineWorker -->>- BulkImportEntityWorker: perform_async(entity)
Note over BulkImportsPipelineWorker, BulkImportEntityWorker: If there is a next stage <br>Process the pipelines of the next stage
participant BulkImportEntityWorker as BulkImports::EntityWorker
participant BulkImportsPipelineWorker as BulkImports::PipelineWorker
```
</details>
## Next steps
Based on the learnings in the spike I propose to iterate in the following steps: \
Each step would be a different issue/MR
1. Add `pipeline_name`, `stage` and `status` to `BulkImports::Tracker`
- Make `BulkImports::Tracker#relation` column nullable;
- Replace `BulkImports::Tracker#relation` logic by `BulkImports::Tracker#pipeline_name`
- Pass the `BulkImports::Tracker` to the pipeline, via `BulkImports::Pipeline::Context`
- Update the `BulkImports::Tracker#status` accordingly
1. Introduce the concurrency by running each pipeline on its own job
- Replace `BulkImports::Groups::Importer` by `BulkImports::Groups::Stages` to keep the Stages definition
- Create the current Entity's trackers for each pipeline, something like
```ruby
# app/workers/bulk_import_worker.rb
def perform(bulk_import_id)
# ...
created_entities.first(next_batch_size).each do |entity|
BulkImports::Groups::Stages.create_trackers_for(entity)
BulkImports::EntityWorker.perform_async(entity.id)
entity.start!
end
end
```
- Create the `BulkImports::PipelineWorker`
1. Remove the `BulkImports::Tracker#relation` column
epic