[Update] 修改信号

apps/assets/api.py

@@ -25,7 +25,7 @@ from .hands import IsSuperUser, IsValidUser, IsSuperUserOrAppUser, \
     get_user_granted_assets
 from .models import AssetGroup, Asset, Cluster, SystemUser, AdminUser
 from . import serializers
-from .tasks import update_assets_hardware_info_manual, test_admin_user_connectability_util, \
+from .tasks import update_asset_hardware_info_manual, test_admin_user_connectability_util, \
     test_asset_connectability_manual, push_system_user_to_cluster_assets_manual, \
     test_system_user_connectability_manual
 
@@ -222,7 +222,7 @@ class AssetRefreshHardwareApi(generics.RetrieveAPIView):
     def retrieve(self, request, *args, **kwargs):
         asset_id = kwargs.get('pk')
         asset = get_object_or_404(Asset, pk=asset_id)
-        summary = update_assets_hardware_info_manual([asset])[1]
+        summary = update_asset_hardware_info_manual(asset)[1]
         if summary.get('dark'):
             return Response(summary['dark'].values(), status=501)
         else:

apps/assets/apps.py

@@ -7,7 +7,5 @@ class AssetsConfig(AppConfig):
     name = 'assets'
 
     def ready(self):
-        from .signals import on_app_ready
-        from . import tasks
-        on_app_ready.send(self.__class__)
+        from . import signals_handler
         super().ready()

apps/assets/const.py

 # -*- coding: utf-8 -*-
 #
-from django.utils.translation import ugettext as _
 
-# PUSH_SYSTEM_USER_PERIOD_LOCK_KEY = "PUSH_SYSTEM_USER_PERIOD_KEY"
-PUSH_SYSTEM_USER_PERIOD_TASK_NAME = _("PUSH SYSTEM USER TO CLUSTER PERIOD: {}")
-PUSH_SYSTEM_USER_MANUAL_TASK_NAME = _("PUSH SYSTEM USER TO CLUSTER MANUALLY: {}")
-PUSH_SYSTEM_USER_TASK_NAME = _("PUSH SYSTEM USER TO CLUSTER: {}")
-# PUSH_SYSTEM_USER_LOCK_KEY = "PUSH_SYSTEM_USER_TO_CLUSTER_LOCK_{}"
-PUSH_SYSTEM_USER_ON_CHANGE_TASK_NAME = _("PUSH SYSTEM USER ON CHANGE: {}")
-PUSH_SYSTEM_USER_ON_CREATE_TASK_NAME = _("PUSH SYSTEM USER ON CREATE: {}")
-PUSH_SYSTEM_USERS_ON_ASSET_CREATE_TASK_NAME = _("PUSH SYSTEM USERS ON ASSET CREAT: {}")
-
-
-UPDATE_ASSETS_HARDWARE_TASK_NAME = _('UPDATE ASSETS HARDWARE INFO')
-UPDATE_ASSETS_HARDWARE_MANUAL_TASK_NAME = _('UPDATE ASSETS HARDWARE INFO MANUALLY')
-UPDATE_ASSETS_HARDWARE_ON_CREATE_TASK_NAME = _('UPDATE ASSETS HARDWARE INFO ON CREATE')
-# UPDATE_ASSETS_HARDWARE_PERIOD_LOCK_KEY = "UPDATE_ASSETS_HARDWARE_PERIOD_LOCK_KEY"
-UPDATE_ASSETS_HARDWARE_PERIOD_TASK_NAME = _('UPDATE ASSETS HARDWARE INFO PERIOD')
 UPDATE_ASSETS_HARDWARE_TASKS = [
    {
-       'name': UPDATE_ASSETS_HARDWARE_TASK_NAME,
+       'name': "setup",
        'action': {
            'module': 'setup'
        }
    }
 ]
 
-# TEST_ADMIN_USER_CONN_PERIOD_LOCK_KEY = "TEST_ADMIN_USER_CONN_PERIOD_KEY"
-TEST_ADMIN_USER_CONN_PERIOD_TASK_NAME = _("TEST ADMIN USER CONN PERIOD: {}")
-TEST_ADMIN_USER_CONN_MANUAL_TASK_NAME = _("TEST ADMIN USER CONN MANUALLY: {}")
-TEST_ADMIN_USER_CONN_TASK_NAME = _("TEST ADMIN USER CONN: {}")
 ADMIN_USER_CONN_CACHE_KEY = "ADMIN_USER_CONN_{}"
 TEST_ADMIN_USER_CONN_TASKS = [
     {
-        "name": "TEST_ADMIN_CONNECTIVE",
+        "name": "ping",
         "action": {
             "module": "ping",
         }
@@ -41,15 +21,11 @@ TEST_ADMIN_USER_CONN_TASKS = [
 ]
 
 ASSET_ADMIN_CONN_CACHE_KEY = "ASSET_ADMIN_USER_CONN_{}"
-TEST_ASSET_CONN_TASK_NAME = _("ASSET CONN TEST MANUAL")
 
-TEST_SYSTEM_USER_CONN_PERIOD_LOCK_KEY = "TEST_SYSTEM_USER_CONN_PERIOD_KEY"
-TEST_SYSTEM_USER_CONN_PERIOD_TASK_NAME = _("TEST SYSTEM USER CONN PERIOD: {}")
-TEST_SYSTEM_USER_CONN_MANUAL_TASK_NAME = _("TEST SYSTEM USER CONN MANUALLY: {}")
 SYSTEM_USER_CONN_CACHE_KEY = "SYSTEM_USER_CONN_{}"
 TEST_SYSTEM_USER_CONN_TASKS = [
    {
-       "name": "TEST_SYSTEM_USER_CONNECTIVE",
+       "name": "ping",
        "action": {
            "module": "ping",
        }

apps/assets/forms.py

@@ -23,6 +23,7 @@ class AssetCreateForm(forms.ModelForm):
             'groups': forms.SelectMultiple(attrs={'class': 'select2', 'data-placeholder': _('Select asset groups')}),
             'cluster': forms.Select(attrs={'class': 'select2', 'data-placeholder': _('Select cluster')}),
             'admin_user': forms.Select(attrs={'class': 'select2', 'data-placeholder': _('Select admin user')}),
+            'port': forms.TextInput()
         }
         help_texts = {
             'hostname': '* required',
@@ -32,6 +33,13 @@ class AssetCreateForm(forms.ModelForm):
             'admin_user': _('Host level admin user, If not set using cluster admin user default')
         }
 
+    def clean_admin_user(self):
+        cluster = self.cleaned_data.get('cluster')
+        admin_user = self.cleaned_data.get('admin_user')
+        if not cluster.admin_user and not admin_user:
+            raise forms.ValidationError(_("You need set a admin user if cluster not have"))
+        return self.cleaned_data['admin_user']
+
 
 class AssetUpdateForm(forms.ModelForm):
     class Meta:
@@ -53,6 +61,13 @@ class AssetUpdateForm(forms.ModelForm):
             'admin_user': _('Host level admin user, If not set using cluster admin user default')
         }
 
+    def clean_admin_user(self):
+        cluster = self.cleaned_data.get('cluster')
+        admin_user = self.cleaned_data.get('admin_user')
+        if not cluster.admin_user and not admin_user:
+            raise forms.ValidationError(_("You need set a admin user if cluster not have"))
+        return self.cleaned_data['admin_user']
+
 
 class AssetBulkUpdateForm(forms.ModelForm):
     assets = forms.ModelMultipleChoiceField(
@@ -283,7 +298,6 @@ class SystemUserUpdateForm(SystemUserForm):
         system_user = super(forms.ModelForm, self).save()
 
         if private_key_file:
-            print(private_key_file)
             private_key = private_key_file.read().strip().decode('utf-8')
             public_key = ssh_pubkey_gen(private_key=private_key)
         else:

apps/assets/models/user.py

@@ -207,13 +207,11 @@ class AdminUser(AssetUser):
 
 
 class SystemUser(AssetUser):
+    SSH_PROTOCOL = 'ssh'
     PROTOCOL_CHOICES = (
-        ('ssh', 'ssh'),
-    )
-    AUTH_METHOD_CHOICES = (
-        ('P', 'Password'),
-        ('K', 'Public key'),
+        (SSH_PROTOCOL, 'ssh'),
     )
+
     cluster = models.ManyToManyField('assets.Cluster', blank=True, verbose_name=_("Cluster"))
     priority = models.IntegerField(default=10, verbose_name=_("Priority"))
     protocol = models.CharField(max_length=16, choices=PROTOCOL_CHOICES, default='ssh', verbose_name=_('Protocol'))
@@ -229,6 +227,12 @@ class SystemUser(AssetUser):
         clusters = self.cluster.all()
         return Asset.objects.filter(cluster__in=clusters)
 
+    def get_clusters(self):
+        return self.cluster.all()
+
+    def get_clusters_joined(self):
+        return ', '.join([cluster.name for cluster in self.get_clusters()])
+
     @property
     def assets_amount(self):
         return len(self.get_clusters_assets())

apps/assets/signals_handler.py

+# -*- coding: utf-8 -*-
+#
+
+
+from django.db.models.signals import post_save, post_init, m2m_changed, pre_save
+from django.dispatch import receiver
+from django.utils.translation import gettext as _
+
+from common.utils import get_logger
+from .models import Asset, SystemUser, Cluster
+from .tasks import update_assets_hardware_info_util, \
+    test_asset_connectability_util, \
+    push_system_user_util
+
+
+logger = get_logger(__file__)
+
+
+def update_asset_hardware_info_on_created(asset):
+    logger.debug("Update asset `{}` hardware info".format(asset))
+    update_assets_hardware_info_util.delay([asset])
+
+
+def test_asset_conn_on_created(asset):
+    logger.debug("Test asset `{}` connectability".format(asset))
+    test_asset_connectability_util.delay(asset)
+
+
+def push_cluster_system_users_to_asset(asset):
+    logger.info("Push cluster system user to asset: {}".format(asset))
+    task_name = _("Push cluster system users to asset")
+    system_users = asset.cluster.systemuser_set.all()
+    push_system_user_util.delay(system_users, [asset], task_name)
+
+
+@receiver(post_save, sender=Asset, dispatch_uid="my_unique_identifier")
+def on_asset_created(sender, instance=None, created=False, **kwargs):
+    if instance and created:
+        logger.info("Asset `` create signal received".format(instance))
+        update_asset_hardware_info_on_created(instance)
+        test_asset_conn_on_created(instance)
+        push_cluster_system_users_to_asset(instance)
+
+
+@receiver(post_init, sender=Asset)
+def on_asset_init(sender, instance, created=False, **kwargs):
+    if instance and created is False:
+        instance.__original_cluster = instance.cluster
+
+
+@receiver(post_save, sender=Asset)
+def on_asset_cluster_changed(sender, instance=None, created=False, **kwargs):
+    if instance and created is False and instance.cluster != instance.__original_cluster:
+        logger.info("Asset cluster changed signal received")
+        push_cluster_system_users_to_asset(instance)
+
+
+def push_to_cluster_assets_on_system_user_created_or_update(system_user):
+    if not system_user.auto_push:
+        return
+    logger.debug("Push system user `{}` to cluster assets".format(system_user.name))
+    for cluster in system_user.cluster.all():
+        task_name = _("Push system user to cluster assets: {}->{}").format(
+            cluster.name, system_user.name
+        )
+        assets = cluster.assets.all()
+        push_system_user_util.delay([system_user], assets, task_name)
+
+
+@receiver(post_save, sender=SystemUser)
+def on_system_user_created_or_updated(sender, instance=None, **kwargs):
+    if instance and instance.auto_push:
+        logger.info("System user `{}` create or update signal received".format(instance))
+        push_to_cluster_assets_on_system_user_created_or_update(instance)
+
+
+@receiver(post_init, sender=Cluster, dispatch_uid="my_unique_identifier")
+def on_cluster_init(sender, instance, **kwargs):
+    logger.debug("On cluster init")
+    instance.__original_assets = tuple(instance.assets.values_list('pk', flat=True))
+    # instance.__origin_system_users = tuple(instance.systemuser_set.all())
+
+
+@receiver(post_save, sender=Cluster, dispatch_uid="my_unique_identifier")
+def on_cluster_assets_changed(sender, instance, **kwargs):
+    assets_origin = instance.__original_assets
+    assets_new = instance.assets.values_list('pk', flat=True)
+    assets_added = set(assets_new) - set(assets_origin)
+    if assets_added:
+        logger.debug("Receive cluster change assets signal")
+        logger.debug("Push cluster `{}` system users to: {}".format(
+            instance, ', '.join([str(asset) for asset in assets_added])
+        ))
+        assets = []
+        for asset_id in assets_added:
+            try:
+                asset = Asset.objects.get(pk=asset_id)
+            except Asset.DoesNotExist:
+                continue
+            else:
+                assets.append(asset)
+        system_users = [s for s in instance.systemuser_set.all() if s.auto_push]
+        task_name = _("Push system user to assets")
+        push_system_user_util.delay(system_users, assets, task_name)
+
+
+@receiver(post_save, sender=Cluster, dispatch_uid="my_unique_identifier")
+def on_cluster_system_user_changed(sender, instance, **kwargs):
+    system_users_origin = instance.__origin_system_users
+    system_user_new = instance.systemuser_set.values_list('pk', flat=True)
+    system_users_added = set(system_user_new) - system_users_origin
+    if system_users_added:
+        logger.debug("Receive cluster change system users signal")
+        system_users = []
+        for system_user_id in system_users_added:
+            try:
+                system_user = SystemUser.objects.get(pk=system_user_id)
+            except SystemUser.DoesNotExist:
+                continue
+            else:
+                system_users.append(system_user)
+        logger.debug("Push new system users `{}` to cluster `{}` assets".format(
+            ','.join([s.name for s in system_users]), instance
+        ))
+        task_name = _(
+            "Push system user to cluster assets: {}->{}").format(
+             instance.name, ', '.join(s.name for s in system_users)
+        )
+        push_system_user_util.delay(system_users, instance.assets.all(), task_name)
+

apps/assets/views/asset.py

@@ -28,7 +28,6 @@ from common.utils import get_object_or_none, get_logger, is_uuid
 from .. import forms
 from ..models import Asset, AssetGroup, AdminUser, Cluster, SystemUser
 from ..hands import AdminUserRequiredMixin
-from ..tasks import update_assets_hardware_info_util
 
 
 __all__ = [
@@ -162,10 +161,6 @@ class AssetUpdateView(AdminUserRequiredMixin, UpdateView):
         kwargs.update(context)
         return super(AssetUpdateView, self).get_context_data(**kwargs)
 
-    def form_invalid(self, form):
-        logger.error(form.errors)
-        return super().form_invalid(form)
-
 
 class AssetDeleteView(AdminUserRequiredMixin, DeleteView):
     model = Asset

apps/jumpserver/settings.py

@@ -336,11 +336,15 @@ CELERY_RESULT_SERIALIZER = 'pickle'
 CELERY_RESULT_BACKEND = CELERY_BROKER_URL
 CELERY_ACCEPT_CONTENT = ['json', 'pickle']
 CELERY_RESULT_EXPIRES = 3600
-CELERY_WORKER_LOG_FORMAT = '%(asctime)s [%(module)s %(levelname)s] %(message)s'
-CELERY_WORKER_TASK_LOG_FORMAT = '%(asctime)s [%(module)s %(levelname)s] %(message)s'
+# CELERY_WORKER_LOG_FORMAT = '%(asctime)s [%(module)s %(levelname)s] %(message)s'
+CELERY_WORKER_LOG_FORMAT = '%(message)s'
+# CELERY_WORKER_TASK_LOG_FORMAT = '%(asctime)s [%(module)s %(levelname)s] %(message)s'
+CELERY_WORKER_TASK_LOG_FORMAT = '%(message)s'
+# CELERY_WORKER_LOG_FORMAT = '%(asctime)s [%(module)s %(levelname)s] %(message)s'
 CELERY_TASK_EAGER_PROPAGATES = True
-# CELERY_TIMEZONE = TIME_ZONE
-# CELERY_ENABLE_UTC = True
+CELERY_REDIRECT_STDOUTS = True
+CELERY_REDIRECT_STDOUTS_LEVEL = "INFO"
+CELERY_WORKER_HIJACK_ROOT_LOGGER = False
 
 
 # Cache use redis

apps/ops/ansible/callback.py

@@ -51,7 +51,6 @@ class AdHocResultCallback(CallbackModule):
                 contacted.remove(host)
 
     def v2_runner_on_failed(self, result, ignore_errors=False):
-        print("#######RUN FAILED" * 19)
         self.gather_result("failed", result)
         super().v2_runner_on_failed(result, ignore_errors=ignore_errors)
 

run_server.py

@@ -5,8 +5,8 @@ import subprocess
 import threading
 import time
 import argparse
-import platform
 import sys
+import signal
 
 from apps import __version__
 
@@ -25,9 +25,7 @@ LOG_LEVEL = CONFIG.LOG_LEVEL
 WORKERS = 4
 
 EXIT_EVENT = threading.Event()
-EXIT_MSGS = []
-
-
+processes = {}
 
 try:
     os.makedirs(os.path.join(BASE_DIR, "data", "static"))
@@ -97,7 +95,6 @@ def start_service(services):
         __version__))
     print('Quit the server with CONTROL-C.')
 
-    processes = {}
     services_all = {
          "gunicorn": start_gunicorn,
          "celery": start_celery,
@@ -126,6 +123,12 @@ def start_service(services):
         time.sleep(5)
 
 
+def stop_service():
+    for name, proc in processes.items():
+        print("Stop service {}".format(name))
+        proc.terminate()
+
+
 if __name__ == '__main__':
     parser = argparse.ArgumentParser(description="Jumpserver start tools")
     parser.add_argument("services", type=str, nargs='+', default="all",
@@ -133,6 +136,9 @@ if __name__ == '__main__':
                         help="The service to start",
                         )
     args = parser.parse_args()
-    start_service(args.services)
+    try:
+        start_service(args.services)
+    except KeyboardInterrupt:
+        stop_service()