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