src/nepi/util/sfaapi.py

   1 #
   2 #    NEPI, a framework to manage network experiments
   3 #    Copyright (C) 2013 INRIA
   4 #
   5 #    This program is free software: you can redistribute it and/or modify
   6 #    it under the terms of the GNU General Public License as published by
   7 #    the Free Software Foundation, either version 3 of the License, or
   8 #    (at your option) any later version.
   9 #
  10 #    This program is distributed in the hope that it will be useful,
  11 #    but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 #    GNU General Public License for more details.
  14 #
  15 #    You should have received a copy of the GNU General Public License
  16 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 #
  18 # Author: Lucia Guevgeozian Odizzio <lucia.guevgeozian_odizzio@inria.fr>
  19
  20 import threading
  21 import hashlib
  22 import re
  23 import os
  24
  25 from nepi.util.logger import Logger
  26
  27 try:
  28     from sfa.client.sfi import Sfi
  29     from sfa.util.xrn import hrn_to_urn
  30 except ImportError:
  31     log = Logger("SFA API")
  32     log.debug("Packages sfa-common or sfa-client not installed.\
  33          Could not import sfa.client.sfi or sfa.util.xrn")
  34
  35 from nepi.util.sfarspec_proc import SfaRSpecProcessing
  36
  37 class SFAAPI(object):
  38     """
  39     API for quering the SFA service.
  40     """
  41     def __init__(self, sfi_user, sfi_auth, sfi_registry, sfi_sm, private_key,
  42         timeout):
  43
  44         self.sfi_user = sfi_user
  45         self.sfi_auth = sfi_auth
  46         self.sfi_registry = sfi_registry
  47         self.sfi_sm = sfi_sm
  48         self.private_key = private_key
  49         self.timeout = timeout
  50
  51         self._blacklist = set()
  52         self._reserved = set()
  53         self._resources_cache = None
  54         self._already_cached = False
  55         self._log = Logger("SFA API")
  56         self.api = Sfi()
  57         self.rspec_proc = SfaRSpecProcessing()
  58         self.lock = threading.Lock()
  59
  60     def _sfi_exec_method(self, command, slicename=None, rspec=None, urn=None):
  61         """
  62         Execute sfi method.
  63         """
  64         if command in ['describe', 'delete', 'allocate', 'provision']:
  65             if not slicename:
  66                 raise TypeError("The slice hrn is expected for this method %s" % command)
  67             if command == 'allocate' and not rspec:
  68                 raise TypeError("RSpec is expected for this method %s" % command)
  69
  70             if command == 'allocate':
  71                 args_list = [slicename, rspec]
  72             elif command == 'delete':
  73                 args_list = [slicename, urn]
  74             else: args_list = [slicename, '-o', '/tmp/rspec_output']
  75
  76         elif command == 'resources':
  77             args_list = ['-o', '/tmp/rspec_output']
  78
  79         else: raise TypeError("Sfi method not supported")
  80
  81         self.api.options.timeout = self.timeout
  82         self.api.options.raw = None
  83         self.api.options.user = self.sfi_user
  84         self.api.options.auth = self.sfi_auth
  85         self.api.options.registry = self.sfi_registry
  86         self.api.options.sm = self.sfi_sm
  87         self.api.options.user_private_key = self.private_key
  88
  89         self.api.command = command
  90         self.api.command_parser = self.api.create_parser_command(self.api.command)
  91         (command_options, command_args) = self.api.command_parser.parse_args(args_list)
  92         self.api.command_options = command_options
  93         self.api.read_config()
  94         self.api.bootstrap()
  95
  96         self.api.dispatch(command, command_options, command_args)
  97         with open("/tmp/rspec_output.rspec", "r") as result_file:
  98             result = result_file.read()
  99         return result
 100
 101     def get_resources_info(self):
 102         """
 103         Get all resources and its attributes from aggregate.
 104         """
 105         try:
 106             rspec_slice = self._sfi_exec_method('resources')
 107         except:
 108             raise RuntimeError("Fail to list resources")
 109
 110         self._resources_cache = self.rspec_proc.parse_sfa_rspec(rspec_slice)
 111         self._already_cached = True
 112         return self._resources_cache
 113
 114     def get_resources_hrn(self, resources=None):
 115         """
 116         Get list of resources hrn, without the resource info.
 117         """
 118         if not resources:
 119             if not self._already_cached:
 120                 resources = self.get_resources_info()['resource']
 121             else:
 122                 resources = self._resources_cache['resource']
 123
 124         component_tohrn = dict()
 125         for resource in resources:
 126             hrn = resource['hrn'].replace('\\', '')
 127             component_tohrn[resource['component_name']] = hrn
 128
 129         return component_tohrn
 130
 131     def get_slice_resources(self, slicename):
 132         """
 133         Get resources and info from slice.
 134         """
 135         try:
 136             with self.lock:
 137                 rspec_slice = self._sfi_exec_method('describe', slicename)
 138         except:
 139             raise RuntimeError("Fail to allocate resource for slice %s" % slicename)
 140
 141         result = self.rspec_proc.parse_sfa_rspec(rspec_slice)
 142         return result
 143
 144
 145     def add_resource_to_slice(self, slicename, resource_hrn, leases=None):
 146         """
 147         Get the list of resources' urn, build the rspec string and call the allocate
 148         and provision method.
 149         """
 150         resources_hrn_new = list()
 151         resource_parts = resource_hrn.split('.')
 152         resource_hrn = '.'.join(resource_parts[:2]) + '.' + '\\.'.join(resource_parts[2:])
 153         resources_hrn_new.append(resource_hrn)
 154
 155         slice_resources = self.get_slice_resources(slicename)['resource']
 156
 157         with self.lock:
 158             if slice_resources:
 159                 slice_resources_hrn = self.get_resources_hrn(slice_resources)
 160                 for s_hrn_key, s_hrn_value in slice_resources_hrn.iteritems():
 161                     s_parts = s_hrn_value.split('.')
 162                     s_hrn = '.'.join(s_parts[:2]) + '.' + '\\.'.join(s_parts[2:])
 163                     resources_hrn_new.append(s_hrn)
 164
 165             resources_urn = self._get_resources_urn(resources_hrn_new)
 166             rspec = self.rspec_proc.build_sfa_rspec(slicename, resources_urn, leases)
 167             f = open("/tmp/rspec_input.rspec", "w")
 168             f.truncate(0)
 169             f.write(rspec)
 170             f.close()
 171
 172             if not os.path.getsize("/tmp/rspec_input.rspec") > 0:
 173                 raise RuntimeError("Fail to create rspec file to allocate resource in slice %s" % slicename)
 174
 175             try:
 176                 self._sfi_exec_method('allocate', slicename, "/tmp/rspec_input.rspec")
 177             except:
 178                 raise RuntimeError("Fail to allocate resource for slice %s" % slicename)
 179             try:
 180                 self._sfi_exec_method('provision', slicename)
 181             except:
 182                 raise RuntimeError("Fail to provision resource for slice %s" % slicename)
 183             return True
 184
 185     def remove_resource_from_slice(self, slicename, resource_hrn, leases=None):
 186         """
 187         Get the list of resources' urn, build the rspec string and call the allocate
 188         and provision method.
 189         """
 190         resource_urn = self._get_resources_urn([resource_hrn]).pop()
 191         try:
 192             self._sfi_exec_method('delete', slicename, urn=resource_urn)
 193         except:
 194             raise RuntimeError("Fail to delete resource for slice %s" % slicename)
 195         return True
 196
 197
 198     def _get_resources_urn(self, resources_hrn):
 199         """
 200         Builds list of resources' urn based on hrn.
 201         """
 202         resources_urn = list()
 203
 204         for resource in resources_hrn:
 205             resources_urn.append(hrn_to_urn(resource, 'node'))
 206
 207         return resources_urn
 208
 209     def blacklist_resource(self, resource_hrn):
 210         self._blacklist.add(resource_hrn)
 211
 212     def blacklisted(self):
 213         return self._blacklist
 214
 215     def unblacklist_resource(self, resource_hrn):
 216         del self._blacklist[resource_hrn]
 217
 218     def reserve_resource(self, resource_hrn):
 219         self._reserved.add(resource_hrn)
 220
 221     def reserved(self):
 222         return self._reserved
 223
 224
 225 class SFAAPIFactory(object):
 226     """
 227     API Factory to manage a map of SFAAPI instances as key-value pairs, it
 228     instanciate a single instance per key. The key represents the same SFA,
 229     credentials.
 230     """
 231
 232     _lock = threading.Lock()
 233     _apis = dict()
 234
 235
 236     @classmethod
 237     def get_api(cls, sfi_user, sfi_auth, sfi_registry, sfi_sm, private_key,
 238             timeout = None):
 239
 240         if sfi_user and sfi_sm:
 241             key = cls.make_key(sfi_user, sfi_sm)
 242             with cls._lock:
 243                 api = cls._apis.get(key)
 244
 245                 if not api:
 246                     api = SFAAPI(sfi_user, sfi_auth, sfi_registry, sfi_sm, private_key,
 247                         timeout)
 248                     cls._apis[key] = api
 249
 250                 return api
 251
 252         return None
 253
 254     @classmethod
 255     def make_key(cls, *args):
 256         skey = "".join(map(str, args))
 257         return hashlib.md5(skey).hexdigest()
 258