Merge pull request #68 from hhyo/github

优化报错体验

archer/settings.py

@@ -15,11 +15,11 @@
 # Build paths inside the project like this: os.path.join(BASE_DIR, ...)
 import os
 import pymysql
+
 pymysql.install_as_MySQLdb()
 
 BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 
-
 # Quick-start development settings - unsuitable for production
 # See https://docs.djangoproject.com/en/1.8/howto/deployment/checklist/
 
@@ -44,6 +44,7 @@
     'django.contrib.sessions',
     'django.contrib.messages',
     'django.contrib.staticfiles',
+    'django_apscheduler',
     'sql',
 )
 
@@ -103,15 +104,15 @@
 STATIC_URL = '/static/'
 STATIC_ROOT = os.path.join(BASE_DIR, 'static')
 
-#扩展django admin里users字段用到，指定了sql/models.py里的class users
-AUTH_USER_MODEL="sql.users"
+# 扩展django admin里users字段用到，指定了sql/models.py里的class users
+AUTH_USER_MODEL = "sql.users"
 
 ###############以下部分需要用户根据自己环境自行修改###################
 
 # Database
 # https://docs.djangoproject.com/en/1.8/ref/settings/#databases
 
-#该项目本身的mysql数据库地址
+# 该项目本身的mysql数据库地址
 DATABASES = {
     'default': {
         'ENGINE': 'django.db.backends.mysql',
@@ -123,16 +124,16 @@
     }
 }
 
-#inception组件所在的地址
+# inception组件所在的地址
 INCEPTION_HOST = '192.168.1.11'
 INCEPTION_PORT = '6100'
 
-#查看回滚SQL时候会用到，这里要告诉archer去哪个mysql里读取inception备份的回滚信息和SQL.
-#注意这里要和inception组件的inception.conf里的inception_remote_XX部分保持一致.
-INCEPTION_REMOTE_BACKUP_HOST='192.168.1.12'
-INCEPTION_REMOTE_BACKUP_PORT=5621
-INCEPTION_REMOTE_BACKUP_USER='inception'
-INCEPTION_REMOTE_BACKUP_PASSWORD='inception'
+# 查看回滚SQL时候会用到，这里要告诉archer去哪个mysql里读取inception备份的回滚信息和SQL.
+# 注意这里要和inception组件的inception.conf里的inception_remote_XX部分保持一致.
+INCEPTION_REMOTE_BACKUP_HOST = '192.168.1.12'
+INCEPTION_REMOTE_BACKUP_PORT = 5621
+INCEPTION_REMOTE_BACKUP_USER = 'inception'
+INCEPTION_REMOTE_BACKUP_PASSWORD = 'inception'
 
 # 账户登录失败锁定时间(秒)
 LOCK_TIME_THRESHOLD = 300
@@ -145,18 +146,19 @@
     import ldap
     # from django_auth_ldap.config import LDAPSearch, GroupOfNamesType
     from django_auth_ldap.config import LDAPSearch, GroupOfUniqueNamesType
+
     # if use self signed certificate, Remove AUTH_LDAP_GLOBAL_OPTIONS annotations
-    #AUTH_LDAP_GLOBAL_OPTIONS={
+    # AUTH_LDAP_GLOBAL_OPTIONS={
     #    ldap.OPT_X_TLS_REQUIRE_CERT: ldap.OPT_X_TLS_NEVER
-    #}
+    # }
     AUTH_LDAP_BIND_DN = "cn=ro,dc=xxx,dc=cn"
     AUTH_LDAP_BIND_PASSWORD = "xxxxxx"
     AUTH_LDAP_SERVER_URI = "ldap://auth.xxx.com"
     AUTH_LDAP_BASEDN = "ou=users,dc=xxx,dc=cn"
     AUTH_LDAP_USER_DN_TEMPLATE = "cn=%(user)s,ou=users,dc=xxx,dc=cn"
     AUTH_LDAP_GROUP_SEARCH = LDAPSearch("ou=groups,dc=xxx,dc=cn",
-        ldap.SCOPE_SUBTREE, "(objectClass=groupOfUniqueNames)"
-    )
+                                        ldap.SCOPE_SUBTREE, "(objectClass=groupOfUniqueNames)"
+                                        )
     AUTH_LDAP_GROUP_TYPE = GroupOfUniqueNamesType()
     AUTH_LDAP_USER_ATTRLIST = ["cn", "sn", "mail"]
     AUTH_LDAP_USER_ATTR_MAP = {
@@ -170,7 +172,7 @@
     # AUTH_LDAP_CACHE_GROUPS = True  # 如打开FIND_GROUP_PERMS后，此配置生效，对组关系进行缓存，不用每次请求都调用ldap
     # AUTH_LDAP_GROUP_CACHE_TIMEOUT = 600  # 缓存时间
 
-#开启以下配置注释，可以帮助调试ldap集成
+# 开启以下配置注释，可以帮助调试ldap集成
 LDAP_LOGS = '/tmp/ldap.log'
 DEFAULT_LOGS = '/tmp/default.log'
 stamdard_format = '[%(asctime)s][%(threadName)s:%(thread)d]' + \
@@ -229,18 +231,18 @@
     }
 }
 
-#是否开启邮件提醒功能：发起SQL上线后会发送邮件提醒审核人审核，执行完毕会发送给DBA. on是开，off是关，配置为其他值均会被archer认为不开启邮件功能
-MAIL_ON_OFF='on'
-
-MAIL_REVIEW_SMTP_SERVER='mail.xxx.com'
-MAIL_REVIEW_SMTP_PORT=25
-MAIL_REVIEW_FROM_ADDR='[email protected]'                                               #发件人，也是登录SMTP server需要提供的用户名
-MAIL_REVIEW_FROM_PASSWORD=''                                                         #发件人邮箱密码，如果为空则不需要login SMTP server
-MAIL_REVIEW_DBA_ADDR=['[email protected]', '[email protected]']        #DBA地址，执行完毕会发邮件给DBA，以list形式保存
-MAIL_REVIEW_SECURE_ADDR=['[email protected]', '[email protected]']     #登录失败，等安全相关发送地址
-#是否过滤【DROP DATABASE】|【DROP TABLE】|【TRUNCATE PARTITION】|【TRUNCATE TABLE】等高危DDL操作：
-#on是开，会首先用正则表达式匹配sqlContent，如果匹配到高危DDL操作，则判断为“自动审核不通过”；off是关，直接将所有的SQL语句提交给inception，对于上述高危DDL操作，只备份元数据
-CRITICAL_DDL_ON_OFF='off'
+# 是否开启邮件提醒功能：发起SQL上线后会发送邮件提醒审核人审核，执行完毕会发送给DBA. on是开，off是关，配置为其他值均会被archer认为不开启邮件功能
+MAIL_ON_OFF = 'on'
+
+MAIL_REVIEW_SMTP_SERVER = 'mail.xxx.com'
+MAIL_REVIEW_SMTP_PORT = 25
+MAIL_REVIEW_FROM_ADDR = '[email protected]'  # 发件人，也是登录SMTP server需要提供的用户名
+MAIL_REVIEW_FROM_PASSWORD = ''  # 发件人邮箱密码，如果为空则不需要login SMTP server
+MAIL_REVIEW_DBA_ADDR = ['[email protected]', '[email protected]']  # DBA地址，执行完毕会发邮件给DBA，以list形式保存
+MAIL_REVIEW_SECURE_ADDR = ['[email protected]', '[email protected]']  # 登录失败，等安全相关发送地址
+# 是否过滤【DROP DATABASE】|【DROP TABLE】|【TRUNCATE PARTITION】|【TRUNCATE TABLE】等高危DDL操作：
+# on是开，会首先用正则表达式匹配sqlContent，如果匹配到高危DDL操作，则判断为“自动审核不通过”；off是关，直接将所有的SQL语句提交给inception，对于上述高危DDL操作，只备份元数据
+CRITICAL_DDL_ON_OFF = 'off'
 
 # 在线查询当inception语法树打印失败时的控制，on是开启校验，失败不允许继续执行并返回错误，off是关闭校验，继续执行，允许执行会导致解析失败的查询表权限验证和脱敏功能失效
 CHECK_QUERY_ON_OFF = True

requirements.txt

@@ -14,6 +14,7 @@ urllib3==1.22
 django-admin-bootstrapped==2.5.7
 gunicorn==19.7.1
 django-auth-ldap==1.3.0
+django-apscheduler==0.2.8
 aliyun-python-sdk-core==2.3.5
 aliyun-python-sdk-core-v3==2.5.3
 aliyun-python-sdk-rds==2.1.1

sql/const.py

@@ -7,10 +7,16 @@ class Const(object):
         'autoreviewing': '自动审核中',
         'manreviewing': '等待审核人审核',
         'pass': '审核通过',
+        'tasktiming': '定时执行',
         'executing': '执行中',
         'autoreviewwrong': '自动审核不通过',
         'exception': '执行有异常',
     }
+    # 定时任务id的前缀
+    workflowJobprefix = {
+        'query': 'query',
+        'sqlreview': 'sqlreview',
+    }
 
 
 class WorkflowDict:

sql/dao.py

@@ -4,26 +4,28 @@
 
 from django.db import connection
 
+
 class Dao(object):
     _CHART_DAYS = 90
 
-    #连进指定的mysql实例里，读取所有databases并返回
+    # 连进指定的mysql实例里，读取所有databases并返回
     def getAlldbByCluster(self, masterHost, masterPort, masterUser, masterPassword):
         listDb = []
         conn = None
         cursor = None
 
         try:
-            conn=MySQLdb.connect(host=masterHost, port=masterPort, user=masterUser, passwd=masterPassword, charset='utf8mb4')
+            conn = MySQLdb.connect(host=masterHost, port=masterPort, user=masterUser, passwd=masterPassword,
+                                   charset='utf8mb4')
             cursor = conn.cursor()
             sql = "show databases"
             n = cursor.execute(sql)
             listDb = [row[0] for row in cursor.fetchall() 
                          if row[0] not in ('information_schema', 'performance_schema', 'mysql', 'test')]
         except MySQLdb.Warning as w:
-            print(str(w))
+            raise Exception(w)
         except MySQLdb.Error as e:
-            print(str(e))
+            raise Exception(e)
         finally:
             if cursor is not None:
                 cursor.close()
@@ -49,9 +51,9 @@ def getAllTableByDb(self, masterHost, masterPort, masterUser, masterPassword, db
                       if row[0] not in (
                           'test')]
         except MySQLdb.Warning as w:
-            print(str(w))
+            raise Exception(w)
         except MySQLdb.Error as e:
-            print(str(e))
+            raise Exception(e)
         finally:
             if cursor is not None:
                 cursor.close()
@@ -76,9 +78,9 @@ def getAllColumnsByTb(self, masterHost, masterPort, masterUser, masterPassword,
             n = cursor.execute(sql)
             listCol = [row[0] for row in cursor.fetchall()]
         except MySQLdb.Warning as w:
-            print(str(w))
+            raise Exception(w)
         except MySQLdb.Error as e:
-            print(str(e))
+            raise Exception(e)
         finally:
             if cursor is not None:
                 cursor.close()
@@ -120,20 +122,25 @@ def mysql_query(self, masterHost, masterPort, masterUser, masterPassword, dbName
             if cursor is not None:
                 cursor.close()
             if conn is not None:
-                conn.rollback()
-                conn.close()
+                try:
+                    conn.rollback()
+                    conn.close()
+                except:
+                    conn.close()
         return result
 
     def getWorkChartsByMonth(self):
         cursor = connection.cursor()
-        sql = "select date_format(create_time, '%%m-%%d'),count(*) from sql_workflow where create_time>=date_add(now(),interval -%s day) group by date_format(create_time, '%%m-%%d') order by 1 asc;" % (Dao._CHART_DAYS)
+        sql = "select date_format(create_time, '%%m-%%d'),count(*) from sql_workflow where create_time>=date_add(now(),interval -%s day) group by date_format(create_time, '%%m-%%d') order by 1 asc;" % (
+            Dao._CHART_DAYS)
         cursor.execute(sql)
         result = cursor.fetchall()
         return result
 
     def getWorkChartsByPerson(self):
         cursor = connection.cursor()
-        sql = "select engineer, count(*) as cnt from sql_workflow where create_time>=date_add(now(),interval -%s day) group by engineer order by cnt desc limit 50;" % (Dao._CHART_DAYS)
+        sql = "select engineer, count(*) as cnt from sql_workflow where create_time>=date_add(now(),interval -%s day) group by engineer order by cnt desc limit 50;" % (
+            Dao._CHART_DAYS)
         cursor.execute(sql)
         result = cursor.fetchall()
         return result

sql/data_masking.py

@@ -12,7 +12,13 @@ class Masking(object):
     def data_masking(self, cluster_name, db_name, sql, sql_result):
         result = {'status': 0, 'msg': 'ok', 'data': []}
         # 通过inception获取语法树,并进行解析
-        print_info = self.query_tree(sql, cluster_name, db_name)
+        try:
+            print_info = self.query_tree(sql, cluster_name, db_name)
+        except Exception as msg:
+            result['status'] = 1
+            result['msg'] = str(msg)
+            return result
+
         if print_info is None:
             result['status'] = 1
             result['msg'] = 'inception返回的结果集为空！可能是SQL语句有语法错误'
@@ -79,7 +85,13 @@ def query_tree(self, sqlContent, cluster_name, dbName):
     # 解析语法树，获取语句涉及的表，用于查询权限限制
     def query_table_ref(self, sqlContent, cluster_name, dbName):
         result = {'status': 0, 'msg': 'ok', 'data': []}
-        print_info = self.query_tree(sqlContent, cluster_name, dbName)
+        try:
+            print_info = self.query_tree(sqlContent, cluster_name, dbName)
+        except Exception as msg:
+            result['status'] = 1
+            result['msg'] = str(msg)
+            return result
+
         if print_info is None:
             result['status'] = 1
             result['msg'] = 'inception返回的结果集为空！可能是SQL语句有语法错误'

sql/inception.py

@@ -213,8 +213,8 @@ def _fetchall(self, sql, paramHost, paramPort, paramUser, paramPasswd, paramDb):
             cur=conn.cursor()
             ret=cur.execute(sql)
             result=cur.fetchall()
-        except MySQLdb.Error as e:
-            print("Mysql Error %d: %s" % (e.args[0], e.args[1]))
+        except Exception as e:
+            raise Exception(e)
         finally:
             if cur is not None:
                 cur.close()
@@ -261,8 +261,5 @@ def query_print(self, sqlContent, clusterName, dbName):
                           %s\
                           inception_magic_commit;" % (
             masterUser, masterPassword, masterHost, str(masterPort), dbName, sqlContent)
-        try:
-            result = self._fetchall(sql, self.inception_host, self.inception_port, '', '', '')
-        except Exception:
-            result = None
+        result = self._fetchall(sql, self.inception_host, self.inception_port, '', '', '')
         return result

sql/jobs.py

@@ -0,0 +1,83 @@
+# -*- coding:utf-8 -*-
+import datetime
+import time
+
+from apscheduler.schedulers.background import BackgroundScheduler
+from apscheduler.schedulers import SchedulerAlreadyRunningError, SchedulerNotRunningError
+from django.core.urlresolvers import reverse
+from django.http import HttpResponseRedirect
+from django.shortcuts import render
+from django_apscheduler.jobstores import DjangoJobStore, register_events, register_job
+
+from sql.const import Const
+from sql.models import workflow
+from .sqlreview import execute_job, getDetailUrl
+
+import logging
+
+logging.basicConfig()
+logging.getLogger('apscheduler').setLevel(logging.DEBUG)
+
+logger = logging.getLogger('default')
+
+# 初始化scheduler
+scheduler = BackgroundScheduler()
+scheduler.add_jobstore(DjangoJobStore(), "default")
+register_events(scheduler)
+try:
+    scheduler.start()
+    logger.debug("Scheduler started!")
+except SchedulerAlreadyRunningError:
+    logger.debug("Scheduler is already running!")
+
+
+# 添加/修改sql执行任务
+def add_sqlcronjob(request):
+    workflowId = request.POST.get('workflowid')
+    run_date = request.POST.get('run_date')
+    if run_date is None or workflowId is None:
+        context = {'errMsg': '时间不能为空'}
+        return render(request, 'error.html', context)
+    elif run_date < datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'):
+        context = {'errMsg': '时间不能小于当前时间'}
+        return render(request, 'error.html', context)
+    workflowDetail = workflow.objects.get(id=workflowId)
+    if workflowDetail.status not in ['审核通过', '定时执行']:
+        context = {'errMsg': '必须为审核通过或者定时执行状态'}
+        return render(request, 'error.html', context)
+
+    run_date = datetime.datetime.strptime(run_date, "%Y-%m-%d %H:%M:%S")
+    url = getDetailUrl(request) + str(workflowId) + '/'
+    job_id = Const.workflowJobprefix['sqlreview'] + '-' + str(workflowId)
+
+    try:
+        scheduler = BackgroundScheduler()
+        scheduler.add_jobstore(DjangoJobStore(), "default")
+        scheduler.add_job(execute_job, 'date', run_date=run_date, args=[workflowId, url], id=job_id,
+                          replace_existing=True)
+        register_events(scheduler)
+        try:
+            scheduler.start()
+            logger.debug("Scheduler started!")
+        except SchedulerAlreadyRunningError:
+            logger.debug("Scheduler is already running!")
+        workflowDetail.status = Const.workflowStatus['tasktiming']
+        workflowDetail.save()
+    except Exception as e:
+        context = {'errMsg': '任务添加失败，错误信息：' + str(e)}
+        return render(request, 'error.html', context)
+    else:
+        logger.debug('add_sqlcronjob:' + job_id + "run_date:" + run_date.strftime('%Y-%m-%d %H:%M:%S'))
+
+    return HttpResponseRedirect(reverse('sql:detail', args=(workflowId,)))
+
+
+# 删除sql执行任务
+def del_sqlcronjob(job_id):
+    logger.debug('del_sqlcronjob:' + job_id)
+    return scheduler.remove_job(job_id)
+
+
+# 获取任务详情
+def job_info(job_id):
+    return scheduler.get_job(job_id)