doris支持会话管理功能

Author: weideguo

README.md

@@ -34,7 +34,7 @@
 | ODPS       | √ | ×  | ×  | × | × | × | × | × | × | × |
 | ClickHouse | √ | √  | √  | × | × | × | × | × | × | × |
 | Cassandra  | √ | ×  | √  | × | × | × | × | × | × | × |
-| Doris      | √ | ×  | √  | × | × | × | × | × | × | × |
+| Doris      | √ | ×  | √  | × | × | × | √ | × | × | × |
 
 
 

common/static/dbdiagnostic/js/db_info.js

@@ -103,6 +103,92 @@ const mysqlDiagnosticInfo = {
     ],
 }
 
+const dorisDiagnosticInfo = {
+    fieldsProcesslist: [
+        'doris',
+        ["All","Not Sleep","Query"],
+        [{
+            title: '',
+            field: 'checkbox',
+            checkbox: true
+        }, {
+            title: 'THEEAD ID',
+            field: 'id',
+            sortable: true
+        }, {
+            title: 'USER',
+            field: 'user',
+            sortable: true
+        }, {
+            title: 'HOST',
+            field: 'host',
+            sortable: true
+        }, {
+            title: 'CATALOG',
+            field: 'catalog',
+            sortable: true
+        }, {
+            title: 'DATABASE',
+            field: 'db',
+            sortable: true
+        }, {
+            title: 'TIME(s)',
+            field: 'time',
+            sortable: true
+        }, {
+            title: 'COMMAND',
+            field: 'command',
+            sortable: true
+        }, {
+            title: 'STATE',
+            field: 'state',
+            sortable: true
+        }, {
+            title: 'INFO',
+            field: 'info',
+            sortable: true,
+            formatter: function (value, row, index) {
+                if (value.length > 20) {
+                    var sql = value.substr(0, 20) + '...';
+                    return sql;
+                } else {
+                    return value
+                }
+            }
+        }, {
+            title: 'QUERYID',
+            field: 'query_id',
+            sortable: true,
+            visible: false // 默认不显示
+        }, {
+            title: '完整INFO',
+            field: 'info',
+            sortable: true,
+            visible: false // 默认不显示
+        }, {
+            title: 'FE',
+            field: 'fe',
+            sortable: true,
+            visible: false // 默认不显示
+        }],
+        function (index, row) {
+            var html = [];
+            $.each(row, function (key, value) {
+                if (key === 'info') {
+                    var sql = window.sqlFormatter.format(value);
+                    //替换所有的换行符
+                    sql = sql.replace(/\r\n/g, "<br>");
+                    sql = sql.replace(/\n/g, "<br>");
+                    //替换所有的空格
+                    sql = sql.replace(/\s/g, "&nbsp;");
+                    html.push('<span>' + sql + '</span>');
+                }
+            });
+            return html.join('');
+        }
+    ]
+}
+
 const mongoDiagnosticInfo = {
     fieldsProcesslist: [
         'mongo',

sql/db_diagnostic.py

@@ -64,14 +64,9 @@ def create_kill_session(request):
 
     result = {"status": 0, "msg": "ok", "data": []}
     query_engine = get_engine(instance=instance)
-    if instance.db_type == "mysql":
-        result["data"] = query_engine.get_kill_command(json.loads(thread_ids))
-    elif instance.db_type == "mongo":
-        kill_command = query_engine.get_kill_command(json.loads(thread_ids))
-        result["data"] = kill_command
-    elif instance.db_type == "oracle":
+    try:
         result["data"] = query_engine.get_kill_command(json.loads(thread_ids))
-    else:
+    except AttributeError:
         result = {
             "status": 1,
             "msg": "暂时不支持{}类型数据库通过进程id构建请求".format(instance.db_type),
@@ -100,7 +95,7 @@ def kill_session(request):
 
     engine = get_engine(instance=instance)
     r = None
-    if instance.db_type == "mysql":
+    if instance.db_type in ["mysql", "doris"]:
         r = engine.kill(json.loads(thread_ids))
     elif instance.db_type == "mongo":
         r = engine.kill_op(json.loads(thread_ids))
@@ -136,11 +131,9 @@ def tablespace(request):
         return HttpResponse(json.dumps(result), content_type="application/json")
 
     query_engine = get_engine(instance=instance)
-    if instance.db_type == "mysql":
-        query_result = query_engine.tablespace(offset, limit)
-    elif instance.db_type == "oracle":
+    try:
         query_result = query_engine.tablespace(offset, limit)
-    else:
+    except AttributeError:
         result = {
             "status": 1,
             "msg": "暂时不支持{}类型数据库的表空间信息查询".format(instance.db_type),
@@ -212,9 +205,9 @@ def innodb_trx(request):
         return HttpResponse(json.dumps(result), content_type="application/json")
 
     query_engine = get_engine(instance=instance)
-    if instance.db_type == "mysql":
+    try:
         query_result = query_engine.get_long_transaction()
-    else:
+    except AttributeError:
         result = {
             "status": 1,
             "msg": "暂时不支持{}类型数据库的长事务查询".format(instance.db_type),

sql/engines/doris.py

@@ -29,6 +29,7 @@ def server_version(self):
         version = result.rows[0][-1].split("-")[0]
         return tuple([int(n) for n in version.split(".")[:3]])
 
+    '''
     def query(self, db_name=None, sql="", limit_num=0, close_conn=True, **kwargs):
         """返回 ResultSet"""
         result_set = ResultSet(full_sql=sql)
@@ -52,13 +53,28 @@ def query(self, db_name=None, sql="", limit_num=0, close_conn=True, **kwargs):
             if close_conn:
                 self.close()
         return result_set
+    '''
 
     forbidden_databases = [
         "__internal_schema",
         "INFORMATION_SCHEMA",
         "information_schema",
     ]
 
+    def processlist(
+        self,
+        command_type,
+        base_sql="select id, user, host, catalog, db, command, time, state, query_id, ifnull(info,'') as info, fe from information_schema.processlist",
+        **kwargs,
+    ):
+        return super(DorisEngine, self).processlist(command_type, base_sql)
+
+    def get_kill_command(self, thread_ids, thread_ids_check=False):
+        return super(DorisEngine, self).get_kill_command(thread_ids, thread_ids_check)
+
+    def kill(self, thread_ids, thread_ids_check=False):
+        return super(DorisEngine, self).kill(thread_ids, thread_ids_check)
+
     def execute_check(self, db_name=None, sql=""):
         """上线单执行前的检查, 返回Review set"""
         check_result = ReviewSet(full_sql=sql)
@@ -186,3 +202,10 @@ def execute(self, db_name=None, sql="", close_conn=True):
         if close_conn:
             self.close()
         return execute_result
+    
+    def get_long_transaction(self, *args, **kwargs):
+        # 不支持的方法需要抛出错误
+        raise AttributeError(f"{self.__class__.__name__} has no attribute 'get_long_transaction'")
+
+    def trxandlocks(self, *args, **kwargs):
+        raise AttributeError(f"{self.__class__.__name__} has no attribute 'trxandlocks'")

sql/engines/mysql.py

@@ -62,6 +62,8 @@ class MysqlForkType(Enum):
 
 
 class MysqlEngine(EngineBase):
+    name = "MySQL"
+    info = "MySQL engine"
     test_query = "SELECT 1"
     _server_version = None
     _server_fork_type = None
@@ -103,10 +105,6 @@ def get_connection(self, db_name=None):
         self.thread_id = self.conn.thread_id()
         return self.conn
 
-    name = "MySQL"
-
-    info = "MySQL engine"
-
     def escape_string(self, value: str) -> str:
         """字符串参数转义"""
         return pymysql.escape_string(value)
@@ -540,7 +538,7 @@ def query(
                 result_set = self.result_set_binary_as_hex(result_set)
         except Exception as e:
             logger.warning(
-                f"MySQL语句执行报错，语句：{sql}，错误信息{traceback.format_exc()}"
+                f"{self.name}语句执行报错，语句：{sql}，错误信息{traceback.format_exc()}"
             )
             result_set.error = str(e)
         finally:
@@ -721,7 +719,7 @@ def execute(self, db_name=None, sql="", close_conn=True, parameters=None):
             cursor.close()
         except Exception as e:
             logger.warning(
-                f"MySQL语句执行报错，语句：{sql}，错误信息{traceback.format_exc()}"
+                f"{self.name}语句执行报错，语句：{sql}，错误信息{traceback.format_exc()}"
             )
             result.error = str(e)
         if close_conn:
@@ -762,9 +760,13 @@ def osc_control(self, **kwargs):
         """
         return self.inc_engine.osc_control(**kwargs)
 
-    def processlist(self, command_type, **kwargs):
+    def processlist(
+        self,
+        command_type,
+        base_sql="select id, user, host, db, command, time, state, ifnull(info,'') as info from information_schema.processlist",
+        **kwargs,
+    ):
         """获取连接信息"""
-        base_sql = "select id, user, host, db, command, time, state, ifnull(info,'') as info from information_schema.processlist"
         # escape
         command_type = self.escape_string(command_type)
         if not command_type:
@@ -778,11 +780,12 @@ def processlist(self, command_type, **kwargs):
 
         return self.query("information_schema", sql)
 
-    def get_kill_command(self, thread_ids):
+    def get_kill_command(self, thread_ids, thread_ids_check=True):
         """由传入的线程列表生成kill命令"""
         # 校验传参
-        if [i for i in thread_ids if not isinstance(i, int)]:
-            return None
+        if thread_ids_check:
+            if [i for i in thread_ids if not isinstance(i, int)]:
+                return None
         sql = "select concat('kill ', id, ';') from information_schema.processlist where id in ({});".format(
             ",".join(str(tid) for tid in thread_ids)
         )
@@ -793,11 +796,12 @@ def get_kill_command(self, thread_ids):
 
         return kill_sql
 
-    def kill(self, thread_ids):
+    def kill(self, thread_ids, thread_ids_check=True):
         """kill线程"""
         # 校验传参
-        if [i for i in thread_ids if not isinstance(i, int)]:
-            return ResultSet(full_sql="")
+        if thread_ids_check:
+            if [i for i in thread_ids if not isinstance(i, int)]:
+                return ResultSet(full_sql="")
         sql = "select concat('kill ', id, ';') from information_schema.processlist where id in ({});".format(
             ",".join(str(tid) for tid in thread_ids)
         )

sql/templates/dbdiagnostic.html

@@ -29,6 +29,7 @@
                         <optgroup id="optgroup-oracle" label="Oracle"></optgroup>
                         <optgroup id="optgroup-redis" label="Redis"></optgroup>
                         <optgroup id="optgroup-pgsql" label="PgSQL"></optgroup>
+                        <optgroup id="optgroup-doris" label="Doris"></optgroup>
                 </select>
             </div>
             <div id="command-div" class="form-group">
@@ -123,7 +124,9 @@ <h4 class="modal-title text-danger">确定要终止所选会话吗？</h4>
         if (typeof oracleDiagnosticInfo !== "undefined" && Array.isArray(oracleDiagnosticInfo.fieldsProcesslist)) {
             processListTableInfos.push(oracleDiagnosticInfo?.fieldsProcesslist);
         }
-        
+        if (typeof dorisDiagnosticInfo !== "undefined" && Array.isArray(dorisDiagnosticInfo.fieldsProcesslist)) {
+            processListTableInfos.push(dorisDiagnosticInfo?.fieldsProcesslist);
+        }
 
 
         // 问题诊断--进程列表
@@ -246,6 +249,8 @@ <h4 class="modal-title text-danger">确定要终止所选会话吗？</h4>
                 }]
             ]
         ]
+        let dorisTableInfo = ['doris',tablespaceListTableInfos[0][1]]
+        tablespaceListTableInfos.push(dorisTableInfo)
 
         // 问题诊断--表空间列表
         function get_space_list() {
@@ -729,7 +734,7 @@ <h4 class="modal-title text-danger">确定要终止所选会话吗？</h4>
                 //获取用户实例列表
                 $(function () {
                     // 会话管理-支持的数据库类型
-                    supportedDbType=['mysql','mongo', 'oracle','redis','pgsql']
+                    supportedDbType=['mysql','mongo', 'oracle','redis','pgsql','doris']
                     $.ajax({
                         type: "get",
                         url: "/group/user_all_instances/",