sfa/plc/nodes.py

   1 ### $Id$
   2 ### $URL$
   3
   4 import os
   5 import time
   6 import datetime
   7 import sys
   8 import traceback
   9
  10 from sfa.util.misc import *
  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.debug import log
  16 from sfa.util.rspec import *
  17 from sfa.util.specdict import *
  18 from sfa.util.policy import Policy
  19 from sfa.server.aggregate import Aggregates
  20
  21 class Nodes(SimpleStorage):
  22
  23     def __init__(self, api, ttl = 1, origin_gid_caller=None):
  24         self.api = api
  25         self.ttl = ttl
  26         self.threshold = None
  27         path = self.api.config.SFA_DATA_DIR
  28         filename = ".".join([self.api.interface, self.api.hrn, "nodes"])
  29         filepath = path + os.sep + filename
  30         self.nodes_file = filepath
  31         SimpleStorage.__init__(self, self.nodes_file)
  32         self.policy = Policy(api)
  33         self.load()
  34         self.origin_hrn = None
  35         if origin_gid_caller:
  36             self.origin_hrn=origin_gid_caller.get_hrn()
  37
  38
  39     def refresh(self):
  40         """
  41         Update the cached list of nodes
  42         """
  43
  44         # Reload components list
  45         now = datetime.datetime.now()
  46         if not self.has_key('threshold') or not self.has_key('timestamp') or \
  47            now > datetime.datetime.fromtimestamp(time.mktime(time.strptime(self['threshold'], self.api.time_format))):
  48             if self.api.interface in ['aggregate']:
  49                 self.refresh_nodes_aggregate()
  50             elif self.api.interface in ['slicemgr']:
  51                 self.refresh_nodes_smgr()
  52
  53     def refresh_nodes_aggregate(self):
  54         rspec = RSpec()
  55         rspec.parseString(self.get_rspec())
  56
  57         # filter nodes according to policy
  58         blist = self.policy['node_blacklist']
  59         wlist = self.policy['node_whitelist']
  60         rspec.filter('NodeSpec', 'name', blacklist=blist, whitelist=wlist)
  61
  62         # extract ifspecs from rspec to get ips'
  63         ips = []
  64         ifspecs = rspec.getDictsByTagName('IfSpec')
  65         for ifspec in ifspecs:
  66             if ifspec.has_key('addr') and ifspec['addr']:
  67                 ips.append(ifspec['addr'])
  68
  69         # extract nodespecs from rspec to get dns names
  70         hostnames = []
  71         nodespecs = rspec.getDictsByTagName('NodeSpec')
  72         for nodespec in nodespecs:
  73             if nodespec.has_key('name') and nodespec['name']:
  74                 hostnames.append(nodespec['name'])
  75
  76         # update timestamp and threshold
  77         timestamp = datetime.datetime.now()
  78         hr_timestamp = timestamp.strftime(self.api.time_format)
  79         delta = datetime.timedelta(hours=self.ttl)
  80         threshold = timestamp + delta
  81         hr_threshold = threshold.strftime(self.api.time_format)
  82
  83         node_details = {}
  84         node_details['rspec'] = rspec.toxml()
  85         node_details['ip'] = ips
  86         node_details['dns'] = hostnames
  87         node_details['timestamp'] = hr_timestamp
  88         node_details['threshold'] = hr_threshold
  89         # save state
  90         self.update(node_details)
  91         self.write()
  92
  93     def get_remote_resources(self, hrn = None):
  94         # convert and threshold to ints
  95         if self.has_key('timestamp') and self['timestamp']:
  96             hr_timestamp = self['timestamp']
  97             timestamp = datetime.datetime.fromtimestamp(time.mktime(time.strptime(hr_timestamp, self.api.time_format)))
  98             hr_threshold = self['threshold']
  99             threshold = datetime.datetime.fromtimestamp(time.mktime(time.strptime(hr_threshold, self.api.time_format)))
 100         else:
 101             timestamp = datetime.datetime.now()
 102             hr_timestamp = timestamp.strftime(self.api.time_format)
 103             delta = datetime.timedelta(hours=self.ttl)
 104             threshold = timestamp + delta
 105             hr_threshold = threshold.strftime(self.api.time_format)
 106
 107         start_time = int(timestamp.strftime("%s"))
 108         end_time = int(threshold.strftime("%s"))
 109         duration = end_time - start_time
 110
 111         aggregates = Aggregates(self.api)
 112         rspecs = {}
 113         networks = []
 114         rspec = RSpec()
 115         credential = self.api.getCredential()
 116         for aggregate in aggregates:
 117           if aggregate not in [self.api.auth.client_cred.get_gid_caller().get_hrn()]:
 118             try:
 119                 origin_hrn = self.origin_hrn
 120                 # get the rspec from the aggregate
 121                 try:
 122                     request_hash=None
 123                     agg_rspec = aggregates[aggregate].get_resources(credential, hrn, request_hash, origin_hrn)
 124                 except:
 125                     arg_list = [credential, hrn]
 126                     request_hash = self.api.key.compute_hash(arg_list)
 127                     agg_rspec = aggregates[aggregate].get_resources(credential, hrn, request_hash, origin_hrn)
 128                 # extract the netspec from each aggregates rspec
 129                 rspec.parseString(agg_rspec)
 130                 networks.extend([{'NetSpec': rspec.getDictsByTagName('NetSpec')}])
 131             except:
 132                 # XX print out to some error log
 133                 print >> log, "Error getting resources at aggregate %s" % aggregate
 134                 traceback.print_exc(log)
 135                 print >> log, "%s" % (traceback.format_exc())
 136         # create the rspec dict
 137         resources = {'networks': networks, 'start_time': start_time, 'duration': duration}
 138         resourceDict = {'RSpec': resources}
 139         # convert rspec dict to xml
 140         rspec.parseDict(resourceDict)
 141         return rspec
 142
 143     def refresh_nodes_smgr(self):
 144
 145         rspec = self.get_remote_resources()
 146         # filter according to policy
 147         blist = self.policy['node_blacklist']
 148         wlist = self.policy['node_whitelist']
 149         rspec.filter('NodeSpec', 'name', blacklist=blist, whitelist=wlist)
 150
 151         # update timestamp and threshold
 152         timestamp = datetime.datetime.now()
 153         hr_timestamp = timestamp.strftime(self.api.time_format)
 154         delta = datetime.timedelta(hours=self.ttl)
 155         threshold = timestamp + delta
 156         hr_threshold = threshold.strftime(self.api.time_format)
 157
 158         nodedict = {'rspec': rspec.toxml(),
 159                     'timestamp': hr_timestamp,
 160                     'threshold':  hr_threshold}
 161
 162         self.update(nodedict)
 163         self.write()
 164
 165     def get_rspec(self, hrn = None):
 166
 167         if self.api.interface in ['slicemgr']:
 168             return self.get_rspec_smgr(hrn)
 169         elif self.api.interface in ['aggregate']:
 170             return self.get_rspec_aggregate(hrn)
 171
 172     def get_rspec_smgr(self, hrn = None):
 173
 174         rspec = self.get_remote_resources(hrn)
 175         return rspec.toxml()
 176
 177     def get_rspec_aggregate(self, hrn = None):
 178         """
 179         Get resource information from PLC
 180         """
 181
 182         slicename = None
 183         # Get the required nodes
 184         if not hrn:
 185             nodes = self.api.plshell.GetNodes(self.api.plauth, {'peer_id': None})
 186             try:  linkspecs = self.api.plshell.GetLinkSpecs() # if call is supported
 187             except:  linkspecs = []
 188         else:
 189             slicename = hrn_to_pl_slicename(hrn)
 190             slices = self.api.plshell.GetSlices(self.api.plauth, [slicename])
 191             if not slices:
 192                 nodes = []
 193             else:
 194                 slice = slices[0]
 195                 node_ids = slice['node_ids']
 196                 nodes = self.api.plshell.GetNodes(self.api.plauth, {'peer_id': None, 'node_id': node_ids})
 197
 198         # Filter out whitelisted nodes
 199         public_nodes = lambda n: n.has_key('slice_ids_whitelist') and not n['slice_ids_whitelist']
 200
 201         # ...only if they are not already assigned to this slice.
 202         if (not slicename):
 203             nodes = filter(public_nodes, nodes)
 204
 205         # Get all network interfaces
 206         interface_ids = []
 207         for node in nodes:
 208             # The field name has changed in plcapi 4.3
 209             if self.api.plshell_version in ['4.2']:
 210                 interface_ids.extend(node['nodenetwork_ids'])
 211             elif self.api.plshell_version in ['4.3']:
 212                 interface_ids.extend(node['interface_ids'])
 213             else:
 214                 raise GeniAPIError, "Unsupported plcapi version ", \
 215                                  self.api.plshell_version
 216
 217         if self.api.plshell_version in ['4.2']:
 218             interfaces = self.api.plshell.GetNodeNetworks(self.api.plauth, interface_ids)
 219         elif self.api.plshell_version in ['4.3']:
 220             interfaces = self.api.plshell.GetInterfaces(self.api.plauth, interface_ids)
 221         else:
 222             raise GeniAPIError, "Unsupported plcapi version ", \
 223                                 self.api.plshell_version
 224         interface_dict = {}
 225         for interface in interfaces:
 226             if self.api.plshell_version in ['4.2']:
 227                 interface_dict[interface['nodenetwork_id']] = interface
 228             elif self.api.plshell_version in ['4.3']:
 229                 interface_dict[interface['interface_id']] = interface
 230             else:
 231                 raise GeniAPIError, "Unsupported plcapi version", \
 232                                     self.api.plshell_version
 233
 234         # join nodes with thier interfaces
 235         for node in nodes:
 236             node['interfaces'] = []
 237             if self.api.plshell_version in ['4.2']:
 238                 for nodenetwork_id in node['nodenetwork_ids']:
 239                     node['interfaces'].append(interface_dict[nodenetwork_id])
 240             elif self.api.plshell_version in ['4.3']:
 241                 for interface_id in node['interface_ids']:
 242                     node['interfaces'].append(interface_dict[interface_id])
 243             else:
 244                 raise GeniAPIError, "Unsupported plcapi version", \
 245                                     self.api.plshell_version
 246
 247         # convert and threshold to ints
 248         if self.has_key('timestamp') and self['timestamp']:
 249             timestamp = datetime.datetime.fromtimestamp(time.mktime(time.strptime(self['timestamp'], self.api.time_format)))
 250             threshold = datetime.datetime.fromtimestamp(time.mktime(time.strptime(self['threshold'], self.api.time_format)))
 251         else:
 252             timestamp = datetime.datetime.now()
 253             delta = datetime.timedelta(hours=self.ttl)
 254             threshold = timestamp + delta
 255
 256         start_time = int(timestamp.strftime("%s"))
 257         end_time = int(threshold.strftime("%s"))
 258         duration = end_time - start_time
 259
 260         # create the plc dict
 261         networks = [{'nodes': nodes,
 262                      'name': self.api.hrn,
 263                      'start_time': start_time,
 264                      'duration': duration}]
 265         if not hrn:
 266             networks[0]['links'] = linkspecs
 267         resources = {'networks': networks, 'start_time': start_time, 'duration': duration}
 268
 269         # convert the plc dict to an rspec dict
 270         resourceDict = RSpecDict(resources)
 271         # convert the rspec dict to xml
 272         rspec = RSpec()
 273         rspec.parseDict(resourceDict)
 274         return rspec.toxml()
 275