celery&flower&supervisor

apps/cmdb/tasks.py

@@ -0,0 +1,56 @@
+import time
+import logging
+
+from celery import shared_task
+from celery_once import QueueOnce
+
+from utils.sandbox_utils import SandboxScan, LoginExecution
+from .models import DeviceScanInfo
+
+info_logger = logging.getLogger('sandbox_info')
+
+
+@shared_task(base=QueueOnce)
+def scan_execution():
+    scan = SandboxScan()
+    execution = LoginExecution()
+    scan_type = execution.get_scan_type()
+    auth_type = execution.get_auth_type()
+    start_time = time.time()
+    if scan_type == 'basic_scan':
+        hosts = scan.basic_scan()
+        for host in hosts:
+            DeviceScanInfo.objects.update_or_create(
+                hostname=host,
+            )
+    else:
+        hosts = scan.os_scan()
+        login_hosts = [host for host in hosts if host['os'] in ['Linux', 'embedded']]
+        nologin_hosts = [host for host in hosts if host not in login_hosts]
+        for host in nologin_hosts:
+            DeviceScanInfo.objects.update_or_create(
+                hostname=host['host'],
+                defaults={
+                    'os_type': host['os']
+                }
+            )
+        for host in login_hosts:
+            kwargs = {
+                'hostname': host['host'],
+                'username': execution.get_ssh_username(),
+                'port': execution.get_ssh_port(),
+                'password': execution.get_ssh_password(),
+                'private_key': execution.get_ssh_private_key()
+            }
+            defaults = execution.login_execution(auth_type=auth_type, **kwargs)
+            DeviceScanInfo.objects.update_or_create(
+                hostname=host['host'],
+                defaults=defaults
+            )
+    end_time = time.time()
+    msg = 'Scan task has been completed, execution time: %(time)s, %(num)s hosts are up.' % {
+        'time': end_time - start_time,
+        'num': len(hosts)
+    }
+    info_logger.info(msg)
+    return msg

apps/cmdb/views_scan.py

@@ -10,12 +10,14 @@
 from django.http import JsonResponse
 from django.shortcuts import render, get_object_or_404
 
+from celery_once import AlreadyQueued
 
 from system.mixin import LoginRequiredMixin
 from custom import BreadcrumbMixin, SandboxListView, SandboxDeleteView
 from utils.sandbox_utils import ConfigFileMixin
 from system.models import Menu
 from .models import DeviceScanInfo
+from .tasks import scan_execution
 
 error_logger = logging.getLogger('sandbox_error')
 
@@ -81,50 +83,10 @@ class DeviceScanDeleteView(SandboxDeleteView):
 class DeviceScanExecView(LoginRequiredMixin, View):
 
     def get(self, request):
-        import time
-        from utils.sandbox_utils import SandboxScan, LoginExecution
-        info_logger = logging.getLogger('sandbox_info')
-        ret = dict(result=False)
-        scan = SandboxScan()
-        execution = LoginExecution()
-        scan_type = execution.get_scan_type()
-        auth_type = execution.get_auth_type()
-        start_time = time.time()
-        if scan_type == 'basic_scan':
-            hosts = scan.basic_scan()
-            for host in hosts:
-                DeviceScanInfo.objects.update_or_create(
-                    hostname=host,
-                )
-        else:
-            hosts = scan.os_scan()
-            login_hosts = [host for host in hosts if host['os'] in ['Linux', 'embedded']]
-            nologin_hosts = [host for host in hosts if host not in login_hosts]
-            for host in nologin_hosts:
-                DeviceScanInfo.objects.update_or_create(
-                    hostname=host['host'],
-                    defaults={
-                        'os_type': host['os']
-                    }
-                )
-            for host in login_hosts:
-                kwargs = {
-                    'hostname': host['host'],
-                    'username': execution.get_ssh_username(),
-                    'port': execution.get_ssh_port(),
-                    'password': execution.get_ssh_password(),
-                    'private_key': execution.get_ssh_private_key()
-                }
-                defaults = execution.login_execution(auth_type=auth_type, **kwargs)
-                DeviceScanInfo.objects.update_or_create(
-                    hostname=host['host'],
-                    defaults=defaults
-                )
-        end_time = time.time()
-        msg = 'Scan task has been completed, execution time: %(time)s, %(num)s hosts are up.' % {
-            'time': end_time - start_time,
-            'num': len(hosts)
-        }
-        info_logger.info(msg)
-        ret['result'] = True
+        ret = dict(status='fail')
+        try:
+            scan_execution.delay()
+            ret['status'] = 'success'
+        except AlreadyQueued:
+            ret['status'] = 'already_queued'
         return JsonResponse(ret)

sandboxMP/__init__.py

@@ -0,0 +1,5 @@
+from __future__ import absolute_import, unicode_literals
+
+from .celery import app as celery_app
+
+__all__ = ('celery_app')

sandboxMP/celery.py

@@ -0,0 +1,38 @@
+from __future__ import absolute_import, unicode_literals
+import os
+from celery import Celery
+
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'sandboxMP.settings')
+
+app = Celery('sandbox')
+
+app.config_from_object('django.conf:settings')
+
+app.autodiscover_tasks()
+
+BROKER_URL = 'redis://localhost:6379/0'
+
+CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'
+
+CELERY_TIMEZONE = 'Asia/Shanghai'
+
+CELERY_ENABLE_UTC = False
+
+CELERYD_FORCE_EXECV = True
+
+CELERYD_CONCURRENCY = 5
+
+CELERY_ACKS_LATE = True
+
+CELERYD_MAX_TASKS_PER_CHILD = 100
+
+CELERYD_TASK_TIME_LIMIT = 60 * 5
+
+
+app.conf.ONCE = {
+    'backend': 'celery_once.backends.Redis',
+    'settings': {
+        'url': 'redis://localhost:6379/2',
+        'default_timeout': 60 * 5
+    }
+}

sandboxMP/settings.py

@@ -13,6 +13,8 @@
 import os
 import sys
 
+from .celery import *
+
 # Build paths inside the project like this: os.path.join(BASE_DIR, ...)
 BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 

templates/cmdb/device_scan.html

@@ -279,8 +279,12 @@
                     },
                 success: function (msg) {
                     layer.closeAll('loading');
-                    if (msg.result) {
-                        layer.alert('扫描已完成', {icon: 1});
+                    if (msg.status == 'success') {
+                        layer.alert('扫描任务已下发', {icon: 1});
+                        oDataTable.ajax.reload();
+                    }
+                    else if (msg.status == 'already_queued') {
+                        layer.alert('当前已有扫描任务正在执行', {icon: 4});
                         oDataTable.ajax.reload();
                     }
                     else {