1 files changed, 33 insertions, 14 deletions
diff --git a/lib/backup/gitaly_backup.rb b/lib/backup/gitaly_backup.rb
index b104beed39c..8ac09e94004 100644
--- a/lib/backup/gitaly_backup.rb
+++ b/lib/backup/gitaly_backup.rb
@@ -2,11 +2,17 @@
 
 module Backup
   # Backup and restores repositories using gitaly-backup
+  #
+  # gitaly-backup can work in parallel and accepts a list of repositories
+  # through input pipe using a specific json format for both backup and restore
   class GitalyBackup
-    def initialize(progress, parallel: nil, parallel_storage: nil)
+    # @param [StringIO] progress IO interface to output progress
+    # @param [Integer] max_parallelism max parallelism when running backups
+    # @param [Integer] storage_parallelism max parallelism per storage (is affected by max_parallelism)
+    def initialize(progress, max_parallelism: nil, storage_parallelism: nil)
       @progress = progress
-      @parallel = parallel
-      @parallel_storage = parallel_storage
+      @max_parallelism = max_parallelism
+      @storage_parallelism = storage_parallelism
     end
 
     def start(type)
@@ -22,20 +28,20 @@ module Backup
                 end
 
       args = []
-      args += ['-parallel', @parallel.to_s] if @parallel
-      args += ['-parallel-storage', @parallel_storage.to_s] if @parallel_storage
+      args += ['-parallel', @max_parallelism.to_s] if @max_parallelism
+      args += ['-parallel-storage', @storage_parallelism.to_s] if @storage_parallelism
 
-      @stdin, stdout, @thread = Open3.popen2(build_env, bin_path, command, '-path', backup_repos_path, *args)
+      @input_stream, stdout, @thread = Open3.popen2(build_env, bin_path, command, '-path', backup_repos_path, *args)
 
       @out_reader = Thread.new do
         IO.copy_stream(stdout, @progress)
       end
     end
 
-    def wait
+    def finish!
       return unless started?
 
-      @stdin.close
+      @input_stream.close
       [@thread, @out_reader].each(&:join)
       status =  @thread.value
 
@@ -49,12 +55,7 @@ module Backup
 
       repository = repo_type.repository_for(container)
 
-      @stdin.puts({
-        storage_name: repository.storage,
-        relative_path: repository.relative_path,
-        gl_project_path: repository.gl_project_path,
-        always_create: repo_type.project?
-      }.merge(Gitlab::GitalyClient.connection_data(repository.storage)).to_json)
+      schedule_backup_job(repository, always_create: repo_type.project?)
     end
 
     def parallel_enqueue?
@@ -63,6 +64,24 @@ module Backup
 
     private
 
+    # Schedule a new backup job through a non-blocking JSON based pipe protocol
+    #
+    # @see https://gitlab.com/gitlab-org/gitaly/-/blob/master/doc/gitaly-backup.md
+    def schedule_backup_job(repository, always_create:)
+      connection_params = Gitlab::GitalyClient.connection_data(repository.storage)
+
+      json_job = {
+        address: connection_params['address'],
+        token: connection_params['token'],
+        storage_name: repository.storage,
+        relative_path: repository.relative_path,
+        gl_project_path: repository.gl_project_path,
+        always_create: always_create
+      }.to_json
+
+      @input_stream.puts(json_job)
+    end
+
     def build_env
       {
         'SSL_CERT_FILE' => OpenSSL::X509::DEFAULT_CERT_FILE,