7 from util.geniserver import *
8 from util.cert import *
9 from util.trustedroot import *
10 from util.excep import *
11 from util.misc import *
12 from util.config import Config
14 class Aggregate(GeniServer):
17 components_file = None
28 # Create a new aggregate object.
30 # @param ip the ip address to listen on
31 # @param port the port to listen on
32 # @param key_file private key filename of registry
33 # @param cert_file certificate filename containing public key (could be a GID file)
35 def __init__(self, ip, port, key_file, cert_file, config = "/usr/share/geniwrapper/util/geni_config"):
36 GeniServer.__init__(ip, port, keyfile, cert_file)
39 basedir = conf.GENI_BASE_DIR + os.sep
40 server_basedir = basedir + os.sep + "plc" + os.sep
41 self.hrn = conf.GENI_INTERFACE_HRN
42 self.components_file = os.sep.join([server_basedir, 'components', hrn + '.comp'])
43 self.whitelist_file = os.sep.join([server_basedir, 'policy', 'whitelist'])
44 self.blacklist_file = os.sep.join([server_basedir, 'policy', 'blacklist'])
45 self.timestamp_file = os.sep.join([server_basedir, 'components', hrn + '.timestamp'])
46 self.components_ttl = components_ttl
47 self.policy['whitelist'] = []
48 self.policy['blacklist'] = []
53 Connect to the plc api interface. First attempt to impor thte shell, if that fails
54 try to connect to the xmlrpc server.
56 self.auth = {'Username': conf.GENI_PLC_USER,
57 'AuthMethod': 'password',
58 'AuthString': conf.GENI_PLC_PASSWORD}
61 # try to import PLC.Shell directly
62 sys.path.append(conf.GENI_PLC_SHELL_PATH)
64 self.shell = PLC.Shell.Shell(globals())
65 self.shell.AuthCheck()
67 # connect to plc api via xmlrpc
68 plc_host = conf.GENI_PLC_HOST
69 plc_port = conf.GENI_PLC_PORT
70 plc_api_path = conf.GENI_PLC_API_PATH
71 url = "https://%(plc_host)s:%(plc_port)s/%(plc_api_path)s/" % locals()
72 self.auth = {'Username': conf.GENI_PLC_USER,
73 'AuthMethod': 'password',
74 'AuthString': conf.GENI_PLC_PASSWORD}
76 self.shell = xmlrpclib.Server(url, verbose = 0, allow_none = True)
77 self.shell.AuthCheck(self.auth)
79 def hostname_to_hrn(self, login_base, hostname):
81 Convert hrn to plantelab name.
83 genihostname = "_".join(hostname.split("."))
84 return ".".join([self.hrn, login_base, genihostname])
86 def slicename_to_hrn(self, slicename):
88 Convert hrn to planetlab name.
90 slicename = slicename.replace("_", ".")
91 return ".".join([self.hrn, slicename])
93 def refresh_components(self):
95 Update the cached list of nodes.
98 # resolve component hostnames
99 nodes = self.shell.GetNodes(self.auth, {}, ['hostname', 'site_id'])
101 # resolve site login_bases
102 site_ids = [node['site_id'] for node in nodes]
103 sites = self.shell.GetSites(self.auth, site_ids, ['site_id', 'login_base'])
106 site_dict[site['site_id']] = site['login_base']
108 # convert plc names to geni hrn
109 self.components = [self.hostname_to_hrn(site_dict[node['site_id']], node['hostname']) for node in nodes]
111 # apply policy. Do not allow nodes found in blacklist, only allow nodes found in whitelist
112 whitelist_policy = lambda node: node in self.policy['whitelist']
113 blacklist_policy = lambda node: node not in self.policy['blacklist']
115 if self.policy['blacklist']:
116 self.components = blacklist_policy(self.components)
117 if self.policy['whitelist']:
118 self.components = whitelist_policy(self.components)
120 # update timestamp and threshold
121 self.timestamp = datetime.datetime.now()
122 delta = datetime.timedelta(hours=self.components_ttl)
123 self.threshold = self.timestamp + delta
125 f = open(self.components_file, 'w')
126 f.write(str(self.components))
128 f = open(self.timestamp_file, 'w')
129 f.write(str(self.threshold))
132 def load_components(self):
134 Read cached list of nodes.
136 print "loading components"
137 # Read component list from cached file
138 if os.path.exists(self.components_file):
139 f = open(self.components_file, 'r')
140 self.components = eval(f.read())
143 time_format = "%Y-%m-%d %H:%M:%S"
144 if os.path.exists(self.timestamp_file):
145 f = open(self.timestamp_file, 'r')
146 timestamp = str(f.read()).split(".")[0]
147 self.timestamp = datetime.datetime.fromtimestamp(time.mktime(time.strptime(timestamp, time_format)))
148 delta = datetime.timedelta(hours=self.components_ttl)
149 self.threshold = self.timestamp + delta
152 def load_policy(self):
154 Read the list of blacklisted and whitelisted nodes.
158 if os.path.exists(self.whitelist_file):
159 f = open(self.whitelist_file, 'r')
160 lines = f.readlines()
163 line = line.strip().replace(" ", "").replace("\n", "")
164 whitelist.extend(line.split(","))
167 if os.path.exists(self.blacklist_file):
168 f = open(self.blacklist_file, 'r')
169 lines = f.readlines()
172 line = line.strip().replace(" ", "").replace("\n", "")
173 blacklist.extend(line.split(","))
175 self.policy['whitelist'] = whitelist
176 self.policy['blacklist'] = blacklist
178 def get_components(self):
180 Return a list of components at this aggregate.
182 # Reload components list
183 now = datetime.datetime.now()
184 #self.load_components()
185 if not self.threshold or not self.timestamp or now > self.threshold:
186 self.refresh_components()
187 elif now < self.threshold and not self.components:
188 self.load_components()
189 return self.components
191 def get_rspec(self, hrn, type):
193 #rspec['nodespec'] = {'name': self.conf.GENI_INTERFACE_HRN}
194 #rsepc['nodespec']['nodes'] = []
196 nodes = self.shell.GetNodes(self.auth)
197 elif type in ['slice']:
198 slicename = hrn_to_pl_slicename(hrn)
199 slices = self.shell.GetSlices(self.auth, [slicename])
200 node_ids = slices[0]['node_ids']
201 nodes = self.shell.GetNodes(self.auth, node_ids)
203 nodespec = {'name': node['hostname'], 'type': 'std'}
204 # rspec['nodespec']['nodes'].append(nodespec)
206 elif type in ['aggregate']:
211 def get_resources(self, slice_hrn):
213 Return the current rspec for the specified slice.
215 slicename = hrn_to_plcslicename(slice_hrn)
216 rspec = self.get_rspec(slicenamem, 'slice')
220 def create_slice(self, slice_hrn, rspec, attributes):
222 Instantiate the specified slice according to whats defined in the rspec.
224 slicename = self.hrn_to_plcslicename(slice_hrn)
226 #nodespec = spec['networks']['nodes']
227 #nodes = [nspec['name'] for nspec in nodespec]
228 #self.shell.AddSliceToNodes(self.auth, slicename, nodes)
229 #for attribute in attributes:
230 #type, value, node, nodegroup = attribute['type'], attribute['value'], attribute['node'], attribute['nodegroup']
231 #shell.AddSliceAttribute(self.auth, slicename, type, value, node, nodegroup)
235 def update_slice(self, slice_hrn, rspec, attributes):
237 Update the specified slice.
240 slicename = self.hrn_to_plcslicename(slice_hrn)
241 slices = self.shell.GetSlices(self.auth, [slicename], ['node_ids'])
243 raise RecordNotFound(slice_hrn)
246 # find out where this slice is currently running
247 nodes = self.shell.GetNodes(self.auth, slice['node_ids'], ['hostname'])
248 hostnames = [node['hostname'] for node in nodes]
250 # get netspec details
252 #nodespec = spec['networks']['nodes']
253 #nodes = [nspec['name'] for nspec in nodespec]
255 # remove nodes not in rspec
256 #delete_nodes = set(hostnames).difference(nodes)
257 # add nodes from rspec
258 #added_nodes = set(nodes).difference(hostnames)
260 #shell.AddSliceToNodes(self.auth, slicename, added_nodes)
261 #shell.DeleteSliceFromNodes(self.auth, slicename, deleted_nodes)
263 #for attribute in attributes:
264 #type, value, node, nodegroup = attribute['type'], attribute['value'], attribute['node'], attribute['nodegroup']
265 #shell.AddSliceAttribute(self.auth, slicename, type, value, node, nodegroup)
267 def delete_slice_(self, slice_hrn):
269 Remove this slice from all components it was previouly associated with and
270 free up the resources it was using.
272 slicename = self.hrn_to_plcslicename(slice_hrn)
273 slices = shell.GetSlices(self.auth, [slicename])
275 raise RecordNotFound(slice_hrn)
278 shell.DeleteSliceFromNodes(self.auth, slicename, slice['node_ids'])
281 def start_slice(self, slice_hrn):
283 Stop the slice at plc.
285 slicename = hrn_to_plcslicename(slice_hrn)
286 slices = self.shell.GetSlices(self.auth, {'name': slicename}, ['slice_id'])
288 raise RecordNotFound(slice_hrn)
290 atrribtes = self.shell.GetSliceAttributes({'slice_id': slice_id, 'name': 'enabled'}, ['slice_attribute_id'])
291 attribute_id = attreibutes[0]
292 self.shell.UpdateSliceAttribute(self.auth, attribute_id, "1" )
295 def stop_slice(self, slice_hrn):
297 Stop the slice at plc
299 slicename = hrn_to_plcslicename(slice_hrn)
300 slices = self.shell.GetSlices(self.auth, {'name': slicename}, ['slice_id'])
302 raise RecordNotFound(slice_hrn)
304 atrribtes = self.shell.GetSliceAttributes({'slice_id': slice_id, 'name': 'enabled'}, ['slice_attribute_id'])
305 attribute_id = attreibutes[0]
306 self.shell.UpdateSliceAttribute(self.auth, attribute_id, "0")
310 def reset_slice(self, slice_hrn):
314 slicename = self.hrn_to_plcslicename(slice_hrn)
317 def get_policy(self):
319 Return this aggregates policy.
326 ##############################
327 ## Server methods here for now
328 ##############################
331 return self..get_components()
334 # return self.get_slices()
336 def resources(self, cred, hrn):
337 self.decode_authentication(cred, 'info')
338 self.verify_object_belongs_to_me(hrn)
340 return self.get_resources(hrn)
342 def create(self, cred, hrn, rspec):
343 self.decode_authentication(cred, 'embed')
344 self.verify_object_belongs_to_me(hrn)
345 return self.create(hrn)
347 def update(self, cred, hrn, rspec):
348 self.decode_authentication(cred, 'embed')
349 self.verify_object_belongs_to_me(hrn)
350 return self.update(hrn)
352 def delete(self, cred, hrn):
353 self.decode_authentication(cred, 'embed')
354 self.verify_object_belongs_to_me(hrn)
355 return self.delete_slice(hrn)
357 def start(self, cred, hrn):
358 self.decode_authentication(cred, 'control')
359 return self.start(hrn)
361 def stop(self, cred, hrn):
362 self.decode_authentication(cred, 'control')
363 return self.stop(hrn)
365 def reset(self, cred, hrn):
366 self.decode_authentication(cred, 'control')
367 return self.reset(hrn)
369 def policy(self, cred):
370 self.decode_authentication(cred, 'info')
371 return self.get_policy()
373 def register_functions(self):
374 GeniServer.register_functions(self)
376 # Aggregate interface methods
377 self.server.register_function(self.components)
378 #self.server.register_function(self.slices)
379 self.server.register_function(self.resources)
380 self.server.register_function(self.create)
381 self.server.register_function(self.delete)
382 self.server.register_function(self.start)
383 self.server.register_function(self.stop)
384 self.server.register_function(self.reset)
385 self.server.register_function(self.policy)