Fix load test checks (#9885)

Author: iddqdex

ydb/tests/olap/lib/allure_utils.py

@@ -9,9 +9,11 @@
 def allure_test_description(
     suite: str,
     test: str,
+    start_time: float,
+    end_time: float,
     addition_table_strings: dict[str, any] = {},
     attachments: tuple[str, str, allure.attachment_type] = [],
-    refference_set: str = ''
+    refference_set: str = '',
 ):
     def _pretty_str(s):
         return ' '.join(s.split('_')).capitalize()
@@ -33,7 +35,7 @@ def _pretty_str(s):
             'table_path': YdbCluster.tables_path,
             'monitoring': (
                 f"<a target='_blank' href='https://monitoring.yandex-team.ru/projects/kikimr/dashboards/mone0310v4dbc6kui89v?"
-                f"p.cluster={monitoring_cluster}&p.database=/{test_info['database']}'>link</a>"
+                f"p.cluster={monitoring_cluster}&p.database=/{test_info['database']}&from={int(start_time * 1000)}&to={int(end_time * 1000)}'>link</a>"
             ),
             'coredumps': f"<a target='_blank' href='{core_link}'>link</a>",
             'db_admin': (

ydb/tests/olap/lib/ydb_cli.py

@@ -6,6 +6,8 @@
 from ydb.tests.olap.lib.ydb_cluster import YdbCluster
 from ydb.tests.olap.lib.utils import get_external_param
 from enum import StrEnum
+from time import time
+from types import TracebackType
 
 
 class WorkloadType(StrEnum):
@@ -37,7 +39,7 @@ class WorkloadRunResult:
         def __init__(
             self, stats: dict[str, dict[str, any]] = {}, query_out: Optional[str] = None, stdout: Optional[str] = None, stderr: Optional[str] = None,
             error_message: Optional[str] = None, plans: Optional[list[YdbCliHelper.QueryPlan]] = None,
-            errors_by_iter: Optional[dict[int, str]] = None, explain_plan: Optional[YdbCliHelper.QueryPlan] = None
+            errors_by_iter: Optional[dict[int, str]] = None, explain_plan: Optional[YdbCliHelper.QueryPlan] = None, traceback: Optional[TracebackType] = None
         ) -> None:
             self.stats = stats
             self.query_out = query_out if str != '' else None
@@ -48,6 +50,7 @@ def __init__(
             self.plans = plans
             self.explain_plan = explain_plan
             self.errors_by_iter = errors_by_iter
+            self.traceback = traceback
 
     @staticmethod
     def workload_run(type: WorkloadType, path: str, query_num: int, iterations: int = 5,
@@ -95,13 +98,22 @@ def _load_plans(plan_path: str, name: str) -> YdbCliHelper.QueryPlan:
                     result.svg = f.read()
             return result
 
+        def _get_nodes_info() -> dict[str, dict[str, int]]:
+            nodes, _ = YdbCluster.get_cluster_nodes()
+            return {
+                n['SystemState']['Host']: {
+                    'start_time': int(n['SystemState'].get('StartTime', int(time()) * 1000) / 1000)
+                }
+                for n in nodes
+            }
+
         errors_by_iter = {}
         try:
             wait_error = YdbCluster.wait_ydb_alive(300, path)
             if wait_error is not None:
                 return YdbCliHelper.WorkloadRunResult(error_message=f'Ydb cluster is dead: {wait_error}')
 
-            cluster_start_time = YdbCluster.get_cluster_info().get('max_start_time', 0)
+            nodes_info = _get_nodes_info()
 
             json_path = yatest.common.work_path(f'q{query_num}.json')
             qout_path = yatest.common.work_path(f'q{query_num}.out')
@@ -132,8 +144,23 @@ def _load_plans(plan_path: str, name: str) -> YdbCliHelper.QueryPlan:
                         err = f'Invalid return code: {exec.returncode} instesd 0.'
             except (yatest.common.process.TimeoutError, yatest.common.process.ExecutionTimeoutError):
                 err = f'Timeout {timeout}s expeared.'
-            if YdbCluster.get_cluster_info().get('max_start_time', 0) != cluster_start_time:
-                err = ('' if err is None else f'{err}\n\n') + 'Some nodes were restart'
+
+            node_errors = []
+            for node, info in _get_nodes_info().items():
+                if node in nodes_info:
+                    if info['start_time'] > nodes_info[node]['start_time']:
+                        node_errors.append(f'Node {node} was restarted')
+                    nodes_info[node]['processed'] = True
+            for node, info in nodes_info.items():
+                if not info.get('processed', False):
+                    node_errors.append(f'Node {node} is down')
+            if len(node_errors) > 0:
+                if err is None:
+                    err = ''
+                else:
+                    err += '\n\n'
+                err += '\n'.join(node_errors)
+
             stats = {}
             if (os.path.exists(json_path)):
                 with open(json_path, 'r') as r:
@@ -162,4 +189,4 @@ def _load_plans(plan_path: str, name: str) -> YdbCliHelper.QueryPlan:
                 errors_by_iter=errors_by_iter
             )
         except BaseException as e:
-            return YdbCliHelper.WorkloadRunResult(error_message=str(e))
+            return YdbCliHelper.WorkloadRunResult(error_message=str(e), traceback=e.__traceback__)

ydb/tests/olap/lib/ydb_cluster.py

@@ -29,7 +29,7 @@ def _get_service_url(cls):
         return f'http://{host}:{port}'
 
     @classmethod
-    def _get_cluster_nodes(cls, path=None):
+    def get_cluster_nodes(cls, path=None):
         try:
             url = f'{cls._get_service_url()}/viewer/json/nodes?'
             if path is not None:
@@ -52,11 +52,9 @@ def get_cluster_info(cls):
             version = ''
             cluster_name = ''
             nodes_wilcard = ''
-            max_start_time = 0
-            nodes, node_count = cls._get_cluster_nodes()
+            nodes, node_count = cls.get_cluster_nodes()
             for node in nodes:
                 n = node.get('SystemState', {})
-                max_start_time = max(max_start_time, int(n.get('StartTime', 0)))
                 cluster_name = n.get('ClusterName', cluster_name)
                 version = n.get('Version', version)
                 for tenant in n.get('Tenants', []):
@@ -69,7 +67,6 @@ def get_cluster_info(cls):
                 'name': cluster_name,
                 'nodes_wilcard': nodes_wilcard,
                 'service_url': cls._get_service_url(),
-                'max_start_time': max_start_time,
             }
         return deepcopy(cls._cluster_info)
 
@@ -184,7 +181,7 @@ def _check_node(n):
 
         errors = []
         try:
-            nodes, node_count = cls._get_cluster_nodes()
+            nodes, node_count = cls.get_cluster_nodes()
             if node_count == 0:
                 errors.append('nodes_count == 0')
             if len(nodes) < node_count:
@@ -216,7 +213,7 @@ def _check_node(n):
                 for path in balanced_paths:
                     paths_to_balance += cls._get_tables(path)
             for p in paths_to_balance:
-                table_nodes, _ = cls._get_cluster_nodes(p)
+                table_nodes, _ = cls.get_cluster_nodes(p)
                 min = None
                 max = None
                 for tn in table_nodes:

ydb/tests/olap/load/conftest.py

@@ -40,16 +40,16 @@ def _attach_plans(plan: YdbCliHelper.QueryPlan) -> None:
                 allure.attach(plan.svg, 'Plan svg', attachment_type=allure.attachment_type.SVG)
 
         test = f'Query{query_num:02d}'
-        allure_test_description(self.suite, test, refference_set=self.refference)
         allure_listener = next(filter(lambda x: isinstance(x, AllureListener), plugin_manager.get_plugin_manager().get_plugins()))
         allure_test_result = allure_listener.allure_logger.get_test(None)
         query_num_param = next(filter(lambda x: x.name == 'query_num', allure_test_result.parameters), None)
         if query_num_param:
             query_num_param.mode = allure.parameter_mode.HIDDEN.value
-
+        start_time = time()
         result = YdbCliHelper.workload_run(
             path=path, query_num=query_num, iterations=self.iterations, type=self.workload_type, timeout=self.timeout
         )
+        allure_test_description(self.suite, test, refference_set=self.refference, start_time=start_time, end_time=time())
         stats = result.stats.get(test)
         if stats is not None:
             allure.attach(json.dumps(stats, indent=2), 'Stats', attachment_type=allure.attachment_type.JSON)
@@ -103,7 +103,7 @@ def _attach_plans(plan: YdbCliHelper.QueryPlan) -> None:
             kind='Load',
             suite=self.suite,
             test=test,
-            timestamp=time(),
+            timestamp=start_time,
             is_successful=success,
             min_duration=_get_duraton(stats, 'Min'),
             max_duration=_get_duraton(stats, 'Max'),
@@ -112,4 +112,7 @@ def _attach_plans(plan: YdbCliHelper.QueryPlan) -> None:
             statistics=stats,
         )
         if not success:
-            pytest.fail(error_message)
+            exc = pytest.fail.Exception(error_message)
+            if result.traceback is not None:
+                exc = exc.with_traceback(result.traceback)
+            raise exc

ydb/tests/olap/scenario/conftest.py

@@ -28,7 +28,6 @@ def teardown_class(cls):
             ScenarioTestHelper(None).remove_path(cls.get_suite_name())
 
     def test(self, ctx: TestContext):
-        allure_test_description(ctx.suite, ctx.test)
         start_time = time.time()
         try:
             ctx.executable(self, ctx)
@@ -41,6 +40,7 @@ def test(self, ctx: TestContext):
                 is_successful=True,
             )
         except pytest.skip.Exception:
+            allure_test_description(ctx.suite, ctx.test, start_time=start_time, end_time=time.time())
             raise
         except BaseException:
             ResultsProcessor.upload_results(
@@ -51,7 +51,9 @@ def test(self, ctx: TestContext):
                 duration=time.time() - start_time,
                 is_successful=False,
             )
+            allure_test_description(ctx.suite, ctx.test, start_time=start_time, end_time=time.time())
             raise
+        allure_test_description(ctx.suite, ctx.test, start_time=start_time, end_time=time.time())
 
 
 def pytest_generate_tests(metafunc):