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