Add a test+fix for upload concurrency. (#1701) (#1741)

The request can have a key for the repository and that must be deleted for the underlying
tasks to NOT attempt to create a new repository version, as that is where race conditions most
often occur.

No-Issue

Signed-off-by: James Tanner <[email protected]>
(cherry picked from commit 9900a96)

dev/docker-compose.yml

@@ -82,6 +82,8 @@ services:
       - _base
       - postgres
       - redis
+    deploy:
+      replicas: 2
 
   content-app:
     image: "localhost/galaxy_ng/galaxy_ng:${COMPOSE_PROFILE}${DEV_IMAGE_SUFFIX:-}"

galaxy_ng/app/tasks/publishing.py

@@ -44,8 +44,24 @@ def _upload_collection(**kwargs):
     except AnsibleRepository.DoesNotExist:
         raise RuntimeError(_('Could not find staging repository: "%s"') % STAGING_NAME)
 
+    # The data key can also contain a repository field, which will
+    # trigger the creation functions to attempt to add the CV
+    # to a new repo version and therefore break upload concurrency
+    kwargs['data'].pop('repository')
+
     # kick off the upload and import task via the
     # pulp_ansible.app.serializers.CollectionVersionUploadSerializer serializer
+    # Example structure:
+    #   args: [ansible, CollectionVersionUploadSerializer]
+    #   kwargs:
+    #       repository_pk: <should be removed>
+    #       data:
+    #           sha256:
+    #           artifact:
+    #           repository: <should be removed>
+    #       context:
+    #           filename:
+    #           filename_ns:
     general_create(*general_args, **kwargs)
 
     return repo

galaxy_ng/tests/integration/api/test_upload_concurrency.py

@@ -0,0 +1,70 @@
+import pytest
+
+import concurrent.futures
+
+from ..utils import (
+    AnsibleDistroAndRepo,
+    ansible_galaxy,
+    build_collection,
+    create_unused_namespace,
+    get_client, gen_string,
+)
+
+
+@pytest.mark.standalone_only
+def test_upload_concurrency(ansible_config, settings, galaxy_client):
+
+    total = 10
+
+    config = ansible_config(profile="admin")
+    client = get_client(
+        config=config
+    )
+
+    # make a repo
+    repo = AnsibleDistroAndRepo(
+        client,
+        gen_string(),
+    )
+    repo_data = repo.get_repo()
+    repo_name = repo_data['name']
+
+    # make 10 namespaces
+    namespaces = [create_unused_namespace(client) for x in range(0, total + 1)]
+
+    # make a collection for each namespace
+    artifacts = []
+    for namespace in namespaces:
+        artifact = build_collection(namespace=namespace, name='foo')
+        artifacts.append(artifact)
+
+    server_url = config.get('url').rstrip('/') + '/content/' + repo_data['name'] + '/'
+
+    args_list = [f"collection publish -vvvv {x.filename}" for x in artifacts]
+    kwargs_list = [{'ansible_config': config, 'server_url': server_url} for x in artifacts]
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=total) as executor:
+
+        future_to_args_kwargs = {
+            executor.submit(ansible_galaxy, args, **kwargs): (args, kwargs)
+            for args, kwargs in zip(args_list, kwargs_list)
+        }
+
+        for future in concurrent.futures.as_completed(future_to_args_kwargs):
+            args, kwargs = future_to_args_kwargs[future]
+            try:
+                result = future.result()
+            except Exception as exc:
+                print(f"Function raised an exception: {exc}")
+            else:
+                print(f"Function returned: {result}")
+
+    gc = galaxy_client("admin")
+    cvs = gc.get(
+        (
+            "/api/automation-hub/v3/plugin/ansible/search/collection-versions/"
+            + f"?repository_name={repo_name}"
+        )
+    )
+
+    assert cvs['meta']['count'] == len(artifacts)

galaxy_ng/tests/integration/utils/client_ansible_galaxy_cli.py

@@ -18,6 +18,7 @@ def ansible_galaxy(
     retries=3,
     check_retcode=0,
     server="automation_hub",
+    server_url=None,
     ansible_config=None,
     token=None,
     force_token=False,
@@ -42,7 +43,10 @@ def ansible_galaxy(
         f.write(f'server_list = {server}\n')
         f.write('\n')
         f.write(f'[galaxy_server.{server}]\n')
-        f.write(f"url={ansible_config.get('url')}\n")
+        if server_url is None:
+            f.write(f"url={ansible_config.get('url')}\n")
+        else:
+            f.write(f"url={server_url}\n")
         if ansible_config.get('auth_url'):
             f.write(f"auth_url={ansible_config.get('auth_url')}\n")
         f.write('validate_certs=False\n')