新增test_connection方法，用于检测实例连接 (#1670)

* 新增test_connection方法，用于检测实例连接，fix #1451

移除Inception检测

* 移除test_inception_check

* 处理Keyspace信息为空的场景

common/check.py

@@ -14,29 +14,6 @@
 logger = logging.getLogger('default')
 
 
-# 检测inception配置
-@superuser_required
-def inception(request):
-    result = {'status': 0, 'msg': 'ok', 'data': []}
-    inception_host = request.POST.get('inception_host', '')
-    inception_port = request.POST.get('inception_port', '')
-    try:
-        conn = MySQLdb.connect(host=inception_host, port=int(inception_port), charset='utf8mb4',
-                               connect_timeout=5)
-        cur = conn.cursor()
-    except Exception as e:
-        logger.error(traceback.format_exc())
-        result['status'] = 1
-        result['msg'] = '无法连接Inception\n{}'.format(str(e))
-        return HttpResponse(json.dumps(result), content_type='application/json')
-    else:
-        cur.close()
-        conn.close()
-
-    # 返回结果
-    return HttpResponse(json.dumps(result), content_type='application/json')
-
-
 # 检测inception配置
 @superuser_required
 def go_inception(request):
@@ -128,11 +105,10 @@ def instance(request):
     instance = Instance.objects.get(id=instance_id)
     try:
         engine = get_engine(instance=instance)
-        engine.get_connection()
-        dbs = engine.get_all_databases()
-        if dbs.error:
+        test_result = engine.test_connection()
+        if test_result.error:
             result['status'] = 1
-            result['msg'] = '无法连接实例,\n{}'.format(dbs.error)
+            result['msg'] = '无法连接实例,\n{}'.format(test_result.error)
     except Exception as e:
         result['status'] = 1
         result['msg'] = '无法连接实例,\n{}'.format(str(e))

common/tests.py

@@ -324,22 +324,6 @@ def testInstanceCheck(self, _get_engine, _conn):
         r_json = r.json()
         self.assertEqual(r_json['status'], 1)
 
-    @patch('MySQLdb.connect')
-    def test_inception_check(self, _conn):
-        c = Client()
-        c.force_login(self.superuser1)
-        data = {
-            "inception_host": "inception",
-            "inception_port": "6669",
-            "inception_remote_backup_host": "mysql",
-            "inception_remote_backup_port": 3306,
-            "inception_remote_backup_user": "mysql",
-            "inception_remote_backup_password": "123456"
-        }
-        r = c.post('/check/inception/', data=data)
-        r_json = r.json()
-        self.assertEqual(r_json['status'], 0)
-
     @patch('MySQLdb.connect')
     def test_go_inception_check(self, _conn):
         c = Client()

sql/engines/__init__.py

@@ -1,10 +1,11 @@
 """engine base库, 包含一个``EngineBase`` class和一个get_engine函数"""
-from sql.engines.models import ResultSet
+from sql.engines.models import ResultSet, ReviewSet
 from sql.utils.ssh_tunnel import SSHConnection
 
 
 class EngineBase:
     """enginebase 只定义了init函数和若干方法的名字, 具体实现用mysql.py pg.py等实现"""
+    test_query = None
 
     def __init__(self, instance=None):
         self.conn = None
@@ -31,7 +32,7 @@ def __init__(self, instance=None):
                     instance.tunnel.pkey,
                     instance.tunnel.pkey_password,
                 )
-                self.host,self.port = self.ssh.get_ssh()
+                self.host, self.port = self.ssh.get_ssh()
 
     def __del__(self):
         if hasattr(self, 'ssh'):
@@ -65,6 +66,10 @@ def remote_instance_conn(self, instance=None):
     def get_connection(self, db_name=None):
         """返回一个conn实例"""
 
+    def test_connection(self):
+        """测试实例链接是否正常"""
+        return self.query(sql=self.test_query)
+
     @property
     def name(self):
         """返回engine名称"""
@@ -138,6 +143,7 @@ def filter_sql(self, sql='', limit_num=0):
 
     def query(self, db_name=None, sql='', limit_num=0, close_conn=True, **kwargs):
         """实际查询 返回一个ResultSet"""
+        return ResultSet()
 
     def query_masking(self, db_name=None, sql='', resultset=None):
         """传入 sql语句, db名, 结果集,
@@ -146,15 +152,18 @@ def query_masking(self, db_name=None, sql='', resultset=None):
 
     def execute_check(self, db_name=None, sql=''):
         """执行语句的检查 返回一个ReviewSet"""
+        return ReviewSet()
 
     def execute(self):
         """执行语句 返回一个ReviewSet"""
+        return ReviewSet()
 
     def get_execute_percentage(self):
         """获取执行进度"""
 
     def get_rollback(self, workflow):
         """获取工单回滚语句"""
+        return list()
 
     def get_variables(self, variables=None):
         """获取实例参数，返回一个 ResultSet"""

sql/engines/clickhouse.py

@@ -13,6 +13,7 @@
 
 
 class ClickHouseEngine(EngineBase):
+    test_query = "SELECT 1"
 
     def __init__(self, instance=None):
         super(ClickHouseEngine, self).__init__(instance=instance)

sql/engines/goinception.py

@@ -15,6 +15,8 @@
 
 
 class GoInceptionEngine(EngineBase):
+    test_query = "INCEPTION GET VARIABLES"
+
     @property
     def name(self):
         return 'GoInception'

sql/engines/mongo.py

@@ -253,6 +253,9 @@ class MongoEngine(EngineBase):
     warning = None
     methodStr = None
 
+    def test_connection(self):
+        return self.get_all_databases()
+
     def exec_cmd(self, sql, db_name=None, slave_ok=''):
         """审核时执行的语句"""
 

sql/engines/mssql.py

@@ -13,6 +13,8 @@
 
 
 class MssqlEngine(EngineBase):
+    test_query = "SELECT 1"
+
     def get_connection(self, db_name=None):
         connstr = """DRIVER=ODBC Driver 17 for SQL Server;SERVER={0},{1};UID={2};PWD={3};
 client charset = UTF-8;connect timeout=10;CHARSET={4};""".format(self.host, self.port, self.user, self.password,

sql/engines/mysql.py

@@ -20,6 +20,7 @@
 
 
 class MysqlEngine(EngineBase):
+    test_query = "SELECT 1"
 
     def __init__(self, instance=None):
         super().__init__(instance=instance)

sql/engines/odps.py

@@ -14,6 +14,7 @@
 
 
 class ODPSEngine(EngineBase):
+    test_query = "SELECT 1"
 
     def get_connection(self, db_name=None):
         if self.conn:

sql/engines/oracle.py

@@ -20,6 +20,7 @@
 
 
 class OracleEngine(EngineBase):
+    test_query = "SELECT 1 FROM DUAL"
 
     def __init__(self, instance=None):
         super(OracleEngine, self).__init__(instance=instance)

sql/engines/pgsql.py

@@ -24,6 +24,8 @@
 
 
 class PgSQLEngine(EngineBase):
+    test_query = "SELECT 1"
+
     def get_connection(self, db_name=None):
         db_name = db_name or self.db_name or 'postgres'
         if self.conn:

sql/engines/phoenix.py

@@ -12,6 +12,8 @@
 
 
 class PhoenixEngine(EngineBase):
+    test_query = "SELECT 1"
+
     def get_connection(self, db_name=None):
         if self.conn:
             return self.conn

sql/engines/redis.py

@@ -27,10 +27,11 @@ def get_connection(self, db_name=None):
         db_name = db_name or self.db_name
         if self.mode == 'cluster':
             return redis.cluster.RedisCluster(host=self.host, port=self.port, password=self.password,
-                                encoding_errors='ignore', decode_responses=True, socket_connect_timeout=10)
+                                              encoding_errors='ignore', decode_responses=True,
+                                              socket_connect_timeout=10)
         else:
             return redis.Redis(host=self.host, port=self.port, db=db_name, password=self.password,
-                         encoding_errors='ignore', decode_responses=True, socket_connect_timeout=10)
+                               encoding_errors='ignore', decode_responses=True, socket_connect_timeout=10)
 
     @property
     def name(self):
@@ -40,6 +41,9 @@ def name(self):
     def info(self):
         return 'Redis engine'
 
+    def test_connection(self):
+        return self.get_all_databases()
+
     def get_all_databases(self, **kwargs):
         """
         获取数据库列表
@@ -51,8 +55,8 @@ def get_all_databases(self, **kwargs):
             rows = conn.config_get('databases')['databases']
         except Exception as e:
             logger.warning(f"Redis CONFIG GET databases 执行报错，异常信息：{e}")
-            rows = 16
-            result.error = str(e)
+            dbs = [int(i.split('db')[1]) for i in conn.info('Keyspace').keys() if len(i.split('db')) == 2]
+            rows = max(dbs, [16])
 
         db_list = [str(x) for x in range(int(rows))]
         result.rows = db_list
@@ -64,7 +68,7 @@ def query_check(self, db_name=None, sql='', limit_num=0):
         safe_cmd = ["scan", "exists", "ttl", "pttl", "type", "get", "mget", "strlen",
                     "hgetall", "hexists", "hget", "hmget", "hkeys", "hvals",
                     "smembers", "scard", "sdiff", "sunion", "sismember", "llen", "lrange", "lindex",
-                    "zrange","zrangebyscore","zscore","zcard","zcount","zrank"]
+                    "zrange", "zrangebyscore", "zscore", "zcard", "zcount", "zrank"]
         # 命令校验，仅可以执行safe_cmd内的命令
         for cmd in safe_cmd:
             if re.match(fr'^{cmd}', sql.strip(), re.I):

sql/urls.py

@@ -78,7 +78,6 @@
     path('workflow/log/', workflow.log),
     path('config/change/', config.change_config),
 
-    path('check/inception/', check.inception),
     path('check/go_inception/', check.go_inception),
     path('check/email/', check.email),
     path('check/instance/', check.instance),