src/nepi/execution/resource.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: Alina Quereilhac <alina.quereilhac@inria.fr>
  19
  20 from nepi.util.timefuncs import tnow, tdiff, tdiffsec, stabsformat
  21 from nepi.util.logger import Logger
  22 from nepi.execution.attribute import Attribute, Flags, Types
  23 from nepi.execution.trace import TraceAttr
  24
  25 import copy
  26 import functools
  27 import logging
  28 import os
  29 import pkgutil
  30 import sys
  31 import threading
  32 import weakref
  33
  34 reschedule_delay = "0.5s"
  35
  36 class ResourceAction:
  37     """ Action that a user can order to a Resource Manager
  38
  39     """
  40     DEPLOY = 0
  41     START = 1
  42     STOP = 2
  43
  44 class ResourceState:
  45     """ State of a Resource Manager
  46
  47     """
  48     NEW = 0
  49     DISCOVERED = 1
  50     PROVISIONED = 2
  51     READY = 3
  52     STARTED = 4
  53     STOPPED = 5
  54     FAILED = 6
  55     RELEASED = 7
  56
  57 ResourceState2str = dict({
  58     ResourceState.NEW : "NEW",
  59     ResourceState.DISCOVERED : "DISCOVERED",
  60     ResourceState.PROVISIONED : "PROVISIONED",
  61     ResourceState.READY : "READY",
  62     ResourceState.STARTED : "STARTED",
  63     ResourceState.STOPPED : "STOPPED",
  64     ResourceState.FAILED : "FAILED",
  65     ResourceState.RELEASED : "RELEASED",
  66     })
  67
  68 def clsinit(cls):
  69     """ Initializes template information (i.e. attributes and traces)
  70     on classes derived from the ResourceManager class.
  71
  72     It is used as a decorator in the class declaration as follows:
  73
  74         @clsinit
  75         class MyResourceManager(ResourceManager):
  76
  77             ...
  78
  79      """
  80
  81     cls._clsinit()
  82     return cls
  83
  84 def clsinit_copy(cls):
  85     """ Initializes template information (i.e. attributes and traces)
  86     on classes direved from the ResourceManager class.
  87     It differs from the clsinit method in that it forces inheritance
  88     of attributes and traces from the parent class.
  89
  90     It is used as a decorator in the class declaration as follows:
  91
  92         @clsinit
  93         class MyResourceManager(ResourceManager):
  94
  95             ...
  96
  97
  98     clsinit_copy should be prefered to clsinit when creating new
  99     ResourceManager child classes.
 100
 101     """
 102
 103     cls._clsinit_copy()
 104     return cls
 105
 106 def failtrap(func):
 107     """ Decorator function for instance methods that should set the
 108     RM state to FAILED when an error is raised. The methods that must be
 109     decorated are: discover, provision, deploy, start, stop.
 110
 111     """
 112     def wrapped(self, *args, **kwargs):
 113         try:
 114             return func(self, *args, **kwargs)
 115         except:
 116             self.fail()
 117
 118             import traceback
 119             err = traceback.format_exc()
 120             logger = Logger(self._rtype)
 121             logger.error(err)
 122             logger.error("SETTING guid %d to state FAILED" % self.guid)
 123             raise
 124
 125     return wrapped
 126
 127 @clsinit
 128 class ResourceManager(Logger):
 129     """ Base clase for all ResourceManagers.
 130
 131     A ResourceManger is specific to a resource type (e.g. Node,
 132     Switch, Application, etc) on a specific backend (e.g. PlanetLab,
 133     OMF, etc).
 134
 135     The ResourceManager instances are responsible for interacting with
 136     and controlling concrete (physical or virtual) resources in the
 137     experimental backends.
 138
 139     """
 140     _rtype = "Resource"
 141     _attributes = None
 142     _traces = None
 143     _help = None
 144     _backend = None
 145
 146     @classmethod
 147     def _register_attribute(cls, attr):
 148         """ Resource subclasses will invoke this method to add a
 149         resource attribute
 150
 151         """
 152
 153         cls._attributes[attr.name] = attr
 154
 155     @classmethod
 156     def _remove_attribute(cls, name):
 157         """ Resource subclasses will invoke this method to remove a
 158         resource attribute
 159
 160         """
 161
 162         del cls._attributes[name]
 163
 164     @classmethod
 165     def _register_trace(cls, trace):
 166         """ Resource subclasses will invoke this method to add a
 167         resource trace
 168
 169         """
 170
 171         cls._traces[trace.name] = trace
 172
 173     @classmethod
 174     def _remove_trace(cls, name):
 175         """ Resource subclasses will invoke this method to remove a
 176         resource trace
 177
 178         """
 179
 180         del cls._traces[name]
 181
 182     @classmethod
 183     def _register_attributes(cls):
 184         """ Resource subclasses will invoke this method to register
 185         resource attributes.
 186
 187         This method should be overriden in the RMs that define
 188         attributes.
 189
 190         """
 191         critical = Attribute("critical",
 192                 "Defines whether the resource is critical. "
 193                 "A failure on a critical resource will interrupt "
 194                 "the experiment. ",
 195                 type = Types.Bool,
 196                 default = True,
 197                 flags = Flags.Design)
 198         hard_release = Attribute("hardRelease",
 199                 "Forces removal of all result files and directories associated "
 200                 "to the RM upon resource release. After release the RM will "
 201                 "be removed from the EC and the results will not longer be "
 202                 "accessible",
 203                 type = Types.Bool,
 204                 default = False,
 205                 flags = Flags.Design)
 206
 207         cls._register_attribute(critical)
 208         cls._register_attribute(hard_release)
 209
 210     @classmethod
 211     def _register_traces(cls):
 212         """ Resource subclasses will invoke this method to register
 213         resource traces
 214
 215         This method should be overriden in the RMs that define traces.
 216
 217         """
 218
 219         pass
 220
 221     @classmethod
 222     def _clsinit(cls):
 223         """ ResourceManager classes have different attributes and traces.
 224         Attribute and traces are stored in 'class attribute' dictionaries.
 225         When a new ResourceManager class is created, the _clsinit method is
 226         called to create a new instance of those dictionaries and initialize
 227         them.
 228
 229         The _clsinit method is called by the clsinit decorator method.
 230
 231         """
 232
 233         # static template for resource attributes
 234         cls._attributes = dict()
 235         cls._register_attributes()
 236
 237         # static template for resource traces
 238         cls._traces = dict()
 239         cls._register_traces()
 240
 241     @classmethod
 242     def _clsinit_copy(cls):
 243         """ Same as _clsinit, except that after creating new instances of the
 244         dictionaries it copies all the attributes and traces from the parent
 245         class.
 246
 247         The _clsinit_copy method is called by the clsinit_copy decorator method.
 248
 249         """
 250         # static template for resource attributes
 251         cls._attributes = copy.deepcopy(cls._attributes)
 252         cls._register_attributes()
 253
 254         # static template for resource traces
 255         cls._traces = copy.deepcopy(cls._traces)
 256         cls._register_traces()
 257
 258     @classmethod
 259     def get_rtype(cls):
 260         """ Returns the type of the Resource Manager
 261
 262         """
 263         return cls._rtype
 264
 265     @classmethod
 266     def get_attributes(cls):
 267         """ Returns a copy of the attributes
 268
 269         """
 270         return copy.deepcopy(cls._attributes.values())
 271
 272     @classmethod
 273     def get_attribute(cls, name):
 274         """ Returns a copy of the attribute with name 'name'
 275
 276         """
 277         return copy.deepcopy(cls._attributes[name])
 278
 279     @classmethod
 280     def get_traces(cls):
 281         """ Returns a copy of the traces
 282
 283         """
 284         return copy.deepcopy(cls._traces.values())
 285
 286     @classmethod
 287     def get_help(cls):
 288         """ Returns the description of the type of Resource
 289
 290         """
 291         return cls._help
 292
 293     @classmethod
 294     def get_backend(cls):
 295         """ Returns the identified of the backend (i.e. testbed, environment)
 296         for the Resource
 297
 298         """
 299         return cls._backend
 300
 301     @classmethod
 302     def get_global(cls, name):
 303         """ Returns the value of a global attribute
 304             Global attribute meaning an attribute for
 305             all the resources from a rtype
 306
 307         :param name: Name of the attribute
 308         :type name: str
 309         :rtype: str
 310         """
 311         global_attr = cls._attributes[name]
 312         return global_attr.value
 313
 314     @classmethod
 315     def set_global(cls, name, value):
 316         """ Set value for a global attribute
 317
 318         :param name: Name of the attribute
 319         :type name: str
 320         :param name: Value of the attribute
 321         :type name: str
 322         """
 323         global_attr = cls._attributes[name]
 324         global_attr.value = value
 325         return value
 326
 327     def __init__(self, ec, guid):
 328         super(ResourceManager, self).__init__(self.get_rtype())
 329
 330         self._guid = guid
 331         self._ec = weakref.ref(ec)
 332         self._connections = set()
 333         self._conditions = dict()
 334
 335         # the resource instance gets a copy of all attributes
 336         self._attrs = copy.deepcopy(self._attributes)
 337
 338         # the resource instance gets a copy of all traces
 339         self._trcs = copy.deepcopy(self._traces)
 340
 341         # Each resource is placed on a deployment group by the EC
 342         # during deployment
 343         self.deployment_group = None
 344
 345         self._start_time = None
 346         self._stop_time = None
 347         self._discover_time = None
 348         self._provision_time = None
 349         self._ready_time = None
 350         self._release_time = None
 351         self._failed_time = None
 352
 353         self._state = ResourceState.NEW
 354
 355         # instance lock to synchronize exclusive state change methods (such
 356         # as deploy and release methods), in order to prevent them from being
 357         # executed at the same time and corrupt internal resource state
 358         self._release_lock = threading.Lock()
 359
 360     @property
 361     def guid(self):
 362         """ Returns the global unique identifier of the RM """
 363         return self._guid
 364
 365     @property
 366     def ec(self):
 367         """ Returns the Experiment Controller of the RM """
 368         return self._ec()
 369
 370     @property
 371     def connections(self):
 372         """ Returns the set of guids of connected RMs """
 373         return self._connections
 374
 375     @property
 376     def conditions(self):
 377         """ Returns the conditions to which the RM is subjected to.
 378
 379         This method returns a dictionary of conditions lists indexed by
 380         a ResourceAction.
 381
 382         """
 383         return self._conditions
 384
 385     @property
 386     def start_time(self):
 387         """ Returns the start time of the RM as a timestamp """
 388         return self._start_time
 389
 390     @property
 391     def stop_time(self):
 392         """ Returns the stop time of the RM as a timestamp """
 393         return self._stop_time
 394
 395     @property
 396     def discover_time(self):
 397         """ Returns the discover time of the RM as a timestamp """
 398         return self._discover_time
 399
 400     @property
 401     def provision_time(self):
 402         """ Returns the provision time of the RM as a timestamp """
 403         return self._provision_time
 404
 405     @property
 406     def ready_time(self):
 407         """ Returns the deployment time of the RM as a timestamp """
 408         return self._ready_time
 409
 410     @property
 411     def release_time(self):
 412         """ Returns the release time of the RM as a timestamp """
 413         return self._release_time
 414
 415     @property
 416     def failed_time(self):
 417         """ Returns the time failure occured for the RM as a timestamp """
 418         return self._failed_time
 419
 420     @property
 421     def state(self):
 422         """ Get the current state of the RM """
 423         return self._state
 424
 425     def log_message(self, msg):
 426         """ Returns the log message formatted with added information.
 427
 428         :param msg: text message
 429         :type msg: str
 430         :rtype: str
 431
 432         """
 433         return " %s guid %d - %s " % (self._rtype, self.guid, msg)
 434
 435     def register_connection(self, guid):
 436         """ Registers a connection to the RM identified by guid
 437
 438         This method should not be overriden. Specific functionality
 439         should be added in the do_connect method.
 440
 441         :param guid: Global unique identified of the RM to connect to
 442         :type guid: int
 443
 444         """
 445         if self.valid_connection(guid):
 446             self.do_connect(guid)
 447             self._connections.add(guid)
 448
 449     def unregister_connection(self, guid):
 450         """ Removes a registered connection to the RM identified by guid
 451
 452         This method should not be overriden. Specific functionality
 453         should be added in the do_disconnect method.
 454
 455         :param guid: Global unique identified of the RM to connect to
 456         :type guid: int
 457
 458         """
 459         if guid in self._connections:
 460             self.do_disconnect(guid)
 461             self._connections.remove(guid)
 462
 463     @failtrap
 464     def discover(self):
 465         """ Performs resource discovery.
 466
 467         This  method is responsible for selecting an individual resource
 468         matching user requirements.
 469
 470         This method should not be overriden directly. Specific functionality
 471         should be added in the do_discover method.
 472
 473         """
 474         with self._release_lock:
 475             if self._state != ResourceState.RELEASED:
 476                 self.do_discover()
 477
 478     @failtrap
 479     def provision(self):
 480         """ Performs resource provisioning.
 481
 482         This  method is responsible for provisioning one resource.
 483         After this method has been successfully invoked, the resource
 484         should be accessible/controllable by the RM.
 485
 486         This method should not be overriden directly. Specific functionality
 487         should be added in the do_provision method.
 488
 489         """
 490         with self._release_lock:
 491             if self._state != ResourceState.RELEASED:
 492                 self.do_provision()
 493
 494     @failtrap
 495     def start(self):
 496         """ Starts the RM (e.g. launch remote process).
 497
 498         There is no standard start behavior. Some RMs will not need to perform
 499         any actions upon start.
 500
 501         This method should not be overriden directly. Specific functionality
 502         should be added in the do_start method.
 503
 504         """
 505
 506         if not self.state in [ResourceState.READY, ResourceState.STOPPED]:
 507             self.error("Wrong state %s for start" % self.state)
 508             return
 509
 510         with self._release_lock:
 511             if self._state != ResourceState.RELEASED:
 512                 self.do_start()
 513
 514     @failtrap
 515     def stop(self):
 516         """ Interrupts the RM, stopping any tasks the RM was performing.
 517
 518         There is no standard stop behavior. Some RMs will not need to perform
 519         any actions upon stop.
 520
 521         This method should not be overriden directly. Specific functionality
 522         should be added in the do_stop method.
 523
 524         """
 525         if not self.state in [ResourceState.STARTED]:
 526             self.error("Wrong state %s for stop" % self.state)
 527             return
 528
 529         with self._release_lock:
 530             self.do_stop()
 531
 532     @failtrap
 533     def deploy(self):
 534         """ Execute all steps required for the RM to reach the state READY.
 535
 536         This method is responsible for deploying the resource (and invoking
 537         the discover and provision methods).
 538
 539         This method should not be overriden directly. Specific functionality
 540         should be added in the do_deploy method.
 541
 542         """
 543         if self.state > ResourceState.READY:
 544             self.error("Wrong state %s for deploy" % self.state)
 545             return
 546
 547         with self._release_lock:
 548             if self._state != ResourceState.RELEASED:
 549                 self.do_deploy()
 550
 551     def release(self):
 552         """ Perform actions to free resources used by the RM.
 553
 554         This  method is responsible for releasing resources that were
 555         used during the experiment by the RM.
 556
 557         This method should not be overriden directly. Specific functionality
 558         should be added in the do_release method.
 559
 560         """
 561         with self._release_lock:
 562             try:
 563                 self.do_release()
 564             except:
 565                 self.set_released()
 566
 567                 import traceback
 568                 err = traceback.format_exc()
 569                 msg = " %s guid %d ----- FAILED TO RELEASE ----- \n %s " % (
 570                         self._rtype, self.guid, err)
 571                 logger = Logger(self._rtype)
 572                 logger.debug(msg)
 573
 574     def fail(self):
 575         """ Sets the RM to state FAILED.
 576
 577         This method should not be overriden directly. Specific functionality
 578         should be added in the do_fail method.
 579
 580         """
 581         with self._release_lock:
 582             if self._state != ResourceState.RELEASED:
 583                 self.do_fail()
 584
 585     def set(self, name, value):
 586         """ Set the value of the attribute
 587
 588         :param name: Name of the attribute
 589         :type name: str
 590         :param name: Value of the attribute
 591         :type name: str
 592         """
 593         attr = self._attrs[name]
 594         attr.value = value
 595         return value
 596
 597     def get(self, name):
 598         """ Returns the value of the attribute
 599
 600         :param name: Name of the attribute
 601         :type name: str
 602         :rtype: str
 603         """
 604         attr = self._attrs[name]
 605
 606         """
 607         A.Q. Commenting due to performance impact
 608         if attr.has_flag(Flags.Global):
 609             self.warning( "Attribute %s is global. Use get_global instead." % name)
 610         """
 611
 612         return attr.value
 613
 614     def has_changed(self, name):
 615         """ Returns the True is the value of the attribute
 616             has been modified by the user.
 617
 618         :param name: Name of the attribute
 619         :type name: str
 620         :rtype: str
 621         """
 622         attr = self._attrs[name]
 623         return attr.has_changed
 624
 625     def has_flag(self, name, flag):
 626         """ Returns true if the attribute has the flag 'flag'
 627
 628         :param flag: Flag to be checked
 629         :type flag: Flags
 630         """
 631         attr = self._attrs[name]
 632         return attr.has_flag(flag)
 633
 634     def has_attribute(self, name):
 635         """ Returns true if the RM has an attribute with name
 636
 637         :param name: name of the attribute
 638         :type name: string
 639         """
 640         return name in self._attrs
 641
 642     def enable_trace(self, name):
 643         """ Explicitly enable trace generation
 644
 645         :param name: Name of the trace
 646         :type name: str
 647         """
 648         trace = self._trcs[name]
 649         trace.enabled = True
 650
 651     def trace_enabled(self, name):
 652         """Returns True if trace is enables
 653
 654         :param name: Name of the trace
 655         :type name: str
 656         """
 657         trace = self._trcs[name]
 658         return trace.enabled
 659
 660     def trace(self, name, attr = TraceAttr.ALL, block = 512, offset = 0):
 661         """ Get information on collected trace
 662
 663         :param name: Name of the trace
 664         :type name: str
 665
 666         :param attr: Can be one of:
 667                          - TraceAttr.ALL (complete trace content),
 668                          - TraceAttr.STREAM (block in bytes to read starting at offset),
 669                          - TraceAttr.PATH (full path to the trace file),
 670                          - TraceAttr.SIZE (size of trace file).
 671         :type attr: str
 672
 673         :param block: Number of bytes to retrieve from trace, when attr is TraceAttr.STREAM
 674         :type name: int
 675
 676         :param offset: Number of 'blocks' to skip, when attr is TraceAttr.STREAM
 677         :type name: int
 678
 679         :rtype: str
 680         """
 681         pass
 682
 683     def register_condition(self, action, group, state, time = None):
 684         """ Registers a condition on the resource manager to allow execution
 685         of 'action' only after 'time' has elapsed from the moment all resources
 686         in 'group' reached state 'state'
 687
 688         :param action: Action to restrict to condition (either 'START' or 'STOP')
 689         :type action: str
 690         :param group: Group of RMs to wait for (list of guids)
 691         :type group: int or list of int
 692         :param state: State to wait for on all RM in group. (either 'STARTED', 'STOPPED' or 'READY')
 693         :type state: str
 694         :param time: Time to wait after 'state' is reached on all RMs in group. (e.g. '2s')
 695         :type time: str
 696
 697         """
 698
 699         if not action in self.conditions:
 700             self._conditions[action] = list()
 701
 702         conditions = self.conditions.get(action)
 703
 704         # For each condition to register a tuple of (group, state, time) is
 705         # added to the 'action' list
 706         if not isinstance(group, list):
 707             group = [group]
 708
 709         conditions.append((group, state, time))
 710
 711     def unregister_condition(self, group, action = None):
 712         """ Removed conditions for a certain group of guids
 713
 714         :param action: Action to restrict to condition (either 'START', 'STOP' or 'READY')
 715         :type action: str
 716
 717         :param group: Group of RMs to wait for (list of guids)
 718         :type group: int or list of int
 719
 720         """
 721         # For each condition a tuple of (group, state, time) is
 722         # added to the 'action' list
 723         if not isinstance(group, list):
 724             group = [group]
 725
 726         for act, conditions in self.conditions.iteritems():
 727             if action and act != action:
 728                 continue
 729
 730             for condition in list(conditions):
 731                 (grp, state, time) = condition
 732
 733                 # If there is an intersection between grp and group,
 734                 # then remove intersected elements
 735                 intsec = set(group).intersection(set(grp))
 736                 if intsec:
 737                     idx = conditions.index(condition)
 738                     newgrp = set(grp)
 739                     newgrp.difference_update(intsec)
 740                     conditions[idx] = (newgrp, state, time)
 741
 742     def get_connected(self, rtype = None):
 743         """ Returns the list of RM with the type 'rtype'
 744
 745         :param rtype: Type of the RM we look for
 746         :type rtype: str
 747         :return: list of guid
 748         """
 749         connected = []
 750         rclass = ResourceFactory.get_resource_type(rtype)
 751         for guid in self.connections:
 752             rm = self.ec.get_resource(guid)
 753             if not rtype or isinstance(rm, rclass):
 754                 connected.append(rm)
 755         return connected
 756
 757     def is_rm_instance(self, rtype):
 758         """ Returns True if the RM is instance of 'rtype'
 759
 760         :param rtype: Type of the RM we look for
 761         :type rtype: str
 762         :return: True|False
 763         """
 764         rclass = ResourceFactory.get_resource_type(rtype)
 765         if isinstance(self, rclass):
 766             return True
 767         return False
 768
 769     @failtrap
 770     def _needs_reschedule(self, group, state, time):
 771         """ Internal method that verify if 'time' has elapsed since
 772         all elements in 'group' have reached state 'state'.
 773
 774         :param group: Group of RMs to wait for (list of guids)
 775         :type group: int or list of int
 776         :param state: State to wait for on all RM in group. (either 'STARTED', 'STOPPED' or 'READY')
 777         :type state: str
 778         :param time: Time to wait after 'state' is reached on all RMs in group. (e.g. '2s')
 779         :type time: str
 780
 781         .. note : time should be written like "2s" or "3m" with s for seconds, m for minutes, h for hours, ...
 782         If for example, you need to wait 2min 30sec, time could be "150s" or "2.5m".
 783         For the moment, 2m30s is not a correct syntax.
 784
 785         """
 786         reschedule = False
 787         delay = reschedule_delay
 788
 789         # check state and time elapsed on all RMs
 790         for guid in group:
 791             rm = self.ec.get_resource(guid)
 792
 793             # If one of the RMs this resource needs to wait for has FAILED
 794             # and is critical we raise an exception
 795             if rm.state == ResourceState.FAILED:
 796                 if not rm.get('critical'):
 797                     continue
 798                 msg = "Resource can not wait for FAILED RM %d. Setting Resource to FAILED"
 799                 raise RuntimeError, msg
 800
 801             # If the RM state is lower than the requested state we must
 802             # reschedule (e.g. if RM is READY but we required STARTED).
 803             if rm.state < state:
 804                 reschedule = True
 805                 break
 806
 807             # If there is a time restriction, we must verify the
 808             # restriction is satisfied
 809             if time:
 810                 if state == ResourceState.DISCOVERED:
 811                     t = rm.discover_time
 812                 if state == ResourceState.PROVISIONED:
 813                     t = rm.provision_time
 814                 elif state == ResourceState.READY:
 815                     t = rm.ready_time
 816                 elif state == ResourceState.STARTED:
 817                     t = rm.start_time
 818                 elif state == ResourceState.STOPPED:
 819                     t = rm.stop_time
 820                 elif state == ResourceState.RELEASED:
 821                     t = rm.release_time
 822                 else:
 823                     break
 824
 825                 # time already elapsed since RM changed state
 826                 waited = "%fs" % tdiffsec(tnow(), t)
 827
 828                 # time still to wait
 829                 wait = tdiffsec(stabsformat(time), stabsformat(waited))
 830
 831                 if wait > 0.001:
 832                     reschedule = True
 833                     delay = "%fs" % wait
 834                     break
 835
 836         return reschedule, delay
 837
 838     def set_with_conditions(self, name, value, group, state, time):
 839         """ Set value 'value' on attribute with name 'name' when 'time'
 840         has elapsed since all elements in 'group' have reached state
 841         'state'
 842
 843         :param name: Name of the attribute to set
 844         :type name: str
 845         :param name: Value of the attribute to set
 846         :type name: str
 847         :param group: Group of RMs to wait for (list of guids)
 848         :type group: int or list of int
 849         :param state: State to wait for on all RM in group. (either 'STARTED', 'STOPPED' or 'READY')
 850         :type state: str
 851         :param time: Time to wait after 'state' is reached on all RMs in group. (e.g. '2s')
 852         :type time: str
 853         """
 854
 855         reschedule = False
 856         delay = reschedule_delay
 857
 858         ## evaluate if set conditions are met
 859
 860         # only can set with conditions after the RM is started
 861         if self.state != ResourceState.STARTED:
 862             reschedule = True
 863         else:
 864             reschedule, delay = self._needs_reschedule(group, state, time)
 865
 866         if reschedule:
 867             callback = functools.partial(self.set_with_conditions,
 868                     name, value, group, state, time)
 869             self.ec.schedule(delay, callback)
 870         else:
 871             self.set(name, value)
 872
 873     def start_with_conditions(self):
 874         """ Starts RM when all the conditions in self.conditions for
 875         action 'START' are satisfied.
 876
 877         """
 878         #import pdb;pdb.set_trace()
 879
 880         reschedule = False
 881         delay = reschedule_delay
 882
 883
 884         ## evaluate if conditions to start are met
 885         if self.ec.abort:
 886             return
 887
 888         # Can only start when RM is either STOPPED or READY
 889         if self.state not in [ResourceState.STOPPED, ResourceState.READY]:
 890             reschedule = True
 891             self.debug("---- RESCHEDULING START ---- state %s " % self.state )
 892         else:
 893             start_conditions = self.conditions.get(ResourceAction.START, [])
 894
 895             self.debug("---- START CONDITIONS ---- %s" % start_conditions)
 896
 897             # Verify all start conditions are met
 898             for (group, state, time) in start_conditions:
 899                 # Uncomment for debug
 900                 #unmet = []
 901                 #for guid in group:
 902                 #    rm = self.ec.get_resource(guid)
 903                 #    unmet.append((guid, rm._state))
 904                 #
 905                 #self.debug("---- WAITED STATES ---- %s" % unmet )
 906
 907                 reschedule, delay = self._needs_reschedule(group, state, time)
 908                 if reschedule:
 909                     break
 910
 911         if reschedule:
 912             self.ec.schedule(delay, self.start_with_conditions)
 913         else:
 914             self.debug("----- STARTING ---- ")
 915             self.start()
 916
 917     def stop_with_conditions(self):
 918         """ Stops RM when all the conditions in self.conditions for
 919         action 'STOP' are satisfied.
 920
 921         """
 922         reschedule = False
 923         delay = reschedule_delay
 924
 925         ## evaluate if conditions to stop are met
 926         if self.ec.abort:
 927             return
 928
 929         # only can stop when RM is STARTED
 930         if self.state != ResourceState.STARTED:
 931             reschedule = True
 932             self.debug("---- RESCHEDULING STOP ---- state %s " % self.state )
 933         else:
 934             self.debug(" ---- STOP CONDITIONS ---- %s" %
 935                     self.conditions.get(ResourceAction.STOP))
 936
 937             stop_conditions = self.conditions.get(ResourceAction.STOP, [])
 938             for (group, state, time) in stop_conditions:
 939                 reschedule, delay = self._needs_reschedule(group, state, time)
 940                 if reschedule:
 941                     break
 942
 943         if reschedule:
 944             callback = functools.partial(self.stop_with_conditions)
 945             self.ec.schedule(delay, callback)
 946         else:
 947             self.debug(" ----- STOPPING ---- ")
 948             self.stop()
 949
 950     def deploy_with_conditions(self):
 951         """ Deploy RM when all the conditions in self.conditions for
 952         action 'READY' are satisfied.
 953
 954         """
 955         reschedule = False
 956         delay = reschedule_delay
 957
 958         ## evaluate if conditions to deploy are met
 959         if self.ec.abort:
 960             return
 961
 962         # only can deploy when RM is either NEW, DISCOVERED or PROVISIONED
 963         if self.state not in [ResourceState.NEW, ResourceState.DISCOVERED,
 964                 ResourceState.PROVISIONED]:
 965             #### XXX: A.Q. IT SHOULD FAIL IF DEPLOY IS CALLED IN OTHER STATES!
 966             reschedule = True
 967             self.debug("---- RESCHEDULING DEPLOY ---- state %s " % self.state )
 968         else:
 969             deploy_conditions = self.conditions.get(ResourceAction.DEPLOY, [])
 970
 971             self.debug("---- DEPLOY CONDITIONS ---- %s" % deploy_conditions)
 972
 973             # Verify all start conditions are met
 974             for (group, state, time) in deploy_conditions:
 975                 # Uncomment for debug
 976                 #unmet = []
 977                 #for guid in group:
 978                 #    rm = self.ec.get_resource(guid)
 979                 #    unmet.append((guid, rm._state))
 980
 981                 #self.debug("---- WAITED STATES ---- %s" % unmet )
 982
 983                 reschedule, delay = self._needs_reschedule(group, state, time)
 984                 if reschedule:
 985                     break
 986
 987         if reschedule:
 988             self.ec.schedule(delay, self.deploy_with_conditions)
 989         else:
 990             self.debug("----- DEPLOYING ---- ")
 991             self.deploy()
 992
 993     def do_connect(self, guid):
 994         """ Performs actions that need to be taken upon associating RMs.
 995         This method should be redefined when necessary in child classes.
 996         """
 997         pass
 998
 999     def do_disconnect(self, guid):
1000         """ Performs actions that need to be taken upon disassociating RMs.
1001         This method should be redefined when necessary in child classes.
1002         """
1003         pass
1004
1005     def valid_connection(self, guid):
1006         """Checks whether a connection with the other RM
1007         is valid.
1008         This method need to be redefined by each new Resource Manager.
1009
1010         :param guid: Guid of the current Resource Manager
1011         :type guid: int
1012         :rtype:  Boolean
1013
1014         """
1015         # TODO: Validate!
1016         return True
1017
1018     def do_discover(self):
1019         self.set_discovered()
1020
1021     def do_provision(self):
1022         self.set_provisioned()
1023
1024     def do_start(self):
1025         self.set_started()
1026
1027     def do_stop(self):
1028         self.set_stopped()
1029
1030     def do_deploy(self):
1031         self.set_ready()
1032
1033     def do_release(self):
1034         self.set_released()
1035
1036     def do_fail(self):
1037         self.set_failed()
1038         self.ec.inform_failure(self.guid)
1039
1040     def set_started(self, time = None):
1041         """ Mark ResourceManager as STARTED """
1042         self.set_state(ResourceState.STARTED, "_start_time", time)
1043         self.debug("----- STARTED ---- ")
1044
1045     def set_stopped(self, time = None):
1046         """ Mark ResourceManager as STOPPED """
1047         self.set_state(ResourceState.STOPPED, "_stop_time", time)
1048         self.debug("----- STOPPED ---- ")
1049
1050     def set_ready(self, time = None):
1051         """ Mark ResourceManager as READY """
1052         self.set_state(ResourceState.READY, "_ready_time", time)
1053         self.debug("----- READY ---- ")
1054
1055     def set_released(self, time = None):
1056         """ Mark ResourceManager as REALEASED """
1057         self.set_state(ResourceState.RELEASED, "_release_time", time)
1058
1059         msg = " %s guid %d ----- RELEASED ----- " % (self._rtype, self.guid)
1060         logger = Logger(self._rtype)
1061         logger.debug(msg)
1062
1063     def set_failed(self, time = None):
1064         """ Mark ResourceManager as FAILED """
1065         self.set_state(ResourceState.FAILED, "_failed_time", time)
1066
1067         msg = " %s guid %d ----- FAILED ----- " % (self._rtype, self.guid)
1068         logger = Logger(self._rtype)
1069         logger.debug(msg)
1070
1071     def set_discovered(self, time = None):
1072         """ Mark ResourceManager as DISCOVERED """
1073         self.set_state(ResourceState.DISCOVERED, "_discover_time", time)
1074         self.debug("----- DISCOVERED ---- ")
1075
1076     def set_provisioned(self, time = None):
1077         """ Mark ResourceManager as PROVISIONED """
1078         self.set_state(ResourceState.PROVISIONED, "_provision_time", time)
1079         self.debug("----- PROVISIONED ---- ")
1080
1081     def set_state(self, state, state_time_attr, time = None):
1082         """ Set the state of the RM while keeping a trace of the time """
1083
1084         # Ensure that RM state will not change after released
1085         if self._state == ResourceState.RELEASED:
1086             return
1087
1088         time = time or tnow()
1089         self.set_state_time(state, state_time_attr, time)
1090
1091     def set_state_time(self, state, state_time_attr, time):
1092         """ Set the time for the RM state change """
1093         setattr(self, state_time_attr, time)
1094         self._state = state
1095
1096 class ResourceFactory(object):
1097     _resource_types = dict()
1098
1099     @classmethod
1100     def resource_types(cls):
1101         """Return the type of the Class"""
1102         return cls._resource_types
1103
1104     @classmethod
1105     def get_resource_type(cls, rtype):
1106         """Return the type of the Class"""
1107         return cls._resource_types.get(rtype)
1108
1109     @classmethod
1110     def register_type(cls, rclass):
1111         """Register a new Ressource Manager"""
1112         cls._resource_types[rclass.get_rtype()] = rclass
1113
1114     @classmethod
1115     def create(cls, rtype, ec, guid):
1116         """Create a new instance of a Ressource Manager"""
1117         rclass = cls._resource_types[rtype]
1118         return rclass(ec, guid)
1119
1120 def populate_factory():
1121     """Find and rgister all available RMs
1122     """
1123     # Once the factory is populated, don't repopulate
1124     if not ResourceFactory.resource_types():
1125         for rclass in find_types():
1126             ResourceFactory.register_type(rclass)
1127
1128 def find_types():
1129     """Look into the different folders to find all the
1130     availables Resources Managers
1131     """
1132     search_path = os.environ.get("NEPI_SEARCH_PATH", "")
1133     search_path = set(search_path.split(" "))
1134
1135     import inspect
1136     import nepi.resources
1137     path = os.path.dirname(nepi.resources.__file__)
1138     search_path.add(path)
1139
1140     types = set()
1141
1142     for importer, modname, ispkg in pkgutil.walk_packages(search_path,
1143             prefix = "nepi.resources."):
1144
1145         loader = importer.find_module(modname)
1146
1147         try:
1148             # Notice: Repeated calls to load_module will act as a reload of the module
1149             if modname in sys.modules:
1150                 module = sys.modules.get(modname)
1151             else:
1152                 module = loader.load_module(modname)
1153
1154             for attrname in dir(module):
1155                 if attrname.startswith("_"):
1156                     continue
1157
1158                 attr = getattr(module, attrname)
1159
1160                 if attr == ResourceManager:
1161                     continue
1162
1163                 if not inspect.isclass(attr):
1164                     continue
1165
1166                 if issubclass(attr, ResourceManager):
1167                     types.add(attr)
1168
1169                     if not modname in sys.modules:
1170                         sys.modules[modname] = module
1171
1172         except:
1173             import traceback
1174             import logging
1175             err = traceback.format_exc()
1176             logger = logging.getLogger("Resource.find_types()")
1177             logger.error("Error while loading Resource Managers %s" % err)
1178
1179     return types
1180