Add concurrency test support for tempest plugins

Add run_concurrent_tasks utility function in tempest.common.concurrency
to enable tempest plugins to write concurrency tests for parallel
operations. The function handles multiprocessing, exception collection,
and cleanup automatically.

All tempest plugins (cinder, manila, glance, etc.) can consume this
utility by passing their service-specific resource_count configuration.

Changes:
 - Add run_concurrent_tasks helper function
 - Add unit tests for the new utility
 - Add release note

Change-Id: I6071a30069aa0e83c079dec9da90e8ebb9f3b254
Signed-off-by: lkuchlan <lkuchlan@redhat.com>

releasenotes/notes/add-concurrency-utility-function-e8d4f2a1c9b5e7f3.yaml

@@ -0,0 +1,14 @@
+---
+features:
+  - |
+    Add ``run_concurrent_tasks`` helper function in ``tempest.common.concurrency``
+    module to simplify writing concurrency tests for OpenStack services. This
+    utility uses multiprocessing to execute operations in parallel and collect
+    results in a shared list. It automatically handles process management,
+    exception collection, and cleanup. Tempest plugins can use this function
+    to test concurrent operations such as parallel volume creation, snapshot
+    creation, or server operations. The number of concurrent processes is
+    specified via the ``resource_count`` parameter, allowing plugins to pass
+    their service-specific configuration values (e.g.,
+    ``CONF.volume.concurrent_resource_count`` for cinder,
+    ``CONF.share.concurrent_resource_count`` for manila).

tempest/common/concurrency.py

@@ -0,0 +1,61 @@
+# Copyright 2025 Red Hat, Inc.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+
+import multiprocessing
+
+
+def run_concurrent_tasks(target, resource_count, **kwargs):
+    """Run a target function concurrently using multiprocessing.
+
+    :param target: Function to execute concurrently. Must accept
+                   (index, resource_ids, **kwargs) as parameters.
+    :param resource_count: Number of concurrent processes to spawn.
+    :param kwargs: Additional keyword arguments passed to the target function.
+    :return: List of results collected from all processes.
+    :raises RuntimeError: If any worker process fails during execution.
+    """
+    manager = multiprocessing.Manager()
+    resource_ids = manager.list()
+    errors = manager.list()  # Capture exceptions from workers
+
+    def wrapped_target(index, resource_ids, **kwargs):
+        try:
+            target(index, resource_ids, **kwargs)
+        except Exception as exc:
+            errors.append(f"Worker {index} failed: {exc}")
+
+    processes = []
+    for i in range(resource_count):
+        p = multiprocessing.Process(
+            target=wrapped_target,
+            args=(i, resource_ids),
+            kwargs=kwargs
+        )
+        processes.append(p)
+
+    # Start all processes
+    for p in processes:
+        p.start()
+
+    # Wait for all processes to finish
+    for p in processes:
+        p.join()
+
+    if errors:
+        raise RuntimeError(
+            "One or more concurrent tasks failed:\n" + "\n".join(errors)
+        )
+
+    return list(resource_ids)

tempest/tests/common/test_concurrency.py

@@ -0,0 +1,99 @@
+# Copyright 2025 Red Hat, Inc.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from tempest.common import concurrency
+from tempest.tests import base
+
+
+class TestConcurrency(base.TestCase):
+
+    def test_run_concurrent_tasks_success(self):
+        """Test successful concurrent task execution."""
+        def target_func(index, resource_ids, prefix='resource'):
+            resource_ids.append(f"{prefix}_{index}")
+
+        result = concurrency.run_concurrent_tasks(
+            target_func,
+            resource_count=3,
+            prefix='test_resource'
+        )
+
+        self.assertEqual(len(result), 3)
+        self.assertIn('test_resource_0', result)
+        self.assertIn('test_resource_1', result)
+        self.assertIn('test_resource_2', result)
+
+    def test_run_concurrent_tasks_multiple_workers(self):
+        """Test concurrent task execution with multiple workers."""
+        def target_func(index, resource_ids):
+            resource_ids.append(f"item_{index}")
+
+        result = concurrency.run_concurrent_tasks(
+            target_func,
+            resource_count=4
+        )
+
+        self.assertEqual(len(result), 4)
+        self.assertIn('item_0', result)
+        self.assertIn('item_1', result)
+        self.assertIn('item_2', result)
+        self.assertIn('item_3', result)
+
+    def test_run_concurrent_tasks_single_process(self):
+        """Test concurrent task execution with single process."""
+        def target_func(index, resource_ids, value):
+            resource_ids.append(value * 2)
+
+        result = concurrency.run_concurrent_tasks(
+            target_func,
+            resource_count=1,
+            value=5
+        )
+
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0], 10)
+
+    def test_run_concurrent_tasks_with_exception(self):
+        """Test that exceptions in tasks are properly captured and raised."""
+        def failing_target(index, resource_ids):
+            if index == 1:
+                raise ValueError("Test error in worker 1")
+            resource_ids.append(f"resource_{index}")
+
+        error = self.assertRaises(
+            RuntimeError,
+            concurrency.run_concurrent_tasks,
+            failing_target,
+            resource_count=3
+        )
+        self.assertIn("Worker 1 failed", str(error))
+        self.assertIn("Test error in worker 1", str(error))
+
+    def test_run_concurrent_tasks_dict_return_values(self):
+        """Test concurrent task execution with dict return values."""
+        def target_returning_dict(index, resource_ids):
+            resource_ids.append({'id': index, 'name': f'resource_{index}'})
+
+        result = concurrency.run_concurrent_tasks(
+            target_returning_dict,
+            resource_count=3
+        )
+
+        self.assertEqual(len(result), 3)
+        # Verify that dicts are present
+        ids = [r['id'] for r in result]
+        self.assertIn(0, ids)
+        self.assertIn(1, ids)
+        self.assertIn(2, ids)