openGauss-OM/script/upgrade_checker/verifier.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
#############################################################################
# Copyright (c) 2023 Huawei Technologies Co.,Ltd.
#
# openGauss is licensed under Mulan PSL v2.
# You can use this software according to the terms
# and conditions of the Mulan PSL v2.
# You may obtain a copy of Mulan PSL v2 at:
#
#          http://license.coscl.org.cn/MulanPSL2
#
# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS,
# WITHOUT WARRANTIES OF ANY KIND,
# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
# See the Mulan PSL v2 for more details.
# ----------------------------------------------------------------------------
# Description  : gs_upgradechk is a utility to check meta data in gaussdb after upgrade.
#############################################################################
"""
校验组件模块，由三个模块进行配合，进行整体的流程。
    collector：负责数据收集。连接数据库，构造校验规则Rule，并执行Rule进行数据的收集。
    analyzer：负责数据分析。依据vmap，对输入的携带有数据的Rule进行分析，生成结论Conclusion
    exporter：负责数据简化导出。将Rule的结果进行简化，记录到一个新的vmap内，最终导出生成vmap。
"""


from upgrade_checker.utils.param import Action
from upgrade_checker.rules.rule_maker import RuleMaker
from upgrade_checker.rules.vmap import VMapHeader, VerifyMap
from upgrade_checker.opengauss import og
from upgrade_checker.log import logger


class Collector(object):
    """
    负责数据收集。连接数据库，构造校验规则Rule，并执行Rule进行数据的收集。
    """
    @staticmethod
    def prepare_db_list(param):
        # 如果参数指明了则使用指明的数据库
        if param.database.value is not None:
            logger.info(f'仅操作参数指明的数据库：{param.database.value}')
            search_db_sql = f"select datname from pg_database " \
                            f"where datname = '{param.database.value}'"

            qres = og.query(search_db_sql)
            if qres.row_count() == 0:
                logger.err(f"未检测到数据库 {param.database.value}")

            return [param.database.value]

        # 否则自动检测所有库，且仅支持A库
        if param.action == Action.VERIFY:
            search_db_sql = "select datname from pg_database " \
                            "where datname not in ('template0', 'template1') and " \
                            "      upper(datcompatibility) != 'A'"
            qres = og.query(search_db_sql)
            if qres.row_count() != 0:
                ignore_list = ','.join([line[0] for line in qres])
                logger.warning(f'暂不支持非A库的校验，{ignore_list} 将被跳过。')
            
            search_db_sql = "select datname from pg_database " \
                            "where datname not in ('template0', 'template1') and " \
                            "      upper(datcompatibility) = 'A'"
            qres = og.query(search_db_sql)
            if qres.row_count() == 0:
                logger.err('没有需要校验的库。')
                
            return [row[0] for row in qres]
        elif param.action == Action.EXPORT:
            search_db_sql = "select datname from pg_database " \
                            "where datname = 'postgres' and upper(datcompatibility) = 'A'"
            qres = og.query(search_db_sql)
            if qres.row_count() == 0:
                logger.err('暂不支持非A兼容性的postgres库。')
            
            logger.info('使用postgres库生成校验地图。')
            return ['postgres']
        else:
            assert False

    def __init__(self, db_list):
        """

        :param db_list: 需要收集的数据库名称列表
        """
        self._database_list = db_list
        self._idx = 0

        self._rule_buffer = []
        self._estimate = None

        logger.log('verifier.collector 数据库采集列表：[%s]。' % (', '.join(self._database_list)))
        logger.log('verifier.collector 初始化完成。')

    def __str__(self):
        return 'Collector: dblist [{0}]， current is {1}'.format(
            ', '.join(self._database_list),
            self._database_list[self._idx - 1]
        )

    def __iter__(self):
        return self

    def __next__(self):
        if len(self._rule_buffer) == 0 and self._collect_next_db() is None:
            raise StopIteration
        rule = self._rule_buffer.pop()
        rule.run()
        return rule

    def _estimate_workload(self):
        self._estimate = len(self._rule_buffer)

    def _collect_next_db(self):
        if self._idx >= len(self._database_list):
            return

        self._rule_buffer = RuleMaker.make_rules(self._database_list[self._idx])
        self._estimate_workload()
        logger.log('数据库%s的所有校验规则准备完毕，共计%d条，开始执行数据采集....' % (self._database_list[self._idx], len(self._rule_buffer)))

        self._idx += 1
        return self._rule_buffer

    def current_db(self):
        return self._database_list[self._idx - 1]

    def check_next_rule(self):
        if len(self._rule_buffer) == 0 and self._collect_next_db() is None:
            return

        rule = self._rule_buffer.pop()
        rule.run()
        return rule

    def get_progress(self):
        """
        返回当前采集的预估进度。
        """
        finished_db_percentage = (self._idx - 1) / len(self._database_list) * 100.0
        curr_db_percentage = (1.0 - len(self._rule_buffer) / self._estimate) / len(self._database_list) * 100.0

        percentage = int(finished_db_percentage + curr_db_percentage)
        if percentage < 0:
            percentage = 0
        elif percentage > 100:
            percentage = 100

        return percentage


class Analyzer(object):
    """
    负责数据分析。依据vmap，对输入的携带有数据的Rule进行分析，生成结论Conclusion
    """
    def __init__(self, vmap):
        self.vmap = VerifyMap(vmap)
        self.vmap.load()
        VMapHeader.check_availability(self.vmap.head)
        logger.log('verifier.analyzer 初始化完成。')

    def __del__(self):
        self.close()

    def __str__(self):
        return 'Analyzer: using vmap ({0})'.format(self.vmap.__str__())

    def analyze(self, rule):
        """
        在vmap内找到对应的规则预期输出，并进行分析处理。
        :param rule:
        :return:
        """
        expect = self.vmap.get(rule.sql)
        conclusion = rule.analyze(expect)
        logger.debug('分析完毕一条规则，共计{0}条小项，{1}告警，规则内容为：{2}。'.format(
            len(conclusion.details),
            len(conclusion.warnings),
            rule.sql)
        )
        return conclusion

    def close(self):
        pass


class Exporter(object):
    """
    负责数据简化导出。将Rule的结果进行简化，记录到一个新的vmap内，最终导出生成vmap。
    """
    def __init__(self, export_vmap_file):
        self.vmap = VerifyMap(export_vmap_file)
        logger.log('verifier.exporter 初始化完成。')

    def __str__(self):
        return 'Exporter: export to vmap ({0})'.format(self.vmap.show_info())

    def record(self, rule):
        self.vmap.push(rule)
        logger.debug('Exporter完成记录一条规则：%s。' % rule.sql)

    def export(self):
        self.vmap.dump()
        logger.debug('Exporter导出校验地图完成。')

    def close(self):
        self.vmap.dump()