VMScheduling/code/main.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
'''
@Auther :liuyuqi.gov@msn.cn
@Time :2018/7/4 15:55
@File :main.py
'''

import time
from configparser import ConfigParser

import pandas as pd

import utils.save_conf
import utils.save_result


class Scheduling():
    '''
    调度
    '''

    # 参数
    alpha = 10
    beta = 0.5
    T = 98
    EXEC_LIMIT = 100000

    # 静态数据 n:app数 N:inst数 m:machine数 k:资源种类
    num_app = num_inst = num_mac = 0
    num_k = 200
    cpuIter = list()
    appIndex = {}
    machineIndex = {}
    inst2AppIndex = {}

    # apps = list()
    # machines = list()
    appResources = list()  # app表，
    machineResources = list()  # machine表
    instanceDeploy = list()  # instance表
    appInterference = list()  # app_interfence冲突表

    # 动态数据
    inst2MachineRemine = {}
    machineResourcesUsed = list()
    machineHasApp = list()  # 6000 [{}, {},{6004=1, 9126=1, 1598=1}, {}, {}, {},
    inst2Machine = list()
    result = pd.DataFrame(columns=list(["instanceid", "machineid"]), data=list())

    def __init__(self, **kw):
        '''
        初始化参数
        :param kw:
        '''
        for k, v in kw.items():
            setattr(self, k, v)

    def loadData(self):
        for i in range(self.T):
            self.cpuIter.append(i)
        app_interference, app_resources, instance_deploy, machine_resources = self.getConfig()
        # 1.app_resources 9338*201
        self.appResources = pd.read_csv(app_resources, header=None,
                                        names=list(["appid", "cpu", "mem", "disk", "P", "M", "PM"]), encoding="utf-8")
        tmp_cpu = self.appResources["cpu"].str.split('|', expand=True).astype('float')
        tmp_mem = self.appResources["mem"].str.split('|', expand=True).astype('float')
        for i in range(self.T):
            self.appResources["cpu_" + str(i)] = tmp_cpu[i]
            self.appResources["mem_" + str(i)] = tmp_mem[i]
        # 去掉cpu/men两列
        self.appResources.pop("cpu")
        self.appResources.pop("mem")
        self.num_app, col = self.appResources.shape  # 9338*201 201列：appid,cpu_1,cpu_2,...mem_1,men_2....,P,M,PM
        self.appResources["appid"] = pd.to_numeric(self.appResources["appid"].str.split("_", expand=True)[1].values)

        # 2.machine_resources 6000*201
        self.machineResources = pd.read_csv(machine_resources, header=None, names=list(
            ["machineid", "cpu", "mem", "disk", "P", "M", "PM"]), encoding="utf-8")
        self.num_mac, col = self.machineResources.shape
        for i in range(self.T):
            self.machineResources["cpu_" + str(i)] = self.machineResources["cpu"]
            self.machineResources["mem_" + str(i)] = self.machineResources["mem"]
        self.machineResources.pop("cpu")
        self.machineResources.pop("mem")
        self.machineResources["machineid"] = pd.to_numeric(
            self.machineResources["machineid"].str.split("_", expand=True)[1].values)
        self.machineResourcesUsed = self.machineResources.copy()
        for i in range(200):
            self.machineResourcesUsed.iloc[:, i + 1] = 0
        # 初始化 6000个空字典组成的list[{},{}....]
        for i in range(self.num_mac):
            self.machineHasApp.append({})

        # 3.instance_deploy
        self.instanceDeploy = pd.read_csv(instance_deploy, header=None,
                                          names=list(["instanceid", "appid", "machineid"]), encoding="utf-8")
        # 增加一个字段标注是否部署
        self.instanceDeploy["isdeploy"] = False
        self.instanceDeploy["instanceid"] = pd.to_numeric(
            self.instanceDeploy["instanceid"].str.split("_", expand=True)[1].values)
        self.instanceDeploy["appid"] = pd.to_numeric(self.instanceDeploy["appid"].str.split("_", expand=True)[1].values)
        self.instanceDeploy["machineid"] = pd.to_numeric(
            self.instanceDeploy["machineid"].str.split("_", expand=True)[1].values)

        # 4.app_interference 冲突表
        self.appInterference = pd.read_csv(app_interference, header=None,
                                           names=list(["appid1", "appid2", "max_interference"]), encoding="utf-8")
        self.appInterference["appid1"] = pd.to_numeric(
            self.appInterference["appid1"].str.split("_", expand=True)[1].values)
        self.appInterference["appid2"] = pd.to_numeric(
            self.appInterference["appid2"].str.split("_", expand=True)[1].values)

        # instance按照磁盘消耗排序
        self.num_app, col = self.appResources.shape
        self.num_inst, col = self.instanceDeploy.shape

    def getConfig(self):
        '''
        step1: 数据参数初始化
        :return:
        '''
        # 生成配置文件
        # self.init_conf()

        # 读取配置文件
        cf = ConfigParser()
        config_path = "../conf/config.ini"
        section_name = "data_file_name"
        cf.read(config_path)

        app_interference = cf.get(section_name, "app_interference")
        app_resources = cf.get(section_name, "app_resources")
        instance_deploy = cf.get(section_name, "instance_deploy")
        machine_resources = cf.get(section_name, "machine_resources")
        return app_interference, app_resources, instance_deploy, machine_resources

    def init_conf(self):
        '''
        初始化配置文件
        :retur
        '''
        utils.save_conf.write()

    def sort_dynamic(self):
        print("ss")

    def pickInstance(self, instanceid):
        '''
        先将instance从部署的主机中删除,删除一行，释放资源
        :return:
        '''
        if instanceid not in self.inst2Machine["instance"]:
            return
        appid = self.inst2Machine[self.inst2Machine["instanceid"] == instanceid]["appid"].values[0]

        # 更新machineResourcesUsed
        for i in range(self.num_k):
            machineResourcesUsed[fromMachine][j] -= appResources[appIt][i]
        fromMachine = self.inst2AppIndex
        self.inst2Machine.pop(instanceid)

    def toMachine(self, instanceid, machineid, doCheck=False):
        '''
        检查互斥条件,然后把instance放入主机
        :rtype: object
        :param instanceid: 实例id
        :param machineid: 主机id
        :param doCheck: 是否检测资源限制
        :return: True和False
        '''
        # instanceid所属的appid
        appid = self.instanceDeploy[self.instanceDeploy["instanceid"] == instanceid]["appid"].values[0]
        # machineid从1开始，而index从0开始
        hasApp = self.machineHasApp[int(machineid - 1)]
        if doCheck:
            # 检查互斥


            # 检查资源限制
            for i in range(self.num_k):
                if (
                                self.machineResourcesUsed[self.machineResourcesUsed["machineid"] == machineid].iloc[:,
                                i + 1].values[0] +
                                self.appResources[self.appResources["appid"] == appid].iloc[:, i + 1].values[0]
                            >
                            self.machineResources[self.machineResources["machineid"] == machineid].iloc[:,
                            i + 1].values[0]):
                    print("Resource Limit: instance: ", instanceid, ",", "machine:", machineid,
                          self.machineResourcesUsed[self.machineResourcesUsed["machineid"] == machineid].iloc[:,
                          i + 1].values[0], "+",
                          self.appResources[self.appResources["appid"] == appid].iloc[:, i + 1].values[0], " >",
                          self.machineResources[self.machineResources["machineid"] == machineid].iloc[:,
                          i + 1].values[0])
                    # 如果不符合则 return False
                    return False
        # 将inst放入新的machine,占用资源
        self.inst2Machine.append([{"instanceid": instanceid, "machineid": machineid}])
        if appid not in hasApp:
            hasApp.update({appid: 1})
        else:
            hasApp.update({appid: hasApp.get(appid) + 1})
        for i in range(self.num_k):
            self.machineResourcesUsed[self.machineResourcesUsed["machineid"] == machineid].iloc[:, i + 1].values[
                0] += \
                self.appResources[self.appResources["appid"] == appid].iloc[:, i + 1].values[0]
        return True

    def run(self, start):
        '''
        执行部署
        :return:
        '''
        # 已经部署的instance
        deployed_Instance = self.instanceDeploy.loc[pd.isna(self.instanceDeploy["machineid"]) == False]
        count_deployed_Instance, col = deployed_Instance.shape
        deployed_Instance.reset_index(drop=True, inplace=True)

        # 将已经部署的instance放置到对应主机中，占用相应资源，这一块代码比java慢了太多
        for i in range(count_deployed_Instance):
            instanceid = deployed_Instance["instanceid"][i]
            machineid = deployed_Instance["machineid"][i]
            self.toMachine(instanceid, machineid, doCheck=False)
            print("初始部署第", i, "个，持续耗时", time.time() - start, "秒")

        # 对instance同样按照disk消耗排序
        self.instanceDeploy = self.instanceDeploy.sort_values(ascending=False, by="disk")

        # 然后通过ff方法，把instance放入machine中。每次放入instance后，队列删除，每次消耗主机i，删除主机i.先使用大主机，磁盘优先计算限制条件
        row1, col = self.instanceDeploy.shape
        while row1 > 0:
            # 先对主机列表按照disk剩余进行排序，降序
            self.machineResourcesUsed = self.machineResourcesUsed.sort_values(ascending=False, by="disk")

            # 每部署一次，消耗一个主机
            self.deployInstance()
            # 筛选未部署的
            self.instanceDeploy = self.instanceDeploy[self.instanceDeploy["isdeploy"] == False]
            row, col = self.instanceDeploy.shape
            self.instanceDeploy.reset_index(drop=True, inplace=True)
            j = j + 1

            print("已经部署：", 68219 - row, "剩余部署Instance数据：", row)
            print("已经消耗Machine主机数据：", j)
        print("部署方案前几条示意：", self.result.head())
        utils.save_result.save_result(self.result)

    def dcmp(self, x):
        '''
        将结果映射到-1，0，1
        :param x:
        :return:
        '''
        if abs(x) < 1e-9:
            return 0
        elif x > 0:
            return 1
        else:
            return -1

    def deployInstance(self):
        '''
        部署逻辑
        :return:
        '''
        # 主机：self.machineResourcesUsed["machineid"][0]
        for row in self.instanceDeploy.itertuples():
            i = row.Index
            # 当前row实例尝试部署到新主机，如果可以部署则部署，如果初始已经部署，则算迁移，释放原来主机资源
            if self.toMachine(row, machineid="", doCheck=True):
                machineHasApp = machineHasApp.append(pd.DataFrame(
                    [{"instanceid": row.instanceid,
                      "machineid": "machine_" + str(j)}]))


if __name__ == '__main__':
    print("------------开始部署啦--------------")
    start = time.time()
    scheduling = Scheduling()
    # 加载数据
    scheduling.loadData()
    print("加载数据耗时:", time.time() - start)
    # 开始调度
    scheduling.run(start)
    # 部署完事
    print("------------部署完啦--------------")
    end = time.time()
    print("总共耗时：", end - start, "秒")