Extend error msg to specify operation name for missing index

esrally/driver/driver.py

@@ -1370,7 +1370,7 @@ def schedule_for(current_track, task, client_index):
     sched = scheduler.scheduler_for(task.schedule, task.params)
     logger.info("Choosing [%s] for [%s].", sched, task)
     runner_for_op = runner.runner_for(op.type)
-    params_for_op = track.operation_parameters(current_track, op).partition(client_index, num_clients)
+    params_for_op = track.operation_parameters(current_track, task).partition(client_index, num_clients)
 
     if requires_time_period_schedule(task, runner_for_op, params_for_op):
         warmup_time_period = task.warmup_time_period if task.warmup_time_period else 0

esrally/track/loader.py

@@ -310,11 +310,12 @@ def track_file(self, track_name):
         return self._track_file
 
 
-def operation_parameters(t, op):
+def operation_parameters(t, task):
+    op = task.operation
     if op.param_source:
         return params.param_source_for_name(op.param_source, t, op.params)
     else:
-        return params.param_source_for_operation(op.type, t, op.params)
+        return params.param_source_for_operation(op.type, t, op.params, task.name)
 
 
 def used_corpora(t, cfg):
@@ -323,7 +324,7 @@ def used_corpora(t, cfg):
         challenge = t.find_challenge_or_default(cfg.opts("track", "challenge.name"))
         for task in challenge.schedule:
             for sub_task in task:
-                param_source = operation_parameters(t, sub_task.operation)
+                param_source = operation_parameters(t, sub_task)
                 if hasattr(param_source, "corpora"):
                     for c in param_source.corpora:
                         # We might have the same corpus *but* they contain different doc sets. Therefore also need to union over doc sets.

esrally/track/params.py

@@ -32,12 +32,12 @@
 __PARAM_SOURCES_BY_NAME = {}
 
 
-def param_source_for_operation(op_type, track, params):
+def param_source_for_operation(op_type, track, params, task_name):
     try:
         # we know that this can only be a Rally core parameter source
-        return __PARAM_SOURCES_BY_OP[op_type](track, params)
+        return __PARAM_SOURCES_BY_OP[op_type](track, params, operation_name=task_name)
     except KeyError:
-        return ParamSource(track, params)
+        return ParamSource(track, params, operation_name=task_name)
 
 
 def param_source_for_name(name, track, params):
@@ -365,7 +365,7 @@ def __init__(self, track, params, **kwargs):
         }
 
         if not index_name:
-            raise exceptions.InvalidSyntax("'index' is mandatory")
+            raise exceptions.InvalidSyntax("'index' is mandatory and is missing for operation '{}'".format(kwargs.get("operation_name")))
 
         if pages:
             self.query_params["pages"] = pages

tests/track/params_test.py

@@ -1604,6 +1604,19 @@ def test_passes_cache(self):
             }
         }, p["body"])
 
+    def test_create_without_index(self):
+        with self.assertRaises(exceptions.InvalidSyntax) as ctx:
+            params.SearchParamSource(track=track.Track(name="unit-test"), params={
+                "type": "type1",
+                "body": {
+                    "query": {
+                        "match_all": {}
+                        }
+                    }
+            }, operation_name="test_operation")
+
+        self.assertEqual("'index' is mandatory and is missing for operation 'test_operation'", ctx.exception.args[0])
+
     def test_passes_request_parameters(self):
         index1 = track.Index(name="index1", types=["type1"])