AB逻辑

Posted 2020-12-12 bao9687426

# AB机新逻辑

"""
1. 查看是否存在有效的远程灾备计划
2. 先给计划创建一个空的任务。
3. 到A端中填充任务，构造成真正可以执行的任务。（在这个过程需要到A端释放以前的以前锁定的资源，锁定现在的资源）（原子性）
这个过程容易失败（失败后我们怎么处理较好呢？？）
   1. 需要释放上一次任务的资源停止之前的任务并等待结束
   1. 重试到一定次数。
   2. 重试完成后继续失败该怎么办？？无限重试？（要不无限重试吧，这样也就不用释放资源了，但是之前到A端锁定的资源该怎么办呢？？？？对了，
   可以直接用特定的标志去释放，成功过去了就会释放，也就不用每一次失败都需要去释放了，并且不能保证一定能释放成功）这儿就这么处理
4. 执行远程容灾任务，该怎么去执行呢？？首先需要明确远程容灾任务是要做什么吧？
   1. 远程容灾任务要做什么？：
      1. 是不是我在前面每一个任务对应的磁盘级别的任务的参数都准备好，这个时候之前去执行就好了

"""
import threading
import time
import json
import copy
import uuid
from apiv1 import models
from box_dashboard import xlogging
from django.db import transaction

_logger = xlogging.getLogger(__name__)


class RemoteScheduleThreading(threading.Thread):
    def __init__(self):
        super(RemoteScheduleThreading, self).__init__(name=‘RemoteScheduleThreading‘)
        self.is_running = list()

    def run(self):
        while True:
            try:
                self.work()
            except Exception as e:
                _logger.debug(‘RemoteScheduleThreading Exception, because of {}‘.format(e))

    def work(self):
        for remote_schedule in models.RemoteBackupSchedule.objects.filter(enabled=True, deleted=False):
            if not self.check_remote_schedule_can_exc(remote_schedule):
                continue
            self.create_remote_task_empty(remote_schedule)

    def check_remote_schedule_can_exc(self, remote_schedule):
        """
        检测远程灾备计划此时是否可以执行
        :param remote_schedule:
        :return:

        1. 需要判断计划是否被禁用或者删除
        2. 判断计划类型，间隔类型查看是否到达执行时间，持续类型无条件继续执行，也就是不需要休眠
        3. 需要判断是否已经存在执行中的任务（通过计划去找任务）
        """
        # type: models.RemoteBackupSchedule, models.RemoteBackupTask
        if not remote_schedule.enabled or remote_schedule.deleted:
            return False
        now = time.time()
        if remote_schedule.get_remote_schedule_type == ‘interval‘:
            if now < remote_schedule.next_run_date:
                return False
        if self.is_have_running_task(remote_schedule):
            return False
        return True

    def create_remote_task_empty(self, remote_schedule):
        """
        创建远程容灾空任务，这个过程只在B端这边组装数据
        :param remote_schedule:
        :return:

        创建远程空任务我们需要什么？？
        1. 需要上一次任务的信息。因为我们需要根据上一次任务的信息找到本次需要同步快照，那么找到本次需要同步的快照，需要什么去找呢？？
           1. 需要上一次同步的主机快照，
           2. 需要上一次同步的时间点，
           3. 需要同步的主机
           4. 需要上一次任务的uuid（需要释放资源）
        """
        # type: models.RemoteBackupTask, models.RemoteBackupSchedule
        prev_info = self.fetch_prev_sync_info(remote_schedule)
        remote_host_ident = remote_schedule.host.ident
        remote_task = models.RemoteBackupTask.objects.create(schedule=remote_schedule)
        current_info = self.generate_current_task_info(remote_task)
        exc_config = {‘prev_info‘: prev_info, ‘remote_host_ident‘: remote_host_ident, ‘current_info‘: current_info}
        remote_task.ext_config = json.dumps(exc_config)
        remote_task.save(update_fields=[‘ext_config‘, ])

    @staticmethod
    def generate_current_task_info(remote_task):
        """
        生成本次任务的一些信息（task_uuid）
        :param remote_task:
        :return:
        """
        task_uuid = uuid.uuid4().hex
        lock_type = ‘remote_schedule_{}‘.format(remote_task.schedule.id)
        return {‘task_uuid‘: task_uuid, ‘remote_task_id‘: remote_task.id, ‘lock_type‘: lock_type}

    @staticmethod
    def fetch_prev_sync_info(remote_schedule):
        """
        获取上一次同步的信息
        :param remote_schedule:
        :return:
        """
        ext_config_dict = json.loads(remote_schedule.ext_config)
        prev_info = ext_config_dict.get(‘current_info‘, dict())
        return prev_info

    @staticmethod
    def is_have_running_task(remote_schedule):
        """
        检测计划是否存在执行中的任务
        :param remote_schedule:
        :return:
        """
        remote_task = remote_schedule.remote_backup_tasks.filter(finish_datetime__isnull=True)
        if remote_task:
            return True
        return False


class RemoteBackupTaskMonitorThreading(threading.Thread):
    def __init__(self):
        super(RemoteBackupTaskMonitorThreading, self).__init__(name=‘RemoteBackupTaskMonitorThreading‘)
        self.running_list = list()

    def run(self):
        while True:
            try:
                self.task_work()
            except Exception as e:
                _logger.debug(‘RemoteBackupTaskMonitorThreading Exception, because of {}‘.format(e))

    def task_work(self):
        for remote_backup_task in models.RemoteBackupTask.objects.filter(finish_datetime__isnull=True):
            self.clear_finish_task()
            if self.is_running(remote_backup_task):
                continue
            remote_task_thread = RemoteBackupTaskThreading(remote_backup_task)
            remote_task_thread.setDaemon(True)
            remote_task_thread.start()
            self.running_list.append(remote_task_thread)

    def is_running(self, remote_backup_task):
        for running_task_thread in self.running_list:
            if running_task_thread.remote_backup_task == remote_backup_task:
                if running_task_thread.is_alive():
                    return True
        return False

    def clear_finish_task(self):
        _running_task = copy.copy(self.running_list)
        for running_task_thread in _running_task:
            if running_task_thread.is_alive():
                continue
            self.running_list.remove(running_task_thread)


class RemoteBackupTaskThreading(threading.Thread):
    def __init__(self, remote_backup_task):
        super(RemoteBackupTaskThreading, self).__init__(name=‘RemoteBackupTaskThreading‘)
        self.remote_backup_task = remote_backup_task
        self.running_sub_task_thread = list()

    def run(self):
        with transaction.atomic():
            self.stop_all_sub_task_and_wait(self.remote_backup_task)
            update_type, sync_info = self.query_need_sync_info_from_A()  # 到A端去找需要同步的信息
            self.check_prev_and_release_other_to_A()
            self.update_local_database_info(update_type, sync_info)
        # 此时子任务已经创建完毕，需要放到线程中去执行
        self.exc_remote_sub_task()
        self.finish_current_task()

    def exc_remote_sub_task(self):
        """
        通过remote_task找到对应的子任务，然后拿到队列中去执行？
        :return:
        """
        # type: models.RemoteBackupTask
        for remote_sub_task in self.remote_backup_task.remote_backup_sub_tasks.all():
            remote_sub_task_thread = RemoteSubTaskThreading(remote_sub_task)
            remote_sub_task_thread.setDaemon(True)
            remote_sub_task_thread.start()
            self.running_sub_task_thread.append(remote_sub_task_thread)

        for remote_sub_task_thread in self.running_sub_task_thread:
            remote_sub_task_thread.join()

    def finish_current_task(self):
        """
        结束本次任务，更新任务，计划等信息
        :return:
        """

    def check_prev_and_release_other_to_A(self):
        """
        到A端去释放资源和检测上一次的资源是否被意外释放
        :return:
        1. 需要上一次任务的信息和这次任务的信息
        """
        ext_config_dict = json.loads(self.remote_backup_task.ext_config)
        prev_info = ext_config_dict.get(‘prev_info‘, dict())
        current_info = ext_config_dict[‘current_info‘]
        return {‘prev_info‘: prev_info, current_info: current_info}

    def update_local_database_info(self, update_type, sync_info):
        """
        通过A端查到的同步信息来更新本地的信息
        :param update_type:
        :param sync_info:
        :return:
        """
        self.update_local_hostsnapshot(update_type, sync_info)
        self.update_local_disksnapshot(update_type, sync_info)
        self.update_local_snapshot_storage(update_type, sync_info)
        self.update_current_info_to_schedule_and_task(sync_info)
        self.create_remote_sub_task(sync_info)

    def update_local_hostsnapshot(self, update_type, sync_info):
        """
        更新本地的主机快照级别的信息，可能是新建，可能是更新
        :param update_type:
        :param sync_info:
        :return:
        """

    def update_local_disksnapshot(self, update_type, sync_info):
        """
         更新本地的磁盘快照级别的信息，可能是新建，可能是更新
        :param update_type:
        :param sync_info:
        :return:
        """

    def update_local_snapshot_storage(self, update_type, sync_info):
        """
        更新本地的快照存储级别的信息，可能是新建，可能是更新
        :param update_type:
        :param sync_info:
        :return:
        """

    def update_current_info_to_schedule_and_task(self, sync_info):
        """
        需要更新当前任务的一些信息
        :param sync_info:
        :return:
        如 hostsnapshot，timestmap
        """

    def create_remote_sub_task(self, sync_info):
        """
        创建远程灾备子任务（一个磁盘一个任务）
        :param sync_info:
        :return:
        1. 子任务执行需要（上一次任务的快照点，上一次同步到的timestamp，本次的快照点，本次的timestamp，本次任务对应的task_uuid）
        因此我们从remote_task拿到上一次的信息prev_info和current_info, r
        还需要生成一个task_uuid保存到子任务数据库记录中
        """

    def stop_all_sub_task_and_wait(self, remote_task):  #
        """
        停止计划上一次任务对应的所有子任务，并等待结束，但是这个过程需要到A端去停止，是不是就有机会失败？
        :param remote_task:
        :return:
        去A端停止子任务，我需要什么？其实这个时候只需要每一个子任务的task_uuid。
        通过当前的任务找到上一次的RemotebackupTask, 然后找到其对应的所有子任务，最后将子任务的所有的task_uuid从数据库读取出来到A端停止并等待任务结束

        """
        # type: models.RemoteBackupSubTask

    def query_need_sync_info_from_A(self):
        """
        到A端去找需要同步的信息
        :return:
        1. 需要知道上一次任务的信息（上一次的主机快照和时间点）
        2. 需要知道远程主机
        3. 需要知道本次同步的类型(间隔还是持续)
        """
        # type: models.RemoteBackupTask
        ext_config_dict = json.loads(self.remote_backup_task.ext_config)
        prev_info = ext_config_dict.get(‘prev_info‘, dict())
        current_task_uuid = ext_config_dict[‘current_info‘][‘task_uuid‘]
        current_lock_type = ext_config_dict[‘current_info‘][‘lock_type‘]
        remote_host_ident = ext_config_dict[‘remote_host_ident‘]
        sync_type = self.get_sync_type()
        return {‘sync_type‘: sync_type, ‘prev_info‘: prev_info, ‘remote_host_ident‘: remote_host_ident,
                ‘task_uuid‘: current_task_uuid, ‘lock_type‘: current_lock_type}

    def get_sync_type(self):
        """
        根据任务找到计划，就可以知道同步的类型
        :return:
        """
        remote_schedule = self.remote_backup_task.schedule
        return remote_schedule.get_remote_schedule_type


class RemoteSubTaskThreading(threading.Thread):
    def __init__(self, remote_sub_task):
        super(RemoteSubTaskThreading, self).__init__(name=‘RemoteSubTaskThreading‘)
        self.remote_sub_task = remote_sub_task

    def run(self):
        self.check_schedule_valid()
        self.start_remote_logic()
        self.start_local_rs_module()
        self.check_remote_logic()
        self.stop_and_wait_task()
        self.finish_sub_task()

    def check_schedule_valid(self):
        """
        检测该任务对应的计划是否还有效
        :return:
        """

    def start_remote_logic(self):
        """
        开始远端的发送数据的任务， 把之前的和现在的任务的主机快照点和时刻，task_uuid传输到A端开始执行任务
        :return:
        """

    def start_local_rs_module(self):
        """
        开始本地的接收数据的任务，
        :return:
        """

    def check_remote_logic(self):
        """
        检测远端任务的状态，查看传输任务是否结束 task_uuid
        :return:
        """
        task = {‘status‘: False}
        while not task[‘status‘]:
            time.sleep(5)

    def stop_and_wait_task(self):
        """
        停止A端的传输任务并等待停止成功，不成功就一直循环，直到停止成功，通过task_uuid
        :return:
        """
        while True:
            try:
                stop_and_wait_task(task_uuid)  # 到A端停止任务
                return
            except Exception as e:
                time.sleep(5)
                _logger.info(‘stop_and_wait_task exception, because of {}‘.format(e))
                continue

    def finish_sub_task(self):
        """
        结束子任务，更新子任务数据库信息
        :return:
        """


# A端

def query_need_sync_info(host_ident, prev_info, sync_type, lock_uuid, lock_type):
    """
    找到本次需要同步的快照
    :param host_ident:
    :param prev_info:
    :param sync_type:
    :param lock_uuid
    :return: 需要返回快照点之间的关系？和本次同步的信息，而至于是否为cdp类型可以在同步信息中得到
    """
    prev_hostsnapshot = prev_info.get(‘hostsnapshot‘)
    host_obj = models.Host.objects.filter(ident=host_ident)
    if not host_obj:
        return ‘update‘, {}
    if prev_hostsnapshot is None:
        # 代表是第一次来，
        # 不管什么同步类型，都直接通过主机找到最后的一个主机快照点？需要区分A端主机快照的类型吗？
        pass
    else:
        if sync_type == ‘interval‘:
            # 间隔备份，
            # 1. 通过主机找到最新的快照点
            # 2. 和前一个任务所同步的信息作比较，（hostsnapshot 和 timestamp）一样，就返回空的同步信息回去如果比以前同步的新，
            #    就需要使用lock_uuid锁住当前的主机快照点
            pass
        else:
            # 持续备份
            # 获取上一次同步到的快照点，通过上一次同步的快照点，找到离他最近的一个快照点，然后分别打开两个快照点，
            # 1. 对于qcow类型算出他们的位图？然后同步，return create, 同步信息
            # 2. 对于cdp类型需要找到两个快照点的关系
            #    1. 追加 (在同一个链上，也在同一个主机快照上) return update ，同步信息
            #    2. 增量 (在同一个链上，但不在同一个主机快照上) return create ，同步信息
            #    3. 无关 (不在同一个链上) return create ，同步信息
            pass
        pass


def check_prev_and_release_other(prev_info, current_info):
    """
    检测上一次任务和释放资源
    :param prev_info:
    :param current_info:
    :return:
    1. 通过lock_type找到所有的主机快照锁
    2. 通过prev的task_uuid是否从主机快照锁找出数据，如果可以找到，就说明还被锁着的，没有找到呢？没有上一次任务不管
    3. 释放除current_info和prev_info中通过task_uuid生成的锁
    """
    lock_type = current_info[‘lock_type‘]
    _ = prev_info
    _ = lock_type