tasks.py

# ~*~ coding: utf-8 ~*~
import json

from celery import shared_task
from django.core.cache import cache

from common.utils import get_object_or_none, capacity_convert, sum_capacity, encrypt_password, get_logger
from .models import SystemUser, AdminUser, Asset
from .const import ADMIN_USER_CONN_CACHE_KEY_PREFIX, SYSTEM_USER_CONN_CACHE_KEY_PREFIX


FORKS = 10
TIMEOUT = 60
logger = get_logger(__file__)


@shared_task
def update_assets_hardware_info(assets):
    """
    Using ansible api to update asset hardware info
    :param assets:  asset seq
    :return: result summary ['contacted': {}, 'dark': {}]
    """
    from ops.utils import run_adhoc
    name = "GET_ASSETS_HARDWARE_INFO"
    tasks = [
        {
            'name': name,
            'action': {
                'module': 'setup'
            }
        }
    ]
    hostname_list = [asset.hostname for asset in assets]
    result = run_adhoc(hostname_list, pattern='all', tasks=tasks,
                       name=name, run_as_admin=True)
    summary, result_raw = result.results_summary, result.results_raw
    for hostname, info in result_raw['ok'].items():
        if info:
            info = info[name]['ansible_facts']
        else:
            continue
        asset = get_object_or_none(Asset, hostname=hostname)
        if not asset:
            continue

        ___vendor = info['ansible_system_vendor']
        ___model = info['ansible_product_version']
        ___sn = info['ansible_product_serial']

        for ___cpu_model in info['ansible_processor']:
            if ___cpu_model.endswith('GHz'):
                break
        else:
            ___cpu_model = 'Unknown'
        ___cpu_count = info['ansible_processor_count']
        ___cpu_cores = info['ansible_processor_cores']
        ___memory = '%s %s' % capacity_convert('{} MB'.format(info['ansible_memtotal_mb']))
        disk_info = {}
        for dev, dev_info in info['ansible_devices'].items():
            if dev_info['removable'] == '0':
                disk_info[dev] = dev_info['size']
        ___disk_total = '%s %s' % sum_capacity(disk_info.values())
        ___disk_info = json.dumps(disk_info)

        ___platform = info['ansible_system']
        ___os = info['ansible_distribution']
        ___os_version = info['ansible_distribution_version']
        ___os_arch = info['ansible_architecture']
        ___hostname_raw = info['ansible_hostname']

        for k, v in locals().items():
            if k.startswith('___'):
                setattr(asset, k.strip('_'), v)
        asset.save()

    for hostname, task in summary['dark'].items():
        logger.warn("Update {} hardware info error: {}".format(
            hostname, task[name],
        ))

    return summary


@shared_task
def update_assets_hardware_period():
    """
    Update asset hardware period task
    :return:
    """
    assets = Asset.objects.filter(type__in=['Server', 'VM'])
    update_assets_hardware_info(assets)


@shared_task
def test_admin_user_connectability(admin_user):
    """
    Test asset admin user can connect or not. Using ansible api do that
    :param admin_user:
    :return:
    """
    from ops.utils import run_adhoc
    assets = admin_user.get_related_assets()
    hosts = [asset.hostname for asset in assets]
    tasks = [
        {
            "name": "TEST_ADMIN_CONNECTIVE",
            "action": {
                "module": "ping",
            }
        }
    ]
    result = run_adhoc(hosts, tasks=tasks, pattern="all", run_as_admin=True)
    return result.results_summary


@shared_task
def test_admin_user_connectability_period():
    # assets = Asset.objects.filter(type__in=['Server', 'VM'])
    admin_users = AdminUser.objects.all()
    for admin_user in admin_users:
        summary = test_admin_user_connectability(admin_user)

        cache.set(ADMIN_USER_CONN_CACHE_KEY_PREFIX + admin_user.name, summary, 60*60*60)
        for i in summary['contacted']:
            cache.set(ADMIN_USER_CONN_CACHE_KEY_PREFIX + i, 1, 60*60*60)

        for i in summary['dark']:
            cache.set(ADMIN_USER_CONN_CACHE_KEY_PREFIX + i, 0, 60*60*60)


@shared_task
def test_admin_user_connectability_manual(asset):
    from ops.utils import run_adhoc
    # assets = Asset.objects.filter(type__in=['Server', 'VM'])
    hosts = [asset.hostname]
    tasks = [
        {
            "name": "TEST_ADMIN_CONNECTIVE",
            "action": {
                "module": "ping",
            }
        }
    ]
    result = run_adhoc(hosts, tasks=tasks, pattern="all", run_as_admin=True)
    if result.results_summary['dark']:
        cache.set(ADMIN_USER_CONN_CACHE_KEY_PREFIX + asset.hostname, 0, 60*60*60)
        return False
    else:
        cache.set(ADMIN_USER_CONN_CACHE_KEY_PREFIX + asset.hostname, 1, 60*60* 60)
        return True


@shared_task
def test_system_user_connectability(system_user):
    """
    Test system cant connect his assets or not.
    :param system_user:
    :return:
    """
    from ops.utils import run_adhoc
    assets = system_user.get_clusters_assets()
    hosts = [asset.hostname for asset in assets]
    tasks = [
        {
            "name": "TEST_SYSTEM_USER_CONNECTIVE",
            "action": {
                "module": "ping",
            }
        }
    ]
    result = run_adhoc(hosts, tasks=tasks, pattern="all", run_as=system_user.name)
    return result.results_summary


@shared_task
def test_system_user_connectability_period():
    for system_user in SystemUser.objects.all():
        summary = test_system_user_connectability(system_user)
        cache.set(SYSTEM_USER_CONN_CACHE_KEY_PREFIX + system_user.name, summary, 60*60*60)


def get_push_system_user_tasks(system_user):
    tasks = [
        {
            'name': 'Add user',
            'action': {
                'module': 'user',
                'args': 'name={} shell={} state=present password={}'.format(
                    system_user.username, system_user.shell,
                    encrypt_password(system_user.password),
                ),
            }
        },
        {
            'name': 'Set authorized key',
            'action': {
                'module': 'authorized_key',
                'args': "user={} state=present key='{}'".format(
                    system_user.username, system_user.public_key
                )
            }
        },
        {
            'name': 'Set sudoers',
            'action': {
                'module': 'lineinfile',
                'args': "dest=/etc/sudoers state=present regexp='^{0} ALL=' "
                        "line='{0} ALL=(ALL) NOPASSWD: {1}' "
                        "validate='visudo -cf %s'".format(
                    system_user.username,
                    system_user.sudo,
                )
            }
        }
    ]
    return tasks


PUSH_SYSTEM_USER_PERIOD_TASK_NAME = 'PUSH SYSTEM USER [{}] PERIOD...'
PUSH_SYSTEM_USER_TASK_NAME = 'PUSH SYSTEM USER [{}] ASSETS'


def push_system_user(system_user, assets, name):
    from ops.utils import get_task_by_name, run_adhoc_object, \
        create_task, create_adhoc

    if system_user.auto_push and assets:
        task = get_task_by_name(name)
        if not task:
            task = create_task(name, created_by="System")
            task.save()
        tasks = get_push_system_user_tasks(system_user)
        hosts = [asset.hostname for asset in assets]
        options = {'forks': FORKS, 'timeout': TIMEOUT}

        adhoc = task.get_latest_adhoc()
        if not adhoc or adhoc.task != tasks or adhoc.hosts != hosts:
            adhoc = create_adhoc(task=task, tasks=tasks, pattern='all',
                                 options=options, hosts=hosts, run_as_admin=True)
        return run_adhoc_object(adhoc)


@shared_task
def push_system_user_period():
    logger.debug("Push system user period")
    for s in SystemUser.objects.filter(auto_push=True):
        assets = s.get_clusters_assets()

        name = PUSH_SYSTEM_USER_PERIOD_TASK_NAME.format(s.name)
        push_system_user(s, assets, name)


def push_system_user_to_assets_if_need(system_user, assets=None, asset_groups=None):
    assets_to_push = []
    system_user_assets = system_user.assets.all()
    if assets:
        assets_to_push.extend(assets)
    if asset_groups:
        for group in asset_groups:
            assets_to_push.extend(group.assets.all())

    assets_need_push = set(assets_to_push) - set(system_user_assets)
    if not assets_need_push:
        return
    logger.debug("Push system user {} to {} assets".format(
        system_user.name, ', '.join([asset.hostname for asset in assets_need_push])
    ))
    result = push_system_user(system_user, assets_need_push, PUSH_SYSTEM_USER_TASK_NAME)
    system_user.assets.add(*tuple(assets_need_push))
    return result