1. from __future__ import absolute_import, unicode_literals
    2. import celery
    3. from celery import task
    4. import json
    5. import time
    6. import logging
    7. import traceback
    8. import requests
    9. from utils.ansible_api import AnsibleRunner
    12. logger = logging.getLogger('django')
    15. def initialize_callback(logger, callback_url, container_id, status):
    16.     '''容器初始化状态回调
    17.     status:
    18.         1: 新增成功,  2: 新增失败,  3: 删除成功,  4:删除失败
    19.     '''
    20.     with requests.Session() as s:
    21.         url = f"{callback_url}/{container_id}/{status}"
    22.         logger.info(url)
    23.         a = requests.adapters.HTTPAdapter(max_retries=60)
    24.         s.mount('http://', a)
    26.         _retries = 3
    27.         while _retries > 0:
    28.             r = s.get(url)
    29.             if r.status_code == 200:
    30.                 logger.info(f"notify success - {container_id}/{status} - {str(r.content)}")
    31.                 break
    32.             _retries -= 1
    33.             time.sleep(10)
    34.         else:
    35.             logger.error(f"notify failed - {container_id}/{status} - {str(r.content)}")
    38. class InitializeCallbackTask(celery.Task):
    39.     '''容器初始化回调'''
    41.     def __init__(self):
    42.         super(InitializeCallbackTask, self).__init__()
    44.     def on_success(self, retval, task_id, args, kwargs):
    45.         need_callback = kwargs.get('need_callback')
    46.         do_action = kwargs['DOACTION']
    47.         if need_callback:
    48.             try:
    49.                 callback_url = kwargs.get('callback_url')
    50.                 container_id = kwargs.get('container_id')
    51.                 status = 1 if do_action == 'INITCON' else 3
    52.                 # user_name = kwargs.get('user_name')
    53.                 initialize_callback(logger, callback_url, container_id, status)
    54.             except Exception:
    55.                 logger.error("initialize success, notify failed.\n" + traceback.format_exc())
    57.     def on_failure(self, exc, task_id, args, kwargs, einfo):
    58.         need_callback = kwargs.get('need_callback')
    59.         logger.error(('Task {0} raised exception: {1} {2}'.format(task_id, exc, einfo.traceback)))
    60.         if need_callback:
    61.             try:
    62.                 callback_url = kwargs.get('callback_url')
    63.                 container_id = kwargs.get('container_id')
    64.                 do_action = kwargs['DOACTION']
    65.                 status = 2 if do_action == 'INITCON' else 4
    66.                 # user_name = kwargs.get('user_name')
    67.                 initialize_callback(logger, callback_url, container_id, status)
    68.             except Exception:
    69.                 logger.error("initialize failed, notify failed.\n" + traceback.format_exc())
    72. @task(base=InitializeCallbackTask)
    73. def initialize_task_v1(ansible_hosts, playbook, need_callback=False, callback_url=None, **kwargs):
    74.     '''容器初始化任务'''
    75.     logger.info(f"ansible-playbook: {playbook} - {json.dumps(kwargs)}")
    77.     def do_action(_resource, playbook, kwargs):
    78.         rbt = AnsibleRunner(_resource)
    79.         rbt.run_playbook(playbook, kwargs)
    81.         _res = rbt.get_playbook_result()
    83.         failed_msg = _res.get('failed')
    84.         skip_msg = _res.get('skipped')
    85.         ok_msg = _res.get('ok')
    86.         status_msg = _res.get('status')
    87.         unreach_msg = _res.get('unreachable')
    88.         changed_msg = _res.get('changed')
    90.         if failed_msg:
    91.             raise Exception(failed_msg)
    93.         if unreach_msg:
    94.             raise Exception(unreach_msg)
    96.         if not failed_msg and not skip_msg and not ok_msg and not status_msg and not unreach_msg and not changed_msg:
    97.             raise Exception({'msg': 'ansible-playbook task was not executed'})
    99.         return _res
    101.     if isinstance(ansible_hosts, dict):
    102.         return do_action(ansible_hosts, playbook, kwargs)
    103.     elif isinstance(ansible_hosts, list):
    104.         _result = []
    105.         for _resource in ansible_hosts:
    106.             for k, _ in _resource.items():
    107.                 kwargs.update({"host": k})
    108.             _result.append(do_action(_resource, playbook, kwargs))
    109.         return _result
    110.     else:
    111.         raise Exception('wrong inventory struct')