Repository: elastic/rally

Issue:
Mechanism needed for benchmarking update-heavy workflows against create-only or index-only
Currently, rally can be configured to generate metadata that contains conflicting ids for 25% of the documents in a corpus. However, the action specified in the metadata is always `index`. Since the performance characteristics of `update` changed significantly in 5.0+, this seems like a blind spot in the current suite. 

While I understand that metadata can be interleaved with the documents in a corpus instead of being generated by rally, side-by-side comparison of performance of the same corpus with conflicting ids would be considerably simpler if the create/update workflow could use generated metadata.

Discussion:

Patch:
diff --git a/docs/telemetry.rst b/docs/telemetry.rst
index 15cf7743..39924664 100644
--- a/docs/telemetry.rst
+++ b/docs/telemetry.rst
@@ -16,12 +16,13 @@ You probably want to gain additional insights from a race. Therefore, we have ad
 
    Available telemetry devices:
 
-   Command    Name                   Description
-   ---------  ---------------------  -----------------------------------------------------
-   jit        JIT Compiler Profiler  Enables JIT compiler logs.
-   gc         GC log                 Enables GC logs.
-   jfr        Flight Recorder        Enables Java Flight Recorder (requires an Oracle JDK)
-   perf       perf stat              Reads CPU PMU counters (requires Linux and perf)
+   Command     Name                   Description
+   ----------  ---------------------  -----------------------------------------------------
+   jit         JIT Compiler Profiler  Enables JIT compiler logs.
+   gc          GC log                 Enables GC logs.
+   jfr         Flight Recorder        Enables Java Flight Recorder (requires an Oracle JDK)
+   perf        perf stat              Reads CPU PMU counters (requires Linux and perf)
+   node-stats  Node Stats             Regularly samples node stats
 
    Keep in mind that each telemetry device may incur a runtime overhead which can skew results.
 
@@ -67,4 +68,26 @@ The ``gc`` telemetry device enables GC logs for the benchmark candidate. You can
 perf
 ----
 
-The ``perf`` telemetry device runs ``perf stat`` on each benchmarked node and writes the output to a log file. It can be used to capture low-level CPU statistics. Note that the perf tool, which is only available on Linux, must be installed before using this telemetry device.
\ No newline at end of file
+The ``perf`` telemetry device runs ``perf stat`` on each benchmarked node and writes the output to a log file. It can be used to capture low-level CPU statistics. Note that the perf tool, which is only available on Linux, must be installed before using this telemetry device.
+
+node-stats
+----------
+
+.. warning::
+
+    This telemetry device will record a lot of metrics and likely skew your measurement results.
+
+The node-stats telemetry devices regularly calls the node-stats API and records metrics from the following sections:
+
+* Indices stats (key ``indices`` in the node-stats API)
+* Thread pool stats (key ``jvm.thread_pool`` in the node-stats API)
+* JVM buffer pool stats (key ``jvm.buffer_pools`` in the node-stats API)
+* Circuit breaker stats (key ``breakers`` in the node-stats API)
+
+Supported telemetry parameters:
+
+* ``node-stats-sample-interval`` (default: 1): A positive number greater than zero denoting the sampling interval in seconds.
+* ``node-stats-include-indices`` (default: ``false``): A boolean indicating whether indices stats should be included.
+* ``node-stats-include-thread-pools`` (default: ``true``): A boolean indicating whether thread pool stats should be included.
+* ``node-stats-include-buffer-pools`` (default: ``true``): A boolean indicating whether buffer pool stats should be included.
+* ``node-stats-include-breakers`` (default: ``true``): A boolean indicating whether circuit breaker stats should be included.
\ No newline at end of file
diff --git a/docs/track.rst b/docs/track.rst
index 75b6858e..a39b6b15 100644
--- a/docs/track.rst
+++ b/docs/track.rst
@@ -313,6 +313,8 @@ With the operation type ``bulk`` you can execute `bulk requests <http://www.elas
 * ``batch-size`` (optional): Defines how many documents Rally will read at once. This is an expert setting and only meant to avoid accidental bottlenecks for very small bulk sizes (e.g. if you want to benchmark with a bulk-size of 1, you should set ``batch-size`` higher).
 * ``pipeline`` (optional): Defines the name of an (existing) ingest pipeline that should be used (only supported from Elasticsearch 5.0).
 * ``conflicts`` (optional): Type of index conflicts to simulate. If not specified, no conflicts will be simulated. Valid values are: 'sequential' (A document id is replaced with a document id with a sequentially increasing id), 'random' (A document id is replaced with a document id with a random other id).
+* ``conflict-probability`` (optional, defaults to 25 percent): A number between (0, 100] that defines how many of the documents will get replaced.
+* ``on-conflict`` (optional, defaults to ``index``): Determines whether Rally should use the action ``index`` or ``update`` on id conflicts.
 * ``detailed-results`` (optional, defaults to ``false``): Records more detailed meta-data for bulk requests. As it analyzes the corresponding bulk response in more detail, this might incur additional overhead which can skew measurement results.
 
 Example::
diff --git a/esrally/mechanic/launcher.py b/esrally/mechanic/launcher.py
index 968fbcd7..a946e177 100644
--- a/esrally/mechanic/launcher.py
+++ b/esrally/mechanic/launcher.py
@@ -31,17 +31,43 @@ def wait_for_rest_layer(es, max_attempts=20):
 
 
 class ClusterLauncher:
-    def __init__(self, cfg, metrics_store, client_factory_class=client.EsClientFactory):
+    """
+    The cluster launcher performs cluster-wide tasks that need to be done in the startup / shutdown phase.
+
+    """
+    def __init__(self, cfg, metrics_store, on_post_launch=None, client_factory_class=client.EsClientFactory):
+        """
+
+        Creates a new ClusterLauncher.
+
+        :param cfg: The config object.
+        :param metrics_store: A metrics store that is configured to receive system metrics.
+        :param on_post_launch: An optional function that takes the Elasticsearch client as a parameter. It is invoked after the
+                               REST API is available.
+        :param client_factory_class: A factory class that can create an Elasticsearch client.
+        """
         self.cfg = cfg
         self.metrics_store = metrics_store
+        self.on_post_launch = on_post_launch
         self.client_factory = client_factory_class
 
     def start(self):
+        """
+        Performs final startup tasks.
+
+        Precondition: All cluster nodes have been started.
+        Postcondition: The cluster is ready to receive HTTP requests or a ``LaunchError`` is raised.
+
+        :return: A representation of the launched cluster.
+        """
+        enabled_devices = self.cfg.opts("mechanic", "telemetry.devices")
+        telemetry_params = self.cfg.opts("mechanic", "telemetry.params")
         hosts = self.cfg.opts("client", "hosts")
         client_options = self.cfg.opts("client", "options")
         es = self.client_factory(hosts, client_options).create()
 
-        t = telemetry.Telemetry(devices=[
+        t = telemetry.Telemetry(enabled_devices, devices=[
+            telemetry.NodeStats(telemetry_params, es, self.metrics_store),
             telemetry.ClusterMetaDataInfo(es),
             telemetry.ClusterEnvironmentInfo(es, self.metrics_store),
             telemetry.GcTimesSummary(es, self.metrics_store),
@@ -61,10 +87,16 @@ class ClusterLauncher:
             logger.error("REST API layer is not yet available. Forcefully terminating cluster.")
             self.stop(c)
             raise exceptions.LaunchError("Elasticsearch REST API layer is not available. Forcefully terminated cluster.")
-
+        if self.on_post_launch:
+            self.on_post_launch(es)
         return c
 
     def stop(self, c):
+        """
+        Performs cleanup tasks. This method should be called before nodes are shut down.
+
+        :param c: The cluster that is about to be stopped.
+        """
         c.telemetry.detach_from_cluster(c)
 
 
diff --git a/esrally/mechanic/mechanic.py b/esrally/mechanic/mechanic.py
index 08fdd111..53fcf054 100644
--- a/esrally/mechanic/mechanic.py
+++ b/esrally/mechanic/mechanic.py
@@ -10,6 +10,8 @@ from esrally.mechanic import supplier, provisioner, launcher, team
 
 logger = logging.getLogger("rally.mechanic")
 
+METRIC_FLUSH_INTERVAL_SECONDS = 30
+
 
 ##############################
 # Public Messages
@@ -214,6 +216,10 @@ def nodes_by_host(ip_port_pairs):
 
 
 class MechanicActor(actor.RallyActor):
+    WAKEUP_RESET_RELATIVE_TIME = "relative_time"
+
+    WAKEUP_FLUSH_METRICS = "flush_metrics"
+
     """
     This actor coordinates all associated mechanics on remote hosts (which do the actual work).
     """
@@ -226,6 +232,7 @@ class MechanicActor(actor.RallyActor):
         self.race_control = None
         self.cluster_launcher = None
         self.cluster = None
+        self.plugins = None
 
     def receiveUnrecognizedMessage(self, msg, sender):
         logger.info("MechanicActor#receiveMessage unrecognized(msg = [%s] sender = [%s])" % (str(type(msg)), str(sender)))
@@ -256,6 +263,7 @@ class MechanicActor(actor.RallyActor):
         cls = metrics.metrics_store_class(self.cfg)
         self.metrics_store = cls(self.cfg)
         self.metrics_store.open(ctx=msg.open_metrics_context)
+        _, self.plugins = load_team(self.cfg, msg.external)
 
         # In our startup procedure we first create all mechanics. Only if this succeeds we'll continue.
         hosts = self.cfg.opts("client", "hosts")
@@ -311,12 +319,19 @@ class MechanicActor(actor.RallyActor):
     @actor.no_retry("mechanic")
     def receiveMsg_ResetRelativeTime(self, msg, sender):
         if msg.reset_in_seconds > 0:
-            self.wakeupAfter(msg.reset_in_seconds)
+            self.wakeupAfter(msg.reset_in_seconds, payload=MechanicActor.WAKEUP_RESET_RELATIVE_TIME)
         else:
             self.reset_relative_time()
 
     def receiveMsg_WakeupMessage(self, msg, sender):
-        self.reset_relative_time()
+        if msg.payload == MechanicActor.WAKEUP_RESET_RELATIVE_TIME:
+            self.reset_relative_time()
+        elif msg.payload == MechanicActor.WAKEUP_FLUSH_METRICS:
+            logger.info("Flushing cluster-wide system metrics store.")
+            self.metrics_store.flush(refresh=False)
+            self.wakeupAfter(METRIC_FLUSH_INTERVAL_SECONDS, payload=MechanicActor.WAKEUP_FLUSH_METRICS)
+        else:
+            raise exceptions.RallyAssertionError("Unknown wakeup reason [{}]".format(msg.payload))
 
     def receiveMsg_BenchmarkFailure(self, msg, sender):
         self.send(self.race_control, msg)
@@ -345,7 +360,8 @@ class MechanicActor(actor.RallyActor):
         self.transition_when_all_children_responded(sender, msg, "cluster_stopping", "cluster_stopped", self.on_all_nodes_stopped)
 
     def on_all_nodes_started(self):
-        self.cluster_launcher = launcher.ClusterLauncher(self.cfg, self.metrics_store)
+        plugin_handler = PostLaunchPluginHandler(self.plugins)
+        self.cluster_launcher = launcher.ClusterLauncher(self.cfg, self.metrics_store, on_post_launch=plugin_handler)
         # Workaround because we could raise a LaunchError here and thespian will attempt to retry a failed message.
         # In that case, we will get a followup RallyAssertionError because on the second attempt, Rally will check
         # the status which is now "nodes_started" but we expected the status to be "nodes_starting" previously.
@@ -366,6 +382,7 @@ class MechanicActor(actor.RallyActor):
                                                 self.cluster.source_revision,
                                                 self.cluster.distribution_version),
                                 self.metrics_store.meta_info))
+        self.wakeupAfter(METRIC_FLUSH_INTERVAL_SECONDS, payload=MechanicActor.WAKEUP_FLUSH_METRICS)
 
     def on_benchmark_started(self):
         self.cluster.on_benchmark_start()
@@ -392,6 +409,21 @@ class MechanicActor(actor.RallyActor):
         # do not self-terminate, let the parent actor handle this
 
 
+class PostLaunchPluginHandler:
+    def __init__(self, plugins, hook_handler_class=team.PluginBootstrapHookHandler):
+        self.handlers = []
+        if plugins:
+            for plugin in plugins:
+                handler = hook_handler_class(plugin)
+                if handler.can_load():
+                    handler.load()
+                    self.handlers.append(handler)
+
+    def __call__(self, client):
+        for handler in self.handlers:
+            handler.invoke(team.PluginBootstrapPhase.post_launch.name, client=client)
+
+
 @thespian.actors.requireCapability('coordinator')
 class Dispatcher(thespian.actors.ActorTypeDispatcher):
     def __init__(self):
@@ -476,8 +508,6 @@ class Dispatcher(thespian.actors.ActorTypeDispatcher):
 
 
 class NodeMechanicActor(actor.RallyActor):
-    METRIC_FLUSH_INTERVAL_SECONDS = 30
-
     """
     One instance of this actor is run on each target host and coordinates the actual work of starting / stopping all nodes that should run
     on this host.
@@ -553,13 +583,13 @@ class NodeMechanicActor(actor.RallyActor):
             elif isinstance(msg, OnBenchmarkStart):
                 self.metrics_store.lap = msg.lap
                 self.mechanic.on_benchmark_start()
-                self.wakeupAfter(NodeMechanicActor.METRIC_FLUSH_INTERVAL_SECONDS)
+                self.wakeupAfter(METRIC_FLUSH_INTERVAL_SECONDS)
                 self.send(sender, BenchmarkStarted())
             elif isinstance(msg, thespian.actors.WakeupMessage):
                 if self.running:
                     logger.info("Flushing system metrics store on host [%s]." % self.host)
                     self.metrics_store.flush(refresh=False)
-                    self.wakeupAfter(NodeMechanicActor.METRIC_FLUSH_INTERVAL_SECONDS)
+                    self.wakeupAfter(METRIC_FLUSH_INTERVAL_SECONDS)
             elif isinstance(msg, OnBenchmarkStop):
                 self.mechanic.on_benchmark_stop()
                 self.metrics_store.flush(refresh=False)
@@ -590,11 +620,7 @@ class NodeMechanicActor(actor.RallyActor):
 # Internal API (only used by the actor and for tests)
 #####################################################
 
-def create(cfg, metrics_store, all_node_ips, cluster_settings=None, sources=False, build=False, distribution=False, external=False,
-           docker=False):
-    races_root = paths.races_root(cfg)
-    challenge_root_path = paths.race_root(cfg)
-    node_ids = cfg.opts("provisioning", "node.ids", mandatory=False)
+def load_team(cfg, external):
     # externally provisioned clusters do not support cars / plugins
     if external:
         car = None
@@ -603,6 +629,15 @@ def create(cfg, metrics_store, all_node_ips, cluster_settings=None, sources=Fals
         team_path = team.team_path(cfg)
         car = team.load_car(team_path, cfg.opts("mechanic", "car.names"), cfg.opts("mechanic", "car.params"))
         plugins = team.load_plugins(team_path, cfg.opts("mechanic", "car.plugins"), cfg.opts("mechanic", "plugin.params"))
+    return car, plugins
+
+
+def create(cfg, metrics_store, all_node_ips, cluster_settings=None, sources=False, build=False, distribution=False, external=False,
+           docker=False):
+    races_root = paths.races_root(cfg)
+    challenge_root_path = paths.race_root(cfg)
+    node_ids = cfg.opts("provisioning", "node.ids", mandatory=False)
+    car, plugins = load_team(cfg, external)
 
     if sources or distribution:
         s = supplier.create(cfg, sources, distribution, build, challenge_root_path, plugins)
diff --git a/esrally/mechanic/provisioner.py b/esrally/mechanic/provisioner.py
index 4a1c3b38..2942caae 100644
--- a/esrally/mechanic/provisioner.py
+++ b/esrally/mechanic/provisioner.py
@@ -2,12 +2,12 @@ import os
 import glob
 import shutil
 import logging
-from enum import Enum
 
 import jinja2
 
 from esrally import exceptions
-from esrally.utils import io, console, process, modules, versions
+from esrally.mechanic import team
+from esrally.utils import io, console, process, versions
 
 logger = logging.getLogger("rally.provisioner")
 
@@ -102,21 +102,6 @@ def cleanup(preserve, install_dir, data_paths):
                 logger.exception("Could not delete [%s]. Skipping..." % install_dir)
 
 
-class ProvisioningPhase(Enum):
-    post_install = 10
-
-    @classmethod
-    def valid(cls, name):
-        for n in ProvisioningPhase.names():
-            if n == name:
-                return True
-        return False
-
-    @classmethod
-    def names(cls):
-        return [p.name for p in list(ProvisioningPhase)]
-
-
 def _apply_config(source_root_path, target_root_path, config_vars):
     for root, dirs, files in os.walk(source_root_path):
         env = jinja2.Environment(loader=jinja2.FileSystemLoader(root))
@@ -173,7 +158,7 @@ class BareProvisioner:
 
         for installer in self.plugin_installers:
             # Never let install hooks modify our original provisioner variables and just provide a copy!
-            installer.invoke_install_hook(ProvisioningPhase.post_install, provisioner_vars.copy())
+            installer.invoke_install_hook(team.PluginBootstrapPhase.post_install, provisioner_vars.copy())
 
         return NodeConfiguration(self.es_installer.car, self.es_installer.node_ip, self.es_installer.node_name,
                                  self.es_installer.node_root_dir, self.es_installer.es_home_path, self.es_installer.node_log_dir,
@@ -297,52 +282,8 @@ class ElasticsearchInstaller:
             return [os.path.join(self.es_home_path, "data")]
 
 
-class InstallHookHandler:
-    def __init__(self, plugin, loader_class=modules.ComponentLoader):
-        self.plugin = plugin
-        # Don't allow the loader to recurse. The subdirectories may contain Elasticsearch specific files which we do not want to add to
-        # Rally's Python load path. We may need to define a more advanced strategy in the future.
-        self.loader = loader_class(root_path=self.plugin.root_path, component_entry_point="plugin", recurse=False)
-        self.hooks = {}
-
-    def can_load(self):
-        return self.loader.can_load()
-
-    def load(self):
-        root_module = self.loader.load()
-        try:
-            # every module needs to have a register() method
-            root_module.register(self)
-        except exceptions.RallyError:
-            # just pass our own exceptions transparently.
-            raise
-        except BaseException:
-            msg = "Could not load install hooks in [%s]" % self.loader.root_path
-            logger.exception(msg)
-            raise exceptions.SystemSetupError(msg)
-
-    def register(self, phase, hook):
-        logger.info("Registering install hook [%s] for phase [%s] in plugin [%s]" % (hook.__name__, phase, self.plugin.name))
-        if not ProvisioningPhase.valid(phase):
-            raise exceptions.SystemSetupError("Provisioning phase [%s] is unknown. Valid phases are: %s." %
-                                              (phase, ProvisioningPhase.names()))
-        if phase not in self.hooks:
-            self.hooks[phase] = []
-        self.hooks[phase].append(hook)
-
-    def invoke(self, phase, variables):
-        if phase in self.hooks:
-            logger.info("Invoking phase [%s] for plugin [%s] in config [%s]" % (phase, self.plugin.name, self.plugin.config))
-            for hook in self.hooks[phase]:
-                logger.info("Invoking hook [%s]." % hook.__name__)
-                # hooks should only take keyword arguments to be forwards compatible with Rally!
-                hook(config_names=self.plugin.config, variables=variables)
-        else:
-            logger.debug("Plugin [%s] in config [%s] has no hook registered for phase [%s]." % (self.plugin.name, self.plugin.config, phase))
-
-
 class PluginInstaller:
-    def __init__(self, plugin, hook_handler_class=InstallHookHandler):
+    def __init__(self, plugin, hook_handler_class=team.PluginBootstrapHookHandler):
         self.plugin = plugin
         self.hook_handler = hook_handler_class(self.plugin)
         if self.hook_handler.can_load():
@@ -371,7 +312,7 @@ class PluginInstaller:
                                         (self.plugin_name, str(return_code)))
 
     def invoke_install_hook(self, phase, variables):
-        self.hook_handler.invoke(phase.name, variables)
+        self.hook_handler.invoke(phase.name, variables=variables)
 
     @property
     def variables(self):
diff --git a/esrally/mechanic/team.py b/esrally/mechanic/team.py
index e37e5c3a..37c4a518 100644
--- a/esrally/mechanic/team.py
+++ b/esrally/mechanic/team.py
@@ -1,11 +1,12 @@
 import os
 import logging
 import configparser
+from enum import Enum
 
 import tabulate
 
 from esrally import exceptions, PROGRAM_NAME
-from esrally.utils import console, repo, io
+from esrally.utils import console, repo, io, modules
 
 logger = logging.getLogger("rally.team")
 
@@ -341,3 +342,62 @@ class PluginDescriptor:
 
     def __eq__(self, other):
         return isinstance(other, type(self)) and (self.name, self.config, self.core_plugin) == (other.name, other.config, other.core_plugin)
+
+
+class PluginBootstrapPhase(Enum):
+    post_install = 10
+    post_launch = 20
+
+    @classmethod
+    def valid(cls, name):
+        for n in PluginBootstrapPhase.names():
+            if n == name:
+                return True
+        return False
+
+    @classmethod
+    def names(cls):
+        return [p.name for p in list(PluginBootstrapPhase)]
+
+
+class PluginBootstrapHookHandler:
+    def __init__(self, plugin, loader_class=modules.ComponentLoader):
+        self.plugin = plugin
+        # Don't allow the loader to recurse. The subdirectories may contain Elasticsearch specific files which we do not want to add to
+        # Rally's Python load path. We may need to define a more advanced strategy in the future.
+        self.loader = loader_class(root_path=self.plugin.root_path, component_entry_point="plugin", recurse=False)
+        self.hooks = {}
+
+    def can_load(self):
+        return self.loader.can_load()
+
+    def load(self):
+        root_module = self.loader.load()
+        try:
+            # every module needs to have a register() method
+            root_module.register(self)
+        except exceptions.RallyError:
+            # just pass our own exceptions transparently.
+            raise
+        except BaseException:
+            msg = "Could not load plugin bootstrap hooks in [{}]".format(self.loader.root_path)
+            logger.exception(msg)
+            raise exceptions.SystemSetupError(msg)
+
+    def register(self, phase, hook):
+        logger.info("Registering plugin bootstrap hook [%s] for phase [%s] in plugin [%s]", hook.__name__, phase, self.plugin.name)
+        if not PluginBootstrapPhase.valid(phase):
+            raise exceptions.SystemSetupError("Phase [{}] is unknown. Valid phases are: {}.".format(phase, PluginBootstrapPhase.names()))
+        if phase not in self.hooks:
+            self.hooks[phase] = []
+        self.hooks[phase].append(hook)
+
+    def invoke(self, phase, **kwargs):
+        if phase in self.hooks:
+            logger.info("Invoking phase [%s] for plugin [%s] in config [%s]", phase, self.plugin.name, self.plugin.config)
+            for hook in self.hooks[phase]:
+                logger.info("Invoking hook [%s].", hook.__name__)
+                # hooks should only take keyword arguments to be forwards compatible with Rally!
+                hook(config_names=self.plugin.config, **kwargs)
+        else:
+            logger.debug("Plugin [%s] in config [%s] has no hook registered for phase [%s].", self.plugin.name, self.plugin.config, phase)
\ No newline at end of file
diff --git a/esrally/mechanic/telemetry.py b/esrally/mechanic/telemetry.py
index 7746a5e0..c78387ea 100644
--- a/esrally/mechanic/telemetry.py
+++ b/esrally/mechanic/telemetry.py
@@ -6,7 +6,7 @@ import subprocess
 import threading
 
 import tabulate
-from esrally import metrics, time
+from esrally import metrics, time, exceptions
 from esrally.utils import io, sysstats, process, console, versions
 
 logger = logging.getLogger("rally.telemetry")
@@ -14,7 +14,7 @@ logger = logging.getLogger("rally.telemetry")
 
 def list_telemetry():
     console.println("Available telemetry devices:\n")
-    devices = [[device.command, device.human_name, device.help] for device in [JitCompiler, Gc, FlightRecorder, PerfStat]]
+    devices = [[device.command, device.human_name, device.help] for device in [JitCompiler, Gc, FlightRecorder, PerfStat, NodeStats]]
     console.println(tabulate.tabulate(devices, ["Command", "Name", "Description"]))
     console.println("\nKeep in mind that each telemetry device may incur a runtime overhead which can skew results.")
 
@@ -116,6 +116,26 @@ class InternalTelemetryDevice(TelemetryDevice):
     internal = True
 
 
+class SamplerThread(threading.Thread):
+    def __init__(self, recorder):
+        threading.Thread.__init__(self)
+        self.stop = False
+        self.recorder = recorder
+
+    def finish(self):
+        self.stop = True
+        self.join()
+
+    def run(self):
+        # noinspection PyBroadException
+        try:
+            while not self.stop:
+                self.recorder.record()
+                time.sleep(self.recorder.sample_interval)
+        except BaseException as e:
+            logger.exception("Could not determine {}".format(self.recorder))
+
+
 class FlightRecorder(TelemetryDevice):
     internal = False
     command = "jfr"
@@ -256,6 +276,133 @@ class PerfStat(TelemetryDevice):
             self.attached = False
 
 
+class NodeStats(TelemetryDevice):
+    internal = False
+    command = "node-stats"
+    human_name = "Node Stats"
+    help = "Regularly samples node stats"
+
+    """
+    Gathers different node stats.
+    """
+    def __init__(self, telemetry_params, client, metrics_store):
+        super().__init__()
+        self.telemetry_params = telemetry_params
+        self.client = client
+        self.metrics_store = metrics_store
+        self.sampler = None
+
+    def attach_to_cluster(self, cluster):
+        super().attach_to_cluster(cluster)
+
+    def on_benchmark_start(self):
+        recorder = NodeStatsRecorder(self.telemetry_params, self.client, self.metrics_store)
+        self.sampler = SamplerThread(recorder)
+        self.sampler.setDaemon(True)
+        self.sampler.start()
+
+    def on_benchmark_stop(self):
+        if self.sampler:
+            self.sampler.finish()
+
+
+class NodeStatsRecorder:
+    def __init__(self, telemetry_params, client, metrics_store):
+        self.sample_interval = telemetry_params.get("node-stats-sample-interval", 1)
+        if self.sample_interval <= 0:
+            raise exceptions.SystemSetupError(
+                "The telemetry parameter 'node-stats-sample-interval' must be greater than zero but was {}.".format(self.sample_interval))
+
+        self.include_indices = telemetry_params.get("node-stats-include-indices", False)
+        self.include_thread_pools = telemetry_params.get("node-stats-include-thread-pools", True)
+        self.include_buffer_pools = telemetry_params.get("node-stats-include-buffer-pools", True)
+        self.include_breakers = telemetry_params.get("node-stats-include-breakers", True)
+        self.client = client
+        self.metrics_store = metrics_store
+
+    def __str__(self):
+        return "node stats"
+
+    def record(self):
+        current_sample = self.sample()
+        for node_stats in current_sample:
+            node_name = node_stats["name"]
+            if self.include_indices:
+                self.record_indices_stats(node_name, node_stats,
+                                          include=["indexing", "search", "merges", "query_cache", "segments", "translog",
+                                                   "request_cache"])
+            if self.include_thread_pools:
+                self.record_thread_pool_stats(node_name, node_stats)
+            if self.include_breakers:
+                self.record_circuit_breaker_stats(node_name, node_stats)
+            if self.include_buffer_pools:
+                self.record_jvm_buffer_pool_stats(node_name, node_stats)
+
+        time.sleep(self.sample_interval)
+
+    def record_indices_stats(self, node_name, node_stats, include):
+        indices_stats = node_stats["indices"]
+        for section in include:
+            if section in indices_stats:
+                for metric_name, metric_value in indices_stats[section].items():
+                    self.put_value(node_name,
+                                   metric_name="indices_{}_{}".format(section, metric_name),
+                                   node_stats_metric_name=metric_name,
+                                   metric_value=metric_value)
+
+    def record_thread_pool_stats(self, node_name, node_stats):
+        thread_pool_stats = node_stats["thread_pool"]
+        for pool_name, pool_metrics in thread_pool_stats.items():
+            for metric_name, metric_value in pool_metrics.items():
+                self.put_value(node_name,
+                               metric_name="thread_pool_{}_{}".format(pool_name, metric_name),
+                               node_stats_metric_name=metric_name,
+                               metric_value=metric_value)
+
+    def record_circuit_breaker_stats(self, node_name, node_stats):
+        breaker_stats = node_stats["breakers"]
+        for breaker_name, breaker_metrics in breaker_stats.items():
+            for metric_name, metric_value in breaker_metrics.items():
+                self.put_value(node_name,
+                               metric_name="breaker_{}_{}".format(breaker_name, metric_name),
+                               node_stats_metric_name=metric_name,
+                               metric_value=metric_value)
+
+    def record_jvm_buffer_pool_stats(self, node_name, node_stats):
+        buffer_pool_stats = node_stats["jvm"]["buffer_pools"]
+        for pool_name, pool_metrics in buffer_pool_stats.items():
+            for metric_name, metric_value in pool_metrics.items():
+                self.put_value(node_name,
+                               metric_name="jvm_buffer_pool_{}_{}".format(pool_name, metric_name),
+                               node_stats_metric_name=metric_name,
+                               metric_value=metric_value)
+
+    def put_value(self, node_name, metric_name, node_stats_metric_name, metric_value):
+        if isinstance(metric_value, (int, float)) and not isinstance(metric_value, bool):
+            # auto-recognize metric keys ending with well-known suffixes
+            if node_stats_metric_name.endswith("in_bytes"):
+                self.metrics_store.put_value_node_level(node_name=node_name,
+                                                        name=metric_name,
+                                                        value=metric_value, unit="byte")
+            elif node_stats_metric_name.endswith("in_millis"):
+                self.metrics_store.put_value_node_level(node_name=node_name,
+                                                        name=metric_name,
+                                                        value=metric_value, unit="ms")
+            else:
+                self.metrics_store.put_count_node_level(node_name=node_name,
+                                                        name=metric_name,
+                                                        count=metric_value)
+
+    def sample(self):
+        import elasticsearch
+        try:
+            stats = self.client.nodes.stats(metric="_all")
+        except elasticsearch.TransportError:
+            logger.exception("Could not retrieve node stats.")
+            return {}
+        return stats["nodes"].values()
+
+
 class StartupTime(InternalTelemetryDevice):
     def __init__(self, metrics_store, stopwatch=time.StopWatch):
         self.metrics_store = metrics_store
@@ -400,7 +547,8 @@ class CpuUsage(InternalTelemetryDevice):
 
     def on_benchmark_start(self):
         if self.node:
-            self.sampler = SampleCpuUsage(self.node, self.metrics_store)
+            recorder = CpuUsageRecorder(self.node, self.metrics_store)
+            self.sampler = SamplerThread(recorder)
             self.sampler.setDaemon(True)
             self.sampler.start()
 
@@ -409,30 +557,25 @@ class CpuUsage(InternalTelemetryDevice):
             self.sampler.finish()
 
 
-class SampleCpuUsage(threading.Thread):
+class CpuUsageRecorder:
     def __init__(self, node, metrics_store):
-        threading.Thread.__init__(self)
-        self.stop = False
         self.node = node
         self.process = sysstats.setup_process_stats(node.process.pid)
         self.metrics_store = metrics_store
+        # the call is blocking already; there is no need for additional waiting in the sampler thread.
+        self.sample_interval = 0
 
-    def finish(self):
-        self.stop = True
-        self.join()
-
-    def run(self):
+    def record(self):
         import psutil
-        # noinspection PyBroadException
         try:
-            while not self.stop:
-                self.metrics_store.put_value_node_level(node_name=self.node.node_name, name="cpu_utilization_1s",
-                                                        value=sysstats.cpu_utilization(self.process), unit="%")
+            self.metrics_store.put_value_node_level(node_name=self.node.node_name, name="cpu_utilization_1s",
+                                                    value=sysstats.cpu_utilization(self.process), unit="%")
         # this can happen when the Elasticsearch process has been terminated already and we were not quick enough to stop.
         except psutil.NoSuchProcess:
             pass
-        except BaseException:
-            logger.exception("Could not determine CPU utilization")
+
+    def __str__(self):
+        return "cpu utilization"
 
 
 def store_node_attribute_metadata(metrics_store, nodes_info):
diff --git a/esrally/track/params.py b/esrally/track/params.py
index 37bb2ea0..1287df65 100644
--- a/esrally/track/params.py
+++ b/esrally/track/params.py
@@ -442,6 +442,15 @@ class BulkIndexParamSource(ParamSource):
         else:
             raise exceptions.InvalidSyntax("Unknown 'conflicts' setting [%s]" % id_conflicts)
 
+        if self.id_conflicts != IndexIdConflict.NoConflicts:
+            self.conflict_probability = self.float_param(params, name="conflict-probability", default_value=25, min_value=0, max_value=100)
+            self.on_conflict = params.get("on-conflict", "index")
+            if self.on_conflict not in ["index", "update"]:
+                raise exceptions.InvalidSyntax("Unknown 'on-conflict' setting [{}]".format(self.on_conflict))
+        else:
+            self.conflict_probability = None
+            self.on_conflict = None
+
         self.corpora = self.used_corpora(track, params)
 
         for corpus in self.corpora:
@@ -473,13 +482,17 @@ class BulkIndexParamSource(ParamSource):
         except ValueError:
             raise exceptions.InvalidSyntax("'batch-size' must be numeric")
 
+        self.ingest_percentage = self.float_param(params, name="ingest-percentage", default_value=100, min_value=0, max_value=100)
+
+    def float_param(self, params, name, default_value, min_value, max_value):
         try:
-            self.ingest_percentage = float(params.get("ingest-percentage", 100.0))
-            if self.ingest_percentage <= 0 or self.ingest_percentage > 100.0:
+            value = float(params.get(name, default_value))
+            if value <= min_value or value > max_value:
                 raise exceptions.InvalidSyntax(
-                    "'ingest-percentage' must be in the range (0.0, 100.0] but was {:.1f}".format(self.ingest_percentage))
+                    "'{}' must be in the range ({:.1f}, {:.1f}] but was {:.1f}".format(name, min_value, max_value, value))
+            return value
         except ValueError:
-            raise exceptions.InvalidSyntax("'ingest-percentage' must be numeric")
+            raise exceptions.InvalidSyntax("'{}' must be numeric".format(name))
 
     def used_corpora(self, t, params):
         corpora = []
@@ -503,7 +516,8 @@ class BulkIndexParamSource(ParamSource):
 
     def partition(self, partition_index, total_partitions):
         return PartitionBulkIndexParamSource(self.corpora, partition_index, total_partitions, self.batch_size, self.bulk_size,
-                                             self.ingest_percentage, self.id_conflicts, self.pipeline, self._params)
+                                             self.ingest_percentage, self.id_conflicts, self.conflict_probability, self.on_conflict,
+                                             self.pipeline, self._params)
 
     def params(self):
         raise exceptions.RallyError("Do not use a BulkIndexParamSource without partitioning")
@@ -513,8 +527,8 @@ class BulkIndexParamSource(ParamSource):
 
 
 class PartitionBulkIndexParamSource:
-    def __init__(self, corpora, partition_index, total_partitions, batch_size, bulk_size, ingest_percentage, id_conflicts=None,
-                 pipeline=None, original_params=None):
+    def __init__(self, corpora, partition_index, total_partitions, batch_size, bulk_size, ingest_percentage,
+                 id_conflicts, conflict_probability, on_conflict, pipeline=None, original_params=None):
         """
 
         :param corpora: Specification of affected document corpora.
@@ -524,6 +538,8 @@ class PartitionBulkIndexParamSource:
         :param bulk_size: The size of bulk index operations (number of documents per bulk).
         :param ingest_percentage: A number between (0.0, 100.0] that defines how much of the whole corpus should be ingested.
         :param id_conflicts: The type of id conflicts.
+        :param conflict_probability: A number between (0.0, 100.0] that defines the probability that a document is replaced by another one.
+        :param on_conflict: A string indicating which action should be taken on id conflicts (either "index" or "update").
         :param pipeline: The name of the ingest pipeline to run.
         :param original_params: The original dict passed to the parent parameter source.
         """
@@ -536,7 +552,7 @@ class PartitionBulkIndexParamSource:
         self.id_conflicts = id_conflicts
         self.pipeline = pipeline
         self.internal_params = bulk_data_based(total_partitions, partition_index, corpora, batch_size,
-                                               bulk_size, id_conflicts, pipeline, original_params)
+                                               bulk_size, id_conflicts, conflict_probability, on_conflict, pipeline, original_params)
 
     def partition(self, partition_index, total_partitions):
         raise exceptions.RallyError("Cannot partition a PartitionBulkIndexParamSource further")
@@ -593,18 +609,20 @@ def chain(*iterables):
                 yield element
 
 
-def create_default_reader(docs, offset, num_lines, num_docs, batch_size, bulk_size, id_conflicts):
+def create_default_reader(docs, offset, num_lines, num_docs, batch_size, bulk_size, id_conflicts, conflict_probability, on_conflict):
     source = Slice(io.FileSource, offset, num_lines)
 
     if docs.includes_action_and_meta_data:
         am_handler = SourceActionMetaData(source)
     else:
-        am_handler = GenerateActionMetaData(docs.target_index, docs.target_type, build_conflicting_ids(id_conflicts, num_docs, offset))
+        am_handler = GenerateActionMetaData(docs.target_index, docs.target_type,
+                                            build_conflicting_ids(id_conflicts, num_docs, offset), conflict_probability, on_conflict)
 
     return IndexDataReader(docs.document_file, batch_size, bulk_size, source, am_handler, docs.target_index, docs.target_type)
 
 
-def create_readers(num_clients, client_index, corpora, batch_size, bulk_size, id_conflicts, create_reader):
+def create_readers(num_clients, client_index, corpora, batch_size, bulk_size, id_conflicts, conflict_probability, on_conflict,
+                   create_reader):
     readers = []
     for corpus in corpora:
         for docs in corpus.documents:
@@ -613,7 +631,8 @@ def create_readers(num_clients, client_index, corpora, batch_size, bulk_size, id
             if num_docs > 0:
                 logger.info("Task-relative client at index [%d] will bulk index [%d] docs starting from line offset [%d] for [%s/%s] "
                             "from corpus [%s]." % (client_index, num_docs, offset, docs.target_index, docs.target_type, corpus.name))
-                readers.append(create_reader(docs, offset, num_lines, num_docs, batch_size, bulk_size, id_conflicts))
+                readers.append(create_reader(docs, offset, num_lines, num_docs, batch_size, bulk_size, id_conflicts, conflict_probability,
+                                             on_conflict))
             else:
                 logger.info("Task-relative client at index [%d] skips [%s] (no documents to read)." % (client_index, corpus.name))
     return readers
@@ -673,8 +692,8 @@ def bulk_generator(readers, client_index, pipeline, original_params):
             yield params
 
 
-def bulk_data_based(num_clients, client_index, corpora, batch_size, bulk_size, id_conflicts, pipeline, original_params,
-                    create_reader=create_default_reader):
+def bulk_data_based(num_clients, client_index, corpora, batch_size, bulk_size, id_conflicts, conflict_probability, on_conflict, pipeline,
+                    original_params, create_reader=create_default_reader):
     """
     Calculates the necessary schedule for bulk operations.
 
@@ -684,22 +703,31 @@ def bulk_data_based(num_clients, client_index, corpora, batch_size, bulk_size, i
     :param batch_size: The number of documents to read in one go.
     :param bulk_size: The size of bulk index operations (number of documents per bulk).
     :param id_conflicts: The type of id conflicts to simulate.
+    :param conflict_probability: A number between (0.0, 100.0] that defines the probability that a document is replaced by another one.
+    :param on_conflict: A string indicating which action should be taken on id conflicts (either "index" or "update").
     :param pipeline: Name of the ingest pipeline to use. May be None.
     :param original_params: A dict of original parameters that were passed from the track. They will be merged into the returned parameters.
     :param create_reader: A function to create the index reader. By default a file based index reader will be created. This parameter is
                       intended for testing only.
     :return: A generator for the bulk operations of the given client.
     """
-    readers = create_readers(num_clients, client_index, corpora, batch_size, bulk_size, id_conflicts, create_reader)
+    readers = create_readers(num_clients, client_index, corpora, batch_size, bulk_size, id_conflicts, conflict_probability, on_conflict,
+                             create_reader)
     return bulk_generator(chain(*readers), client_index, pipeline, original_params)
 
 
 class GenerateActionMetaData:
-    def __init__(self, index_name, type_name, conflicting_ids, rand=random.randint):
+    def __init__(self, index_name, type_name, conflicting_ids=None, conflict_probability=None, on_conflict=None,
+                 rand=random.random, randint=random.randint):
         self.index_name = index_name
         self.type_name = type_name
         self.conflicting_ids = conflicting_ids
+        self.on_conflict = on_conflict
+        # random() produces numbers between 0 and 1 and the user denotes the probability in percentage between 0 and 100.
+        self.conflict_probability = conflict_probability / 100.0 if conflict_probability else None
+
         self.rand = rand
+        self.randint = randint
         self.id_up_to = 0
 
     def __iter__(self):
@@ -707,15 +735,22 @@ class GenerateActionMetaData:
 
     def __next__(self):
         if self.conflicting_ids is not None:
-            # 25% of the time we replace a doc:
-            if self.id_up_to > 0 and self.rand(0, 3) == 3:
-                doc_id = self.conflicting_ids[self.rand(0, self.id_up_to - 1)]
+            if self.id_up_to > 0 and self.rand() <= self.conflict_probability:
+                doc_id = self.conflicting_ids[self.randint(0, self.id_up_to - 1)]
+                action = self.on_conflict
             else:
                 doc_id = self.conflicting_ids[self.id_up_to]
                 self.id_up_to += 1
-            return '{"index": {"_index": "%s", "_type": "%s", "_id": "%s"}}' % (self.index_name, self.type_name, doc_id)
+                action = "index"
+
+            if action == "index":
+                return "index", '{"index": {"_index": "%s", "_type": "%s", "_id": "%s"}}' % (self.index_name, self.type_name, doc_id)
+            elif action == "update":
+                return "update", '{"update": {"_index": "%s", "_type": "%s", "_id": "%s"}}' % (self.index_name, self.type_name, doc_id)
+            else:
+                raise exceptions.RallyAssertionError("Unknown action [{}]".format(action))
         else:
-            return '{"index": {"_index": "%s", "_type": "%s"}}' % (self.index_name, self.type_name)
+            return "index", '{"index": {"_index": "%s", "_type": "%s"}}' % (self.index_name, self.type_name)
 
 
 class SourceActionMetaData:
@@ -726,7 +761,7 @@ class SourceActionMetaData:
         return self
 
     def __next__(self):
-        return next(self.source)
+        return "source", next(self.source)
 
 
 class Slice:
@@ -815,11 +850,16 @@ class IndexDataReader:
     def read_bulk(self):
         docs_in_bulk = 0
         current_bulk = []
-
-        for action_metadata_line, document in zip(self.action_metadata, self.file_source):
-            if action_metadata_line:
+        for action_metadata_item, document in zip(self.action_metadata, self.file_source):
+            if action_metadata_item:
+                action_type, action_metadata_line = action_metadata_item
                 current_bulk.append(action_metadata_line)
-            current_bulk.append(document)
+                if action_type == "update":
+                    current_bulk.append("{\"doc\":%s}" % document)
+                else:
+                    current_bulk.append(document)
+            else:
+                current_bulk.append(document)
             docs_in_bulk += 1
             if docs_in_bulk == self.bulk_size:
                 break


Test patch:
diff --git a/tests/mechanic/launcher_test.py b/tests/mechanic/launcher_test.py
index 7e9d89bf..8c54bfae 100644
--- a/tests/mechanic/launcher_test.py
+++ b/tests/mechanic/launcher_test.py
@@ -1,6 +1,6 @@
-from unittest import TestCase
+from unittest import TestCase, mock
 
-from esrally import config
+from esrally import config, exceptions
 from esrally.mechanic import launcher
 
 
@@ -11,14 +11,15 @@ class MockMetricsStore:
 
 class MockClientFactory:
     def __init__(self, hosts, client_options):
-        pass
+        self.client_options = client_options
 
     def create(self):
-        return MockClient()
+        return MockClient(self.client_options)
 
 
 class MockClient:
-    def __init__(self):
+    def __init__(self, client_options):
+        self.client_options = client_options
         self.cluster = SubClient({
             "cluster_name": "rally-benchmark-cluster",
             "nodes": {
@@ -54,8 +55,14 @@ class MockClient:
         }
 
     def info(self):
+        if self.client_options.get("raise-error-on-info", False):
+            import elasticsearch
+            raise elasticsearch.ConnectionError("Unittest error")
         return self._info
 
+    def search(self, *args, **kwargs):
+        return {}
+
 
 class SubClient:
     def __init__(self, info):
@@ -73,7 +80,7 @@ class ExternalLauncherTests(TestCase):
         cfg = config.Config()
         cfg.add(config.Scope.application, "mechanic", "telemetry.devices", [])
         cfg.add(config.Scope.application, "client", "hosts", ["10.0.0.10:9200", "10.0.0.11:9200"])
-        cfg.add(config.Scope.application, "client", "options", [])
+        cfg.add(config.Scope.application, "client", "options", {})
 
         m = launcher.ExternalLauncher(cfg, MockMetricsStore(), client_factory_class=MockClientFactory)
         m.start()
@@ -85,10 +92,60 @@ class ExternalLauncherTests(TestCase):
         cfg = config.Config()
         cfg.add(config.Scope.application, "mechanic", "telemetry.devices", [])
         cfg.add(config.Scope.application, "client", "hosts", ["10.0.0.10:9200", "10.0.0.11:9200"])
-        cfg.add(config.Scope.application, "client", "options", [])
+        cfg.add(config.Scope.application, "client", "options", {})
         cfg.add(config.Scope.application, "mechanic", "distribution.version", "2.3.3")
 
         m = launcher.ExternalLauncher(cfg, MockMetricsStore(), client_factory_class=MockClientFactory)
         m.start()
         # did not change user defined value
         self.assertEqual(cfg.opts("mechanic", "distribution.version"), "2.3.3")
+
+
+class ClusterLauncherTests(TestCase):
+    def test_launches_cluster_with_post_launch_handler(self):
+        on_post_launch = mock.Mock()
+        cfg = config.Config()
+        cfg.add(config.Scope.application, "client", "hosts", ["10.0.0.10:9200", "10.0.0.11:9200"])
+        cfg.add(config.Scope.application, "client", "options", {})
+        cfg.add(config.Scope.application, "mechanic", "telemetry.devices", [])
+        cfg.add(config.Scope.application, "mechanic", "telemetry.params", {})
+
+        cluster_launcher = launcher.ClusterLauncher(cfg, MockMetricsStore(),
+                                                    on_post_launch=on_post_launch, client_factory_class=MockClientFactory)
+        cluster = cluster_launcher.start()
+
+        self.assertEqual(["10.0.0.10:9200", "10.0.0.11:9200"], cluster.hosts)
+        self.assertIsNotNone(cluster.telemetry)
+        # this requires at least Python 3.6
+        # on_post_launch.assert_called_once()
+        self.assertEqual(1, on_post_launch.call_count)
+
+    def test_launches_cluster_without_post_launch_handler(self):
+        cfg = config.Config()
+        cfg.add(config.Scope.application, "client", "hosts", ["10.0.0.10:9200", "10.0.0.11:9200"])
+        cfg.add(config.Scope.application, "client", "options", {})
+        cfg.add(config.Scope.application, "mechanic", "telemetry.devices", [])
+        cfg.add(config.Scope.application, "mechanic", "telemetry.params", {})
+
+        cluster_launcher = launcher.ClusterLauncher(cfg, MockMetricsStore(), client_factory_class=MockClientFactory)
+        cluster = cluster_launcher.start()
+
+        self.assertEqual(["10.0.0.10:9200", "10.0.0.11:9200"], cluster.hosts)
+        self.assertIsNotNone(cluster.telemetry)
+
+    @mock.patch("time.sleep")
+    def test_error_on_cluster_launch(self, sleep):
+        on_post_launch = mock.Mock()
+        cfg = config.Config()
+        cfg.add(config.Scope.application, "client", "hosts", ["10.0.0.10:9200", "10.0.0.11:9200"])
+        # Simulate that the client will raise an error upon startup
+        cfg.add(config.Scope.application, "client", "options", {"raise-error-on-info": True})
+        cfg.add(config.Scope.application, "mechanic", "telemetry.devices", [])
+        cfg.add(config.Scope.application, "mechanic", "telemetry.params", {})
+
+        cluster_launcher = launcher.ClusterLauncher(cfg, MockMetricsStore(),
+                                                    on_post_launch=on_post_launch, client_factory_class=MockClientFactory)
+        with self.assertRaisesRegex(exceptions.LaunchError,
+                                    "Elasticsearch REST API layer is not available. Forcefully terminated cluster."):
+            cluster_launcher.start()
+        on_post_launch.assert_not_called()
\ No newline at end of file
diff --git a/tests/mechanic/provisioner_test.py b/tests/mechanic/provisioner_test.py
index 7c29095a..b06ead1f 100644
--- a/tests/mechanic/provisioner_test.py
+++ b/tests/mechanic/provisioner_test.py
@@ -439,60 +439,11 @@ class PluginInstallerTests(TestCase):
         installer = provisioner.PluginInstaller(plugin, hook_handler_class=PluginInstallerTests.NoopHookHandler)
 
         self.assertEqual(0, len(installer.hook_handler.hook_calls))
-        installer.invoke_install_hook(provisioner.ProvisioningPhase.post_install, {"foo": "bar"})
+        installer.invoke_install_hook(team.PluginBootstrapPhase.post_install, {"foo": "bar"})
         self.assertEqual(1, len(installer.hook_handler.hook_calls))
         self.assertEqual({"foo": "bar"}, installer.hook_handler.hook_calls["post_install"])
 
 
-class InstallHookHandlerTests(TestCase):
-    class UnitTestComponentLoader:
-        def __init__(self, root_path, component_entry_point, recurse):
-            self.root_path = root_path
-            self.component_entry_point = component_entry_point
-            self.recurse = recurse
-            self.registration_function = None
-
-        def load(self):
-            return self.registration_function
-
-    class UnitTestHook:
-        def __init__(self, phase="post_install"):
-            self.phase = phase
-            self.call_counter = 0
-
-        def post_install_hook(self, config_names, variables, **kwargs):
-            self.call_counter += variables["increment"]
-
-        def register(self, handler):
-            # we can register multiple hooks here
-            handler.register(self.phase, self.post_install_hook)
-            handler.register(self.phase, self.post_install_hook)
-
-    def test_loads_module(self):
-        plugin = team.PluginDescriptor("unittest-plugin")
-        hook = InstallHookHandlerTests.UnitTestHook()
-        handler = provisioner.InstallHookHandler(plugin, loader_class=InstallHookHandlerTests.UnitTestComponentLoader)
-
-        handler.loader.registration_function = hook
-        handler.load()
-
-        handler.invoke("post_install", {"increment": 4})
-
-        # we registered our hook twice. Check that it has been called twice.
-        self.assertEqual(hook.call_counter, 2 * 4)
-
-    def test_cannot_register_for_unknown_phase(self):
-        plugin = team.PluginDescriptor("unittest-plugin")
-        hook = InstallHookHandlerTests.UnitTestHook(phase="this_is_an_unknown_install_phase")
-        handler = provisioner.InstallHookHandler(plugin, loader_class=InstallHookHandlerTests.UnitTestComponentLoader)
-
-        handler.loader.registration_function = hook
-        with self.assertRaises(exceptions.SystemSetupError) as ctx:
-            handler.load()
-        self.assertEqual("Provisioning phase [this_is_an_unknown_install_phase] is unknown. Valid phases are: ['post_install'].",
-                         ctx.exception.args[0])
-
-
 class DockerProvisionerTests(TestCase):
     @mock.patch("esrally.utils.sysstats.total_memory")
     @mock.patch("uuid.uuid4")
diff --git a/tests/mechanic/team_test.py b/tests/mechanic/team_test.py
index 37f80635..b8a94ecb 100644
--- a/tests/mechanic/team_test.py
+++ b/tests/mechanic/team_test.py
@@ -118,3 +118,52 @@ class PluginLoaderTests(TestCase):
             "var": "0",
             "hello": "true"
         }, plugin.variables)
+
+
+class PluginBootstrapHookHandlerTests(TestCase):
+    class UnitTestComponentLoader:
+        def __init__(self, root_path, component_entry_point, recurse):
+            self.root_path = root_path
+            self.component_entry_point = component_entry_point
+            self.recurse = recurse
+            self.registration_function = None
+
+        def load(self):
+            return self.registration_function
+
+    class UnitTestHook:
+        def __init__(self, phase="post_install"):
+            self.phase = phase
+            self.call_counter = 0
+
+        def post_install_hook(self, config_names, variables, **kwargs):
+            self.call_counter += variables["increment"]
+
+        def register(self, handler):
+            # we can register multiple hooks here
+            handler.register(self.phase, self.post_install_hook)
+            handler.register(self.phase, self.post_install_hook)
+
+    def test_loads_module(self):
+        plugin = team.PluginDescriptor("unittest-plugin")
+        hook = PluginBootstrapHookHandlerTests.UnitTestHook()
+        handler = team.PluginBootstrapHookHandler(plugin, loader_class=PluginBootstrapHookHandlerTests.UnitTestComponentLoader)
+
+        handler.loader.registration_function = hook
+        handler.load()
+
+        handler.invoke("post_install", variables={"increment": 4})
+
+        # we registered our hook twice. Check that it has been called twice.
+        self.assertEqual(hook.call_counter, 2 * 4)
+
+    def test_cannot_register_for_unknown_phase(self):
+        plugin = team.PluginDescriptor("unittest-plugin")
+        hook = PluginBootstrapHookHandlerTests.UnitTestHook(phase="this_is_an_unknown_install_phase")
+        handler = team.PluginBootstrapHookHandler(plugin, loader_class=PluginBootstrapHookHandlerTests.UnitTestComponentLoader)
+
+        handler.loader.registration_function = hook
+        with self.assertRaises(exceptions.SystemSetupError) as ctx:
+            handler.load()
+        self.assertEqual("Phase [this_is_an_unknown_install_phase] is unknown. Valid phases are: ['post_install', 'post_launch'].",
+                         ctx.exception.args[0])
diff --git a/tests/mechanic/telemetry_test.py b/tests/mechanic/telemetry_test.py
index a6266c0c..63d9b1d4 100644
--- a/tests/mechanic/telemetry_test.py
+++ b/tests/mechanic/telemetry_test.py
@@ -3,7 +3,7 @@ import collections
 import unittest.mock as mock
 from unittest import TestCase
 
-from esrally import config, metrics
+from esrally import config, metrics, exceptions
 from esrally.mechanic import telemetry, team, cluster
 
 
@@ -200,6 +200,340 @@ class GcTests(TestCase):
             env["ES_JAVA_OPTS"])
 
 
+class NodStatsRecorderTests(TestCase):
+    def test_negative_sample_interval_forbidden(self):
+        client = Client()
+        cfg = create_config()
+        metrics_store = metrics.EsMetricsStore(cfg)
+        telemetry_params = {
+            "node-stats-sample-interval": -1 * random.random()
+        }
+        with self.assertRaisesRegex(exceptions.SystemSetupError,
+                                    "The telemetry parameter 'node-stats-sample-interval' must be greater than zero but was .*\."):
+            telemetry.NodeStatsRecorder(telemetry_params, client, metrics_store=metrics_store)
+
+    @mock.patch("esrally.metrics.EsMetricsStore.put_count_node_level")
+    @mock.patch("esrally.metrics.EsMetricsStore.put_value_node_level")
+    def test_stores_default_nodes_stats(self, metrics_store_put_value, metrics_store_put_count):
+        node_stats_response = {
+            "cluster_name" : "elasticsearch",
+            "nodes" : {
+                "Zbl_e8EyRXmiR47gbHgPfg" : {
+                    "timestamp" : 1524379617017,
+                    "name" : "rally0",
+                    "transport_address" : "127.0.0.1:9300",
+                    "host" : "127.0.0.1",
+                    "ip" : "127.0.0.1:9300",
+                    "roles" : [
+                        "master",
+                        "data",
+                        "ingest"
+                    ],
+                    "indices" : {
+                        "docs" : {
+                            "count" : 0,
+                            "deleted" : 0
+                        },
+                        "store" : {
+                            "size_in_bytes" : 0
+                        },
+                        "indexing" : {
+                            "is_throttled" : False,
+                            "throttle_time_in_millis" : 0
+                        },
+                        "search" : {
+                            "open_contexts" : 0,
+                            "query_total" : 0,
+                            "query_time_in_millis" : 0
+                        },
+                        "merges" : {
+                            "current" : 0,
+                            "current_docs" : 0,
+                            "current_size_in_bytes" : 0
+                        },
+                        "query_cache" : {
+                            "memory_size_in_bytes" : 0,
+                            "total_count" : 0,
+                            "hit_count" : 0,
+                            "miss_count" : 0,
+                            "cache_size" : 0,
+                            "cache_count" : 0,
+                            "evictions" : 0
+                        },
+                        "completion" : {
+                            "size_in_bytes" : 0
+                        },
+                        "segments" : {
+                            "count" : 0,
+                            "memory_in_bytes" : 0,
+                            "max_unsafe_auto_id_timestamp" : -9223372036854775808,
+                            "file_sizes" : { }
+                        },
+                        "translog" : {
+                            "operations" : 0,
+                            "size_in_bytes" : 0,
+                            "uncommitted_operations" : 0,
+                            "uncommitted_size_in_bytes" : 0
+                        },
+                        "request_cache" : {
+                            "memory_size_in_bytes" : 0,
+                            "evictions" : 0,
+                            "hit_count" : 0,
+                            "miss_count" : 0
+                        },
+                        "recovery" : {
+                            "current_as_source" : 0,
+                            "current_as_target" : 0,
+                            "throttle_time_in_millis" : 0
+                        }
+                    },
+                    "jvm" : {
+                        "buffer_pools" : {
+                            "mapped" : {
+                                "count" : 7,
+                                "used_in_bytes" : 3120,
+                                "total_capacity_in_bytes" : 9999
+                            },
+                            "direct" : {
+                                "count" : 6,
+                                "used_in_bytes" : 73868,
+                                "total_capacity_in_bytes" : 73867
+                            }
+                        },
+                        "classes" : {
+                            "current_loaded_count" : 9992,
+                            "total_loaded_count" : 9992,
+                            "total_unloaded_count" : 0
+                        }
+                    },
+                    "thread_pool" : {
+                        "generic" : {
+                            "threads" : 4,
+                            "queue" : 0,
+                            "active" : 0,
+                            "rejected" : 0,
+                            "largest" : 4,
+                            "completed" : 8
+                        }
+                    },
+                    "breakers" : {
+                        "parent" : {
+                            "limit_size_in_bytes" : 726571417,
+                            "limit_size" : "692.9mb",
+                            "estimated_size_in_bytes" : 0,
+                            "estimated_size" : "0b",
+                            "overhead" : 1.0,
+                            "tripped" : 0
+                        }
+                    }
+                }
+            }
+        }
+
+        client = Client(nodes=SubClient(stats=node_stats_response))
+        cfg = create_config()
+        metrics_store = metrics.EsMetricsStore(cfg)
+        telemetry_params = {}
+        recorder = telemetry.NodeStatsRecorder(telemetry_params, client, metrics_store=metrics_store)
+        recorder.record()
+
+        metrics_store_put_count.assert_has_calls([
+            mock.call(node_name="rally0", name="thread_pool_generic_threads", count=4),
+            mock.call(node_name="rally0", name="thread_pool_generic_queue", count=0),
+            mock.call(node_name="rally0", name="thread_pool_generic_active", count=0),
+            mock.call(node_name="rally0", name="thread_pool_generic_rejected", count=0),
+            mock.call(node_name="rally0", name="thread_pool_generic_largest", count=4),
+            mock.call(node_name="rally0", name="thread_pool_generic_completed", count=8),
+            mock.call(node_name="rally0", name="breaker_parent_overhead", count=1.0),
+            mock.call(node_name="rally0", name="breaker_parent_tripped", count=0),
+            mock.call(node_name="rally0", name="jvm_buffer_pool_mapped_count", count=7),
+            mock.call(node_name="rally0", name="jvm_buffer_pool_direct_count", count=6),
+        ], any_order=True)
+
+        metrics_store_put_value.assert_has_calls([
+            mock.call(node_name="rally0", name="breaker_parent_limit_size_in_bytes", value=726571417, unit="byte"),
+            mock.call(node_name="rally0", name="breaker_parent_estimated_size_in_bytes", value=0, unit="byte"),
+            mock.call(node_name="rally0", name="jvm_buffer_pool_mapped_used_in_bytes", value=3120, unit="byte"),
+            mock.call(node_name="rally0", name="jvm_buffer_pool_mapped_total_capacity_in_bytes", value=9999, unit="byte"),
+            mock.call(node_name="rally0", name="jvm_buffer_pool_direct_used_in_bytes", value=73868, unit="byte"),
+            mock.call(node_name="rally0", name="jvm_buffer_pool_direct_total_capacity_in_bytes", value=73867, unit="byte"),
+        ], any_order=True)
+
+    @mock.patch("esrally.metrics.EsMetricsStore.put_count_node_level")
+    @mock.patch("esrally.metrics.EsMetricsStore.put_value_node_level")
+    def test_stores_all_nodes_stats(self, metrics_store_put_value, metrics_store_put_count):
+        node_stats_response = {
+            "cluster_name" : "elasticsearch",
+            "nodes" : {
+                "Zbl_e8EyRXmiR47gbHgPfg" : {
+                    "timestamp" : 1524379617017,
+                    "name" : "rally0",
+                    "transport_address" : "127.0.0.1:9300",
+                    "host" : "127.0.0.1",
+                    "ip" : "127.0.0.1:9300",
+                    "roles" : [
+                        "master",
+                        "data",
+                        "ingest"
+                    ],
+                    "indices" : {
+                        "docs" : {
+                            "count" : 0,
+                            "deleted" : 0
+                        },
+                        "store" : {
+                            "size_in_bytes" : 0
+                        },
+                        "indexing" : {
+                            "is_throttled" : False,
+                            "throttle_time_in_millis" : 0
+                        },
+                        "search" : {
+                            "open_contexts" : 0,
+                            "query_total" : 0,
+                            "query_time_in_millis" : 0
+                        },
+                        "merges" : {
+                            "current" : 0,
+                            "current_docs" : 0,
+                            "current_size_in_bytes" : 0
+                        },
+                        "query_cache" : {
+                            "memory_size_in_bytes" : 0,
+                            "total_count" : 0,
+                            "hit_count" : 0,
+                            "miss_count" : 0,
+                            "cache_size" : 0,
+                            "cache_count" : 0,
+                            "evictions" : 0
+                        },
+                        "completion" : {
+                            "size_in_bytes" : 0
+                        },
+                        "segments" : {
+                            "count" : 0,
+                            "memory_in_bytes" : 0,
+                            "max_unsafe_auto_id_timestamp" : -9223372036854775808,
+                            "file_sizes" : { }
+                        },
+                        "translog" : {
+                            "operations" : 0,
+                            "size_in_bytes" : 0,
+                            "uncommitted_operations" : 0,
+                            "uncommitted_size_in_bytes" : 0
+                        },
+                        "request_cache" : {
+                            "memory_size_in_bytes" : 0,
+                            "evictions" : 0,
+                            "hit_count" : 0,
+                            "miss_count" : 0
+                        },
+                        "recovery" : {
+                            "current_as_source" : 0,
+                            "current_as_target" : 0,
+                            "throttle_time_in_millis" : 0
+                        }
+                    },
+                    "jvm" : {
+                        "buffer_pools" : {
+                            "mapped" : {
+                                "count" : 7,
+                                "used_in_bytes" : 3120,
+                                "total_capacity_in_bytes" : 9999
+                            },
+                            "direct" : {
+                                "count" : 6,
+                                "used_in_bytes" : 73868,
+                                "total_capacity_in_bytes" : 73867
+                            }
+                        },
+                        "classes" : {
+                            "current_loaded_count" : 9992,
+                            "total_loaded_count" : 9992,
+                            "total_unloaded_count" : 0
+                        }
+                    },
+                    "thread_pool" : {
+                        "generic" : {
+                            "threads" : 4,
+                            "queue" : 0,
+                            "active" : 0,
+                            "rejected" : 0,
+                            "largest" : 4,
+                            "completed" : 8
+                        }
+                    },
+                    "breakers" : {
+                        "parent" : {
+                            "limit_size_in_bytes" : 726571417,
+                            "limit_size" : "692.9mb",
+                            "estimated_size_in_bytes" : 0,
+                            "estimated_size" : "0b",
+                            "overhead" : 1.0,
+                            "tripped" : 0
+                        }
+                    }
+                }
+            }
+        }
+
+        client = Client(nodes=SubClient(stats=node_stats_response))
+        cfg = create_config()
+        metrics_store = metrics.EsMetricsStore(cfg)
+        telemetry_params = {
+            "node-stats-include-indices": True
+        }
+        recorder = telemetry.NodeStatsRecorder(telemetry_params, client, metrics_store=metrics_store)
+        recorder.record()
+
+        metrics_store_put_count.assert_has_calls([
+            mock.call(node_name="rally0", name="indices_search_open_contexts", count=0),
+            mock.call(node_name="rally0", name="indices_search_query_total", count=0),
+            mock.call(node_name="rally0", name="indices_merges_current", count=0),
+            mock.call(node_name="rally0", name="indices_merges_current_docs", count=0),
+            mock.call(node_name="rally0", name="indices_query_cache_total_count", count=0),
+            mock.call(node_name="rally0", name="indices_query_cache_hit_count", count=0),
+            mock.call(node_name="rally0", name="indices_query_cache_miss_count", count=0),
+            mock.call(node_name="rally0", name="indices_query_cache_cache_size", count=0),
+            mock.call(node_name="rally0", name="indices_query_cache_cache_count", count=0),
+            mock.call(node_name="rally0", name="indices_query_cache_evictions", count=0),
+            mock.call(node_name="rally0", name="indices_segments_count", count=0),
+            mock.call(node_name="rally0", name="indices_segments_max_unsafe_auto_id_timestamp", count=-9223372036854775808),
+            mock.call(node_name="rally0", name="indices_translog_operations", count=0),
+            mock.call(node_name="rally0", name="indices_translog_uncommitted_operations", count=0),
+            mock.call(node_name="rally0", name="indices_request_cache_evictions", count=0),
+            mock.call(node_name="rally0", name="indices_request_cache_hit_count", count=0),
+            mock.call(node_name="rally0", name="indices_request_cache_miss_count", count=0),
+            mock.call(node_name="rally0", name="thread_pool_generic_threads", count=4),
+            mock.call(node_name="rally0", name="thread_pool_generic_queue", count=0),
+            mock.call(node_name="rally0", name="thread_pool_generic_active", count=0),
+            mock.call(node_name="rally0", name="thread_pool_generic_rejected", count=0),
+            mock.call(node_name="rally0", name="thread_pool_generic_largest", count=4),
+            mock.call(node_name="rally0", name="thread_pool_generic_completed", count=8),
+            mock.call(node_name="rally0", name="breaker_parent_overhead", count=1.0),
+            mock.call(node_name="rally0", name="breaker_parent_tripped", count=0),
+            mock.call(node_name="rally0", name="jvm_buffer_pool_mapped_count", count=7),
+            mock.call(node_name="rally0", name="jvm_buffer_pool_direct_count", count=6),
+        ], any_order=True)
+
+        metrics_store_put_value.assert_has_calls([
+            mock.call(node_name="rally0", name="indices_indexing_throttle_time_in_millis", value=0, unit="ms"),
+            mock.call(node_name="rally0", name="indices_search_query_time_in_millis", value=0, unit="ms"),
+            mock.call(node_name="rally0", name="indices_merges_current_size_in_bytes", value=0, unit="byte"),
+            mock.call(node_name="rally0", name="indices_query_cache_memory_size_in_bytes", value=0, unit="byte"),
+            mock.call(node_name="rally0", name="indices_segments_memory_in_bytes", value=0, unit="byte"),
+            mock.call(node_name="rally0", name="indices_translog_size_in_bytes", value=0, unit="byte"),
+            mock.call(node_name="rally0", name="indices_translog_uncommitted_size_in_bytes", value=0, unit="byte"),
+            mock.call(node_name="rally0", name="indices_request_cache_memory_size_in_bytes", value=0, unit="byte"),
+            mock.call(node_name="rally0", name="breaker_parent_limit_size_in_bytes", value=726571417, unit="byte"),
+            mock.call(node_name="rally0", name="breaker_parent_estimated_size_in_bytes", value=0, unit="byte"),
+            mock.call(node_name="rally0", name="jvm_buffer_pool_mapped_used_in_bytes", value=3120, unit="byte"),
+            mock.call(node_name="rally0", name="jvm_buffer_pool_mapped_total_capacity_in_bytes", value=9999, unit="byte"),
+            mock.call(node_name="rally0", name="jvm_buffer_pool_direct_used_in_bytes", value=73868, unit="byte"),
+            mock.call(node_name="rally0", name="jvm_buffer_pool_direct_total_capacity_in_bytes", value=73867, unit="byte"),
+        ], any_order=True)
+
+
 class ClusterEnvironmentInfoTests(TestCase):
     @mock.patch("esrally.metrics.EsMetricsStore.add_meta_info")
     def test_stores_cluster_level_metrics_on_attach(self, metrics_store_add_meta_info):
diff --git a/tests/track/params_test.py b/tests/track/params_test.py
index 18328575..39976959 100644
--- a/tests/track/params_test.py
+++ b/tests/track/params_test.py
@@ -1,3 +1,4 @@
+import random
 from unittest import TestCase
 
 from esrally import exceptions
@@ -100,31 +101,53 @@ class ConflictingIdsBuilderTests(TestCase):
 
 class ActionMetaDataTests(TestCase):
     def test_generate_action_meta_data_without_id_conflicts(self):
-        self.assertEqual('{"index": {"_index": "test_index", "_type": "test_type"}}',
-                         next(params.GenerateActionMetaData("test_index", "test_type", conflicting_ids=None)))
+        self.assertEqual(("index", '{"index": {"_index": "test_index", "_type": "test_type"}}'),
+                         next(params.GenerateActionMetaData("test_index", "test_type")))
 
     def test_generate_action_meta_data_with_id_conflicts(self):
-        pseudo_random_sequence = iter([
-            # first column == 3 -> we'll draw a "random" id, second column == "random" id
-            3, 1,
-            3, 3,
-            3, 2,
-            0,
-            3, 0])
-
-        generator = params.GenerateActionMetaData("test_index", "test_type", conflicting_ids=[100, 200, 300, 400],
-                                                  rand=lambda x, y: next(pseudo_random_sequence))
-
-        # first one is always not drawn from a random index
-        self.assertEqual('{"index": {"_index": "test_index", "_type": "test_type", "_id": "100"}}', next(generator))
+        def idx(id):
+            return "index", '{"index": {"_index": "test_index", "_type": "test_type", "_id": "%s"}}' % id
+
+        def conflict(action, id):
+            return action, '{"%s": {"_index": "test_index", "_type": "test_type", "_id": "%s"}}' % (action, id)
+
+        pseudo_random_conflicts = iter([
+            # if this value is <= our chosen threshold of 0.25 (see conflict_probability) we produce a conflict.
+            0.2,
+            0.25,
+            0.2,
+            # no conflict
+            0.3,
+            # conflict again
+            0.0
+        ])
+
+        chosen_index_of_conflicting_ids = iter([
+            # the "random" index of the id in the array `conflicting_ids` that will produce a conflict
+            1,
+            3,
+            2,
+            0])
+
+        conflict_action = random.choice(["index", "update"])
+
+        generator = params.GenerateActionMetaData("test_index", "test_type",
+                                                  conflicting_ids=[100, 200, 300, 400],
+                                                  conflict_probability=25,
+                                                  on_conflict=conflict_action,
+                                                  rand=lambda: next(pseudo_random_conflicts),
+                                                  randint=lambda x, y: next(chosen_index_of_conflicting_ids))
+
+        # first one is always *not* drawn from a random index
+        self.assertEqual(idx("100"), next(generator))
         # now we start using random ids, i.e. look in the first line of the pseudo-random sequence
-        self.assertEqual('{"index": {"_index": "test_index", "_type": "test_type", "_id": "200"}}', next(generator))
-        self.assertEqual('{"index": {"_index": "test_index", "_type": "test_type", "_id": "400"}}', next(generator))
-        self.assertEqual('{"index": {"_index": "test_index", "_type": "test_type", "_id": "300"}}', next(generator))
-        # "random" returns 0 instead of 3 -> we draw the next sequential one, which is 200
-        self.assertEqual('{"index": {"_index": "test_index", "_type": "test_type", "_id": "200"}}', next(generator))
+        self.assertEqual(conflict(conflict_action, "200"), next(generator))
+        self.assertEqual(conflict(conflict_action, "400"), next(generator))
+        self.assertEqual(conflict(conflict_action, "300"), next(generator))
+        # no conflict -> we draw the next sequential one, which is 200
+        self.assertEqual(idx("200"), next(generator))
         # and we're back to random
-        self.assertEqual('{"index": {"_index": "test_index", "_type": "test_type", "_id": "100"}}', next(generator))
+        self.assertEqual(conflict(conflict_action, "100"), next(generator))
 
     def test_source_file_action_meta_data(self):
         source = params.Slice(io.StringAsFileSource, 0, 5)
@@ -139,7 +162,7 @@ class ActionMetaDataTests(TestCase):
         ]
 
         source.open(data, "r")
-        self.assertEqual(data, list(generator))
+        self.assertEqual([("source", doc) for doc in data], list(generator))
         source.close()
 
 
@@ -155,7 +178,7 @@ class IndexDataReaderTests(TestCase):
         bulk_size = 50
 
         source = params.Slice(io.StringAsFileSource, 0, len(data))
-        am_handler = params.GenerateActionMetaData("test_index", "test_type", conflicting_ids=None)
+        am_handler = params.GenerateActionMetaData("test_index", "test_type")
 
         reader = params.IndexDataReader(data, batch_size=bulk_size, bulk_size=bulk_size, file_source=source, action_metadata=am_handler,
                                         index_name="test_index", type_name="test_type")
@@ -176,7 +199,7 @@ class IndexDataReaderTests(TestCase):
         bulk_size = 50
 
         source = params.Slice(io.StringAsFileSource, 3, len(data))
-        am_handler = params.GenerateActionMetaData("test_index", "test_type", conflicting_ids=None)
+        am_handler = params.GenerateActionMetaData("test_index", "test_type")
 
         reader = params.IndexDataReader(data, batch_size=bulk_size, bulk_size=bulk_size, file_source=source, action_metadata=am_handler,
                                         index_name="test_index", type_name="test_type")
@@ -199,7 +222,7 @@ class IndexDataReaderTests(TestCase):
         bulk_size = 3
 
         source = params.Slice(io.StringAsFileSource, 0, len(data))
-        am_handler = params.GenerateActionMetaData("test_index", "test_type", conflicting_ids=None)
+        am_handler = params.GenerateActionMetaData("test_index", "test_type")
 
         reader = params.IndexDataReader(data, batch_size=bulk_size, bulk_size=bulk_size, file_source=source, action_metadata=am_handler,
                                         index_name="test_index", type_name="test_type")
@@ -223,7 +246,7 @@ class IndexDataReaderTests(TestCase):
 
         # only 5 documents to index for this client
         source = params.Slice(io.StringAsFileSource, 0, 5)
-        am_handler = params.GenerateActionMetaData("test_index", "test_type", conflicting_ids=None)
+        am_handler = params.GenerateActionMetaData("test_index", "test_type")
 
         reader = params.IndexDataReader(data, batch_size=bulk_size, bulk_size=bulk_size, file_source=source, action_metadata=am_handler,
                                         index_name="test_index", type_name="test_type")
@@ -263,6 +286,69 @@ class IndexDataReaderTests(TestCase):
         expected_line_sizes = [6, 6, 2]
         self.assert_bulks_sized(reader, expected_bulk_sizes, expected_line_sizes)
 
+    def test_read_bulk_with_id_conflicts(self):
+        pseudo_random_conflicts = iter([
+            # if this value is <= our chosen threshold of 0.25 (see conflict_probability) we produce a conflict.
+            0.2,
+            0.25,
+            0.2,
+            # no conflict
+            0.3
+        ])
+
+        chosen_index_of_conflicting_ids = iter([
+            # the "random" index of the id in the array `conflicting_ids` that will produce a conflict
+            1,
+            3,
+            2])
+
+        data = [
+            '{"key": "value1"}',
+            '{"key": "value2"}',
+            '{"key": "value3"}',
+            '{"key": "value4"}',
+            '{"key": "value5"}'
+        ]
+        bulk_size = 2
+
+        source = params.Slice(io.StringAsFileSource, 0, len(data))
+        am_handler = params.GenerateActionMetaData("test_index", "test_type",
+                                                   conflicting_ids=[100, 200, 300, 400],
+                                                   conflict_probability=25,
+                                                   on_conflict="update",
+                                                   rand=lambda: next(pseudo_random_conflicts),
+                                                   randint=lambda x, y: next(chosen_index_of_conflicting_ids))
+
+        reader = params.IndexDataReader(data, batch_size=bulk_size, bulk_size=bulk_size, file_source=source, action_metadata=am_handler,
+                                        index_name="test_index", type_name="test_type")
+
+        # consume all bulks
+        bulks = []
+        with reader:
+            for index, type, batch in reader:
+                for bulk_size, bulk in batch:
+                    bulks.append(bulk)
+
+        self.assertEqual([
+            [
+                '{"index": {"_index": "test_index", "_type": "test_type", "_id": "100"}}',
+                '{"key": "value1"}',
+                '{"update": {"_index": "test_index", "_type": "test_type", "_id": "200"}}',
+                '{"doc":{"key": "value2"}}'
+            ],
+            [
+                '{"update": {"_index": "test_index", "_type": "test_type", "_id": "400"}}',
+                '{"doc":{"key": "value3"}}',
+                '{"update": {"_index": "test_index", "_type": "test_type", "_id": "300"}}',
+                '{"doc":{"key": "value4"}}'
+            ],
+            [
+                '{"index": {"_index": "test_index", "_type": "test_type", "_id": "200"}}',
+                '{"key": "value5"}'
+            ]
+
+        ], bulks)
+
     def assert_bulks_sized(self, reader, expected_bulk_sizes, expected_line_sizes):
         with reader:
             bulk_index = 0
@@ -475,6 +561,15 @@ class BulkIndexParamSourceTests(TestCase):
 
         self.assertEqual("Unknown 'conflicts' setting [crazy]", ctx.exception.args[0])
 
+    def test_create_with_unknown_on_conflict_setting(self):
+        with self.assertRaises(exceptions.InvalidSyntax) as ctx:
+            params.BulkIndexParamSource(track=track.Track(name="unit-test"), params={
+                "conflicts": "sequential",
+                "on-conflict": "delete"
+            })
+
+        self.assertEqual("Unknown 'on-conflict' setting [delete]", ctx.exception.args[0])
+
     def test_create_with_ingest_percentage_too_low(self):
         with self.assertRaises(exceptions.InvalidSyntax) as ctx:
             params.BulkIndexParamSource(track=track.Track(name="unit-test"), params={
@@ -690,7 +785,9 @@ class BulkDataGeneratorTests(TestCase):
         ])
 
         bulks = params.bulk_data_based(num_clients=1, client_index=0, corpora=[corpus],
-                                       batch_size=5, bulk_size=5, id_conflicts=params.IndexIdConflict.NoConflicts, pipeline=None,
+                                       batch_size=5, bulk_size=5,
+                                       id_conflicts=params.IndexIdConflict.NoConflicts, conflict_probability=None, on_conflict=None,
+                                       pipeline=None,
                                        original_params={
                                            "my-custom-parameter": "foo",
                                            "my-custom-parameter-2": True
@@ -748,7 +845,9 @@ class BulkDataGeneratorTests(TestCase):
             ]
 
         bulks = params.bulk_data_based(num_clients=1, client_index=0, corpora=corpora,
-                                       batch_size=5, bulk_size=5, id_conflicts=params.IndexIdConflict.NoConflicts, pipeline=None,
+                                       batch_size=5, bulk_size=5,
+                                       id_conflicts=params.IndexIdConflict.NoConflicts, conflict_probability=None, on_conflict=None,
+                                       pipeline=None,
                                        original_params={
                                            "my-custom-parameter": "foo",
                                            "my-custom-parameter-2": True
@@ -802,7 +901,8 @@ class BulkDataGeneratorTests(TestCase):
         ])
 
         bulks = params.bulk_data_based(num_clients=1, client_index=0, corpora=[corpus], batch_size=3, bulk_size=3,
-                                       id_conflicts=params.IndexIdConflict.NoConflicts, pipeline=None,
+                                       id_conflicts=params.IndexIdConflict.NoConflicts, conflict_probability=None, on_conflict=None,
+                                       pipeline=None,
                                        original_params={
                                            "body": "foo",
                                            "custom-param": "bar"