Skip to content

C
CONF
Commit 9bbf46f7 authored by Spiros Koulouzis's avatar Spiros Koulouzis
Browse files
Options

Cleanup

parent e56b5026
Show whitespace changes
Inline Side-by-side
Showing with 0 additions and 690 deletions
drip-deployer/cluster_file deleted 100644 → 0
View file @ e56b5026
54.236.37.228 ubuntu /tmp/1 master
54.236.8.78 ubuntu /tmp/2 slave
drip-planner2provisioner/input.json deleted 100644 → 0
View file @ e56b5026
{
"creationDate": 1488459287833,
"parameters": [
{
"url": null,
"attributes": null,
"value": [
{
"name": "2d13d708e3a9441ab8336ce874e08dd1",
"size": "Small",
"docker": "mogswitch/InputDistributor"
},
{
"name": "8fcc1788d9ee462c826572c79fdb2a6a",
"size": "Small",
"docker": "mogswitch/InputDistributor"
},
{
"name": "5e0add703c8a43938a39301f572e46c0",
"size": "Small",
"docker": "mogswitch/InputDistributor"
}
],
"encoding": "UTF-8"
}
]
}
\ No newline at end of file
drip-simple_planner/ICPCP.py deleted 100644 → 0
View file @ e56b5026
'''
Created on Nov 20, 2015
@author: junchao
'''
import os
import sys
import re
import random
import networkx as nx
import numpy as np
import json
from subprocess import call
import time
from NewInstance import NewInstance
import collections
class Workflow():
def add_entry_exit(self, g):
# add entry node to the graph
g.add_node(0, est = -1, eft = -1, lft =-1)
for vertex in g:
if len(g.predecessors(vertex)) == 0 and not vertex == 0:
self.G.add_weighted_edges_from([(0, vertex, 0)])
# add exit node to the graph
g.add_node(self.vertex_num-1, est = -1, eft = -1, lft =-1)
startnodes = []
for vertex in g:
if len(g.successors(vertex)) == 0 and not vertex == self.vertex_num-1:
startnodes.append(vertex)
for node in startnodes:
self.G.add_weighted_edges_from([(node, self.vertex_num-1, 0)])
def init(self, content):
self.G = nx.DiGraph()
self.vertex_num =0
for (key, value) in content['workflow'].items():
if isinstance(value, list) :
if key == 'nodes' :
self.vertex_num = len(value)
for node in value:
self.G.add_node(node['name'], est = -1, eft = -1, lft =-1)
if key == 'links' :
for link in value:
self.G.add_weighted_edges_from([(link['source'], link['target'], link['weight'])])
print self.G.nodes
#parse the performance matrix
p = []
od = collections.OrderedDict(sorted(content['performance'].items()))
for (key, value) in od.items():
row = []
row.append(0)
for i in value.split(','):
row.append(int(i))
row.append(0)
print row
p.append(row)
self.p_table = np.matrix(p)
#parse the price vector
self.vm_price = []
for i in content['price'].split(','):
self.vm_price.append(int(i))
#parse the deadline
self.d_list = []
for (key, value) in content['deadline'].items():
self.d_list.append([int(key), int(value)])
self.d_table = np.matrix(self.d_list)
self.successful = 0
self.vertex_num += 2
self.add_entry_exit(self.G)
#test whether the DAG contains cycles
if len(list(nx.simple_cycles(self.G))) > 0:
print "the DAG contains cycles"
sys.exit()
self.assigned_list = [-1]*(self.vertex_num)
self.instances = []
self.G.node[0]['est'] = 0
self.G.node[0]['eft'] = 0
self.cal_est(0)
self.G.node[self.vertex_num-1]['lft'] = self.d_table[self.d_table.shape[0]-1,1]#self.p_table[0, child], self.p_table.shape[0]
self.cal_lft(self.vertex_num-1)
def init1(self, workflow_file_name, performance_file_name, price_file_name, deadline_file_name):
#Initialization
self.G=nx.DiGraph()
self.vertex_num = 0
self.successful = 0
#Read the workflow information
graph = pydot.graph_from_dot_file(workflow_file_name)
nx_graph = nx.from_pydot(graph)
self.G=nx.DiGraph()
for node in nx_graph:
#print node
self.G.add_node(int(node)+1, est = -1, eft = -1, lft =-1)
self.vertex_num += 1
#print nx_graph.edge
#print workflow_file_name
for link in nx_graph.edges_iter():
#print link[0], link[1]
self.G.add_weighted_edges_from([(int(link[0])+1, int(link[1])+1, int(float(nx_graph[link[0]][link[1]][0]['weight'])))])
self.vertex_num += 2
self.add_entry_exit(self.G)
#test whether the DAG contains cycles
if len(list(nx.simple_cycles(self.G))) > 0:
print "the DAG contains cycles"
sys.exit()
#read performance table
l = [ map(int,line.split(',')) for line in open(performance_file_name, 'r')]
#append the entry and exit node information
for row in l:
row.insert(0, 0)
row.insert(len(row),0)
self.p_table = np.matrix(l)
self.assigned_list = [-1]*(self.vertex_num)
self.vm_price = map(int,open(price_file_name,'r').readline().split(','))
self.instances = []
self.d_list = [ map(int,line.split('\t')) for line in open(deadline_file_name, 'r')]
tmpList = self.d_list[len(self.d_list)-1]
self.d_table = np.matrix(tmpList)
#deadline = open(deadline_file_name, 'r').readline()
self.G.node[0]['est'] = 0
self.G.node[0]['eft'] = 0
self.cal_est(0)
self.G.node[self.vertex_num-1]['lft'] = self.d_table[self.d_table.shape[0]-1,1]#self.p_table[0, child], self.p_table.shape[0]
self.cal_lft(self.vertex_num-1)
#The following two functions are to initialize the EST, EFT and LFT
#calculate the earliest start time and earliest finish time
def cal_est(self, i):
for child in self.G.successors(i):
est = self.G.node[i]['eft']+self.G[i][child]['weight']
if est>self.G.node[child]['est']:
self.G.node[child]['est'] = est
self.G.node[child]['eft'] = est + self.p_table[0, child]
self.cal_est(child)
def cal_lft(self, d):
for parent in self.G.predecessors(d):
lft = self.G.node[d]['lft'] - self.p_table[0, d] - self.G[parent][d]['weight']
d_table_list = []
for deadline in self.d_table:
d_table_list.append(deadline[0, 0])
if parent in d_table_list:
deadline = self.d_table[d_table_list.index([parent]),1]
if deadline < lft:
lft = deadline
if self.G.node[parent]['lft'] == -1 or lft<self.G.node[parent]['lft']:
# parent may not finish later
self.G.node[parent]['lft'] = lft
#print "call graphAssignLFT(",graph.node[parent]['name'],")"
self.cal_lft(parent)
#Finding critical path
def ic_pcp(self):
self.assigned_list[0] = 0
self.assigned_list[self.vertex_num-1] = 0
self.assign_parents(self.vertex_num-1)
def has_unassigned_parent(self, i):
for parent in self.G.predecessors(i):
if (self.assigned_list[parent] == -1):
return True
return False
def assign_parents(self, i):
while (self.has_unassigned_parent(i)):
if self.successful == 1: #resources cannot be met
break
pcp = []
self.find_critical_path(i, pcp)
assigned_vms = self.assign_path(pcp)
if -1 in assigned_vms:
print 'resource cannot be met'
break
self.G.node[pcp[len(pcp)-1]]['eft'] = self.G.node[pcp[len(pcp)-1]]['est'] + self.p_table[assigned_vms[len(pcp)-1], pcp[len(pcp)-1]]
self.update_est(pcp[len(pcp)-1], pcp, assigned_vms)
self.update_lft(pcp[0], pcp, assigned_vms)
#split according to the types of assigned VMs and add it to the new instance
ni = NewInstance(assigned_vms, self.G.node[pcp[len(pcp)-1]]['est'], self.G.node[pcp[0]]['eft'], pcp)
for j in xrange(len(pcp)):
ni.cost = ni.cost + self.vm_price[assigned_vms[j]]
self.instances.append(ni)
for j in reversed(pcp): #also in the paper they didn't mention the order
self.assign_parents(j)
#TODO: A very tricky thing on updating the EST and EFT.
def update_est(self, i, pcp, assigned_vms):
for child in self.G.successors(i):
if child not in pcp:
est = self.G.node[i]['eft']+self.G[i][child]['weight']
if self.assigned_list[i] == -1:
eft = est + self.p_table[0, child]
else:
eft = est + self.p_table[self.assigned_list[child], child]
else:
if pcp.index(child) == len(pcp) - 1:
est = self.G.node[i]['eft']
eft = est + self.p_table[self.assigned_list[child], child]
else:
pos = pcp.index(child)
est = self.G.node[i]['eft'] + self.G[pcp[pos+1]][pcp[pos]]['weight']
eft = est + self.p_table[self.assigned_list[child], child]
#decide whether the assignment will violate other parent data dependency
all_smaller = True
for parent in self.G.predecessors(child):
if not parent == i:
if self.G.node[parent]['eft'] + self.G[parent][child]['weight'] > est:
all_smaller = False
if all_smaller:
self.G.node[child]['est'] = est
self.G.node[child]['eft'] = eft
self.update_est(child, pcp, assigned_vms)
def update_lft(self, d, pcp, assigned_vms):
for parent in self.G.predecessors(d):
if parent not in pcp:
if self.assigned_list[d] == -1:
lft = self.G.node[d]['lft'] - self.p_table[0, d] - self.G[parent][d]['weight']
else:
lft = self.G.node[d]['lft'] - self.p_table[self.assigned_list[d], d] - self.G[parent][d]['weight']
else:
pos = pcp.index(parent)
#if pos < len(pcp) -1:
if assigned_vms[pos] == assigned_vms[pos-1]:#9, 6, 2
lft = self.G.node[d]['lft'] - self.p_table[self.assigned_list[d], d]
else:
lft = self.G.node[d]['lft'] - self.p_table[self.assigned_list[d], d] - self.G[pcp[pos]][pcp[pos-1]]['weight']
if lft < self.G.node[parent]['lft']:
self.G.node[parent]['lft'] = lft
self.update_lft(parent, pcp, assigned_vms)
def find_critical_path(self, i, pcp):
cal_cost = 0
critical_parent = -1
for n in self.G.predecessors(i):
if self.assigned_list[n] == -1: #parent of node i is not assigned
if self.G.node[n]['eft'] + self.G.edge[n][i]['weight'] > cal_cost:
cal_cost = self.G.node[n]['eft'] + self.G.edge[n][i]['weight']
critical_parent = n
if not critical_parent == -1:
pcp.append(critical_parent)
self.find_critical_path(critical_parent, pcp)
def exec_time_sum(self, pcp, vm_type):
sum = 0
for i in pcp:
sum += self.p_table[vm_type, i]
return sum
#look forward one step when assigning a vm to a pcp how the est varies
def est_vary(self, pcp, d):
head_pcp = pcp[len(pcp)-1]
original_est = self.G.node[head_pcp]['est']
biggest_est = -1
biggest_parent = -1
for parent in self.G.predecessors(head_pcp):
if parent == d:
est = self.G.node[parent]['eft']
else:
est = self.G.node[parent]['eft'] + self.G[parent][head_pcp]['weight']
if biggest_est < est:
biggest_est = est
biggest_parent = parent
return original_est-biggest_est
#choose the best existing available instance for the pcp
def choose_exist_instance(self, pcp):
best_vm = None
best_exec_time = -1
best_vary_time = -1
for vm in self.instances:
head_pcp = pcp[len(pcp)-1]
for parent in self.G.predecessors(head_pcp):
head_exist_pcp = vm.task_list[0] #The last node of the previous critical path
if parent == head_exist_pcp:
if best_vm == None:
best_vm = vm
exec_time = self.exec_time_sum(pcp, vm.vm_type)
best_exec_time = exec_time
best_vary_time = self.est_vary(pcp, head_exist_pcp)
else:
best_vm_head = vm.task_list[0]
# if assigned to the vm, what will the est be
exec_time = self.exec_time_sum(pcp, vm.vm_type)
# calculate the lft
varied_time = self.G.node[head_pcp]['est']-self.est_vary(pcp, head_exist_pcp)
lft = varied_time+exec_time
if (exec_time - self.est_vary(pcp, head_exist_pcp))*self.vm_price[vm.vm_type]> \
(best_exec_time - self.est_vary(pcp, best_vm.task_list[0]))*self.vm_price[best_vm.vm_type] \
and lft < self.G.node[head_pcp]['lft']: #also should not violate the lft
best_vm = vm
best_exec_time = exec_time
best_vary_time = varied_time
if not best_vm == None:
best_vm.vm_end = self.G.node[pcp[len(pcp)-1]]['est']-best_vary_time+best_exec_time
return best_vm
def assign_path(self, pcp):
cheapest_vm = -1
cheapest_sum = 9999999 #the initialized value should be a very large number
for i in xrange(self.p_table.shape[0]): # use the the shape of the performance table to identify how many VM types are there
violate_LFT = 0
start = self.G.node[pcp[len(pcp)-1]]['est']
cost_sum = 0
for j in xrange(len(pcp)-1, -1, -1):
cost_sum += self.vm_price[i]
if j == len(pcp)-1:
start = start + self.p_table[i, pcp[j]]
else:
start = start + self.p_table[i, pcp[j]]+ self.G[pcp[j+1]][pcp[j]]['weight']
if self.G.node[pcp[j]]['lft'] < start:
violate_LFT = 1
#launch a new instance of the cheapest service which can finish each task of P before its LFT
if violate_LFT == 0 and cost_sum < cheapest_sum:
cheapest_vm = i
cheapest_sum = cost_sum
for i in xrange(len(pcp)):
self.assigned_list[pcp[i]] = cheapest_vm
return [cheapest_vm]*len(pcp)
def generate_string(self, node):
s = "name "+str(node)+"\n"
for i in xrange(len(self.vm_price)):
s = s+str(self.p_table[i, node])+"\n"
s = s+"assigned vm: " + str(self.assigned_list[node]+1)
return s
def generateJSON(self):
content = {}
for i in xrange(1, self.vertex_num-1, 1):
if self.assigned_list[i] == 0:
content[str(i)] = 'large'
if self.assigned_list[i] == 1:
content[str(i)] = 'Medium'
if self.assigned_list[i] == 2:
content[str(i)] = 'Small'
return content
#calculate the total execution cost
def cal_cost(self):
cost = 0
for vm in self.instances:
cost = cost + vm.cost
return cost
def cal_cost1(self):
cost = 0
for i in range(1, len(self.assigned_list)-1):
cost += self.vm_price[self.assigned_list[i]]
return cost
def has_edge_vm(self, vm1, vm2):
for node1 in vm1.task_list:
for node2 in vm2.task_list:
if self.G.has_edge(node1, node2) or self.G.has_edge(node2, node1):
return True
return False
drip-simple_planner/NewInstance.py deleted 100644 → 0
View file @ e56b5026
class NewInstance(object):
def __init__(self, vm_type, vm_start, vm_end, pcp):
self.vm_type = vm_type
self.vm_start = vm_start
self.vm_end = vm_end
self.task_list = pcp
self.cost = 0
\ No newline at end of file
drip-simple_planner/rpc_server.py deleted 100644 → 0
View file @ e56b5026
#!/usr/bin/env python
import pika
import networkx as nx
import sys
import numpy as np
import sys, argparse
import operator
import os
from toscaparser import *
from toscaparser.tosca_template import ToscaTemplate
import re
import getopt
from ICPCP import Workflow
import random
import time
import json
connection = pika.BlockingConnection(pika.ConnectionParameters(host='172.17.0.2'))
channel = connection.channel()
channel.queue_declare(queue='planner_queue')
def handleDelivery(message):
parsed_json = json.loads(message)
params = parsed_json["parameters"]
for param in params:
name = param["name"]
value = param["value"]
def on_request(ch, method, props, body):
handleDelivery(body)
print(" Message %s" % body)
response = "AAAAAAAAAAAAAAAAAAAAAA"
json1 = response.get('parameters')[0].get('value').get('topology_template').get('node_templates')
deadline = 0
for j in json1:
#print json[j]
if not json1[j]['type'] == "Switch.nodes.Application.Connection":
deadline = int(re.search(r'\d+', json1[j]['properties']['QoS']['response_time']).group())
#get the nodes from the json
nodeDic = {}
nodeDic1 = {}
i = 1
for j in json1:
if not json1[j]['type'] == "Switch.nodes.Application.Connection":
print j, json1[j]
nodeDic[j] = i
nodeDic1[i] = j
i = i + 1
#get the links from the json
links = []
for j in json1:
if json1[j]['type'] == "Switch.nodes.Application.Connection":
print json1[j]['properties']['source']['component_name']
print json1[j]['properties']['target']['component_name']
link= {}
link['source'] = nodeDic[json1[j]['properties']['source']['component_name']]
link['target'] = nodeDic[json1[j]['properties']['target']['component_name']]
link['weight'] = random.randint(1, 10)
links.append(link)
# compose the json as input of the workflow
wfJson = {}
wfJson['workflow'] = {}
nodesList = []
sorted_nodeDic = sorted(nodeDic.items(), key=operator.itemgetter(1))
for key, value in sorted_nodeDic:
v = {}
v['name'] = value
nodesList.append(v)
wfJson['workflow']['nodes'] = nodesList
wfJson['workflow']['links'] = links
#print deadline
wfJson['price'] = "5,2,1"
wfJson['deadline'] = {'2': deadline}
#generate performance
performance = {}
for key, value in sorted_nodeDic:
performance[str(value)] = "1,2,3"
wfJson['performance'] = performance
print wfJson
#send request to the server
start = time.time()
wf = Workflow()
wf.init(wfJson)
wf.ic_pcp()
#print content['workflow']
#return
res = wf.generateJSON()
end = time.time()
print (end - start)
# generate the json files in the corresponding format as the
outcontent = {}
outcontent["creationDate"] = 1487002029722
outcontent["parameters"] = []
par1 = {}
par1["url"] = "null"
par1["encoding"] = "UTF-8"
par1["value"] = res
par1["attributes"] = "null"
outcontent["parameters"].append(par1)
response = outcontent
ch.basic_publish(exchange='',
routing_key=props.reply_to,
properties=pika.BasicProperties(correlation_id = \
props.correlation_id),
body=str(response))
ch.basic_ack(delivery_tag = method.delivery_tag)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(on_request, queue='planner_queue')
print(" [x] Awaiting RPC requests")
channel.start_consuming()
\ No newline at end of file
drip-simple_planner/server.py deleted 100644 → 0
View file @ e56b5026
#https://github.com/skoulouzis/DRIP/blob/package/doc/json_samples/kbExampleMessage.json
import networkx as nx
import sys
import numpy as np
import sys, argparse
import operator
import os
from toscaparser import *
from toscaparser.tosca_template import ToscaTemplate
import re
import getopt
from ICPCP import Workflow
import random
import time
import json
def main(argv):
workflow_file = ""
try:
opts, args = getopt.getopt(argv,"hw:s:",["workflow=", "SDI="])
except getopt.GetoptError:
print 'server.py -w <workflowfile> -s <SDI>'
sys.exit(2)
for opt, arg in opts:
if opt == '-h':
print 'server.py -w <workflowfile> -s <SDI>'
sys.exit()
elif opt in ("-w", "--workflow"):
workflow_file = arg
elif opt in ("-s", "--SDI"):
SDI_file = arg
data = {}
print workflow_file
with open(workflow_file) as data_file:
data = json.load(data_file)
#print data
#path = "input.yaml"
'''
a_file = os.path.isfile(path)
tosca = ToscaTemplate(path)
#print tosca.tpl
json = tosca.tpl.get('topology_template').get('node_templates')
#print json
'''
json1 = data.get('parameters')[0].get('value').get('topology_template').get('node_templates')
deadline = 0
for j in json1:
#print json[j]
if not json1[j]['type'] == "Switch.nodes.Application.Connection":
deadline = int(re.search(r'\d+', json1[j]['properties']['QoS']['response_time']).group())
#get the nodes from the json
nodeDic = {}
nodeDic1 = {}
i = 1
for j in json1:
if not json1[j]['type'] == "Switch.nodes.Application.Connection":
print j, json1[j]
nodeDic[j] = i
nodeDic1[i] = j
i = i + 1
#get the links from the json
links = []
for j in json1:
if json1[j]['type'] == "Switch.nodes.Application.Connection":
print json1[j]['properties']['source']['component_name']
print json1[j]['properties']['target']['component_name']
link= {}
link['source'] = nodeDic[json1[j]['properties']['source']['component_name']]
link['target'] = nodeDic[json1[j]['properties']['target']['component_name']]
link['weight'] = random.randint(1, 10)
links.append(link)
# compose the json as input of the workflow
wfJson = {}
wfJson['workflow'] = {}
nodesList = []
sorted_nodeDic = sorted(nodeDic.items(), key=operator.itemgetter(1))
for key, value in sorted_nodeDic:
v = {}
v['name'] = value
nodesList.append(v)
wfJson['workflow']['nodes'] = nodesList
wfJson['workflow']['links'] = links
#print deadline
wfJson['price'] = "5,2,1"
wfJson['deadline'] = {'2': deadline}
#generate performance
performance = {}
for key, value in sorted_nodeDic:
performance[str(value)] = "1,2,3"
wfJson['performance'] = performance
print wfJson
#send request to the server
start = time.time()
wf = Workflow()
wf.init(wfJson)
wf.ic_pcp()
#print content['workflow']
#return
res = wf.generateJSON()
end = time.time()
print (end - start)
# generate the json files in the corresponding format as the
outcontent = {}
outcontent["creationDate"] = 1487002029722
outcontent["parameters"] = []
par1 = {}
par1["url"] = "null"
par1["encoding"] = "UTF-8"
par1["value"] = res
par1["attributes"] = "null"
outcontent["parameters"].append(par1)
return outcontent
if __name__ == '__main__':
main(sys.argv[1:])
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment