sfa/managers/aggregate_manager_max.py

   1 #!/usr/bin/python
   2
   3 import sys
   4 import pdb
   5 import xml.dom.minidom
   6
   7 from sfa.util.rspec import RSpec
   8 from sfa.util.xrn import urn_to_hrn, hrn_to_urn, get_authority
   9 from sfa.util.plxrn import hrn_to_pl_slicename
  10 from sfa.util.plxrn import hrn_to_pl_slicename
  11 from sfa.util.rspec import *
  12 from sfa.util.specdict import *
  13 from sfa.util.faults import *
  14 from sfa.util.storage import *
  15 from sfa.util.policy import Policy
  16 from sfa.server.aggregate import Aggregates
  17 from sfa.server.registry import Registries
  18 from sfa.util.faults import *
  19 from sfa.util.callids import Callids
  20
  21
  22 SFA_MAX_CONF_FILE = '/etc/sfa/max_allocations'
  23 SFA_MAX_DEFAULT_RSPEC = '/etc/sfa/max_physical.xml'
  24 SFA_MAX_CANNED_RSPEC = '/etc/sfa/max_physical_canned.xml'
  25
  26 topology = {}
  27
  28 class SfaOutOfResource(SfaFault):
  29     def __init__(self, interface):
  30         faultString = "Interface " + interface + " not available"
  31         SfaFault.__init__(self, 100, faultString, '')
  32
  33 class SfaNoPairRSpec(SfaFault):
  34     def __init__(self, interface, interface2):
  35         faultString = "Interface " + interface + " should be paired with " + interface2
  36         SfaFault.__init__(self, 100, faultString, '')
  37
  38 # Returns a mapping from interfaces to the nodes they lie on and their peer interfaces
  39 # i -> node,i_peer
  40
  41 def get_interface_map():
  42     r = RSpec()
  43     r.parseFile(SFA_MAX_DEFAULT_RSPEC)
  44     rspec = r.toDict()
  45     capacity = rspec['rspec']['capacity']
  46     netspec = capacity[0]['netspec'][0]
  47     linkdefs = {}
  48     for n in netspec['nodespec']:
  49         ifspecs = n['ifspec']
  50         nodename = n['node']
  51         for i in ifspecs:
  52             ifname = i['name']
  53             linkid = i['linkid']
  54
  55             if (linkdefs.has_key(linkid)):
  56                 linkdefs[linkid].extend([(nodename,ifname)])
  57             else:
  58                 linkdefs[linkid]=[(nodename,ifname)]
  59
  60     # topology maps interface x interface -> link,node1,node2
  61     topology={}
  62
  63     for k in linkdefs.keys():
  64         (n1,i1) = linkdefs[k][0]
  65         (n2,i2) = linkdefs[k][1]
  66
  67         topology[i1] = (n1, i2)
  68         topology[i2] = (n2, i1)
  69
  70
  71     return topology
  72
  73
  74 def allocations_to_rspec(allocations):
  75     rspec = xml.dom.minidom.parse(SFA_MAX_DEFAULT_RSPEC)
  76     req = rspec.firstChild.appendChild(rspec.createElement("request"))
  77     for (iname,ip) in allocations:
  78         ifspec = req.appendChild(rspec.createElement("ifspec"))
  79         ifspec.setAttribute("name","tns:"+iname)
  80         ifspec.setAttribute("ip",ip)
  81
  82     return rspec.toxml()
  83
  84
  85 def if_endpoints(ifs):
  86     nodes=[]
  87     for l in ifs:
  88         nodes.extend(topology[l][0])
  89     return nodes
  90
  91 def lock_state_file():
  92     # Noop for demo
  93     return True
  94
  95 def unlock_state_file():
  96     return True
  97     # Noop for demo
  98
  99 def read_alloc_dict():
 100     alloc_dict={}
 101     rows = open(SFA_MAX_CONF_FILE).read().split('\n')
 102     for r in rows:
 103         columns = r.split(' ')
 104         if (len(columns)==2):
 105             hrn = columns[0]
 106             allocs = columns[1].split(',')
 107             ipallocs = map(lambda alloc:alloc.split('/'), allocs)
 108             alloc_dict[hrn]=ipallocs
 109     return alloc_dict
 110
 111 def commit_alloc_dict(d):
 112     f = open(SFA_MAX_CONF_FILE, 'w')
 113     for hrn in d.keys():
 114         columns = d[hrn]
 115         ipcolumns = map(lambda x:"/".join(x), columns)
 116         row = hrn+' '+','.join(ipcolumns)+'\n'
 117         f.write(row)
 118     f.close()
 119
 120 def collapse_alloc_dict(d):
 121     ret = []
 122     for k in d.keys():
 123         ret.extend(d[k])
 124     return ret
 125
 126
 127 def alloc_links(api, hrn, links_to_add, links_to_drop):
 128     slicename=hrn_to_pl_slicename(hrn)
 129     for (iface,ip) in links_to_add:
 130         node = topology[iface][0][0]
 131         try:
 132             api.plshell.AddSliceTag(api.plauth, slicename, "ip_addresses", ip, node)
 133             api.plshell.AddSliceTag(api.plauth, slicename, "vsys", "getvlan", node)
 134         except Exception:
 135             # Probably a duplicate tag. XXX July 21
 136             pass
 137     return True
 138
 139 def alloc_nodes(api,hrn, requested_ifs):
 140     requested_nodes = if_endpoints(requested_ifs)
 141     create_slice_max_aggregate(api, hrn, requested_nodes)
 142
 143 # Taken from slices.py
 144
 145 def create_slice_max_aggregate(api, hrn, nodes):
 146     # Get the slice record
 147     global topology
 148     topology = get_interface_map()
 149     slice = {}
 150     registries = Registries(api)
 151     registry = registries[api.hrn]
 152     credential = api.getCredential()
 153     urn = hrn_to_urn(hrn, 'slice')
 154     records = registry.Resolve(urn, credential)
 155     for record in records:
 156         if record.get_type() in ['slice']:
 157             slice = record.as_dict()
 158     if not slice:
 159         raise RecordNotFound(hrn)
 160
 161     # Make sure slice exists at plc, if it doesnt add it
 162     slicename = hrn_to_pl_slicename(hrn)
 163     slices = api.plshell.GetSlices(api.plauth, [slicename], ['node_ids'])
 164     if not slices:
 165         parts = slicename.split("_")
 166         login_base = parts[0]
 167         # if site doesnt exist add it
 168         sites = api.plshell.GetSites(api.plauth, [login_base])
 169         if not sites:
 170             authority = get_authority(hrn)
 171             authority_urn = hrn_to_urn(authority, 'authority')
 172             site_records = registry.Resolve(authority_urn, credential)
 173             site_record = {}
 174             if not site_records:
 175                 raise RecordNotFound(authority)
 176             site_record = site_records[0]
 177             site = site_record.as_dict()
 178
 179             # add the site
 180             site.pop('site_id')
 181             site_id = api.plshell.AddSite(api.plauth, site)
 182         else:
 183             site = sites[0]
 184
 185         slice_fields = {}
 186         slice_keys = ['name', 'url', 'description']
 187         for key in slice_keys:
 188             if key in slice and slice[key]:
 189                 slice_fields[key] = slice[key]
 190         api.plshell.AddSlice(api.plauth, slice_fields)
 191         slice = slice_fields
 192         slice['node_ids'] = 0
 193     else:
 194         slice = slices[0]
 195
 196     # get the list of valid slice users from the registry and make
 197     # they are added to the slice
 198     researchers = record.get('researcher', [])
 199     for researcher in researchers:
 200         person_record = {}
 201         researcher_urn = hrn_to_urn(researcher, 'user')
 202         person_records = registry.Resolve(researcher_urn, credential)
 203         for record in person_records:
 204             if record.get_type() in ['user']:
 205                 person_record = record
 206         if not person_record:
 207             pass
 208         person_dict = person_record.as_dict()
 209         persons = api.plshell.GetPersons(api.plauth, [person_dict['email']],
 210                                          ['person_id', 'key_ids'])
 211
 212         # Create the person record
 213         if not persons:
 214             person_id=api.plshell.AddPerson(api.plauth, person_dict)
 215
 216             # The line below enables the user account on the remote aggregate
 217             # soon after it is created.
 218             # without this the user key is not transfered to the slice
 219             # (as GetSlivers returns key of only enabled users),
 220             # which prevents the user from login to the slice.
 221             # We may do additional checks before enabling the user.
 222
 223             api.plshell.UpdatePerson(api.plauth, person_id, {'enabled' : True})
 224             key_ids = []
 225         else:
 226             key_ids = persons[0]['key_ids']
 227
 228         api.plshell.AddPersonToSlice(api.plauth, person_dict['email'],
 229                                      slicename)
 230
 231         # Get this users local keys
 232         keylist = api.plshell.GetKeys(api.plauth, key_ids, ['key'])
 233         keys = [key['key'] for key in keylist]
 234
 235         # add keys that arent already there
 236         for personkey in person_dict['keys']:
 237             if personkey not in keys:
 238                 key = {'key_type': 'ssh', 'key': personkey}
 239                 api.plshell.AddPersonKey(api.plauth, person_dict['email'], key)
 240
 241     # find out where this slice is currently running
 242     nodelist = api.plshell.GetNodes(api.plauth, slice['node_ids'],
 243                                     ['hostname'])
 244     hostnames = [node['hostname'] for node in nodelist]
 245
 246     # remove nodes not in rspec
 247     deleted_nodes = list(set(hostnames).difference(nodes))
 248     # add nodes from rspec
 249     added_nodes = list(set(nodes).difference(hostnames))
 250
 251     api.plshell.AddSliceToNodes(api.plauth, slicename, added_nodes)
 252     api.plshell.DeleteSliceFromNodes(api.plauth, slicename, deleted_nodes)
 253
 254     return 1
 255
 256
 257 def ListResources(api, creds, options, call_id):
 258     if Callids().already_handled(call_id): return ""
 259     # get slice's hrn from options
 260     xrn = options.get('geni_slice_urn', '')
 261     hrn, type = urn_to_hrn(xrn)
 262     # Eg. config line:
 263     # plc.princeton.sapan vlan23,vlan45
 264
 265     allocations = read_alloc_dict()
 266     if (hrn and allocations.has_key(hrn)):
 267             ret_rspec = allocations_to_rspec(allocations[hrn])
 268     else:
 269         ret_rspec = open(SFA_MAX_CANNED_RSPEC).read()
 270
 271     return (ret_rspec)
 272
 273
 274 def CreateSliver(api, xrn, creds, rspec_xml, users, call_id):
 275     if Callids().already_handled(call_id): return False
 276
 277     global topology
 278     hrn = urn_to_hrn(xrn)[0]
 279     topology = get_interface_map()
 280
 281     # Check if everything in rspec is either allocated by hrn
 282     # or not allocated at all.
 283     r = RSpec()
 284     r.parseString(rspec_xml)
 285     rspec = r.toDict()
 286
 287     lock_state_file()
 288
 289     allocations = read_alloc_dict()
 290     requested_allocations = rspec_to_allocations (rspec)
 291     current_allocations = collapse_alloc_dict(allocations)
 292     try:
 293         current_hrn_allocations=allocations[hrn]
 294     except KeyError:
 295         current_hrn_allocations=[]
 296
 297     # Check request against current allocations
 298     requested_interfaces = map(lambda(elt):elt[0], requested_allocations)
 299     current_interfaces = map(lambda(elt):elt[0], current_allocations)
 300     current_hrn_interfaces = map(lambda(elt):elt[0], current_hrn_allocations)
 301
 302     for a in requested_interfaces:
 303         if (a not in current_hrn_interfaces and a in current_interfaces):
 304             raise SfaOutOfResource(a)
 305         if (topology[a][1] not in requested_interfaces):
 306             raise SfaNoPairRSpec(a,topology[a][1])
 307     # Request OK
 308
 309     # Allocations to delete
 310     allocations_to_delete = []
 311     for a in current_hrn_allocations:
 312         if (a not in requested_allocations):
 313             allocations_to_delete.extend([a])
 314
 315     # Ok, let's do our thing
 316     alloc_nodes(api, hrn, requested_interfaces)
 317     alloc_links(api, hrn, requested_allocations, allocations_to_delete)
 318     allocations[hrn] = requested_allocations
 319     commit_alloc_dict(allocations)
 320
 321     unlock_state_file()
 322
 323     return True
 324
 325 def rspec_to_allocations(rspec):
 326     ifs = []
 327     try:
 328         ifspecs = rspec['rspec']['request'][0]['ifspec']
 329         for l in ifspecs:
 330             ifs.extend([(l['name'].replace('tns:',''),l['ip'])])
 331     except KeyError:
 332         # Bad RSpec
 333         pass
 334     return ifs
 335
 336 def main():
 337     t = get_interface_map()
 338     r = RSpec()
 339     rspec_xml = open(sys.argv[1]).read()
 340     #ListResources(None,'foo')
 341     CreateSliver(None, "plc.princeton.sap0", rspec_xml, 'call-id-sap0')
 342
 343 if __name__ == "__main__":
 344     main()