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 libs.save_conf


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

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

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

    apps = list()
    machines = list()
    appResources = list()
    machineResources = list()

    # 动态数据
    inst2Machine = {}
    machineResourcesUsed = list()
    machineHasApp = pd.DataFrame(columns=list(["instanceid", "machineid"]), data=list())

    def __init__(self, **kw):
        for k, v in kw.items():
            setattr(self, k, v)

    def loadData(self):
        '''   n
               app_resources.csv
               m
               machine_resources.csv
               N
               instance_deploy.csv
               iterference_cnt
               app_interference.csv
            judge framework
        '''
        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):
            # 新添加98列CPU限制
            # self.appResources["cpu_" + str(i)] = None
            # self.appResources["mem_" + str(i)] = None
            # 赋值
            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.n, col = self.appResources.shape  # 9338*201 201列：appid,cpu_1,cpu_2,...mem_1,men_2....,P,M,PM

        # 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")

        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.machineResourcesUsed = self.machineResources.copy()

        for i in range(200):
            self.machineResourcesUsed.iloc[:, i + 1] = 0

        # 3.instance_deploy
        self.inst2Machine = pd.read_csv(instance_deploy, header=None,
                                        names=list(["instanceid", "appid", "machineid"]), encoding="utf-8")
        # 增加一个字段标注是否部署
        self.inst2Machine["isdeploy"] = False
        # # 4.app_interference 冲突表
        self.appIndexference = pd.read_csv(app_interference, header=None,
                                           names=list(["appid1", "appid2", "max_interference"]), encoding="utf-8")

        # instance按照磁盘消耗排序
        self.n, col = self.appResources.shape
        self.N, col = self.machineResources.shape
        self.m, col = self.inst2Machine.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
        '''
        libs.save_conf.write()

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

    def pickInstance(self, instanceid):
        '''
        先将instance从部署的主机中删除,删除一行，释放资源
        :return:
        '''
        self.inst2Machine.pop(instanceid)

    def toMachine(self, instanceid, machineid, doCheck=True):
        '''
        检查互斥条件,然后把instance放入主机
        :param instanceid: 实例id
        :param machineid: 主机id
        :param doCheck: 是否检测资源限制
        :return: True和False
        '''
        appid = self.inst2Machine[self.inst2Machine["instanceid"] == instanceid]["appid"].values[0]
        if doCheck:
            # 检查互斥


            # 检查资源限制
            for i in range(self.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
        # instance占用资源
        for i in range(self.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):
        '''

        :return:
        '''
        # 已经部署的instance
        deployed_Instance = self.inst2Machine.loc[pd.isna(self.inst2Machine["machineid"]) == False]
        count_deployed_Instance, col = deployed_Instance.shape
        deployed_Instance = deployed_Instance.reset_index(drop=True)

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

        # 先对已经部署的主机列表按照资源消耗进行排序

        # 先使用大主机，磁盘优先计算限制条件
        row1, col = self.inst2Machine.shape
        while row1 > 0:
            # 每部署一次，消耗一个主机
            for row2 in self.inst2Machine.itertuples():
                if row2.
                self.toMachine(row2)

            # 筛选未部署的
            self.inst2Machine = self.inst2Machine[self.inst2Machine["isdeploy"] == False]
            row, col = self.inst2Machine.shape
            self.inst2Machine = self.inst2Machine.reset_index(drop=True)
            j = j + 1

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

    def deployInstance(self):
        '''
        部署逻辑
        :return:
        '''
        pass


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