Check node errors in perftests, support multislot (#17474)

4 files changed, 145 insertions, 60 deletions

diff --git a/ydb/tests/olap/lib/allure_utils.py b/ydb/tests/olap/lib/allure_utils.py
index 91222421f64..f58f5df08f7 100644
--- a/ydb/tests/olap/lib/allure_utils.py
+++ b/ydb/tests/olap/lib/allure_utils.py
@@ -8,6 +8,14 @@ from copy import deepcopy
 from pytz import timezone
 
 
+class NodeErrors:
+    def __init__(self, node: YdbCluster.Node, message: str):
+        self.node = node
+        self.core_hashes: list[tuple[str, str]] = []    # id, aggregated hash
+        self.was_oom: bool = False
+        self.message: str = message
+
+
 def _set_monitoring(test_info: dict[str, str], start_time: float, end_time: float) -> None:
     monitoring_start = int((start_time) * 1000)
     monitoring_end = int((end_time) * 1000)
@@ -29,12 +37,32 @@ def _set_monitoring(test_info: dict[str, str], start_time: float, end_time: floa
         ])
 
 
-def _set_coredumps(test_info: dict[str, str]) -> None:
-    if test_info['name'].startswith('ydb-k8s'):
-        core_link = f"https://coredumps.n.yandex-team.ru/index?itype=kikimr&host_list={test_info['nodes_wilcard']}&show_fixed=True"
-    else:
-        core_link = f"https://kikimr-cores.n.yandex-team.ru/show?server={test_info['nodes_wilcard']}%2A"
-    test_info['coredumps'] = f"<a target='_blank' href='{core_link}'>link</a>"
+def _set_coredumps(test_info: dict[str, str], start_time: float, end_time: float) -> None:
+    tz = timezone('Europe/Moscow')
+    params = urlencode([
+        ('filter', f'program_type=kikimr; @cluster_name={test_info["name"]}'),
+        ('since_ts', datetime.fromtimestamp(start_time, tz).isoformat()),
+        ('till_ts', datetime.fromtimestamp(end_time, tz).isoformat()),
+    ])
+    test_info['coredumps'] = f"<a target='_blank' href='https://coredumps.yandex-team.ru/v3/cores?{params}'>link</a>"
+
+
+def _set_node_errors(test_info: dict[str, str], node_errors: list[NodeErrors]) -> None:
+    if len(node_errors) == 0:
+        return
+    html = '<ul>'
+    for node in node_errors:
+        html += f'<li>{node.node.ic_port}@{node.node.host}'
+        if node.message:
+            html += f'<p>Node {node.message}</p>'
+        if node.was_oom:
+            html += '<p>Node was killed by OOM</p>'
+        for core_id, core_hash in node.core_hashes:
+            color = hex(0xFF0000 + hash(core_hash) % 0xFFFF).split('x')[-1]
+            html += f'<p>There was coredump <a target="_blank" href="https://coredumps.yandex-team.ru/v3/cores/{core_id}" style="background-color: #{color}">{core_hash}</a></p>'
+        html += '</li>'
+    html += '</ul>'
+    test_info['<span style="background-color: #FF8888">node errors</span>'] = html
 
 
 def _set_results_plot(test_info: dict[str, str], suite: str, test: str, refference_set: str) -> None:
@@ -81,6 +109,7 @@ def allure_test_description(
     addition_table_strings: dict[str, any] = {},
     attachments: tuple[str, str, allure.attachment_type] = [],
     refference_set: str = '',
+    node_errors: list[NodeErrors] = [],
 ):
     def _pretty_str(s):
         return ' '.join(s.split('_')).capitalize()
@@ -93,7 +122,8 @@ def allure_test_description(
     test_info.update(addition_table_strings)
 
     _set_monitoring(test_info, start_time, end_time)
-    _set_coredumps(test_info)
+    _set_coredumps(test_info, start_time, end_time)
+    _set_node_errors(test_info, node_errors)
     _set_results_plot(test_info, suite, test, refference_set)
     _set_logs_command(test_info, start_time, end_time)
 
@@ -110,7 +140,6 @@ def allure_test_description(
             'time': f"{datetime.fromtimestamp(start_time).strftime('%a %d %b %y %H:%M:%S')} - {datetime.fromtimestamp(end_time).strftime('%H:%M:%S')}",
         }
     )
-    del test_info['nodes_wilcard']
     table_strings = '\n'.join([f'<tr><td>{_pretty_str(k)}</td><td>{v}</td></tr>' for k, v in test_info.items()])
     allure.dynamic.description_html(
         f'''<table border='1' cellpadding='4px'><tbody>
diff --git a/ydb/tests/olap/lib/ydb_cli.py b/ydb/tests/olap/lib/ydb_cli.py
index 547fe7788a8..c4823f4cbfb 100644
--- a/ydb/tests/olap/lib/ydb_cli.py
+++ b/ydb/tests/olap/lib/ydb_cli.py
@@ -148,7 +148,6 @@ class YdbCliHelper:
             self.query_syntax = query_syntax
             self.scale = scale
             self.query_prefix = query_prefix
-            self._nodes_info: dict[str, dict[str, int]] = {}
             self._plan_path = _get_output_path('plan')
             self._query_output_path = _get_output_path('out')
             self._json_path = _get_output_path('json')
@@ -231,27 +230,6 @@ class YdbCliHelper:
                 with open(self._query_output_path, 'r') as r:
                     self.result.query_out = r.read()
 
-        @staticmethod
-        def _get_nodes_info() -> dict[str, dict[str, Any]]:
-            return {
-                n.host: {
-                    'start_time': n.start_time
-                }
-                for n in YdbCluster.get_cluster_nodes(db_only=True)
-            }
-
-        def _check_nodes(self):
-            node_errors = []
-            for node, info in self._get_nodes_info().items():
-                if node in self._nodes_info:
-                    if info['start_time'] > self._nodes_info[node]['start_time']:
-                        node_errors.append(f'Node {node} was restarted')
-                    self._nodes_info[node]['processed'] = True
-            for node, info in self._nodes_info.items():
-                if not info.get('processed', False):
-                    node_errors.append(f'Node {node} is down')
-            self.result.add_error('\n'.join(node_errors))
-
         def _parse_stdout(self, stdout: str) -> None:
             self.result.stdout = stdout
             for line in self.result.stdout.splitlines():
@@ -289,11 +267,9 @@ class YdbCliHelper:
                 if wait_error is not None:
                     self.result.error_message = wait_error
                 else:
-                    self._nodes_info = self._get_nodes_info()
                     process = yatest.common.process.execute(self._get_cmd(), check_exit_code=False)
                     self._parse_stderr(process.stderr.decode('utf-8', 'replace'))
                     self._parse_stdout(process.stdout.decode('utf-8', 'replace'))
-                    self._check_nodes()
                     self._load_stats()
                     self._load_query_out()
                     self._load_plans()
diff --git a/ydb/tests/olap/lib/ydb_cluster.py b/ydb/tests/olap/lib/ydb_cluster.py
index 573a090d368..86b71649221 100644
--- a/ydb/tests/olap/lib/ydb_cluster.py
+++ b/ydb/tests/olap/lib/ydb_cluster.py
@@ -37,8 +37,9 @@ class YdbCluster:
         def __init__(self, desc: dict):
             ss = desc.get('SystemState', {})
             self.host: str = ss.get('Host', '')
+            ports = {e.get('Name', ''): int(e.get('Address', '0').split(':')[-1]) for e in ss.get('Endpoints', [])}
+            self.ic_port: int = ports.get('ic', 0)
             self.disconnected: bool = desc.get('Disconnected', False)
-            self.cluster_name: str = ss.get('ClusterName', '')
             self.version: str = ss.get('Version', '')
             self.start_time: float = 0.001 * int(ss.get('StartTime', time() * 1000))
             if 'Storage' in ss.get('Roles', []):
@@ -110,21 +111,14 @@ class YdbCluster:
     def get_cluster_info(cls):
         if cls._cluster_info is None:
             version = ''
-            cluster_name = ''
-            nodes_wilcard = ''
             nodes = cls.get_cluster_nodes(db_only=True)
             for node in nodes:
-                if not cluster_name:
-                    cluster_name = node.cluster_name
                 if not version:
                     version = node.version
-                if not nodes_wilcard and node.role == YdbCluster.Node.Role.COMPUTE:
-                    nodes_wilcard = node.host.split('.')[0].rstrip('0123456789')
             cls._cluster_info = {
                 'database': cls.ydb_database,
                 'version': version,
-                'name': cluster_name,
-                'nodes_wilcard': nodes_wilcard,
+                'name': cls.ydb_database.strip('/').split('/')[0],
             }
         return deepcopy(cls._cluster_info)
 
diff --git a/ydb/tests/olap/load/lib/conftest.py b/ydb/tests/olap/load/lib/conftest.py
index 6c8f29b048c..2c958dc0fc0 100644
--- a/ydb/tests/olap/load/lib/conftest.py
+++ b/ydb/tests/olap/load/lib/conftest.py
@@ -14,7 +14,7 @@ from time import time
 from typing import Optional
 from ydb.tests.olap.lib.ydb_cli import YdbCliHelper, WorkloadType, CheckCanonicalPolicy
 from ydb.tests.olap.lib.ydb_cluster import YdbCluster
-from ydb.tests.olap.lib.allure_utils import allure_test_description
+from ydb.tests.olap.lib.allure_utils import allure_test_description, NodeErrors
 from ydb.tests.olap.lib.results_processor import ResultsProcessor
 from ydb.tests.olap.lib.utils import get_external_param
 from ydb.tests.olap.scenario.helpers.scenario_tests_helper import ScenarioTestHelper
@@ -37,6 +37,7 @@ class LoadSuiteBase:
     scale: Optional[int] = None
     query_prefix: str = get_external_param('query-prefix', '')
     verify_data: bool = True
+    __nodes_state: Optional[dict[tuple[str, int], YdbCluster.Node]] = None
 
     @classmethod
     def suite(cls) -> str:
@@ -69,7 +70,7 @@ class LoadSuiteBase:
     @allure.step('check tables size')
     def check_tables_size(cls, folder: Optional[str], tables: dict[str, int]):
         wait_error = YdbCluster.wait_ydb_alive(
-            20 * 60, (
+            int(os.getenv('WAIT_CLUSTER_ALIVE_TIMEOUT', 20 * 60)), (
                 f'{YdbCluster.tables_path}/{folder}'
                 if folder is not None
                 else [f'{YdbCluster.tables_path}/{t}' for t in tables.keys()]
@@ -92,8 +93,19 @@ class LoadSuiteBase:
             msg = "\n".join(errors)
             pytest.fail(f'Unexpected tables size in `{folder}`:\n {msg}')
 
+    @staticmethod
+    def __execute_ssh(host: str, cmd: str):
+        ssh_cmd = ['ssh', "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null"]
+        ssh_user = os.getenv('SSH_USER')
+        if ssh_user is not None:
+            ssh_cmd += ['-l', ssh_user]
+        ssh_key_file = os.getenv('SSH_KEY_FILE')
+        if ssh_key_file is not None:
+            ssh_cmd += ['-i', ssh_key_file]
+        return yatest.common.execute(ssh_cmd + [host, cmd], wait=False)
+
     @classmethod
-    def _attach_logs(cls, start_time, attach_name):
+    def __attach_logs(cls, start_time, attach_name):
         hosts = [node.host for node in filter(lambda x: x.role == YdbCluster.Node.Role.STORAGE, YdbCluster.get_cluster_nodes())]
         tz = timezone('Europe/Moscow')
         start = datetime.fromtimestamp(start_time, tz).isoformat()
@@ -102,28 +114,20 @@ class LoadSuiteBase:
             '': {},
         }
         exec_start = deepcopy(exec_kikimr)
-        ssh_cmd = ['ssh', "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null"]
-        ssh_user = os.getenv('SSH_USER')
-        if ssh_user is not None:
-            ssh_cmd += ['-l', ssh_user]
-        ssh_key_file = os.getenv('SSH_KEY_FILE')
-        if ssh_key_file is not None:
-            ssh_cmd += ['-i', ssh_key_file]
         for host in hosts:
             for c in exec_kikimr.keys():
                 try:
-                    exec_kikimr[c][host] = yatest.common.execute(ssh_cmd + [host, cmd.format(
+                    exec_kikimr[c][host] = cls.__execute_ssh(host, cmd.format(
                         storage='kikimr',
-                        container=f' -m k8s_container:{c}' if c else '')],
-                        wait=False)
+                        container=f' -m k8s_container:{c}' if c else ''
+                    ))
                 except BaseException as e:
                     logging.error(e)
             for c in exec_start.keys():
                 try:
-                    exec_start[c][host] = yatest.common.execute(ssh_cmd + [host, cmd.format(
+                    exec_start[c][host] = cls.__execute_ssh(host, cmd.format(
                         storage='kikimr-start',
-                        container=f' -m k8s_container:{c}' if c else '')],
-                        wait=False)
+                        container=f' -m k8s_container:{c}' if c else ''))
                 except BaseException as e:
                     logging.error(e)
 
@@ -147,6 +151,83 @@ class LoadSuiteBase:
             allure.attach.file(archive, f'{attach_name}_{c}_logs', extension='tar.gz')
 
     @classmethod
+    def save_nodes_state(cls) -> None:
+        cls.__nodes_state = {(n.host, n.ic_port): n for n in YdbCluster.get_cluster_nodes(db_only=True)}
+
+    @classmethod
+    def __get_core_hashes_by_pod(cls, hosts: set[str], start_time: float, end_time: float) -> dict[str, list[tuple[str, str]]]:
+        core_processes = {
+            h: cls.__execute_ssh(h, 'sudo flock /tmp/brk_pad /Berkanavt/breakpad/bin/kikimr_breakpad_analizer.sh')
+            for h in hosts
+        }
+
+        core_hashes = {}
+        for h, exec in core_processes.items():
+            exec.wait(check_exit_code=False)
+            if exec.returncode != 0:
+                logging.error(f'Error while process coredumps on host {h}: {exec.stderr.decode("utf-8")}')
+            exec = cls.__execute_ssh(h, ('find /coredumps/ -name "sended_*.json" '
+                                         f'-mmin -{(10 + time() - start_time) / 60} -mmin +{(-10 + time() - end_time) / 60}'
+                                         ' | while read FILE; do cat $FILE; echo -n ","; done'))
+            exec.wait(check_exit_code=False)
+            if exec.returncode == 0:
+                for core in json.loads(f'[{exec.stdout.decode("utf-8").strip(",")}]'):
+                    pod_name = core.get('pod', '')
+                    core_hashes.setdefault(pod_name, [])
+                    core_hashes[pod_name].append((core.get('core_uuid', ''), core.get('core_hash', '')))
+            else:
+                logging.error(f'Error while search coredumps on host {h}: {exec.stderr.decode("utf-8")}')
+        return core_hashes
+
+    @classmethod
+    def __get_hosts_with_omms(cls, hosts: set[str], start_time: float, end_time: float) -> set[str]:
+        tz = timezone('Europe/Moscow')
+        start = datetime.fromtimestamp(start_time, tz).strftime("%Y-%m-%d %H:%M:%S")
+        end = datetime.fromtimestamp(end_time, tz).strftime("%Y-%m-%d %H:%M:%S")
+        oom_cmd = f'sudo journalctl -k -q --no-pager -S {start} -U {end} --grep "Out of memory: Kill process"'
+        ooms = set()
+        for h in hosts:
+            exec = cls.__execute_ssh(h, oom_cmd)
+            exec.wait(check_exit_code=False)
+            if exec.returncode == 0:
+                if exec.stdout.decode('utf-8'):
+                    ooms.add(h)
+            else:
+                logging.error(f'Error while search OOMs on host {h}: {exec.stderr.decode("utf-8")}')
+        return ooms
+
+    @classmethod
+    def check_nodes(cls, result: YdbCliHelper.WorkloadRunResult, end_time: float) -> list[NodeErrors]:
+        if cls.__nodes_state is None:
+            return []
+        node_errors = []
+        fail_hosts = set()
+        for node in YdbCluster.get_cluster_nodes(db_only=True):
+            node_id = (node.host, node.ic_port)
+            saved_node = cls.__nodes_state.get(node_id)
+            if saved_node is not None:
+                if node.start_time > saved_node.start_time:
+                    node_errors.append(NodeErrors(node, 'was restarted'))
+                    fail_hosts.add(node.host)
+                del cls.__nodes_state[node_id]
+        for _, node in cls.__nodes_state.items():
+            node_errors.append(NodeErrors(node, 'is down'))
+            fail_hosts.add(node.host)
+        cls.__nodes_state = None
+        if len(node_errors) == 0:
+            return []
+
+        core_hashes = cls.__get_core_hashes_by_pod(fail_hosts, result.start_time, end_time)
+        ooms = cls.__get_hosts_with_omms(fail_hosts, result.start_time, end_time)
+        for node in node_errors:
+            node.core_hashes = core_hashes.get(f'{node.node.ic_port}@{node.node.host}', [])
+            node.was_oom = node.node.host in ooms
+
+        for err in node_errors:
+            result.add_error(f'Node {err.node.ic_port}@{err.node.host} {err.message}')
+        return node_errors
+
+    @classmethod
     def process_query_result(cls, result: YdbCliHelper.WorkloadRunResult, query_num: int, iterations: int, upload: bool):
         def _get_duraton(stats, field):
             r = stats.get(field)
@@ -210,15 +291,20 @@ class LoadSuiteBase:
         for p in ['Mean']:
             if p in stats:
                 allure.dynamic.parameter(p, _duration_text(stats[p] / 1000.))
+        end_time = time()
         if os.getenv('NO_KUBER_LOGS') is None and not result.success:
-            cls._attach_logs(start_time=result.start_time, attach_name='kikimr')
+            cls.__attach_logs(start_time=result.start_time, attach_name='kikimr')
         allure.attach(json.dumps(stats, indent=2), 'Stats', attachment_type=allure.attachment_type.JSON)
+        allure_test_description(
+            cls.suite(), test, refference_set=cls.refference,
+            start_time=result.start_time, end_time=end_time, node_errors=cls.check_nodes(result, end_time)
+        )
         if upload:
             ResultsProcessor.upload_results(
                 kind='Load',
                 suite=cls.suite(),
                 test=test,
-                timestamp=time(),
+                timestamp=end_time,
                 is_successful=result.success,
                 min_duration=_get_duraton(stats, 'Min'),
                 max_duration=_get_duraton(stats, 'Max'),
@@ -263,6 +349,7 @@ class LoadSuiteBase:
                         if param.name == 'query_num':
                             param.mode = allure.parameter_mode.HIDDEN.value
         qparams = self._get_query_settings(query_num)
+        self.save_nodes_state()
         result = YdbCliHelper.workload_run(
             path=path,
             query_num=query_num,
@@ -274,5 +361,4 @@ class LoadSuiteBase:
             scale=self.scale,
             query_prefix=qparams.query_prefix
         )
-        allure_test_description(self.suite(), self._test_name(query_num), refference_set=self.refference, start_time=result.start_time, end_time=time())
         self.process_query_result(result, query_num, qparams.iterations, True)