Add ability to delete race from metric store

docs/command_line_reference.rst

@@ -127,6 +127,13 @@ This will show the path to the default distribution::
       "elasticsearch": "/Users/dm/.rally/benchmarks/distributions/elasticsearch-6.8.0.tar.gz"
     }
 
+``delete``
+~~~~~~~~~~~
+
+The ``delete`` subcommand is used to delete records for different configuration options:
+
+* race: Will delete a race and all corresponding metrics records from Elasticsearch metric store. ``--id`` is a required command line flag. Use ``list races`` to get the id of the race to be deleted.
+
 ``install``
 ~~~~~~~~~~~
 

esrally/metrics.py

@@ -66,6 +66,12 @@ def template_exists(self, name):
     def delete_template(self, name):
         self.guarded(self._client.indices.delete_template, name=name)
 
+    def delete_by_query(self, index, body):
+        return self.guarded(self._client.delete_by_query, index=index, body=body)
+
+    def delete(self, index, id):
+        return self.guarded(self._client.delete, index=index, id=id, ignore=404)
+
     def get_index(self, name):
         return self.guarded(self._client.indices.get, name=name)
 
@@ -1225,6 +1231,10 @@ def results_store(cfg):
         return NoopResultsStore()
 
 
+def delete_race(cfg):
+    race_store(cfg).delete_race()
+
+
 def list_races(cfg):
     def format_dict(d):
         if d:
@@ -1494,23 +1504,44 @@ def find_by_race_id(self, race_id):
     def list(self):
         raise NotImplementedError("abstract method")
 
+    def delete_race(self):
+        raise NotImplementedError("abstract method")
+
     def store_race(self, race):
         raise NotImplementedError("abstract method")
 
     def _max_results(self):
-        return int(self.cfg.opts("system", "list.races.max_results"))
+        return int(self.cfg.opts("system", "list.max_results"))
 
     def _track(self):
-        return self.cfg.opts("system", "list.races.track", mandatory=False)
+        return self.cfg.opts("system", "admin.track", mandatory=False)
 
     def _benchmark_name(self):
         return self.cfg.opts("system", "list.races.benchmark_name", mandatory=False)
 
+    def _race_timestamp(self):
+        return self.cfg.opts("system", "add.race_timestamp")
+
+    def _message(self):
+        return self.cfg.opts("system", "add.message")
+
+    def _chart_type(self):
+        return self.cfg.opts("system", "add.chart_type", mandatory=False)
+
+    def _chart_name(self):
+        return self.cfg.opts("system", "add.chart_name", mandatory=False)
+
     def _from_date(self):
-        return self.cfg.opts("system", "list.races.from_date", mandatory=False)
+        return self.cfg.opts("system", "list.from_date", mandatory=False)
 
     def _to_date(self):
-        return self.cfg.opts("system", "list.races.to_date", mandatory=False)
+        return self.cfg.opts("system", "list.to_date", mandatory=False)
+
+    def _dry_run(self):
+        return self.cfg.opts("system", "admin.dry_run", mandatory=False)
+
+    def _id(self):
+        return self.cfg.opts("system", "delete.id")
 
 
 # Does not inherit from RaceStore as it is only a delegator with the same API.
@@ -1532,6 +1563,9 @@ def store_race(self, race):
         self.file_store.store_race(race)
         self.es_store.store_race(race)
 
+    def delete_race(self):
+        return self.es_store.delete_race()
+
     def list(self):
         return self.es_store.list()
 
@@ -1547,6 +1581,9 @@ def store_race(self, race):
     def _race_file(self, race_id=None):
         return os.path.join(paths.race_root(cfg=self.cfg, race_id=race_id), "race.json")
 
+    def delete_race(self):
+        raise NotImplementedError("Not supported for in-memory datastore.")
+
     def list(self):
         results = glob.glob(self._race_file(race_id="*"))
         all_races = self._to_races(results)
@@ -1614,6 +1651,25 @@ def index_name(self, race):
         race_timestamp = race.race_timestamp
         return f"{EsRaceStore.INDEX_PREFIX}{race_timestamp:%Y-%m}"
 
+    def delete_race(self):
+        races = self._id().split(",")
+        environment = self.environment_name
+        if self._dry_run():
+            if len(races) == 1:
+                console.println(f"Would delete race with id {races[0]} in environment {environment}.")
+            else:
+                console.println(f"Would delete {len(races)} races: {races} in environment {environment}.")
+        else:
+            for race_id in races:
+                selector = {"query": {"bool": {"filter": [{"term": {"environment": environment}}, {"term": {"race-id": race_id}}]}}}
+                self.client.delete_by_query(index="rally-races-*", body=selector)
+                self.client.delete_by_query(index="rally-metrics-*", body=selector)
+                result = self.client.delete_by_query(index="rally-results-*", body=selector)
+                if result["deleted"] > 0:
+                    console.println(f"Successfully deleted [{race_id}] in environment [{environment}].")
+                else:
+                    console.println(f"Did not find [{race_id}] in environment [{environment}].")
+
     def list(self):
         track = self._track()
         name = self._benchmark_name()

esrally/rally.py

@@ -178,6 +178,19 @@ def add_track_source(subparser):
     )
     add_track_source(list_parser)
 
+    delete_parser = subparsers.add_parser("delete", help="Delete records")
+    delete_parser.add_argument(
+        "configuration",
+        metavar="configuration",
+        help="The configuration for which Rally should delete the available records. " "Possible values are: race",
+        choices=["race"],
+    )
+    delete_parser.add_argument(
+        "--dry-run", help="Just show what would be done but do not apply the operation.", default=False, action="store_true"
+    )
+    delete_parser.add_argument("--id", help="Ids of the items to delete. Separate multiple ids with a comma.", required=True)
+    add_track_source(delete_parser)
+
     info_parser = subparsers.add_parser("info", help="Show info about a track")
     add_track_source(info_parser)
     info_parser.add_argument(
@@ -771,6 +784,7 @@ def add_track_source(subparser):
 
     for p in [
         list_parser,
+        delete_parser,
         race_parser,
         compare_parser,
         build_parser,
@@ -822,6 +836,14 @@ def dispatch_list(cfg):
         raise exceptions.SystemSetupError("Cannot list unknown configuration option [%s]" % what)
 
 
+def dispatch_delete(cfg):
+    what = cfg.opts("system", "delete.config.option")
+    if what == "race":
+        metrics.delete_race(cfg)
+    else:
+        raise exceptions.SystemSetupError("Cannot delete unknown configuration option [%s]" % what)
+
+
 def print_help_on_errors():
     heading = "Getting further help:"
     console.println(console.format.bold(heading))
@@ -1025,14 +1047,19 @@ def dispatch_sub_command(arg_parser, args, cfg):
             reporter.compare(cfg, args.baseline, args.contender)
         elif sub_command == "list":
             cfg.add(config.Scope.applicationOverride, "system", "list.config.option", args.configuration)
-            cfg.add(config.Scope.applicationOverride, "system", "list.races.max_results", args.limit)
-            cfg.add(config.Scope.applicationOverride, "system", "list.races.track", args.track)
+            cfg.add(config.Scope.applicationOverride, "system", "list.max_results", args.limit)
+            cfg.add(config.Scope.applicationOverride, "system", "admin.track", args.track)
             cfg.add(config.Scope.applicationOverride, "system", "list.races.benchmark_name", args.benchmark_name)
-            cfg.add(config.Scope.applicationOverride, "system", "list.races.from_date", args.from_date)
+            cfg.add(config.Scope.applicationOverride, "system", "list.from_date", args.from_date)
             cfg.add(config.Scope.applicationOverride, "system", "list.races.to_date", args.to_date)
             configure_mechanic_params(args, cfg, command_requires_car=False)
             configure_track_params(arg_parser, args, cfg, command_requires_track=False)
             dispatch_list(cfg)
+        elif sub_command == "delete":
+            cfg.add(config.Scope.applicationOverride, "system", "delete.config.option", args.configuration)
+            cfg.add(config.Scope.applicationOverride, "system", "delete.id", args.id)
+            cfg.add(config.Scope.applicationOverride, "system", "admin.dry_run", args.dry_run)
+            dispatch_delete(cfg)
         elif sub_command == "build":
             cfg.add(config.Scope.applicationOverride, "mechanic", "car.plugins", opts.csv_to_list(args.elasticsearch_plugins))
             cfg.add(config.Scope.applicationOverride, "mechanic", "plugin.params", opts.to_dict(args.plugin_params))

tests/metrics_test.py

@@ -862,7 +862,7 @@ def as_dict(self):
 
     def setup_method(self, method):
         self.cfg = config.Config()
-        self.cfg.add(config.Scope.application, "system", "list.races.max_results", 100)
+        self.cfg.add(config.Scope.application, "system", "list.max_results", 100)
         self.cfg.add(config.Scope.application, "system", "env.name", "unittest-env")
         self.cfg.add(config.Scope.application, "system", "time.start", self.RACE_TIMESTAMP)
         self.cfg.add(config.Scope.application, "system", "race.id", self.RACE_ID)
@@ -1001,12 +1001,21 @@ def test_store_race(self):
         }
         self.es_mock.index.assert_called_with(index="rally-races-2016-01", id=self.RACE_ID, item=expected_doc)
 
+    @mock.patch("esrally.utils.console.println")
+    def test_delete_race(self, console):
+        self.es_mock.delete_by_query.return_value = {"deleted": 0}
+        self.cfg.add(config.Scope.application, "system", "delete.id", "0101")
+        self.race_store.delete_race()
+        expected_query = {"query": {"bool": {"filter": [{"term": {"environment": "unittest-env"}}, {"term": {"race-id": "0101"}}]}}}
+        self.es_mock.delete_by_query.assert_called_with(index="rally-results-*", body=expected_query)
+        console.assert_called_with("Did not find [0101] in environment [unittest-env].")
+
     def test_filter_race(self):
         self.es_mock.search.return_value = {"hits": {"total": 0}}
-        self.cfg.add(config.Scope.application, "system", "list.races.track", "unittest")
+        self.cfg.add(config.Scope.application, "system", "admin.track", "unittest")
         self.cfg.add(config.Scope.application, "system", "list.races.benchmark_name", "unittest-test")
-        self.cfg.add(config.Scope.application, "system", "list.races.to_date", "20160131")
-        self.cfg.add(config.Scope.application, "system", "list.races.from_date", "20160230")
+        self.cfg.add(config.Scope.application, "system", "list.to_date", "20160131")
+        self.cfg.add(config.Scope.application, "system", "list.from_date", "20160230")
         self.race_store.list()
         expected_query = {
             "query": {
@@ -1598,7 +1607,7 @@ def setup_method(self):
         self.cfg = config.Config()
         self.cfg.add(config.Scope.application, "node", "root.dir", os.path.join(tempfile.gettempdir(), str(uuid.uuid4())))
         self.cfg.add(config.Scope.application, "system", "env.name", "unittest-env")
-        self.cfg.add(config.Scope.application, "system", "list.races.max_results", 100)
+        self.cfg.add(config.Scope.application, "system", "list.max_results", 100)
         self.cfg.add(config.Scope.application, "system", "time.start", self.RACE_TIMESTAMP)
         self.cfg.add(config.Scope.application, "system", "race.id", self.RACE_ID)
         self.race_store = metrics.FileRaceStore(self.cfg)
@@ -1688,21 +1697,28 @@ def test_filter_race(self):
 
         self.race_store.store_race(race)
         assert len(self.race_store.list()) == 1
-        self.cfg.add(config.Scope.application, "system", "list.races.track", "unittest-2")
+        self.cfg.add(config.Scope.application, "system", "admin.track", "unittest-2")
         assert len(self.race_store.list()) == 0
-        self.cfg.add(config.Scope.application, "system", "list.races.track", "unittest")
+        self.cfg.add(config.Scope.application, "system", "admin.track", "unittest")
         assert len(self.race_store.list()) == 1
         self.cfg.add(config.Scope.application, "system", "list.races.benchmark_name", "unittest-test-2")
         assert len(self.race_store.list()) == 0
         self.cfg.add(config.Scope.application, "system", "list.races.benchmark_name", "unittest-test")
         assert len(self.race_store.list()) == 1
-        self.cfg.add(config.Scope.application, "system", "list.races.to_date", "20160129")
+        self.cfg.add(config.Scope.application, "system", "list.to_date", "20160129")
         assert len(self.race_store.list()) == 0
-        self.cfg.add(config.Scope.application, "system", "list.races.to_date", "20160131")
+        self.cfg.add(config.Scope.application, "system", "list.to_date", "20160131")
         assert len(self.race_store.list()) == 1
-        self.cfg.add(config.Scope.application, "system", "list.races.from_date", "20160131")
+        self.cfg.add(config.Scope.application, "system", "list.from_date", "20160131")
         assert len(self.race_store.list()) == 1
 
+    def test_delete_race(self):
+        self.cfg.add(config.Scope.application, "system", "delete.id", "0101")
+
+        with pytest.raises(NotImplementedError) as ctx:
+            self.race_store.delete_race()
+        assert ctx.value.args[0] == "Not supported for in-memory datastore."
+
 
 class TestStatsCalculator:
     def test_calculate_global_stats(self):
@@ -2374,7 +2390,7 @@ def setup_method(self, method):
         self.cfg.add(config.Scope.application, "node", "root.dir", os.path.join(tempfile.gettempdir(), str(uuid.uuid4())))
         self.cfg.add(config.Scope.application, "node", "rally.root", paths.rally_root())
         self.cfg.add(config.Scope.application, "system", "env.name", "unittest-env")
-        self.cfg.add(config.Scope.application, "system", "list.races.max_results", 100)
+        self.cfg.add(config.Scope.application, "system", "list.max_results", 100)
         self.cfg.add(config.Scope.application, "system", "time.start", TestFileRaceStore.RACE_TIMESTAMP)
         self.cfg.add(config.Scope.application, "system", "race.id", TestFileRaceStore.RACE_ID)