refactor(elastic): remove all elasticsearch related code

since we are now sending performence stats into Argus,
we don't need anything to be sending data to ElasticSearch
anymore

Author: fruch

Diff for: add_new_dc_test.py

@@ -77,15 +77,15 @@ def reconfigure_keyspaces_to_use_network_topology_strategy(self, keyspaces: List
     def prewrite_db_with_data(self) -> None:
         self.log.info("Prewriting database...")
         stress_cmd = self.params.get('prepare_write_cmd')
-        pre_thread = self.run_stress_thread(stress_cmd=stress_cmd, stats_aggregate_cmds=False, round_robin=False)
+        pre_thread = self.run_stress_thread(stress_cmd=stress_cmd, round_robin=False)
         self.verify_stress_thread(cs_thread_pool=pre_thread)
         self.log.info("Database pre write completed")
 
     def start_stress_during_adding_new_dc(self) -> Tuple[CassandraStressThread, CassandraStressThread]:
         self.log.info("Running stress during adding new DC")
         stress_cmds = self.params.get('stress_cmd')
-        read_thread = self.run_stress_thread(stress_cmd=stress_cmds[0], stats_aggregate_cmds=False, round_robin=False)
-        write_thread = self.run_stress_thread(stress_cmd=stress_cmds[1], stats_aggregate_cmds=False, round_robin=False)
+        read_thread = self.run_stress_thread(stress_cmd=stress_cmds[0], round_robin=False)
+        write_thread = self.run_stress_thread(stress_cmd=stress_cmds[1], round_robin=False)
         self.log.info("Stress during adding DC started")
         return read_thread, write_thread
 
@@ -106,7 +106,7 @@ def add_node_in_new_dc(self) -> BaseNode:
     def verify_data_can_be_read_from_new_dc(self, new_node: BaseNode) -> None:
         self.log.info("Veryfing if data has been transferred successfully to the new DC")
         stress_cmd = self.params.get('verify_data_after_entire_test') + f" -node {new_node.ip_address}"
-        end_stress = self.run_stress_thread(stress_cmd=stress_cmd, stats_aggregate_cmds=False, round_robin=False)
+        end_stress = self.run_stress_thread(stress_cmd=stress_cmd, round_robin=False)
         self.verify_stress_thread(cs_thread_pool=end_stress)
 
     def querying_new_node_should_return_no_data(self, new_node: BaseNode) -> None:

Diff for: admission_control_overload_test.py

@@ -27,14 +27,12 @@ def check_prometheus_metrics(self, start_time, now):
     def run_load(self, job_num, job_cmd, is_prepare=False):
         is_ever_triggered = False
         if is_prepare and not skip_optional_stage('prepare_write'):
-            prepare_stress_queue = self.run_stress_thread(stress_cmd=job_cmd, stress_num=job_num, prefix='preload-',
-                                                          stats_aggregate_cmds=False)
+            prepare_stress_queue = self.run_stress_thread(stress_cmd=job_cmd, stress_num=job_num, prefix='preload-')
             self.get_stress_results(prepare_stress_queue)
         elif not is_prepare and not skip_optional_stage('main_load'):
             stress_queue = []
             stress_res = []
-            stress_queue.append(self.run_stress_thread(stress_cmd=job_cmd, stress_num=job_num,
-                                                       stats_aggregate_cmds=False))
+            stress_queue.append(self.run_stress_thread(stress_cmd=job_cmd, stress_num=job_num))
 
             start_time = time.time()
             while stress_queue:

Diff for: docs/bisecting-with-sct.md

@@ -32,11 +32,10 @@ def test_write(self):
     if stress_multiplier_w := self.params.get("stress_multiplier_w"):
         stress_multiplier = stress_multiplier_w
 
-    self.create_test_stats(doc_id_with_timestamp=True)
     self.run_fstrim_on_all_db_nodes()
 
     stress_queue = self.run_stress_thread(
-        stress_cmd=base_cmd_w, stress_num=stress_multiplier, stats_aggregate_cmds=False)
+        stress_cmd=base_cmd_w, stress_num=stress_multiplier)
     results = self.get_stress_results(queue=stress_queue)
 
     # set bisect_ref_value only for the first test run
@@ -46,5 +45,4 @@ def test_write(self):
     self.build_histogram(PerformanceTestWorkload.WRITE, PerformanceTestType.THROUGHPUT)
     self.update_test_details(scylla_conf=True)
     self.display_results(results, test_name='test_write')
-    self.check_regression()
 ```

Diff for: jupyter/provision-test.ipynb

@@ -93,8 +93,6 @@
     "        self.stats[disrupt][key[status]].append(data)\n",
     "        self.stats[disrupt]['cnt'] += 1\n",
     "        self.log.debug('Update nemesis info with: %s', data)\n",
-    "        if self.tester.create_stats:\n",
-    "            self.tester.update({'nemesis': self.stats})\n",
     "        if self.es_publisher:\n",
     "            self.es_publisher.publish(disrupt_name=disrupt, status=status, data=data)\n",
     "\n",

Diff for: microbenchmarking_test.py

@@ -14,7 +14,6 @@
 
 from sdcm.argus_results import send_perf_simple_query_result_to_argus
 from sdcm.tester import ClusterTester, teardown_on_exception, log_run_info
-from sdcm.utils.microbenchmarking.perf_simple_query_reporter import PerfSimpleQueryAnalyzer
 
 
 class PerfSimpleQueryTest(ClusterTester):
@@ -32,12 +31,5 @@ def test_perf_simple_query(self):
         if result.ok:
             output = self.db_cluster.nodes[0].remoter.run("cat perf-simple-query-result.txt").stdout
             results = json.loads(output)
-            self.create_test_stats(
-                specific_tested_stats={"perf_simple_query_result": results},
-                doc_id_with_timestamp=True)
-            if self.create_stats:
-                is_gce = self.params.get('cluster_backend') == 'gce'
-                PerfSimpleQueryAnalyzer(self._test_index, self._es_doc_type).check_regression(
-                    self._test_id, is_gce=is_gce,
-                    extra_jobs_to_compare=self.params.get('perf_extra_jobs_to_compare'))
-            send_perf_simple_query_result_to_argus(self.test_config.argus_client(), results)
+
+        send_perf_simple_query_result_to_argus(self.test_config.argus_client(), results)

Diff for: performance_regression_alternator_test.py

@@ -36,14 +36,9 @@ def _workload(self, stress_cmd, stress_num, test_name=None, sub_type=None, keysp
         if debug_message:
             self.log.debug(debug_message)
 
-        if save_stats:
-            self.create_test_stats(test_name=test_name, sub_type=sub_type,
-                                   doc_id_with_timestamp=True, append_sub_test_to_name=False)
         stress_queue = self.run_stress_thread(stress_cmd=stress_cmd, stress_num=stress_num, keyspace_num=keyspace_num,
-                                              prefix=prefix, stats_aggregate_cmds=False)
+                                              prefix=prefix)
         self.get_stress_results(queue=stress_queue, store_results=True)
-        if save_stats:
-            self.update_test_details(scylla_conf=True, alternator=is_alternator)
 
     def create_alternator_table(self, schema, alternator_write_isolation):
         node = self.db_cluster.nodes[0]
@@ -74,9 +69,6 @@ def preload_data(self, compaction_strategy=None):
         # if test require a pre-population of data
         prepare_write_cmd = self.params.get('prepare_write_cmd')
         if prepare_write_cmd:
-            # create new document in ES with doc_id = test_id + timestamp
-            # allow to correctly save results for future compare
-            self.create_test_stats(sub_type='write-prepare', doc_id_with_timestamp=True)
             stress_queue = []
             params = {'prefix': 'preload-'}
             for stress_type in ['dynamodb', 'cassandra-cql']:
@@ -91,14 +83,13 @@ def preload_data(self, compaction_strategy=None):
                         params.update({'stress_cmd': stress_cmd.replace('dynamodb', stress_type)})
 
                         # Run all stress commands
-                        params.update(dict(stats_aggregate_cmds=False))
                         self.log.debug('RUNNING stress cmd: {}'.format(stress_cmd.replace('dynamodb', stress_type)))
                         stress_queue.append(self.run_stress_thread(**params))
 
                 # One stress cmd command
                 else:
                     stress_queue.append(self.run_stress_thread(stress_cmd=prepare_write_cmd.replace('dynamodb', stress_type), stress_num=1,
-                                                               prefix='preload-', stats_aggregate_cmds=False))
+                                                               prefix='preload-'))
 
             for stress in stress_queue:
                 self.get_stress_results(queue=stress, store_results=False)
@@ -136,8 +127,6 @@ def test_write(self):
             schema=schema, alternator_write_isolation=alternator.enums.WriteIsolation.ALWAYS_USE_LWT)
         self._workload(sub_type='with-lwt', stress_cmd=base_cmd_w, stress_num=stress_multiplier, keyspace_num=1)
 
-        self.check_regression_with_baseline('cql')
-
     def test_read(self):
         """
         Test steps:
@@ -171,8 +160,6 @@ def test_read(self):
         self.alternator.set_write_isolation(node=node, isolation=alternator.enums.WriteIsolation.ALWAYS_USE_LWT)
         self._workload(sub_type='with-lwt', stress_cmd=base_cmd_r, stress_num=stress_multiplier, keyspace_num=1)
 
-        self.check_regression_with_baseline('cql')
-
     def test_mixed(self):
         """
         Test steps:
@@ -207,8 +194,6 @@ def test_mixed(self):
         self.alternator.set_write_isolation(node=node, isolation=alternator.enums.WriteIsolation.ALWAYS_USE_LWT)
         self._workload(sub_type='with-lwt', stress_cmd=base_cmd_m, stress_num=stress_multiplier, keyspace_num=1)
 
-        self.check_regression_with_baseline('cql')
-
     def test_latency(self):
         """
         Test steps:
@@ -259,7 +244,6 @@ def test_latency(self):
         self._workload(
             test_name=self.id() + '_read', sub_type='with-lwt', stress_cmd=base_cmd_r, stress_num=stress_multiplier,
             keyspace_num=1)
-        self.check_regression_with_baseline('cql')
 
         stress_multiplier = 1
         self.run_fstrim_on_all_db_nodes()
@@ -282,7 +266,6 @@ def test_latency(self):
         self._workload(
             test_name=self.id() + '_write', sub_type='with-lwt', stress_cmd=base_cmd_w + " -target 3000",
             stress_num=stress_multiplier, keyspace_num=1)
-        self.check_regression_with_baseline('cql')
 
         stress_multiplier = 1
         self.wait_no_compactions_running(n=120)
@@ -303,5 +286,3 @@ def test_latency(self):
         self.alternator.set_write_isolation(node=node, isolation=alternator.enums.WriteIsolation.ALWAYS_USE_LWT)
         self._workload(test_name=self.id() + '_mixed', sub_type='with-lwt',
                        stress_cmd=base_cmd_m + " -target 5000", stress_num=stress_multiplier, keyspace_num=1)
-
-        self.check_regression_with_baseline('cql')

Diff for: performance_regression_cdc_test.py

@@ -37,7 +37,6 @@ def test_write_with_cdc(self):
                            sub_type="cdc_enabled")
 
         self.wait_no_compactions_running()
-        self.check_regression_with_baseline(subtest_baseline="cdc_disabled")
 
     def test_write_with_cdc_preimage(self):
         write_cmd = self.params.get("stress_cmd_w")
@@ -56,8 +55,6 @@ def test_write_with_cdc_preimage(self):
                            test_name="test_write",
                            sub_type="cdc_preimage_enabled")
 
-        self.check_regression_with_baseline(subtest_baseline="cdc_disabled")
-
     def test_write_with_cdc_postimage(self):
         write_cmd = self.params.get("stress_cmd_w")
 
@@ -75,8 +72,6 @@ def test_write_with_cdc_postimage(self):
                            test_name="test_write",
                            sub_type="cdc_preimage_enabled")
 
-        self.check_regression_with_baseline(subtest_baseline="cdc_disabled")
-
     def test_write_throughput(self):
         self.cdc_workflow()
 
@@ -150,19 +145,11 @@ def cdc_workflow(self, use_cdclog_reader=False):  # pylint: disable=unused-varia
 
         self.wait_no_compactions_running()
 
-        self.check_regression_with_baseline(subtest_baseline="cdc_disabled")
-
     def _workload_cdc(self, stress_cmd, stress_num, test_name, sub_type=None,  # pylint: disable=too-many-arguments
                       save_stats=True, read_cdclog_cmd=None, update_cdclog_stats=False, enable_batching=True):
         cdc_stress_queue = None
 
-        if save_stats:
-            self.create_test_stats(sub_type=sub_type,
-                                   doc_id_with_timestamp=True,
-                                   append_sub_test_to_name=False)
-
-        stress_queue = self.run_stress_thread(stress_cmd=stress_cmd, stress_num=stress_num,
-                                              stats_aggregate_cmds=False)
+        stress_queue = self.run_stress_thread(stress_cmd=stress_cmd, stress_num=stress_num)
 
         if read_cdclog_cmd:
             cdc_stress_queue = self.run_cdclog_reader_thread(stress_cmd=read_cdclog_cmd,

Diff for: performance_regression_gradual_grow_throughput.py

@@ -1,7 +1,5 @@
-import pathlib
 import time
 from enum import Enum
-from collections import defaultdict
 
 import json
 from typing import NamedTuple
@@ -10,9 +8,7 @@
 from sdcm.utils.common import skip_optional_stage
 from sdcm.sct_events import Severity
 from sdcm.sct_events.system import TestFrameworkEvent
-from sdcm.results_analyze import PredefinedStepsTestPerformanceAnalyzer
 from sdcm.utils.decorators import latency_calculator_decorator
-from sdcm.utils.latency import calculate_latency, analyze_hdr_percentiles
 
 
 class CSPopulateDistribution(Enum):
@@ -160,7 +156,6 @@ def preload_data(self, compaction_strategy=None):
         for stress_cmd in population_commands:
             params.update({'stress_cmd': stress_cmd})
             # Run all stress commands
-            params.update(dict(stats_aggregate_cmds=False))
             self.log.debug('RUNNING stress cmd: {}'.format(stress_cmd))
             stress_queue.append(self.run_stress_thread(**params))
 
@@ -174,20 +169,12 @@ def check_latency_during_steps(self, step):
             latency_results = json.load(file)
         self.log.debug('Step %s: latency_results were loaded from file %s and its result is %s',
                        step, self.latency_results_file, latency_results)
-        if latency_results and self.create_stats:
-            latency_results[step]["step"] = step
-            latency_results[step] = calculate_latency(latency_results[step])
-            latency_results = analyze_hdr_percentiles(latency_results)
-            pathlib.Path(self.latency_results_file).unlink()
-            self.log.debug('collected latency values are: %s', latency_results)
-            self.update({"latency_during_ops": latency_results})
-            return latency_results
 
     def run_step(self, stress_cmds, current_throttle, num_threads, step_duration):
         results = []
         stress_queue = []
         for stress_cmd in stress_cmds:
-            params = {"round_robin": True, "stats_aggregate_cmds": False}
+            params = {"round_robin": True}
             stress_cmd_to_run = stress_cmd.replace(
                 "$threads", f"{num_threads}").replace("$throttle", f"{current_throttle}")
             if step_duration is not None:
@@ -215,23 +202,17 @@ def run_gradual_increase_load(self, workload: Workload, stress_num, num_loaders,
             # Wait for 4 minutes after warmup to let for all background processes to finish
             time.sleep(240)
 
-        if not self.exists():
-            self.log.debug("Create test statistics in ES")
-            self.create_test_stats(sub_type=workload.workload_type, doc_id_with_timestamp=True)
         total_summary = {}
 
         for throttle_step in workload.throttle_steps:
             self.log.info("Run cs command with rate: %s Kops", throttle_step)
             current_throttle = f"fixed={int(int(throttle_step) // (num_loaders * stress_num))}/s" if throttle_step != "unthrottled" else ""
             run_step = ((latency_calculator_decorator(legend=f"Gradual test step {throttle_step} op/s",
                                                       cycle_name=throttle_step))(self.run_step))
-            results, _ = run_step(stress_cmds=workload.cs_cmd_tmpl, current_throttle=current_throttle,
-                                  num_threads=workload.num_threads, step_duration=workload.step_duration)
+            run_step(stress_cmds=workload.cs_cmd_tmpl, current_throttle=current_throttle,
+                     num_threads=workload.num_threads, step_duration=workload.step_duration)
 
-            calculate_result = self._calculate_average_max_latency(results)
-            self.update_test_details(scylla_conf=True)
             summary_result = self.check_latency_during_steps(step=throttle_step)
-            summary_result[throttle_step].update({"ops_rate": calculate_result["op rate"] * num_loaders})
             total_summary.update(summary_result)
             if workload.drop_keyspace:
                 self.drop_keyspace()
@@ -242,7 +223,6 @@ def run_gradual_increase_load(self, workload: Workload, stress_num, num_loaders,
                 time.sleep(180 - wait_time)
 
         self.save_total_summary_in_file(total_summary)
-        self.run_performance_analyzer(total_summary=total_summary)
 
     def save_total_summary_in_file(self, total_summary):
         total_summary_json = json.dumps(total_summary, indent=4, separators=(", ", ": "))
@@ -254,54 +234,10 @@ def save_total_summary_in_file(self, total_summary):
         with open(filename, "w", encoding="utf-8") as res_file:
             res_file.write(total_summary_json)
 
-    def run_performance_analyzer(self, total_summary):
-        perf_analyzer = PredefinedStepsTestPerformanceAnalyzer(
-            es_index=self._test_index,
-            es_doc_type=self._es_doc_type,
-            email_recipients=self.params.get('email_recipients'))
-        # Keep next 2 lines for debug purpose
-        self.log.debug("es_index: %s", self._test_index)
-        self.log.debug("total_summary: %s", total_summary)
-        is_gce = bool(self.params.get('cluster_backend') == 'gce')
-        try:
-            perf_analyzer.check_regression(test_id=self.test_id,
-                                           data=total_summary,
-                                           is_gce=is_gce,
-                                           email_subject_postfix=self.params.get('email_subject_postfix'))
-        except Exception as exc:  # noqa: BLE001
-            TestFrameworkEvent(
-                message='Failed to check regression',
-                source=self.__class__.__name__,
-                source_method='check_regression',
-                exception=exc
-            ).publish_or_dump()
-
-    @staticmethod
-    def _calculate_average_max_latency(results):
-        status = defaultdict(float).fromkeys(results[0].keys(), 0.0)
-        max_latency = defaultdict(list)
-
-        for result in results:
-            for key in status:
-                try:
-                    status[key] += float(result.get(key, 0.0))
-                    if key in ["latency 95th percentile", "latency 99th percentile"]:
-                        max_latency[f"{key} max"].append(float(result.get(key, 0.0)))
-                except ValueError:
-                    continue
-
-        for key in status:
-            status[key] = round(status[key] / len(results), 2)
-
-        for key, latency in max_latency.items():
-            status[key] = max(latency)
-
-        return status
-
     def warmup_cache(self, stress_cmd_templ, num_threads):
         stress_queue = []
         for stress_cmd in stress_cmd_templ:
-            params = {"round_robin": True, "stats_aggregate_cmds": False}
+            params = {"round_robin": True}
             stress_cmd_to_run = stress_cmd.replace("$threads", str(num_threads))
             params.update({'stress_cmd': stress_cmd_to_run})
             # Run all stress commands