Repository: elastic/rally

Issue:
Index-append operation only indexing bulk-size * clients documents
<!--

*******************************************************************************

Please ask questions in our forum at 

https://discuss.elastic.co/c/elasticsearch/rally

GitHub is reserved for bug reports and feature requests only and we will 
close your issue immediately if it is just a question.
 
*******************************************************************************

If you are in fact posting a bug report or a feature request, please include 
one and only one of the below blocks in your new issue.

-->

<!--
If you are filing a bug report, please remove the below feature
request block and provide responses for all of the below items.
-->

**Rally version** (get with `esrally --version`):
Latest from master, 425d8f6e8828f944607807d192aa7a98a45672f3

**Invoked command**:
```
./rally --track-path=/home/hinmanm/es/mytrack --target-hosts=127.0.0.1:9200 --pipeline=benchmark-only
```

**Configuration file (located in `~/.rally/rally.ini`))**:

```
[meta]
config.version = 12

[system]
env.name = local

[node]
root.dir = /home/hinmanm/.rally/benchmarks
src.root.dir = /home/hinmanm/es

[source]
remote.repo.url = https://github.com/elastic/elasticsearch.git
elasticsearch.src.subdir = elasticsearch

[build]
gradle.bin = /home/hinmanm/.sdkman/candidates/gradle/current/bin/gradle

[runtime]
java.home = /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.151-1.b12.fc26.x86_64

[benchmarks]
local.dataset.cache = ${node:root.dir}/data

[reporting]
datastore.type = elasticsearch
datastore.host = localhost
datastore.port = 9900
datastore.secure = False
datastore.user = 
datastore.password = 

[tracks]
default.url = https://github.com/elastic/rally-tracks

[teams]
default.url = https://github.com/elastic/rally-teams

[defaults]
preserve_benchmark_candidate = False

[distributions]
release.1.url = https://download.elasticsearch.org/elasticsearch/elasticsearch/elasticsearch-{{VERSION}}.tar.gz
release.2.url = https://download.elasticsearch.org/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/{{VERSION}}/elasticsearch-{{VERSION}}.tar.gz
release.url = https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-{{VERSION}}.tar.gz
release.cache = true
```

**JVM version**:
JDK 8

**OS version**:
Fedora 26

**Description of the problem including expected versus actual behavior**:

I have a track with an index-append operation defined inline in the challenge like so:

```json
      "schedule": [
        {
          "operation": {
            "name": "index-append",
            "operation-type": "bulk",
            "bulk-size": {{bulk_size | default(100)}}
          },
          "clients": 4
        },
```

The documents.json contains 1967 documents, however, only 400 are actually indexed.

**Steps to reproduce**:
 1. Using a track with many documents, add a challenge schedule with a low bulk-size and multiple clients
 2. Run the track
 3. Observe that only `bulk-size x clients` documents are indexed, in my case, `100 x 4 = 400` documents actually indexed.

I've noticed that this didn't affect me when the indexing was defined in a separate operation, it only started affecting me when I defined it inline in the challenge.

**Provide logs (if relevant)**:
The data is from a private repo, so I cannot provide it here.

Discussion:

Patch:
diff --git a/esrally/driver/driver.py b/esrally/driver/driver.py
index cfb50622..112471b4 100644
--- a/esrally/driver/driver.py
+++ b/esrally/driver/driver.py
@@ -1230,9 +1230,16 @@ def schedule_for(current_track, task, client_index):
                     "time period of [%s] seconds." % (task.schedule, task, str(warmup_time_period), str(task.time_period)))
         return time_period_based(sched, warmup_time_period, task.time_period, runner_for_op, params_for_op)
     else:
+        warmup_iterations = task.warmup_iterations if task.warmup_iterations else 0
+        if task.iterations:
+            iterations = task.iterations
+        elif params_for_op.size():
+            iterations = params_for_op.size() - warmup_iterations
+        else:
+            iterations = 1
         logger.info("Creating iteration-count based schedule with [%s] distribution for [%s] with [%d] warmup iterations and "
-                    "[%d] iterations." % (task.schedule, op, task.warmup_iterations, task.iterations))
-        return iteration_count_based(sched, task.warmup_iterations, task.iterations, runner_for_op, params_for_op)
+                    "[%d] iterations." % (task.schedule, op, warmup_iterations, iterations))
+        return iteration_count_based(sched, warmup_iterations, iterations, runner_for_op, params_for_op)
 
 
 def time_period_based(sched, warmup_time_period, time_period, runner, params):
diff --git a/esrally/track/loader.py b/esrally/track/loader.py
index 013a9875..c8e7b401 100644
--- a/esrally/track/loader.py
+++ b/esrally/track/loader.py
@@ -625,11 +625,11 @@ def post_process_for_test_mode(t):
             # we need iterate over leaf tasks and await iterating over possible intermediate 'parallel' elements
             for leaf_task in task:
                 # iteration-based schedules are divided among all clients and we should provide at least one iteration for each client.
-                if leaf_task.warmup_iterations > leaf_task.clients:
+                if leaf_task.warmup_iterations is not None and leaf_task.warmup_iterations > leaf_task.clients:
                     count = leaf_task.clients
                     logger.info("Resetting warmup iterations to %d for [%s]" % (count, str(leaf_task)))
                     leaf_task.warmup_iterations = count
-                if leaf_task.iterations > leaf_task.clients:
+                if leaf_task.iterations is not None and leaf_task.iterations > leaf_task.clients:
                     count = leaf_task.clients
                     logger.info("Resetting measurement iterations to %d for [%s]" % (count, str(leaf_task)))
                     leaf_task.iterations = count
@@ -1054,8 +1054,8 @@ class TrackSpecificationReader:
 
     def parse_parallel(self, ops_spec, ops, challenge_name):
         # use same default values as #parseTask() in case the 'parallel' element did not specify anything
-        default_warmup_iterations = self._r(ops_spec, "warmup-iterations", error_ctx="parallel", mandatory=False, default_value=0)
-        default_iterations = self._r(ops_spec, "iterations", error_ctx="parallel", mandatory=False, default_value=1)
+        default_warmup_iterations = self._r(ops_spec, "warmup-iterations", error_ctx="parallel", mandatory=False)
+        default_iterations = self._r(ops_spec, "iterations", error_ctx="parallel", mandatory=False)
         default_warmup_time_period = self._r(ops_spec, "warmup-time-period", error_ctx="parallel", mandatory=False)
         default_time_period = self._r(ops_spec, "time-period", error_ctx="parallel", mandatory=False)
         clients = self._r(ops_spec, "clients", error_ctx="parallel", mandatory=False)
@@ -1079,7 +1079,7 @@ class TrackSpecificationReader:
                             "this name exists." % (challenge_name, completed_by))
         return track.Parallel(tasks, clients)
 
-    def parse_task(self, task_spec, ops, challenge_name, default_warmup_iterations=0, default_iterations=1,
+    def parse_task(self, task_spec, ops, challenge_name, default_warmup_iterations=None, default_iterations=None,
                    default_warmup_time_period=None, default_time_period=None, completed_by_name=None):
 
         op_spec = task_spec["operation"]
@@ -1106,10 +1106,10 @@ class TrackSpecificationReader:
                           schedule=schedule,
                           # this is to provide scheduler-specific parameters for custom schedulers.
                           params=task_spec)
-        if task.warmup_iterations != default_warmup_iterations and task.time_period is not None:
+        if task.warmup_iterations is not None and task.time_period is not None:
             self._error("Operation '%s' in challenge '%s' defines '%d' warmup iterations and a time period of '%d' seconds. Please do not "
                         "mix time periods and iterations." % (op.name, challenge_name, task.warmup_iterations, task.time_period))
-        elif task.warmup_time_period is not None and task.iterations != default_iterations:
+        elif task.warmup_time_period is not None and task.iterations is not None:
             self._error("Operation '%s' in challenge '%s' defines a warmup time period of '%d' seconds and '%d' iterations. Please do not "
                         "mix time periods and iterations." % (op.name, challenge_name, task.warmup_time_period, task.iterations))
 
diff --git a/esrally/track/params.py b/esrally/track/params.py
index 65257f38..c8b662c2 100644
--- a/esrally/track/params.py
+++ b/esrally/track/params.py
@@ -113,8 +113,8 @@ class ParamSource:
         * It will either run an operation for a pre-determined number of times or
         * It can run until the parameter source is exhausted.
 
-        In the former case, return just 1. In the latter case, you should determine the number of times that `#params()` will be invoked.
-        With that number, Rally can show the progress made so far to the user.
+        In the former case, you should determine the number of times that `#params()` will be invoked. With that number, Rally can show
+        the progress made so far to the user. In the latter case, return ``None``.
 
         :return:  The "size" of this parameter source or ``None`` if should run eternally.
         """
diff --git a/esrally/track/track.py b/esrally/track/track.py
index 9aafbf8c..cfd1ba0a 100644
--- a/esrally/track/track.py
+++ b/esrally/track/track.py
@@ -588,7 +588,7 @@ class Parallel:
 
 
 class Task:
-    def __init__(self, name, operation, meta_data=None, warmup_iterations=0, iterations=1, warmup_time_period=None, time_period=None,
+    def __init__(self, name, operation, meta_data=None, warmup_iterations=None, iterations=None, warmup_time_period=None, time_period=None,
                  clients=1,
                  completes_parent=False, schedule="deterministic", params=None):
         self.name = name


Test patch:
diff --git a/tests/driver/driver_test.py b/tests/driver/driver_test.py
index fbfcc08a..981546e7 100644
--- a/tests/driver/driver_test.py
+++ b/tests/driver/driver_test.py
@@ -452,6 +452,47 @@ class SchedulerTests(ScheduleTestCase):
         ]
         self.assert_schedule(expected_schedule, schedule)
 
+    def test_schedule_param_source_determines_iterations_no_warmup(self):
+        # we neither define any time-period nor any iteration count on the task.
+        task = track.Task("bulk-index", track.Operation("bulk-index", track.OperationType.Bulk.name, params={"body": ["a"], "size": 3},
+                                                        param_source="driver-test-param-source"),
+                          clients=1, params={"target-throughput": 4, "clients": 4})
+
+        invocations = driver.schedule_for(self.test_track, task, 0)
+
+        self.assert_schedule([
+            (0.0, metrics.SampleType.Normal, 1 / 3, {"body": ["a"], "size": 3}),
+            (1.0, metrics.SampleType.Normal, 2 / 3, {"body": ["a"], "size": 3}),
+            (2.0, metrics.SampleType.Normal, 3 / 3, {"body": ["a"], "size": 3}),
+        ], list(invocations))
+
+    def test_schedule_param_source_determines_iterations_including_warmup(self):
+        task = track.Task("bulk-index", track.Operation("bulk-index", track.OperationType.Bulk.name, params={"body": ["a"], "size": 5},
+                                                        param_source="driver-test-param-source"),
+                          warmup_iterations=2, clients=1, params={"target-throughput": 4, "clients": 4})
+
+        invocations = driver.schedule_for(self.test_track, task, 0)
+
+        self.assert_schedule([
+            (0.0, metrics.SampleType.Warmup, 1 / 5, {"body": ["a"], "size": 5}),
+            (1.0, metrics.SampleType.Warmup, 2 / 5, {"body": ["a"], "size": 5}),
+            (2.0, metrics.SampleType.Normal, 3 / 5, {"body": ["a"], "size": 5}),
+            (3.0, metrics.SampleType.Normal, 4 / 5, {"body": ["a"], "size": 5}),
+            (4.0, metrics.SampleType.Normal, 5 / 5, {"body": ["a"], "size": 5}),
+        ], list(invocations))
+
+    def test_schedule_defaults_to_iteration_based(self):
+        # no time-period and no iterations specified on the task. Also, the parameter source does not define a size.
+        task = track.Task("bulk-index", track.Operation("bulk-index", track.OperationType.Bulk.name, params={"body": ["a"]},
+                                                        param_source="driver-test-param-source"),
+                          clients=1, params={"target-throughput": 4, "clients": 4})
+
+        invocations = driver.schedule_for(self.test_track, task, 0)
+
+        self.assert_schedule([
+            (0.0, metrics.SampleType.Normal, 1 / 1, {"body": ["a"]}),
+        ], list(invocations))
+
     def test_schedule_for_warmup_time_based(self):
         task = track.Task("time-based", track.Operation("time-based", track.OperationType.Bulk.name, params={"body": ["a"], "size": 11},
                                                         param_source="driver-test-param-source"),