Allow index pattern for source-index in shrink-index operation

docs/track.rst

@@ -1303,7 +1303,7 @@ shrink-index
 With the operation ``shrink-index`` you can execute the `shrink index API <https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-shrink-index.html>`_. Note that this does not correspond directly to the shrink index API call in Elasticsearch but it is a high-level operation that executes all the necessary low-level operations under the hood to shrink an index. It supports the following parameters:
 
 * ``source-index`` (mandatory): The name of the index that should be shrinked. Multiple indices can be defined using the `Multi-target syntax <https://www.elastic.co/guide/en/elasticsearch/reference/current/multi-index.html>`_.
-* ``target-index`` (mandatory): The name of the index that contains the shrinked shards. If multiple indices match ``source-index``, the common prefix will appended to ``target-index``.
+* ``target-index`` (mandatory): The name of the index that contains the shrinked shards. If multiple indices match ``source-index``, one shrink operation will execute for every matching index. Each shrink operation will use a modified ``target-index``: the unique suffix of the source index (derived by removing the common prefix of all matching source indices) will be appended to ``target-index``. See also the example below.
 * ``target-body`` (mandatory): The body containing settings and aliases for ``target-index``.
 * ``shrink-node`` (optional, defaults to a random data node): As a first step, the source index needs to be fully relocated to a single node. Rally will automatically choose a random data node in the cluster but you can choose one explicitly if needed.
 

esrally/driver/runner.py

@@ -308,6 +308,8 @@ def mandatory(params, key, op):
             f"Add it to your parameter source and try again.")
 
 
+# TODO: remove and use https://docs.python.org/3/library/stdtypes.html#str.removeprefix
+#  once Python 3.9 becomes the minimum version
 def remove_prefix(string, prefix):
     if string.startswith(prefix):
         return string[len(prefix):]
@@ -1339,15 +1341,17 @@ async def __call__(self, es, params):
         target_body = deepcopy(mandatory(params, "target-body", self))
         shrink_node = params.get("shrink-node")
         # Choose a random data node if none is specified
-        node_names = [shrink_node] if shrink_node else []
-        if not shrink_node:
+        if shrink_node:
+            node_names = [shrink_node]
+        else:
+            node_names = []
             # choose a random data node
             node_info = await es.nodes.info()
             for node in node_info["nodes"].values():
                 if "data" in node["roles"]:
                     node_names.append(node["name"])
             if not node_names:
-                raise exceptions.RallyAssertionError("Could not choose a suitable shrink-node automatically. Please specify it explicitly.")
+                raise exceptions.RallyAssertionError("Could not choose a suitable shrink-node automatically. Specify it explicitly.")
 
         for source_index in source_indices:
             shrink_node = random.choice(node_names)
@@ -1365,7 +1369,7 @@ async def __call__(self, es, params):
                                           preserve_existing=True)
 
             self.logger.info("Waiting for relocation to finish for index [%s] ...", source_index)
-            await self._wait_for(es, source_index, "shard relocation for index [{}]".format(source_index))
+            await self._wait_for(es, source_index, f"shard relocation for index [{source_index}]")
             self.logger.info("Shrinking [%s] to [%s].", source_index, target_index)
             if "settings" not in target_body:
                 target_body["settings"] = {}
@@ -1377,10 +1381,10 @@ async def __call__(self, es, params):
             await es.indices.shrink(index=source_index, target=final_target_index, body=target_body)
 
             self.logger.info("Waiting for shrink to finish for index [%s] ...", source_index)
-            await self._wait_for(es, final_target_index, "shrink for index [{}]".format(final_target_index))
+            await self._wait_for(es, final_target_index, f"shrink for index [{final_target_index}]")
             self.logger.info("Shrinking [%s] to [%s] has finished.", source_index, final_target_index)
         # ops_count is not really important for this operation...
-        return 1, "ops"
+        return len(source_indices), "ops"
 
     def __repr__(self, *args, **kwargs):
         return "shrink-index"