5f3138fda5174f28fc02b22f061daa6c3f5e5d50
[sfa.git] / sfa / dummy / dummydriver.py
1 import time
2 import datetime
3 #
4 from sfa.util.faults import MissingSfaInfo, UnknownSfaType, \
5     RecordNotFound, SfaNotImplemented, SliverDoesNotExist
6
7 from sfa.util.sfalogging import logger
8 from sfa.util.defaultdict import defaultdict
9 from sfa.util.sfatime import utcparse, datetime_to_string, datetime_to_epoch
10 from sfa.util.xrn import Xrn, hrn_to_urn, get_leaf
11 from sfa.util.cache import Cache
12
13 # one would think the driver should not need to mess with the SFA db, but..
14 from sfa.storage.alchemy import dbsession
15 from sfa.storage.model import RegRecord
16
17 # used to be used in get_ticket
18 #from sfa.trust.sfaticket import SfaTicket
19
20 from sfa.rspecs.version_manager import VersionManager
21 from sfa.rspecs.rspec import RSpec
22
23 # the driver interface, mostly provides default behaviours
24 from sfa.managers.driver import Driver
25
26 from sfa.dummy.dummyshell import DummyShell
27 from sfa.dummy.dummyaggregate import DummyAggregate
28 from sfa.dummy.dummyslices import DummySlices
29 from sfa.dummy.dummyxrn import DummyXrn, slicename_to_hrn, hostname_to_hrn, hrn_to_dummy_slicename, xrn_to_hostname
30
31
32 def list_to_dict(recs, key):
33     """
34     convert a list of dictionaries into a dictionary keyed on the 
35     specified dictionary key 
36     """
37     return dict ( [ (rec[key],rec) for rec in recs ] )
38
39 #
40 # DummyShell is just an xmlrpc serverproxy where methods can be sent as-is; 
41
42 class DummyDriver (Driver):
43
44     # the cache instance is a class member so it survives across incoming requests
45     cache = None
46
47     def __init__ (self, config):
48         Driver.__init__ (self, config)
49         self.config = config
50         self.hrn = config.SFA_INTERFACE_HRN
51         self.root_auth = config.SFA_REGISTRY_ROOT_AUTH
52         self.shell = DummyShell (config)
53         self.testbedInfo = self.shell.GetTestbedInfo()
54  
55     ########################################
56     ########## registry oriented
57     ########################################
58
59     def augment_records_with_testbed_info (self, sfa_records):
60         return self.fill_record_info (sfa_records)
61
62     ########## 
63     def register (self, sfa_record, hrn, pub_key):
64         type = sfa_record['type']
65         dummy_record = self.sfa_fields_to_dummy_fields(type, hrn, sfa_record)
66         
67         if type == 'authority':
68             pointer = -1
69
70         elif type == 'slice':
71             slices = self.shell.GetSlices({'slice_name': dummy_record['slice_name']})
72             if not slices:
73                  pointer = self.shell.AddSlice(dummy_record)
74             else:
75                  pointer = slices[0]['slice_id']
76
77         elif type == 'user':
78             users = self.shell.GetUsers({'email':sfa_record['email']})
79             if not users:
80                 pointer = self.shell.AddUser(dummy_record)
81             else:
82                 pointer = users[0]['user_id']
83     
84             # Add the user's key
85             if pub_key:
86                 self.shell.AddUserKey({'user_id' : pointer, 'key' : pub_key})
87
88         elif type == 'node':
89             nodes = self.shell.GetNodes(dummy_record['hostname'])
90             if not nodes:
91                 pointer = self.shell.AddNode(dummy_record)
92             else:
93                 pointer = users[0]['node_id']
94     
95         return pointer
96         
97     ##########
98     def update (self, old_sfa_record, new_sfa_record, hrn, new_key):
99         pointer = old_sfa_record['pointer']
100         type = old_sfa_record['type']
101         dummy_record=self.sfa_fields_to_dummy_fields(type, hrn, new_sfa_record)
102
103         # new_key implemented for users only
104         if new_key and type not in [ 'user' ]:
105             raise UnknownSfaType(type)
106
107     
108         if type == "slice":
109             self.shell.UpdateSlice({'slice_id': pointer, 'fields': dummy_record})
110     
111         elif type == "user":
112             self.shell.UpdateUser({'user_id': pointer, 'fields': dummy_record})
113
114             if new_key:
115                 self.shell.AddUserKey({'user_id' : pointer, 'key' : new_key})
116
117         elif type == "node":
118             self.shell.UpdateNode({'node_id': pointer, 'fields': dummy_record})
119
120
121         return True
122         
123
124     ##########
125     def remove (self, sfa_record):
126         type=sfa_record['type']
127         pointer=sfa_record['pointer']
128         if type == 'user':
129             self.shell.DeleteUser({'user_id': pointer})
130         elif type == 'slice':
131             self.shell.DeleteSlice({'slice_id': pointer})
132         elif type == 'node':
133             self.shell.DeleteNode({'node_id': pointer})
134
135         return True
136
137
138
139
140
141     ##
142     # Convert SFA fields to Dummy testbed fields for use when registering or updating
143     # registry record in the dummy testbed
144     #
145
146     def sfa_fields_to_dummy_fields(self, type, hrn, sfa_record):
147
148         dummy_record = {}
149  
150         if type == "slice":
151             dummy_record["slice_name"] = hrn_to_dummy_slicename(hrn)
152         
153         elif type == "node":
154             if "hostname" not in sfa_record:
155                 raise MissingSfaInfo("hostname")
156             dummy_record["hostname"] = sfa_record["hostname"]
157             if "type" in sfa_record:
158                dummy_record["type"] = sfa_record["type"]
159             else:
160                dummy_record["type"] = "dummy_type"
161  
162         elif type == "authority":
163             dummy_record["name"] = hrn
164
165         elif type == "user":
166             dummy_record["user_name"] = sfa_record["email"].split('@')[0]
167             dummy_record["email"] = sfa_record["email"]
168
169         return dummy_record
170
171     ####################
172     def fill_record_info(self, records):
173         """
174         Given a (list of) SFA record, fill in the DUMMY TESTBED specific 
175         and SFA specific fields in the record. 
176         """
177         if not isinstance(records, list):
178             records = [records]
179
180         self.fill_record_dummy_info(records)
181         self.fill_record_hrns(records)
182         self.fill_record_sfa_info(records)
183         return records
184
185     def fill_record_dummy_info(self, records):
186         """
187         Fill in the DUMMY specific fields of a SFA record. This
188         involves calling the appropriate DUMMY method to retrieve the 
189         database record for the object.
190             
191         @param record: record to fill in field (in/out param)     
192         """
193         # get ids by type
194         node_ids, slice_ids, user_ids = [], [], [] 
195         type_map = {'node': node_ids, 'slice': slice_ids, 'user': user_ids}
196                   
197         for record in records:
198             for type in type_map:
199                 if type == record['type']:
200                     type_map[type].append(record['pointer'])
201
202         # get dummy records
203         nodes, slices, users = {}, {}, {}
204         if node_ids:
205             node_list = self.shell.GetNodes({'node_ids':node_ids})
206             nodes = list_to_dict(node_list, 'node_id')
207         if slice_ids:
208             slice_list = self.shell.GetSlices({'slice_ids':slice_ids})
209             slices = list_to_dict(slice_list, 'slice_id')
210         if user_ids:
211             user_list = self.shell.GetUsers({'user_ids': user_ids})
212             users = list_to_dict(user_list, 'user_id')
213
214         dummy_records = {'node': nodes, 'slice': slices, 'user': users}
215
216
217         # fill record info
218         for record in records:
219             # records with pointer==-1 do not have dummy info.
220             if record['pointer'] == -1:
221                 continue
222            
223             for type in dummy_records:
224                 if record['type'] == type:
225                     if record['pointer'] in dummy_records[type]:
226                         record.update(dummy_records[type][record['pointer']])
227                         break
228             # fill in key info
229             if record['type'] == 'user':
230                 record['key_ids'] = []
231                 recors['keys'] = []
232                 for key in dummy_records['user'][record['pointer']]['keys']:
233                      record['key_ids'].append(-1)
234                      recors['keys'].append(key)
235
236         return records
237
238     def fill_record_hrns(self, records):
239         """
240         convert dummy ids to hrns
241         """
242
243         # get ids
244         slice_ids, user_ids, node_ids = [], [], []
245         for record in records:
246             if 'user_ids' in record:
247                 user_ids.extend(record['user_ids'])
248             if 'slice_ids' in record:
249                 slice_ids.extend(record['slice_ids'])
250             if 'node_ids' in record:
251                 node_ids.extend(record['node_ids'])
252
253         # get dummy records
254         slices, users, nodes = {}, {}, {}
255         if user_ids:
256             user_list = self.shell.GetUsers({'user_ids': user_ids})
257             users = list_to_dict(user_list, 'user_id')
258         if slice_ids:
259             slice_list = self.shell.GetSlices({'slice_ids': slice_ids})
260             slices = list_to_dict(slice_list, 'slice_id')       
261         if node_ids:
262             node_list = self.shell.GetNodes({'node_ids': node_ids})
263             nodes = list_to_dict(node_list, 'node_id')
264        
265         # convert ids to hrns
266         for record in records:
267             # get all relevant data
268             type = record['type']
269             pointer = record['pointer']
270             testbed_name = self.testbed_name()
271             auth_hrn = self.hrn
272             if pointer == -1:
273                 continue
274
275             if 'user_ids' in record:
276                 emails = [users[user_id]['email'] for user_id in record['user_ids'] \
277                           if user_id in  users]
278                 usernames = [email.split('@')[0] for email in emails]
279                 user_hrns = [".".join([auth_hrn, testbed_name, username]) for username in usernames]
280                 record['users'] = user_hrns 
281             if 'slice_ids' in record:
282                 slicenames = [slices[slice_id]['slice_name'] for slice_id in record['slice_ids'] \
283                               if slice_id in slices]
284                 slice_hrns = [slicename_to_hrn(auth_hrn, slicename) for slicename in slicenames]
285                 record['slices'] = slice_hrns
286             if 'node_ids' in record:
287                 hostnames = [nodes[node_id]['hostname'] for node_id in record['node_ids'] \
288                              if node_id in nodes]
289                 node_hrns = [hostname_to_hrn(auth_hrn, login_base, hostname) for hostname in hostnames]
290                 record['nodes'] = node_hrns
291
292             
293         return records   
294
295     def fill_record_sfa_info(self, records):
296
297         def startswith(prefix, values):
298             return [value for value in values if value.startswith(prefix)]
299
300         # get user ids
301         user_ids = []
302         for record in records:
303             user_ids.extend(record.get("user_ids", []))
304         
305         # get sfa records for all records associated with these records.   
306         # we'll replace pl ids (person_ids) with hrns from the sfa records
307         # we obtain
308         
309         # get the registry records
310         user_list, users = [], {}
311         user_list = dbsession.query (RegRecord).filter(RegRecord.pointer.in_(user_ids))
312         # create a hrns keyed on the sfa record's pointer.
313         # Its possible for multiple records to have the same pointer so
314         # the dict's value will be a list of hrns.
315         users = defaultdict(list)
316         for user in user_list:
317             users[user.pointer].append(user)
318
319         # get the dummy records
320         dummy_user_list, dummy_users = [], {}
321         dummy_user_list = self.shell.GetUsers({'user_ids': user_ids})
322         dummy_users = list_to_dict(dummy_user_list, 'user_id')
323
324         # fill sfa info
325         for record in records:
326             # skip records with no pl info (top level authorities)
327             #if record['pointer'] == -1:
328             #    continue 
329             sfa_info = {}
330             type = record['type']
331             logger.info("fill_record_sfa_info - incoming record typed %s"%type)
332             if (type == "slice"):
333                 # all slice users are researchers
334                 record['geni_urn'] = hrn_to_urn(record['hrn'], 'slice')
335                 record['PI'] = []
336                 record['researcher'] = []
337                 for user_id in record.get('user_ids', []):
338                     hrns = [user.hrn for user in users[user_id]]
339                     record['researcher'].extend(hrns)                
340
341             elif (type.startswith("authority")):
342                 record['url'] = None
343                 logger.info("fill_record_sfa_info - authority xherex")
344
345             elif (type == "node"):
346                 sfa_info['dns'] = record.get("hostname", "")
347                 # xxx TODO: URI, LatLong, IP, DNS
348     
349             elif (type == "user"):
350                 logger.info('setting user.email')
351                 sfa_info['email'] = record.get("email", "")
352                 sfa_info['geni_urn'] = hrn_to_urn(record['hrn'], 'user')
353                 sfa_info['geni_certificate'] = record['gid'] 
354                 # xxx TODO: PostalAddress, Phone
355             record.update(sfa_info)
356
357
358     ####################
359     def update_relation (self, subject_type, target_type, relation_name, subject_id, target_ids):
360         # hard-wire the code for slice/user for now, could be smarter if needed
361         if subject_type =='slice' and target_type == 'user' and relation_name == 'researcher':
362             subject=self.shell.GetSlices ({'slice_id': subject_id})[0]
363             if 'user_ids' not in subject.keys():
364                  subject['user_ids'] = []
365             current_target_ids = subject['user_ids']
366             add_target_ids = list ( set (target_ids).difference(current_target_ids))
367             del_target_ids = list ( set (current_target_ids).difference(target_ids))
368             logger.debug ("subject_id = %s (type=%s)"%(subject_id,type(subject_id)))
369             for target_id in add_target_ids:
370                 self.shell.AddUserToSlice ({'user_id': target_id, 'slice_id': subject_id})
371                 logger.debug ("add_target_id = %s (type=%s)"%(target_id,type(target_id)))
372             for target_id in del_target_ids:
373                 logger.debug ("del_target_id = %s (type=%s)"%(target_id,type(target_id)))
374                 self.shell.DeleteUserFromSlice ({'user_id': target_id, 'slice_id': subject_id})
375         else:
376             logger.info('unexpected relation %s to maintain, %s -> %s'%(relation_name,subject_type,target_type))
377
378         
379     ########################################
380     ########## aggregate oriented
381     ########################################
382
383     def testbed_name (self): return "dummy"
384
385     # 'geni_request_rspec_versions' and 'geni_ad_rspec_versions' are mandatory
386     def aggregate_version (self):
387         version_manager = VersionManager()
388         ad_rspec_versions = []
389         request_rspec_versions = []
390         for rspec_version in version_manager.versions:
391             if rspec_version.content_type in ['*', 'ad']:
392                 ad_rspec_versions.append(rspec_version.to_dict())
393             if rspec_version.content_type in ['*', 'request']:
394                 request_rspec_versions.append(rspec_version.to_dict()) 
395         return {
396             'testbed':self.testbed_name(),
397             'geni_request_rspec_versions': request_rspec_versions,
398             'geni_ad_rspec_versions': ad_rspec_versions,
399             }
400
401     def list_slices (self, creds, options):
402     
403         slices = self.shell.GetSlices()
404         slice_hrns = [slicename_to_hrn(self.hrn, slice['slice_name']) for slice in slices]
405         slice_urns = [hrn_to_urn(slice_hrn, 'slice') for slice_hrn in slice_hrns]
406     
407         return slice_urns
408         
409     # first 2 args are None in case of resource discovery
410     def list_resources (self, slice_urn, slice_hrn, creds, options):
411     
412         version_manager = VersionManager()
413         # get the rspec's return format from options
414         rspec_version = version_manager.get_version(options.get('geni_rspec_version'))
415         version_string = "rspec_%s" % (rspec_version)
416     
417         aggregate = DummyAggregate(self)
418         rspec =  aggregate.get_rspec(slice_xrn=slice_urn, version=rspec_version, 
419                                      options=options)
420     
421         return rspec
422     
423     def sliver_status (self, slice_urn, slice_hrn):
424         # find out where this slice is currently running
425         slice_name = hrn_to_dummy_slicename(slice_hrn)
426         
427         slice = self.shell.GetSlices({'slice_name': slice_name})
428         if len(slices) == 0:        
429             raise SliverDoesNotExist("%s (used %s as slicename internally)" % (slice_hrn, slicename))
430         
431         # report about the local nodes only
432         nodes = self.shell.GetNodes({'node_ids':slice['node_ids']})
433
434         if len(nodes) == 0:
435             raise SliverDoesNotExist("You have not allocated any slivers here") 
436
437         # get login info
438         user = {}
439         keys = []
440         if slice['user_ids']:
441             users = self.shell.GetUsers({'user_ids': slice['user_ids']})
442             for user in users:
443                  keys.extend(user['keys'])
444
445             user.update({'urn': slice_urn,
446                          'login': slice['slice_name'],
447                          'protocol': ['ssh'],
448                          'port': ['22'],
449                          'keys': keys})
450
451     
452         result = {}
453         top_level_status = 'unknown'
454         if nodes:
455             top_level_status = 'ready'
456         result['geni_urn'] = slice_urn
457         result['dummy_login'] = slice['slice_name']
458         result['dummy_expires'] = datetime_to_string(utcparse(slice['expires']))
459         result['geni_expires'] = datetime_to_string(utcparse(slice['expires']))
460         
461         resources = []
462         for node in nodes:
463             res = {}
464             res['dummy_hostname'] = node['hostname']
465             res['geni_expires'] = datetime_to_string(utcparse(slice['expires']))
466             sliver_id = Xrn(slice_urn, type='slice', id=node['node_id'], authority=self.hrn).urn
467             res['geni_urn'] = sliver_id
468             res['geni_status'] = 'ready'
469             res['geni_error'] = ''
470             res['users'] = [users]  
471     
472             resources.append(res)
473             
474         result['geni_status'] = top_level_status
475         result['geni_resources'] = resources
476         return result
477
478     def create_sliver (self, slice_urn, slice_hrn, creds, rspec_string, users, options):
479
480         aggregate = DummyAggregate(self)
481         slices = DummySlices(self)
482         sfa_peer = slices.get_sfa_peer(slice_hrn)
483         slice_record=None    
484         if users:
485             slice_record = users[0].get('slice_record', {})
486     
487         # parse rspec
488         rspec = RSpec(rspec_string)
489         requested_attributes = rspec.version.get_slice_attributes()
490         
491         # ensure slice record exists
492         slice = slices.verify_slice(slice_hrn, slice_record, sfa_peer, options=options)
493         # ensure user records exists
494         #users = slices.verify_users(slice_hrn, slice, users, sfa_peer, options=options)
495         
496         # add/remove slice from nodes
497         requested_slivers = []
498         for node in rspec.version.get_nodes_with_slivers():
499             hostname = None
500             if node.get('component_name'):
501                 hostname = node.get('component_name').strip()
502             elif node.get('component_id'):
503                 hostname = xrn_to_hostname(node.get('component_id').strip())
504             if hostname:
505                 requested_slivers.append(hostname)
506         requested_slivers_ids = []
507         for hostname in requested_slivers:
508             node_id = self.shell.GetNodes({'hostname': hostname})[0]['node_id']
509             requested_slivers_ids.append(node_id) 
510         nodes = slices.verify_slice_nodes(slice, requested_slivers_ids) 
511     
512         return aggregate.get_rspec(slice_xrn=slice_urn, version=rspec.version)
513
514     def delete_sliver (self, slice_urn, slice_hrn, creds, options):
515         slicename = hrn_to_dummy_slicename(slice_hrn)
516         slices = self.shell.GetSlices({'slice_name': slicename})
517         if not slices:
518             return True
519         slice = slices[0]
520         
521         try:
522             self.shell.DeleteSliceFromNodes({'slice_id': slice['slice_id'], 'node_ids': slice['node_ids']})
523             return True
524         except:
525             return False
526     
527     def renew_sliver (self, slice_urn, slice_hrn, creds, expiration_time, options):
528         slicename = hrn_to_dummy_slicename(slice_hrn)
529         slices = self.shell.GetSlices({'slice_name': slicename})
530         if not slices:
531             raise RecordNotFound(slice_hrn)
532         slice = slices[0]
533         requested_time = utcparse(expiration_time)
534         record = {'expires': int(datetime_to_epoch(requested_time))}
535         try:
536             self.shell.UpdateSlice({'slice_id': slice['slice_id'], 'fields':record})
537             return True
538         except:
539             return False
540
541     # set the 'enabled' tag to True
542     def start_slice (self, slice_urn, slice_hrn, creds):
543         slicename = hrn_to_dummy_slicename(slice_hrn)
544         slices = self.shell.GetSlices({'slice_name': slicename})
545         if not slices:
546             raise RecordNotFound(slice_hrn)
547         slice_id = slices[0]['slice_id']
548         slice_enabled = slices[0]['enabled'] 
549         # just update the slice enabled tag
550         if not slice_enabled:
551             self.shell.UpdateSlice({'slice_id': slice_id, 'fields': {'enabled': True}})
552         return 1
553
554     # set the 'enabled' tag to False
555     def stop_slice (self, slice_urn, slice_hrn, creds):
556         slicename = hrn_to_pl_slicename(slice_hrn)
557         slices = self.shell.GetSlices({'slice_name': slicename})
558         if not slices:
559             raise RecordNotFound(slice_hrn)
560         slice_id = slices[0]['slice_id']
561         slice_enabled = slices[0]['enabled']
562         # just update the slice enabled tag
563         if slice_enabled:
564             self.shell.UpdateSlice({'slice_id': slice_id, 'fields': {'enabled': False}})
565         return 1
566     
567     def reset_slice (self, slice_urn, slice_hrn, creds):
568         raise SfaNotImplemented ("reset_slice not available at this interface")
569     
570     def get_ticket (self, slice_urn, slice_hrn, creds, rspec_string, options):
571         raise SfaNotImplemented,"DummyDriver.get_ticket needs a rewrite"