Repository: elastic/rally

Issue:
Externalize log rotation
In #503 we implemented that Rally's logging can be configured via a file. Our default configuration contains the following handler definition:

```json
    "rally_log_handler": {
      "class": "logging.handlers.TimedRotatingFileHandler",
      "filename": "${LOG_PATH}/rally.log",
      "utc": true,
      "when": "midnight",
      "backupCount": 14,
      "encoding": "UTF-8",
      "formatter": "normal",
      "filters": ["isActorLog"]
    }
```

This is problematic because Rally uses multiple processes and there is a lurking race condition when rolling log files (which we had already found in #263). Hence, we will stop rotating logs from within Rally and leverage established tools like `logrotate` for that.

### Tasks

* [x] Change the current log configuration template to use [`WatchedFileHandler`](https://docs.python.org/3/library/logging.handlers.html#watchedfilehandler) which is meant to be used with tools like `logrotate`
* [x] Determine a suitable `logrotate` configuration for Rally and provide a snippet in our docs.
* [x] Implement a check on Rally startup that ensures that the new configuration file template is used  (backup the old one if present and replace it with the new one)

Note: `WatchedFileHandler` uses the `stat` system call to check whether the inode of the log file has changed to determine whether to reopen the file. The file system metadata are cached. Also, this check is done every time a log line is written. In steady state this happens only a couple of times per minute in the default logging configuration. Therefore, we consider the performance impact of using `WatchedFileHandler` by default negligible.

Discussion:

Patch:
diff --git a/docs/configuration.rst b/docs/configuration.rst
index 7bd36dcb..108c735c 100644
--- a/docs/configuration.rst
+++ b/docs/configuration.rst
@@ -119,7 +119,19 @@ Logging in Rally is configured in ``~/.rally/logging.json``. For more informatio
 * The Python reference documentation on the `logging configuration schema <https://docs.python.org/3/library/logging.config.html#logging-config-dictschema>`_ explains the file format.
 * The `logging handler documentation <https://docs.python.org/3/library/logging.handlers.html>`_ describes how to customize where log output is written to.
 
-By default, Rally will log all output to ``~/.rally/logs/rally.log`` and rotate this file daily.
+By default, Rally will log all output to ``~/.rally/logs/rally.log``.
+
+The log file will not be rotated automatically as this is problematic due to Rally's multi-process architecture. Setup an external tool like `logrotate <https://linux.die.net/man/8/logrotate>`_ to achieve that. See the following example as a starting point for your own ``logrotate`` configuration and ensure to replace the path ``/home/user/.rally/logs/rally.log`` with the proper one::
+
+    /home/user/.rally/logs/rally.log {
+            daily                   # rotate daily
+            rotate 7                # keep the last seven log files
+            maxage 14               # remove logs older than 14 days
+            compress                # compress old logs ...
+            delaycompress           # ... after moving them
+            missingok               # ignore missing log files
+            notifempty              # don't attempt to rotate empty ones
+    }
 
 Example
 ~~~~~~~
diff --git a/docs/migrate.rst b/docs/migrate.rst
index cb8a7d7c..648c3042 100644
--- a/docs/migrate.rst
+++ b/docs/migrate.rst
@@ -1,6 +1,26 @@
 Migration Guide
 ===============
 
+Migrating to Rally 1.0.1
+------------------------
+
+Logs are not rotated
+^^^^^^^^^^^^^^^^^^^^
+
+With Rally 1.0.1 we have disabled automatic rotation of logs by default because it can lead to race conditions due to Rally's multi-process architecture. If you did not change the default out-of-the-box logging configuration, Rally will automatically fix your configuration. Otherwise, you need to replace all instances of ``logging.handlers.TimedRotatingFileHandler`` with ``logging.handlers.WatchedFileHandler`` to disable log rotation.
+
+To rotate logs we recommend to use external tools like `logrotate <https://linux.die.net/man/8/logrotate>`_. See the following example as a starting point for your own ``logrotate`` configuration and ensure to replace the path ``/home/user/.rally/logs/rally.log`` with the proper one::
+
+    /home/user/.rally/logs/rally.log {
+            daily                   # rotate daily
+            rotate 7                # keep the last seven log files
+            maxage 14               # remove logs older than 14 days
+            compress                # compress old logs ...
+            delaycompress           # ... after moving them
+            missingok               # ignore missing log files
+            notifempty              # don't attempt to rotate empty ones
+    }
+
 Migrating to Rally 1.0.0
 ------------------------
 
diff --git a/esrally/actor.py b/esrally/actor.py
index c4eb3697..ac945c5d 100644
--- a/esrally/actor.py
+++ b/esrally/actor.py
@@ -69,7 +69,9 @@ def no_retry(f, actor_name):
             msg = "Error in {}".format(actor_name)
             # log here as the full trace might get lost.
             logging.getLogger(__name__).exception(msg)
-            self.send(sender, BenchmarkFailure(msg, e))
+            # don't forward the exception as is because the main process might not have this class available on the load path
+            # and will fail then while deserializing the cause.
+            self.send(sender, BenchmarkFailure("{} ({})".format(msg, str(e))))
     return guard
 
 
diff --git a/esrally/log.py b/esrally/log.py
index 55e726ac..5320e1da 100644
--- a/esrally/log.py
+++ b/esrally/log.py
@@ -3,6 +3,7 @@ import logging.config
 import json
 import time
 import os
+import hashlib
 
 from esrally.utils import io
 
@@ -31,6 +32,26 @@ def default_log_path():
     return os.path.join(os.path.expanduser("~"), ".rally", "logs")
 
 
+def remove_obsolete_default_log_config():
+    """
+    Log rotation is problematic because Rally uses multiple processes and there is a lurking race condition when
+    rolling log files. Hence, we do not rotate logs from within Rally and leverage established tools like logrotate for that.
+
+    Checks whether the user has a problematic out-of-the-box logging configuration delivered with Rally 1.0.0 which
+    used log rotation and removes it so it can be replaced by a new one in a later step.
+    """
+    log_config = log_config_path()
+    if io.exists(log_config):
+        source_path = io.normalize_path(os.path.join(os.path.dirname(__file__), "resources", "logging_1_0_0.json"))
+        with open(source_path, "r", encoding="UTF-8") as src:
+            contents = src.read().replace("${LOG_PATH}", default_log_path())
+            source_hash = hashlib.sha512(contents.encode()).hexdigest()
+        with open(log_config, "r", encoding="UTF-8") as target:
+            target_hash = hashlib.sha512(target.read().encode()).hexdigest()
+        if source_hash == target_hash:
+            os.rename(log_config, "{}.bak".format(log_config))
+
+
 def install_default_log_config():
     """
     Ensures a log configuration file is present on this machine. The default
diff --git a/esrally/mechanic/telemetry.py b/esrally/mechanic/telemetry.py
index e5f5316f..d9cd9b0c 100644
--- a/esrally/mechanic/telemetry.py
+++ b/esrally/mechanic/telemetry.py
@@ -675,7 +675,8 @@ class DiskIo(InternalTelemetryDevice):
             else:
                 try:
                     self.disk_start = sysstats.disk_io_counters()
-                    self.logger.warning("Process I/O counters are unsupported on this platform. Falling back to less accurate disk I/O counters.")
+                    self.logger.warning("Process I/O counters are not supported on this platform. Falling back to less accurate disk "
+                                        "I/O counters.")
                 except RuntimeError:
                     self.logger.exception("Could not determine I/O stats at benchmark start.")
 
diff --git a/esrally/racecontrol.py b/esrally/racecontrol.py
index 9fafc602..281c3564 100644
--- a/esrally/racecontrol.py
+++ b/esrally/racecontrol.py
@@ -97,9 +97,11 @@ class BenchmarkActor(actor.RallyActor):
     def receiveUnrecognizedMessage(self, msg, sender):
         self.logger.info("BenchmarkActor received unknown message [%s] (ignoring).", (str(msg)))
 
+    @actor.no_retry("race control")
     def receiveMsg_Setup(self, msg, sender):
         self.setup(msg, sender)
 
+    @actor.no_retry("race control")
     def receiveMsg_EngineStarted(self, msg, sender):
         self.logger.info("Mechanic has started engine successfully.")
         self.metrics_store.meta_info = msg.system_meta_info
@@ -114,6 +116,7 @@ class BenchmarkActor(actor.RallyActor):
         # start running we assume that each race has at least one lap
         self.run()
 
+    @actor.no_retry("race control")
     def receiveMsg_TaskFinished(self, msg, sender):
         self.logger.info("Task has finished.")
         self.logger.info("Bulk adding request metrics to metrics store.")
@@ -122,17 +125,20 @@ class BenchmarkActor(actor.RallyActor):
         # other stores (used by driver and mechanic). Hence there is no need to reset the timer in our own metrics store.
         self.send(self.mechanic, mechanic.ResetRelativeTime(msg.next_task_scheduled_in))
 
+    @actor.no_retry("race control")
     def receiveMsg_BenchmarkCancelled(self, msg, sender):
         self.cancelled = True
         # even notify the start sender if it is the originator. The reason is that we call #ask() which waits for a reply.
         # We also need to ask in order to avoid races between this notification and the following ActorExitRequest.
         self.send(self.start_sender, msg)
 
+    @actor.no_retry("race control")
     def receiveMsg_BenchmarkFailure(self, msg, sender):
         self.logger.info("Received a benchmark failure from [%s] and will forward it now.", sender)
         self.error = True
         self.send(self.start_sender, msg)
 
+    @actor.no_retry("race control")
     def receiveMsg_BenchmarkComplete(self, msg, sender):
         self.logger.info("Benchmark is complete.")
         self.logger.info("Bulk adding request metrics to metrics store.")
@@ -141,6 +147,7 @@ class BenchmarkActor(actor.RallyActor):
         self.main_driver = None
         self.send(self.mechanic, mechanic.OnBenchmarkStop())
 
+    @actor.no_retry("race control")
     def receiveMsg_BenchmarkStopped(self, msg, sender):
         self.logger.info("Bulk adding system metrics to metrics store.")
         self.metrics_store.bulk_add(msg.system_metrics)
@@ -153,6 +160,7 @@ class BenchmarkActor(actor.RallyActor):
         else:
             self.teardown()
 
+    @actor.no_retry("race control")
     def receiveMsg_EngineStopped(self, msg, sender):
         self.logger.info("Mechanic has stopped engine successfully.")
         self.logger.info("Bulk adding system metrics to metrics store.")
@@ -170,8 +178,6 @@ class BenchmarkActor(actor.RallyActor):
 
     def setup(self, msg, sender):
         self.start_sender = sender
-        self.mechanic = self.createActor(mechanic.MechanicActor, targetActorRequirements={"coordinator": True})
-
         self.cfg = msg.cfg
         # to load the track we need to know the correct cluster distribution version. Usually, this value should be set but there are rare
         # cases (external pipeline and user did not specify the distribution version) where we need to derive it ourselves. For source
@@ -203,6 +209,7 @@ class BenchmarkActor(actor.RallyActor):
         self.race_store = metrics.race_store(self.cfg)
         self.logger.info("Asking mechanic to start the engine.")
         cluster_settings = challenge.cluster_settings
+        self.mechanic = self.createActor(mechanic.MechanicActor, targetActorRequirements={"coordinator": True})
         self.send(self.mechanic, mechanic.StartEngine(self.cfg, self.metrics_store.open_context, cluster_settings, msg.sources, msg.build,
                                                       msg.distribution, msg.external, msg.docker))
 
diff --git a/esrally/rally.py b/esrally/rally.py
index 51d7f14c..4e22bd51 100644
--- a/esrally/rally.py
+++ b/esrally/rally.py
@@ -483,6 +483,7 @@ def dispatch_sub_command(cfg, sub_command):
 
 def main():
     check_python_version()
+    log.remove_obsolete_default_log_config()
     log.install_default_log_config()
     log.configure_logging()
     logger = logging.getLogger(__name__)
diff --git a/esrally/rallyd.py b/esrally/rallyd.py
index c4eb948d..bb446730 100644
--- a/esrally/rallyd.py
+++ b/esrally/rallyd.py
@@ -52,6 +52,7 @@ def status():
 
 def main():
     check_python_version()
+    log.remove_obsolete_default_log_config()
     log.install_default_log_config()
     log.configure_logging()
     console.init()
diff --git a/esrally/resources/logging.json b/esrally/resources/logging.json
index 8277b033..57965787 100644
--- a/esrally/resources/logging.json
+++ b/esrally/resources/logging.json
@@ -19,11 +19,8 @@
   },
   "handlers": {
     "rally_log_handler": {
-      "class": "logging.handlers.TimedRotatingFileHandler",
+      "class": "logging.handlers.WatchedFileHandler",
       "filename": "${LOG_PATH}/rally.log",
-      "utc": true,
-      "when": "midnight",
-      "backupCount": 14,
       "encoding": "UTF-8",
       "formatter": "normal",
       "filters": ["isActorLog"]
diff --git a/esrally/resources/logging_1_0_0.json b/esrally/resources/logging_1_0_0.json
new file mode 100644
index 00000000..8277b033
--- /dev/null
+++ b/esrally/resources/logging_1_0_0.json
@@ -0,0 +1,55 @@
+{
+  "version": 1,
+  "formatters": {
+    "normal": {
+      "format": "%(asctime)s,%(msecs)d %(actorAddress)s/PID:%(process)d %(name)s %(levelname)s %(message)s",
+      "datefmt": "%Y-%m-%d %H:%M:%S",
+      "()": "esrally.log.configure_utc_formatter"
+    },
+    "profile": {
+      "format": "%(asctime)s,%(msecs)d PID:%(process)d %(name)s %(levelname)s %(message)s",
+      "datefmt": "%Y-%m-%d %H:%M:%S",
+      "()": "esrally.log.configure_utc_formatter"
+    }
+  },
+  "filters": {
+    "isActorLog": {
+      "()": "thespian.director.ActorAddressLogFilter"
+    }
+  },
+  "handlers": {
+    "rally_log_handler": {
+      "class": "logging.handlers.TimedRotatingFileHandler",
+      "filename": "${LOG_PATH}/rally.log",
+      "utc": true,
+      "when": "midnight",
+      "backupCount": 14,
+      "encoding": "UTF-8",
+      "formatter": "normal",
+      "filters": ["isActorLog"]
+    },
+    "rally_profile_handler": {
+      "class": "logging.FileHandler",
+      "filename": "${LOG_PATH}/profile.log",
+      "delay": true,
+      "encoding": "UTF-8",
+      "formatter": "profile"
+    }
+  },
+  "root": {
+    "handlers": ["rally_log_handler"],
+    "level": "INFO"
+  },
+  "loggers": {
+    "elasticsearch": {
+      "handlers": ["rally_log_handler"],
+      "level": "WARNING",
+      "propagate": false
+    },
+    "rally.profile": {
+      "handlers": ["rally_profile_handler"],
+      "level": "INFO",
+      "propagate": false
+    }
+  }
+}
\ No newline at end of file
diff --git a/esrally/track/loader.py b/esrally/track/loader.py
index 8194ec17..e24c4c4b 100644
--- a/esrally/track/loader.py
+++ b/esrally/track/loader.py
@@ -231,14 +231,15 @@ def operation_parameters(t, op):
 
 def used_corpora(t, cfg):
     corpora = {}
-    challenge = t.find_challenge_or_default(cfg.opts("track", "challenge.name"))
-    for task in challenge.schedule:
-        for sub_task in task:
-            param_source = operation_parameters(t, sub_task.operation)
-            if hasattr(param_source, "corpora"):
-                for c in param_source.corpora:
-                    # We might have the same corpus *but* they contain different doc sets. Therefore also need to union over doc sets.
-                    corpora[c.name] = corpora.get(c.name, c).union(c)
+    if t.corpora:
+        challenge = t.find_challenge_or_default(cfg.opts("track", "challenge.name"))
+        for task in challenge.schedule:
+            for sub_task in task:
+                param_source = operation_parameters(t, sub_task.operation)
+                if hasattr(param_source, "corpora"):
+                    for c in param_source.corpora:
+                        # We might have the same corpus *but* they contain different doc sets. Therefore also need to union over doc sets.
+                        corpora[c.name] = corpora.get(c.name, c).union(c)
     return corpora.values()
 
 
@@ -633,7 +634,7 @@ class TrackFileReader:
         except jinja2.exceptions.TemplateNotFound:
             self.logger.exception("Could not load [%s]", track_spec_file)
             raise exceptions.SystemSetupError("Track {} does not exist".format(track_name))
-        except (json.JSONDecodeError, jinja2.exceptions.TemplateError) as e:
+        except Exception as e:
             self.logger.exception("Could not load [%s].", track_spec_file)
             # Convert to string early on to avoid serialization errors with Jinja exceptions.
             raise TrackSyntaxError("Could not load '{}'".format(track_spec_file), str(e))
@@ -775,7 +776,7 @@ class TrackSpecificationReader:
                                        template_name="default",
                                        template_vars=self.track_params)
             return json.loads(rendered)
-        except (json.JSONDecodeError, jinja2.exceptions.TemplateError) as e:
+        except Exception as e:
             self.logger.exception("Could not load file template for %s.", description)
             raise TrackSyntaxError("Could not load file template for '%s'" % description, str(e))
 
diff --git a/esrally/utils/git.py b/esrally/utils/git.py
index b5d044e8..83c90c3f 100644
--- a/esrally/utils/git.py
+++ b/esrally/utils/git.py
@@ -84,6 +84,11 @@ def head_revision(src_dir):
     return process.run_subprocess_with_output("git -C {0} rev-parse --short HEAD".format(src_dir))[0].strip()
 
 
+@probed
+def current_branch(src_dir):
+    return process.run_subprocess_with_output("git -C {0} rev-parse --abbrev-ref HEAD".format(src_dir))[0].strip()
+
+
 @probed
 def branches(src_dir, remote=True):
     if remote:
diff --git a/esrally/utils/repo.py b/esrally/utils/repo.py
index 46eba13b..511fd96d 100644
--- a/esrally/utils/repo.py
+++ b/esrally/utils/repo.py
@@ -57,8 +57,9 @@ class RallyRepository:
                     self.logger.warning(msg)
             branch = versions.best_match(git.branches(self.repo_dir, remote=False), distribution_version)
             if branch:
-                self.logger.info("Checking out [%s] in [%s] for distribution version [%s].", branch, self.repo_dir, distribution_version)
-                git.checkout(self.repo_dir, branch=branch)
+                if git.current_branch(self.repo_dir) != branch:
+                    self.logger.info("Checking out [%s] in [%s] for distribution version [%s].", branch, self.repo_dir, distribution_version)
+                    git.checkout(self.repo_dir, branch=branch)
             else:
                 raise exceptions.SystemSetupError("Cannot find %s for distribution version %s" % (self.resource_name, distribution_version))
         except exceptions.SupplyError as e:


Test patch:
diff --git a/tests/track/loader_test.py b/tests/track/loader_test.py
index f08bd286..bb8e799c 100644
--- a/tests/track/loader_test.py
+++ b/tests/track/loader_test.py
@@ -1367,6 +1367,58 @@ class TrackSpecificationReaderTests(TestCase):
                          "'duplicate-task-name'. Please use the task's name property to assign a unique name for each task.",
                          ctx.exception.args[0])
 
+    def test_load_invalid_index_body(self):
+        track_specification = {
+            "description": "description for unit test",
+            "indices": [
+                {
+                    "name": "index-historical",
+                    "body": "body.json",
+                    "types": ["_doc"]
+                }
+            ],
+            "corpora": [
+                {
+                    "name": "test",
+                    "documents": [
+                        {
+                            "source-file": "documents-main.json.bz2",
+                            "document-count": 10,
+                            "compressed-bytes": 100,
+                            "uncompressed-bytes": 10000
+                        }
+                    ]
+                }
+            ],
+            "schedule": [
+                {
+                    "clients": 8,
+                    "operation": {
+                        "name": "index-append",
+                        "operation-type": "index",
+                        "bulk-size": 5000
+                    }
+                }
+            ]
+        }
+        reader = loader.TrackSpecificationReader(
+            track_params={"number_of_shards": 3},
+            source=io.DictStringFileSourceFactory({
+                "/mappings/body.json": ["""
+            {
+                "settings": {
+                    "number_of_shards": {{ number_of_shards }}
+                },
+                "mappings": {
+                    "_doc": "no closing quotation mark!!,
+                }
+            }
+            """]
+            }))
+        with self.assertRaises(loader.TrackSyntaxError) as ctx:
+            reader("unittest", track_specification, "/mappings")
+        self.assertEqual("Could not load file template for 'definition for index index-historical in body.json'", ctx.exception.args[0])
+
     def test_parse_unique_task_names(self):
         track_specification = {
             "description": "description for unit test",
diff --git a/tests/utils/repo_test.py b/tests/utils/repo_test.py
index 765eddf3..ab67962c 100644
--- a/tests/utils/repo_test.py
+++ b/tests/utils/repo_test.py
@@ -125,7 +125,9 @@ class RallyRepositoryTests(TestCase):
     @mock.patch("esrally.utils.git.branches", autospec=True)
     @mock.patch("esrally.utils.git.checkout", autospec=True)
     @mock.patch("esrally.utils.git.rebase")
-    def test_updates_locally(self, rebase, checkout, branches, fetch, is_working_copy):
+    @mock.patch("esrally.utils.git.current_branch")
+    def test_updates_locally(self, curr_branch, rebase, checkout, branches, fetch, is_working_copy):
+        curr_branch.return_value = "5"
         branches.return_value = ["1", "2", "5", "master"]
         is_working_copy.return_value = True
 
@@ -181,7 +183,9 @@ class RallyRepositoryTests(TestCase):
     @mock.patch("esrally.utils.git.branches", autospec=True)
     @mock.patch("esrally.utils.git.checkout", autospec=True)
     @mock.patch("esrally.utils.git.rebase")
-    def test_does_not_update_unknown_branch_remotely_local_fallback(self, rebase, checkout, branches, fetch, is_working_copy):
+    @mock.patch("esrally.utils.git.current_branch")
+    def test_does_not_update_unknown_branch_remotely_local_fallback(self, curr_branch, rebase, checkout, branches, fetch, is_working_copy):
+        curr_branch.return_value = "master"
         # we have only "master" remotely but a few more branches locally
         branches.side_effect = ["5", ["1", "2", "5", "master"]]
         is_working_copy.return_value = True