A 任务计划新增执行失败钉钉/企业微信通知代替系统通知

spug_api/apps/schedule/models.py

@@ -44,6 +44,7 @@ class Task(models.Model, ModelMixin):
     is_active = models.BooleanField(default=False)
     desc = models.CharField(max_length=255, null=True)
     latest = models.ForeignKey(History, on_delete=models.PROTECT, null=True)
+    rst_notify = models.CharField(max_length=255, null=True)
 
     created_at = models.CharField(max_length=20, default=human_datetime)
     created_by = models.ForeignKey(User, models.PROTECT, related_name='+')
@@ -56,6 +57,7 @@ class Task(models.Model, ModelMixin):
         tmp['latest_status'] = self.latest.status if self.latest else None
         tmp['latest_run_time'] = self.latest.run_time if self.latest else None
         tmp['latest_status_alias'] = self.latest.get_status_display() if self.latest else None
+        tmp['rst_notify'] = json.loads(self.rst_notify) if self.rst_notify else {'mode': '0'}
         if self.trigger == 'cron':
             tmp['trigger_args'] = json.loads(self.trigger_args)
         return tmp

spug_api/apps/schedule/scheduler.py

@@ -11,6 +11,7 @@ from django_redis import get_redis_connection
 from django.utils.functional import SimpleLazyObject
 from django.db import close_old_connections
 from apps.schedule.models import Task, History
+from apps.schedule.utils import send_fail_notify
 from apps.notify.models import Notify
 from apps.schedule.executors import dispatch
 from apps.schedule.utils import auto_clean_schedule_history
@@ -19,10 +20,8 @@ from django.conf import settings
 from libs import AttrDict, human_datetime
 import logging
 import json
-import time
 
 logger = logging.getLogger("django.apps.scheduler")
-counter = dict()
 
 
 class Scheduler:
@@ -68,10 +67,10 @@ class Scheduler:
             Notify.make_notify('schedule', '1', '调度器已关闭', '调度器意外关闭，你可以在github上提交issue')
         elif event.code == EVENT_JOB_MAX_INSTANCES:
             logger.info(f'EVENT_JOB_MAX_INSTANCES: {event}')
-            Notify.make_notify('schedule', '1', f'{obj.name} - 达到调度实例上限', '一般为上个周期的执行任务还未结束，请增加调度间隔或减少任务执行耗时')
+            send_fail_notify(obj, '达到调度实例上限，一般为上个周期的执行任务还未结束，请增加调度间隔或减少任务执行耗时')
         elif event.code == EVENT_JOB_ERROR:
             logger.info(f'EVENT_JOB_ERROR: job_id {event.job_id} exception: {event.exception}')
-            Notify.make_notify('schedule', '1', f'{obj.name} - 执行异常', f'{event.exception}')
+            send_fail_notify(obj, f'执行异常：{event.exception}')
         elif event.code == EVENT_JOB_EXECUTED:
             if event.retval:
                 score = 0
@@ -84,9 +83,8 @@ class Scheduler:
                     output=json.dumps(event.retval)
                 )
                 Task.objects.filter(pk=event.job_id).update(latest=history)
-                if score != 0 and time.time() - counter.get(event.job_id, 0) > 3600:
+                if score != 0:
-                    counter[event.job_id] = time.time()
+                    send_fail_notify(obj)
-                    Notify.make_notify('schedule', '1', f'{obj.name} - 执行失败', '请在任务计划中查看失败详情')
 
     def _init_builtin_jobs(self):
         self.scheduler.add_job(auto_clean_records, 'cron', hour=0, minute=0)

spug_api/apps/schedule/utils.py

@@ -2,6 +2,10 @@
 # Copyright: (c) <spug.dev@gmail.com>
 # Released under the AGPL-3.0 License.
 from apps.schedule.models import Task, History
+from libs.utils import human_datetime
+from threading import Thread
+import requests
+import json
 
 
 def auto_clean_schedule_history():
@@ -11,3 +15,56 @@ def auto_clean_schedule_history():
             History.objects.filter(task_id=task.id, id__lt=record.id).delete()
         except IndexError:
             pass
+
+
+def send_fail_notify(task, msg=None):
+    rst_notify = json.loads(task.rst_notify)
+    mode = rst_notify.get('mode')
+    url = rst_notify.get('value')
+    if mode != '0' and url:
+        Thread(target=_do_notify, args=(task, mode, url, msg)).start()
+
+
+def _do_notify(task, mode, url, msg):
+    if mode == '1':
+        texts = [
+            '## <font color="#f90202">任务执行失败通知</font> ## ',
+            f'**任务名称：** {task.name} ',
+            f'**任务类型：** {task.type} ',
+            f'**描述信息：** {msg or "请在任务计划执行历史中查看详情"} ',
+            f'**发生时间：** {human_datetime()} ',
+            '> 来自 Spug运维平台'
+        ]
+        data = {
+            'msgtype': 'markdown',
+            'markdown': {
+                'title': '任务执行失败通知',
+                'text': '\n\n'.join(texts)
+            }
+        }
+        requests.post(url, json=data)
+    elif mode == '2':
+        data = {
+            'task_id': task.id,
+            'task_name': task.name,
+            'task_type': task.type,
+            'message': msg or '请在任务计划执行历史中查看详情',
+            'created_at': human_datetime()
+        }
+        requests.post(url, json=data)
+    elif mode == '3':
+        texts = [
+            '## <font color="warning">任务执行失败通知</font>',
+            f'**任务名称：** {task.name} ',
+            f'**任务类型：** {task.type} ',
+            f'**描述信息：** {msg or "请在任务计划执行历史中查看详情"} ',
+            f'**发生时间：** {human_datetime()} ',
+            '> 来自 Spug运维平台'
+        ]
+        data = {
+            'msgtype': 'markdown',
+            'markdown': {
+                'content': '\n'.join(texts)
+            }
+        }
+        requests.post(url, json=data)

spug_api/apps/schedule/views.py

@@ -26,6 +26,7 @@ class Schedule(View):
             Argument('type', help='请输入任务类型'),
             Argument('name', help='请输入任务名称'),
             Argument('command', help='请输入任务内容'),
+            Argument('rst_notify', type=dict, help='请选择执行失败通知方式'),
             Argument('targets', type=list, filter=lambda x: len(x), help='请选择执行对象'),
             Argument('trigger', filter=lambda x: x in dict(Task.TRIGGERS), help='请选择触发器类型'),
             Argument('trigger_args', help='请输入触发器参数'),
@@ -33,6 +34,7 @@ class Schedule(View):
         ).parse(request.body)
         if error is None:
             form.targets = json.dumps(form.targets)
+            form.rst_notify = json.dumps(form.rst_notify)
             if form.trigger == 'cron':
                 args = json.loads(form.trigger_args)['rule'].split()
                 if len(args) != 5:

spug_web/src/pages/schedule/Form.js

@@ -148,7 +148,7 @@ class ComForm extends React.Component {
 
   render() {
     const info = store.record;
-    const {getFieldDecorator} = this.props.form;
+    const {getFieldDecorator, getFieldValue} = this.props.form;
     const {page, args, loading, showTmp, nextRunTime} = this.state;
     const [b1, b2, b3] = this.verifyButtonStatus();
     return (
@@ -196,6 +196,21 @@ class ComForm extends React.Component {
                 onChange={val => this.setState({command: val})}
                 height="150px"/>
             </Form.Item>
+            <Form.Item label="失败通知" help="任务执行失败告警通知">
+              {getFieldDecorator('rst_notify.value', {initialValue: info['rst_notify']['value']})(
+                <Input
+                  addonBefore={getFieldDecorator('rst_notify.mode', {initialValue: info['rst_notify']['mode']})(
+                    <Select style={{width: 100}}>
+                      <Select.Option value="0">关闭</Select.Option>
+                      <Select.Option value="1">钉钉</Select.Option>
+                      <Select.Option value="3">企业微信</Select.Option>
+                      <Select.Option value="2">Webhook</Select.Option>
+                    </Select>
+                  )}
+                  disabled={getFieldValue('rst_notify.mode') === '0'}
+                  placeholder="请输入"/>
+              )}
+            </Form.Item>
             <Form.Item label="备注信息">
               {getFieldDecorator('desc', {initialValue: info['desc']})(
                 <Input.TextArea placeholder="请输入模板备注信息"/>