新增定时执行SQL工单

审核通过后可由申请人或者审核人设定工单定时执行，执行前可终止、可修改执行时间

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/jobs.py

@@ -0,0 +1,68 @@
+# -*- coding:utf-8 -*-
+import datetime
+
+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
+
+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)
+
+scheduler = BackgroundScheduler()
+scheduler.add_jobstore(DjangoJobStore(), "default")
+
+register_events(scheduler)
+
+try:
+    scheduler.start()
+    print("Scheduler started!")
+except SchedulerAlreadyRunningError:
+    print("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.add_job(execute_job, 'date', run_date=run_date, args=[workflowId, url], id=job_id,
+                          replace_existing=True)
+        workflowDetail.status = Const.workflowStatus['tasktiming']
+        workflowDetail.save()
+    except Exception as e:
+        context = {'errMsg': '任务添加失败，错误信息：' + str(e)}
+        return render(request, 'error.html', context)
+    return HttpResponseRedirect(reverse('sql:detail', args=(workflowId,)))
+
+
+# 删除sql执行任务
+def del_sqlcronjob(job_id):
+    return scheduler.remove_job(job_id)
+
+
+# 获取任务详情
+def job_info(job_id):
+    return scheduler.get_job(job_id)

sql/sqlreview.py

@@ -0,0 +1,124 @@
+# -*- coding: UTF-8 -*-
+import json
+
+import time
+from threading import Thread
+
+from django.db import connection
+from django.utils import timezone
+from django.conf import settings
+
+from .dao import Dao
+from .const import Const, WorkflowDict
+from .sendmail import MailSender
+from .inception import InceptionDao
+from .aes_decryptor import Prpcrypt
+from .models import users, workflow, master_config
+from .workflow import Workflow
+from .permission import role_required, superuser_required
+import logging
+
+logger = logging.getLogger('default')
+
+dao = Dao()
+inceptionDao = InceptionDao()
+mailSender = MailSender()
+prpCryptor = Prpcrypt()
+workflowOb = Workflow()
+
+
+# 获取当前请求url
+def getDetailUrl(request):
+    scheme = request.scheme
+    host = request.META['HTTP_HOST']
+    return "%s://%s/detail/" % (scheme, host)
+
+
+# 根据实例名获取主库连接字符串，并封装成一个dict
+def getMasterConnStr(clusterName):
+    listMasters = master_config.objects.filter(cluster_name=clusterName)
+
+    masterHost = listMasters[0].master_host
+    masterPort = listMasters[0].master_port
+    masterUser = listMasters[0].master_user
+    masterPassword = prpCryptor.decrypt(listMasters[0].master_password)
+    dictConn = {'masterHost': masterHost, 'masterPort': masterPort, 'masterUser': masterUser,
+                'masterPassword': masterPassword}
+    return dictConn
+
+
+# SQL工单执行回调
+def execute_call_back(workflowId, clusterName, url):
+    workflowDetail = workflow.objects.get(id=workflowId)
+    # 获取审核人
+    try:
+        listAllReviewMen = json.loads(workflowDetail.review_man)
+    except ValueError:
+        listAllReviewMen = (workflowDetail.review_man,)
+
+    dictConn = getMasterConnStr(clusterName)
+    try:
+        # 交给inception先split，再执行
+        (finalStatus, finalList) = inceptionDao.executeFinal(workflowDetail, dictConn)
+
+        # 封装成JSON格式存进数据库字段里
+        strJsonResult = json.dumps(finalList)
+        workflowDetail = workflow.objects.get(id=workflowId)
+        workflowDetail.execute_result = strJsonResult
+        workflowDetail.finish_time = timezone.now()
+        workflowDetail.status = finalStatus
+        workflowDetail.is_manual = 0
+        workflowDetail.audit_remark = ''
+        # 重新获取连接，防止超时
+        connection.close()
+        workflowDetail.save()
+    except Exception as e:
+        logger.error(e)
+
+    # 如果执行完毕了，则根据settings.py里的配置决定是否给提交者和DBA一封邮件提醒.DBA需要知晓审核并执行过的单子
+    if hasattr(settings, 'MAIL_ON_OFF') == True:
+        if getattr(settings, 'MAIL_ON_OFF') == "on":
+            # 给主、副审核人，申请人，DBA各发一封邮件
+            engineer = workflowDetail.engineer
+            reviewMen = workflowDetail.review_man
+            workflowStatus = workflowDetail.status
+            workflowName = workflowDetail.workflow_name
+            objEngineer = users.objects.get(username=engineer)
+            strTitle = "SQL上线工单执行完毕 # " + str(workflowId)
+            strContent = "发起人：" + engineer + "\n审核人：" + reviewMen + "\n工单地址：" + url + "\n工单名称： " + workflowName + "\n执行结果：" + workflowStatus
+            mailSender.sendEmail(strTitle, strContent, [objEngineer.email])
+            mailSender.sendEmail(strTitle, strContent, getattr(settings, 'MAIL_REVIEW_DBA_ADDR'))
+            for reviewMan in listAllReviewMen:
+                if reviewMan == "":
+                    continue
+                objReviewMan = users.objects.get(username=reviewMan)
+                mailSender.sendEmail(strTitle, strContent, [objReviewMan.email])
+
+
+# 给定时任务执行sql
+def execute_job(workflowId, url):
+    workflowDetail = workflow.objects.get(id=workflowId)
+    clusterName = workflowDetail.cluster_name
+
+    # 服务器端二次验证，当前工单状态必须为定时执行过状态
+    if workflowDetail.status != Const.workflowStatus['tasktiming']:
+        raise Exception('工单不是定时执行状态')
+
+    # 将流程状态修改为执行中，并更新reviewok_time字段
+    workflowDetail.status = Const.workflowStatus['executing']
+    workflowDetail.reviewok_time = timezone.now()
+    workflowDetail.save()
+    # 执行之前重新split并check一遍，更新SHA1缓存；因为如果在执行中，其他进程去做这一步操作的话，会导致inception core dump挂掉
+    splitReviewResult = inceptionDao.sqlautoReview(workflowDetail.sql_content, workflowDetail.cluster_name,
+                                                   isSplit='yes')
+    workflowDetail.review_content = json.dumps(splitReviewResult)
+    try:
+        workflowDetail.save()
+    except Exception:
+        # 关闭后重新获取连接，防止超时
+        connection.close()
+        workflowDetail.save()
+
+    # 采取异步回调的方式执行语句，防止出现持续执行中的异常
+    t = Thread(target=execute_call_back, args=(workflowId, clusterName, url))
+    t.start()

sql/static/allWorkflow.html

@@ -5,9 +5,11 @@
     <div id="toolbar" class="bootstrap-select">
         <select id="navStatus" class="form-control dropdown-menu-right selectpicker ">
             <option value="all" selected="selected">全部工单</option>
-            <option value="manreviewing">等待审核人审核</option>
-            <option value="finish">已执行完毕</option>
+            <option value="manreviewing">待审核</option>
+            <option value="pass">审核通过</option>
+            <option value="tasktiming">定时执行</option>
             <option value="executing">执行中</option>
+            <option value="finish">已执行完毕</option>
             <option value="exception">执行有异常</option>
             <option value="abort">人工终止流程</option>
             <option value="autoreviewwrong">自动审核不通过</option>