comment out code that talks to registry until the credential issue is resolved
[sfa.git] / geni / aggregate.py
1 import os
2 import sys
3 import datetime
4 import time
5 import xmlrpclib
6
7 from geni.util.geniserver import GeniServer
8 from geni.util.geniclient import *
9 from geni.util.cert import Keypair, Certificate
10 from geni.util.credential import Credential
11 from geni.util.trustedroot import TrustedRootList
12 from geni.util.excep import *
13 from geni.util.misc import *
14 from geni.util.config import Config
15 from geni.util.rspec import Rspec
16 from geni.util.specdict import *
17 from geni.util.storage import SimpleStorage
18
19 class Aggregate(GeniServer):
20
21     hrn = None
22     nodes_ttl = None
23     nodes = None
24     slices = None 
25     policy = None
26     timestamp = None
27     threshold = None    
28     shell = None
29     registry = None
30     key_file = None
31     cert_file = None
32     credential = None
33   
34     ##
35     # Create a new aggregate object.
36     #
37     # @param ip the ip address to listen on
38     # @param port the port to listen on
39     # @param key_file private key filename of registry
40     # @param cert_file certificate filename containing public key (could be a GID file)     
41
42     def __init__(self, ip, port, key_file, cert_file, config = "/usr/share/geniwrapper/geni/util/geni_config"):
43         GeniServer.__init__(self, ip, port, key_file, cert_file)
44         self.key_file = key_file
45         self.cert_file = cert_file
46         self.config = Config(config)
47         self.basedir = self.config.GENI_BASE_DIR + os.sep
48         self.server_basedir = self.basedir + os.sep + "geni" + os.sep
49         self.hrn = self.config.GENI_INTERFACE_HRN
50         
51         nodes_file = os.sep.join([self.server_basedir, 'agg.' + self.hrn + '.components'])
52         self.nodes = SimpleStorage(nodes_file)
53         self.nodes.load()
54        
55         slices_file = os.sep.join([self.server_basedir, 'agg.' + self.hrn + '.slices'])
56         self.slices = SimpleStorage(slices_file)
57         self.slices.load()
58  
59         policy_file = os.sep.join([self.server_basedir, 'agg.' + self.hrn + '.policy'])
60         self.policy = SimpleStorage(policy_file, {'whitelist': [], 'blacklist': []})
61         self.policy.load()
62         
63         timestamp_file = os.sep.join([self.server_basedir, 'agg.' + self.hrn + '.timestamp']) 
64         self.timestamp = SimpleStorage(timestamp_file)
65
66         self.nodes_ttl = 1
67
68         self.connectPLC()
69         self.connectRegistry()
70         #self.loadCredential()
71
72     def connectRegistry(self):
73         """
74         Connect to the registry
75         """
76         # connect to registry using GeniClient
77         address = self.config.GENI_REGISTRY_HOSTNAME
78         port = self.config.GENI_REGISTRY_PORT
79         url = 'https://%(address)s:%(port)s' % locals()
80         self.registry = GeniClient(url, self.key_file, self.cert_file)
81
82     
83     def connectPLC(self):
84         """
85         Connect to the plc api interface. First attempt to impor thte shell, if that fails
86         try to connect to the xmlrpc server.
87         """
88         self.auth = {'Username': self.config.GENI_PLC_USER,
89                      'AuthMethod': 'password',
90                      'AuthString': self.config.GENI_PLC_PASSWORD}
91
92         try:
93            # try to import PLC.Shell directly
94             sys.path.append(self.config.GENI_PLC_SHELL_PATH) 
95             import PLC.Shell
96             self.shell = PLC.Shell.Shell(globals())
97             self.shell.AuthCheck()
98         except ImportError:
99             # connect to plc api via xmlrpc
100             plc_host = self.config.GENI_PLC_HOST
101             plc_port = self.config.GENI_PLC_PORT
102             plc_api_path = self.config.GENI_PLC_API_PATH                 
103             url = "https://%(plc_host)s:%(plc_port)s/%(plc_api_path)s/" % locals()
104             self.auth = {'Username': self.config.GENI_PLC_USER,
105                  'AuthMethod': 'password',
106                  'AuthString': self.config.GENI_PLC_PASSWORD} 
107
108             self.shell = xmlrpclib.Server(url, verbose = 0, allow_none = True) 
109             self.shell.AuthCheck(self.auth)
110
111     def loadCredential(self):
112         """
113         Attempt to load credential from file if it exists. If it doesnt get 
114         credential from registry.
115         """ 
116
117         self_cred_filename = self.server_basedir + os.sep + "agg." + self.hrn + ".cred"
118         ma_cred_filename = self.server_basedir + os.sep + "agg." + self.hrn + ".ma.cred"
119         
120         # see if this file exists
121         try:
122             cred = Credential(filename = ma_cred_filename, subject=self.hrn)
123             self.credential = cred.save_to_string()
124         except IOError:
125             # get self credential
126             self_cred = self.registry.get_credential(None, 'ma', self.hrn)
127             self_credential = Credential(string = self_cred)
128             self_credential.save_to_file(self_cred_filename)
129
130             # get ma credential
131             ma_cred = self.registry.get_credential(self_cred)
132             ma_credential = Credential(string = ma_cred)
133             ma_credential.save_to_file(ma_cred_filename)
134             self.credential = ma_cred
135
136     def hostname_to_hrn(self, login_base, hostname):
137         """
138         Convert hrn to plantelab name.
139         """
140         genihostname = "_".join(hostname.split("."))
141         return ".".join([self.hrn, login_base, genihostname])
142
143     def slicename_to_hrn(self, slicename):
144         """
145         Convert hrn to planetlab name.
146         """
147         slicename = slicename.replace("_", ".")
148         return ".".join([self.hrn, slicename])
149
150     def refresh_components(self):
151         """
152         Update the cached list of nodes and save in 4 differnt formats
153         (rspec, dns, ip, hrn)
154         """
155
156         node_details = {}
157         # get node list in rspec format
158         rspec = Rspec()
159         rspec.parseString(self.get_rspec(self.hrn, 'aggregate'))
160         # filter nodes according to policy
161         rspec.filter('NodeSpec', 'name', blacklist=self.policy['blacklist'], whitelist=self.policy['whitelist'])
162         # extract ifspec info to get ip's
163         ips = []
164         ifspecs = rspec.getDictsByTagName('IfSpec')
165         for ifspec in ifspecs:
166             if ifspec.has_key('addr') and ifspec['addr']:
167                 ips.append(ifspec['addr']) 
168
169         # resolve component hostnames 
170         nodes = self.shell.GetNodes(self.auth, {}, ['hostname', 'site_id'])
171     
172         # resolve site login_bases
173         site_ids = [node['site_id'] for node in nodes]
174         sites = self.shell.GetSites(self.auth, site_ids, ['site_id', 'login_base'])
175         site_dict = {}
176         for site in sites:
177             site_dict[site['site_id']] = site['login_base']
178
179         # filter nodes according to policy
180         # convert plc names to geni hrn
181         nodedict = {}
182         for node in nodes:
183             node_hrn = self.hostname_to_hrn(site_dict[node['site_id']], node['hostname'])
184             # apply policy. 
185             # Do not allow nodes found in blacklist, only allow nodes found in whitelist
186             if self.policy['whitelist'] and node_hrn not in self.polciy['whitelist']:
187                 continue
188             if self.policy['blacklist'] and node_hrn in self.policy['blacklist']:
189                 continue
190             nodedict[node_hrn] = node['hostname']
191
192         
193         node_details['rspec'] = rspec.toxml()
194         node_details['hrn'] = nodedict.keys()
195         node_details['dns'] = nodedict.values()
196         node_details['ip'] = ips
197         # save state 
198         self.nodes = SimpleStorage(self.nodes.db_filename, node_details)
199         self.nodes.write()
200
201         
202         # update timestamp and threshold
203         self.timestamp['timestamp'] =  datetime.datetime.now()
204         delta = datetime.timedelta(hours=self.nodes_ttl)
205         self.threshold = self.timestamp['timestamp'] + delta 
206         self.timestamp.write()        
207  
208     def load_components(self):
209         """
210         Read cached list of nodes.
211         """
212         # Read component list from cached file 
213         self.nodes.load()
214         self.timestamp.load() 
215         time_format = "%Y-%m-%d %H:%M:%S"
216         timestamp = self.timestamp['timestamp']
217         self.timestamp['timestamp'] = datetime.datetime.fromtimestamp(time.mktime(time.strptime(timestamp, time_format)))
218         delta = datetime.timedelta(hours=self.nodes_ttl)
219         self.threshold = self.timestamp['timestamp'] + delta
220
221     def load_policy(self):
222         """
223         Read the list of blacklisted and whitelisted nodes.
224         """
225         self.policy.load()
226
227
228     def getComponents(self, type = 'rspec'):
229         """
230         Return a list of components at this aggregate.
231         """
232         valid_types = ['rspec', 'hrn', 'dns', 'ip']
233         if type not in valid_types:
234             raise Exception, "Invalid type specified, must be one of the following: %s" \
235                              % ", ".join(valid_types)
236         
237         # Reload components list
238         now = datetime.datetime.now()
239         #self.load_components()
240         if not self.threshold or not self.timestamp['timestamp'] or now > self.threshold:
241             self.refresh_components()
242         elif now < self.threshold and not self.nodes.keys(): 
243             self.load_components()
244         return self.nodes[type]
245     
246     def getSlices(self, hrn):
247         """
248         Return a list of instnatiated managed by this slice manager.
249         """
250
251         # XX list only the slices at the specfied hrn
252         return dict(self.slices)
253  
254     def get_rspec(self, hrn, type):
255         """
256         Get resource information from PLC
257         """
258         
259         # Get the required nodes
260         if type in ['aggregate']:
261             nodes = self.shell.GetNodes(self.auth)
262         elif type in ['slice']:
263             slicename = hrn_to_pl_slicename(hrn)
264             slices = self.shell.GetSlices(self.auth, [slicename])
265             node_ids = slices[0]['node_ids']
266             nodes = self.shell.GetNodes(self.auth, node_ids) 
267         
268         # Get all network interfaces
269         interface_ids = []
270         for node in nodes:
271             interface_ids.extend(node['nodenetwork_ids'])
272         interfaces = self.shell.GetNodeNetworks(self.auth, interface_ids)
273         interface_dict = {}
274         for interface in interfaces:
275             interface_dict[interface['nodenetwork_id']] = interface
276         
277         # join nodes with thier interfaces
278         for node in nodes:
279             node['interfaces'] = []
280             for nodenetwork_id in node['nodenetwork_ids']:
281                 node['interfaces'].append(interface_dict[nodenetwork_id])
282
283         # convert and threshold to ints
284         if self.timestamp.has_key('timestamp') and self.timestamp['timestamp']:
285             timestamp = self.timestamp['timestamp']
286             threshold = self.threshold
287         else:
288             timestamp = datetime.datetime.now()
289             delta = datetime.timedelta(hours=self.nodes_ttl)
290             threshold = timestamp + delta        
291
292     
293         start_time = int(timestamp.strftime("%s"))
294         end_time = int(threshold.strftime("%s"))
295         duration = end_time - start_time
296
297         # create the plc dict
298         networks = [{'nodes': nodes, 'name': self.hrn, 'start_time': start_time, 'duration': duration}] 
299         resources = {'networks': networks, 'start_time': start_time, 'duration': duration}
300
301         # convert the plc dict to an rspec dict
302         resourceDict = RspecDict(resources)
303         # convert the rspec dict to xml
304         rspec = Rspec()
305         rspec.parseDict(resourceDict)
306         return rspec.toxml()
307
308     def getResources(self, slice_hrn):
309         """
310         Return the current rspec for the specified slice.
311         """
312         rspec = self.get_rspec(slice_hrn, 'slice')
313         
314         return rspec
315  
316
317     def createSlice(self, slice_hrn, rspec, attributes = []):
318         """
319         Instantiate the specified slice according to whats defined in the rspec.
320         """
321         
322         spec = Rspec(rspec)
323         # save slice state locally
324         # we can assume that spec object has been validated so its safer to
325         # save this instead of the unvalidated rspec the user gave us
326         self.slices[slice_hrn] = spec.toxml()
327         self.slices.write()
328         
329         # Get slice info
330         slicename = hrn_to_pl_slicename(slice_hrn)
331         slices = self.shell.GetSlices(self.auth, [slicename], ['node_ids'])
332         if not slices:
333             raise RecordNotFound(slice_hrn)
334         slice = slices[0]
335
336         # find out where this slice is currently running
337         nodes = self.shell.GetNodes(self.auth, slice['node_ids'], ['hostname'])
338         hostnames = [node['hostname'] for node in nodes]
339
340         # get netspec details
341         nodespecs = spec.getDictsByTagName('NodeSpec')
342         nodes = [nodespec['name'] for nodespec in nodespecs]    
343        
344         # save slice state locally
345         # we can assume that spec object has been validated so its safer to 
346         # save this instead of the unvalidated rspec the user gave us
347         self.slices[slice_hrn] = spec.toxml()
348         self.slices.write()
349
350         # remove nodes not in rspec
351         deleted_nodes = list(set(hostnames).difference(nodes))
352         # add nodes from rspec
353         added_nodes = list(set(nodes).difference(hostnames))
354     
355         self.shell.AddSliceToNodes(self.auth, slicename, added_nodes)
356         self.shell.DeleteSliceFromNodes(self.auth, slicename, deleted_nodes)
357
358         for attribute in attributes:
359             type, value, node, nodegroup = attribute['type'], attribute['value'], attribute['node'], attribute['nodegroup']
360             self.shell.AddSliceAttribute(self.auth, slicename, type, value, node, nodegroup)
361     
362         # contact registry to get slice users and add them to the slice
363         #slice_record = self.registry.resolve(self.credential, slice_hrn)
364         # persons = slice_record['users']
365         
366         #for person in persons:
367         #    shell.AddPersonToSlice(person['email'], slice_name)
368         return 1
369
370     def updateSlice(self, slice_hrn, rspec, attributes = []):
371         return self.create_slice(slice_hrn, rspec, attributes)
372          
373     def deleteSlice(self, slice_hrn):
374         """
375         Remove this slice from all components it was previouly associated with and 
376         free up the resources it was using.
377         """
378         if self.slices.has_key(slice_hrn):
379             self.slices.pop(slice_hrn)
380             self.slices.write()
381
382         slicename = hrn_to_pl_slicename(slice_hrn)
383         slices = self.shell.GetSlices(self.auth, [slicename])
384         if not slices:
385             return 1  
386         slice = slices[0]
387       
388         self.shell.DeleteSliceFromNodes(self.auth, slicename, slice['node_ids'])
389         return 1
390
391     def startSlice(self, slice_hrn):
392         """
393         Stop the slice at plc.
394         """
395         slicename = hrn_to_pl_slicename(slice_hrn)
396         slices = self.shell.GetSlices(self.auth, {'name': slicename}, ['slice_id'])
397         if not slices:
398             #raise RecordNotFound(slice_hrn)
399             return 1 
400         slice_id = slices[0]
401         atrribtes = self.shell.GetSliceAttributes({'slice_id': slice_id, 'name': 'enabled'}, ['slice_attribute_id'])
402         attribute_id = attreibutes[0] 
403         self.shell.UpdateSliceAttribute(self.auth, attribute_id, "1" )
404         return 1
405
406     def stopSlice(self, slice_hrn):
407         """
408         Stop the slice at plc
409         """
410         slicename = hrn_to_pl_slicename(slice_hrn)
411         slices = self.shell.GetSlices(self.auth, {'name': slicename}, ['slice_id'])
412         if not slices:
413             #raise RecordNotFound(slice_hrn)
414             return 1
415         slice_id = slices[0]
416         atrribtes = self.shell.GetSliceAttributes({'slice_id': slice_id, 'name': 'enabled'}, ['slice_attribute_id'])
417         attribute_id = attreibutes[0]
418         self.shell.UpdateSliceAttribute(self.auth, attribute_id, "0")
419         return 1
420
421
422     def resetSlice(self, slice_hrn):
423         """
424         Reset the slice
425         """
426         # XX not yet implemented
427         return 1
428
429     def getPolicy(self):
430         """
431         Return this aggregates policy.
432         """
433     
434         return self.policy
435         
436     
437
438 ##############################
439 ## Server methods here for now
440 ##############################
441
442     def list_components(self):
443         return self.getComponents()
444
445     def list_slices(self, cred, hrn):
446         self.decode_authentication(cred, 'list')
447         return self.getSlices(hrn)
448
449     def get_resources(self, cred, hrn):
450         self.decode_authentication(cred, 'info')
451         return self.getResources(hrn)
452
453     def get_policy(self, cred):
454         self.decode_authentication(cred, 'info')
455         return self.getPolicy()
456
457     def create_slice(self, cred, hrn, rspec):
458         self.decode_authentication(cred, 'embed')
459         return self.createSlice(hrn, rspec)
460
461     def update_slice(self, cred, hrn, rspec):
462         self.decode_authentication(cred, 'embed')
463         return self.updateSlice(hrn)    
464
465     def delete_slice(self, cred, hrn):
466         self.decode_authentication(cred, 'embed')
467         return self.deleteSlice(hrn)
468
469     def start_slice(self, cred, hrn):
470         self.decode_authentication(cred, 'control')
471         return self.startSlice(hrn)
472
473     def stop_slice(self, cred, hrn):
474         self.decode_authentication(cred, 'control')
475         return self.stopSlice(hrn)
476
477     def reset_slice(self, cred, hrn):
478         self.decode_authentication(cred, 'control')
479         return self.resetSlice(hrn)
480
481     def register_functions(self):
482         GeniServer.register_functions(self)
483
484         # Aggregate interface methods
485         self.server.register_function(self.list_components)
486         self.server.register_function(self.list_slices)
487         self.server.register_function(self.get_resources)
488         self.server.register_function(self.get_policy)
489         self.server.register_function(self.create_slice)
490         self.server.register_function(self.update_slice)
491         self.server.register_function(self.delete_slice)
492         self.server.register_function(self.start_slice)
493         self.server.register_function(self.stop_slice)
494         self.server.register_function(self.reset_slice)
495