Migration

nomad/client/migration.py

@@ -120,8 +120,12 @@ def determine_upload_paths(paths, pattern=None):
 @click.option('--pattern', default=None, type=str, help='Interpret the paths as directory and migrate those subdirectory that match the given regexp')
 @click.option('--parallel', default=1, type=int, help='Use the given amount of parallel processes. Default is 1.')
 def package(upload_paths, pattern, parallel):
-    upload_path_queue = multiprocessing.Queue()
-    for upload_path in determine_upload_paths(upload_paths, pattern):
+    upload_paths = determine_upload_paths(upload_paths, pattern)
+    upload_path_queue = multiprocessing.Queue(len(upload_paths))
+
+    print('Package %d uploads with %d processes.' % (len(upload_paths), parallel))
+
+    for upload_path in upload_paths:
         upload_path_queue.put(upload_path)
 
     def package_paths():
@@ -132,7 +136,7 @@ def package(upload_paths, pattern, parallel):
 
         try:
             while True:
-                upload_path = upload_path_queue.get_nowait()
+                upload_path = upload_path_queue.get()
                 migration.package_index(upload_path)
         except queue.Empty:
             pass
@@ -145,6 +149,7 @@ def package(upload_paths, pattern, parallel):
 
     for process in processes:
         process.join()
+    upload_path_queue.close()
 
 
 @migration.command(help='Get an report over all migrated packages.')
@@ -174,9 +179,14 @@ def pid_prefix(prefix: int):
 @click.option('--pattern', default=None, type=str, help='Interpret the paths as directory and migrate those subdirectory that match the given regexp')
 @click.option('--delete-failed', default='', type=str, help='String from N, U, P to determine if empty (N), failed (U), or failed to publish (P) uploads should be deleted or kept for debugging.')
 @click.option('--parallel', default=1, type=int, help='Use the given amount of parallel processes. Default is 1.')
-def upload(upload_paths: list, pattern: str, parallel: int, delete_failed: str):
+@click.option('--create-packages', is_flag=True, help='Indicate that packages should be created, if they do not already exist.')
+def upload(
+        upload_paths: list, pattern: str, parallel: int, delete_failed: str,
+        create_packages: bool):
 
     infrastructure.setup_logging()
     infrastructure.setup_mongo()
 
-    _Migration(threads=parallel).migrate(*determine_upload_paths(upload_paths, pattern), delete_failed=delete_failed)
+    _Migration(threads=parallel).migrate(
+        *determine_upload_paths(upload_paths, pattern), delete_failed=delete_failed,
+        create_packages=create_packages)