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