test/execution/resource.py

   1 #!/usr/bin/env python
   2 #
   3 #    NEPI, a framework to manage network experiments
   4 #    Copyright (C) 2013 INRIA
   5 #
   6 #    This program is free software: you can redistribute it and/or modify
   7 #    it under the terms of the GNU General Public License version 2 as
   8 #    published by the Free Software Foundation;
   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
  21 from nepi.execution.attribute import Attribute
  22 from nepi.execution.ec import ExperimentController, FailureLevel
  23 from nepi.execution.resource import ResourceManager, ResourceState, \
  24         clsinit_copy, ResourceAction
  25
  26 import random
  27 import time
  28 import unittest
  29
  30 @clsinit_copy
  31 class MyResource(ResourceManager):
  32     _rtype = "MyResource"
  33
  34     @classmethod
  35     def _register_attributes(cls):
  36         cool_attr = Attribute("my_attr", "is a really nice attribute!")
  37         cls._register_attribute(cool_attr)
  38
  39     def __init__(self, ec, guid):
  40         super(MyResource, self).__init__(ec, guid)
  41
  42 @clsinit_copy
  43 class AnotherResource(ResourceManager):
  44     _rtype = "AnotherResource"
  45
  46     def __init__(self, ec, guid):
  47         super(AnotherResource, self).__init__(ec, guid)
  48
  49 class Channel(ResourceManager):
  50     _rtype = "Channel"
  51
  52     def __init__(self, ec, guid):
  53         super(Channel, self).__init__(ec, guid)
  54
  55     def do_deploy(self):
  56         time.sleep(1)
  57         super(Channel, self).do_deploy()
  58         self.logger.debug(" -------- DEPLOYED ------- ")
  59
  60 class Interface(ResourceManager):
  61     _rtype = "Interface"
  62
  63     def __init__(self, ec, guid):
  64         super(Interface, self).__init__(ec, guid)
  65
  66     def do_deploy(self):
  67         node = self.get_connected(Node.get_rtype())[0]
  68         chan = self.get_connected(Channel.get_rtype())[0]
  69
  70         if node.state < ResourceState.PROVISIONED:
  71             self.ec.schedule("0.5s", self.deploy)
  72         elif chan.state < ResourceState.READY:
  73             self.ec.schedule("0.5s", self.deploy)
  74         else:
  75             time.sleep(2)
  76             super(Interface, self).do_deploy()
  77             self.logger.debug(" -------- DEPLOYED ------- ")
  78
  79 class Node(ResourceManager):
  80     _rtype = "Node"
  81
  82     def __init__(self, ec, guid):
  83         super(Node, self).__init__(ec, guid)
  84
  85     def do_deploy(self):
  86         if self.state == ResourceState.NEW:
  87             self.do_discover()
  88             self.do_provision()
  89             self.logger.debug(" -------- PROVISIONED ------- ")
  90             self.ec.schedule("1s", self.deploy)
  91         elif self.state == ResourceState.PROVISIONED:
  92             ifaces = self.get_connected(Interface.get_rtype())
  93             for rm in ifaces:
  94                 if rm.state < ResourceState.READY:
  95                     self.ec.schedule("0.5s", self.deploy)
  96                     return
  97
  98             super(Node, self).do_deploy()
  99             self.logger.debug(" -------- DEPLOYED ------- ")
 100
 101 class Application(ResourceManager):
 102     _rtype = "Application"
 103
 104     def __init__(self, ec, guid):
 105         super(Application, self).__init__(ec, guid)
 106
 107     def do_deploy(self):
 108         node = self.get_connected(Node.get_rtype())[0]
 109         if node.state < ResourceState.READY:
 110             self.ec.schedule("0.5s", self.deploy)
 111         else:
 112             time.sleep(random.random() * 2)
 113             super(Application, self).do_deploy()
 114             self.logger.debug(" -------- DEPLOYED ------- ")
 115
 116     def do_start(self):
 117         super(Application, self).do_start()
 118         time.sleep(random.random() * 3)
 119         self.ec.schedule("0.5s", self.stop)
 120
 121 class ErrorApplication(ResourceManager):
 122     _rtype = "ErrorApplication"
 123
 124     def __init__(self, ec, guid):
 125         super(ErrorApplication, self).__init__(ec, guid)
 126
 127     def do_deploy(self):
 128         node = self.get_connected(Node.get_rtype())[0]
 129         if node.state < ResourceState.READY:
 130             self.ec.schedule("0.5s", self.deploy)
 131         else:
 132             time.sleep(random.random() * 2)
 133             raise RuntimeError("NOT A REAL ERROR. JUST TESTING")
 134
 135 class ResourceFactoryTestCase(unittest.TestCase):
 136     def test_add_resource_factory(self):
 137         from nepi.execution.resource import ResourceFactory
 138
 139         ResourceFactory._resource_types = dict()
 140         ResourceFactory.register_type(MyResource)
 141         ResourceFactory.register_type(AnotherResource)
 142
 143         # Take into account default 'Critical' attribute
 144         self.assertEquals(MyResource.get_rtype(), "MyResource")
 145         self.assertEquals(len(MyResource._attributes), 3)
 146
 147         self.assertEquals(ResourceManager.get_rtype(), "Resource")
 148         self.assertEquals(len(ResourceManager._attributes), 2)
 149
 150         self.assertEquals(AnotherResource.get_rtype(), "AnotherResource")
 151         self.assertEquals(len(AnotherResource._attributes), 2)
 152
 153         self.assertEquals(len(ResourceFactory.resource_types()), 2)
 154
 155         # restore factory state for other tests
 156         from nepi.execution.resource import populate_factory
 157         ResourceFactory._resource_types = dict()
 158         populate_factory()
 159
 160 class ResourceManagerTestCase(unittest.TestCase):
 161     def test_register_condition(self):
 162         ec = ExperimentController()
 163         rm = ResourceManager(ec, 15)
 164
 165         group = [1,3,5,7]
 166         rm.register_condition(ResourceAction.START, group,
 167                 ResourceState.STARTED)
 168
 169         group = [10,8]
 170         rm.register_condition(ResourceAction.START,
 171                 group, ResourceState.STARTED, time = "10s")
 172
 173         waiting_for = []
 174         conditions = rm.conditions.get(ResourceAction.START)
 175         for (group, state, time) in conditions:
 176             waiting_for.extend(group)
 177
 178         self.assertEquals(waiting_for, [1, 3, 5, 7, 10, 8])
 179
 180         group = [1, 2, 3, 4, 6]
 181         rm.unregister_condition(group)
 182
 183         waiting_for = []
 184         conditions = rm.conditions.get(ResourceAction.START)
 185         for (group, state, time) in conditions:
 186             waiting_for.extend(group)
 187
 188         self.assertEquals(waiting_for, [5, 7, 10, 8])
 189
 190     def test_deploy_in_order(self):
 191         """
 192         Test scenario: 2 applications running one on 1 node each.
 193         Nodes are connected to Interfaces which are connected
 194         through a channel between them.
 195
 196          - Application needs to wait until Node is ready to be ready
 197          - Node needs to wait until Interface is ready to be ready
 198          - Interface needs to wait until Node is provisioned to be ready
 199          - Interface needs to wait until Channel is ready to be ready
 200          - The channel doesn't wait for any other resource to be ready
 201
 202         """
 203         from nepi.execution.resource import ResourceFactory
 204
 205         ResourceFactory.register_type(Application)
 206         ResourceFactory.register_type(Node)
 207         ResourceFactory.register_type(Interface)
 208         ResourceFactory.register_type(Channel)
 209
 210         ec = ExperimentController()
 211
 212         app1 = ec.register_resource("Application")
 213         app2 = ec.register_resource("Application")
 214         node1 = ec.register_resource("Node")
 215         node2 = ec.register_resource("Node")
 216         iface1 = ec.register_resource("Interface")
 217         iface2 = ec.register_resource("Interface")
 218         chan = ec.register_resource("Channel")
 219
 220         ec.register_connection(app1, node1)
 221         ec.register_connection(app2, node2)
 222         ec.register_connection(iface1, node1)
 223         ec.register_connection(iface2, node2)
 224         ec.register_connection(iface1, chan)
 225         ec.register_connection(iface2, chan)
 226
 227         ec.deploy()
 228
 229         guids = [app1, app2]
 230         ec.wait_finished(guids)
 231
 232         ec.shutdown()
 233
 234         rmapp1 = ec.get_resource(app1)
 235         rmapp2 = ec.get_resource(app2)
 236         rmnode1 = ec.get_resource(node1)
 237         rmnode2 = ec.get_resource(node2)
 238         rmiface1 = ec.get_resource(iface1)
 239         rmiface2 = ec.get_resource(iface2)
 240         rmchan = ec.get_resource(chan)
 241
 242         ## Validate deploy order
 243         # - Application needs to wait until Node is ready to be ready
 244         self.assertTrue(rmnode1.ready_time < rmapp1.ready_time)
 245         self.assertTrue(rmnode2.ready_time < rmapp2.ready_time)
 246
 247         # - Node needs to wait until Interface is ready to be ready
 248         self.assertTrue(rmnode1.ready_time > rmiface1.ready_time)
 249         self.assertTrue(rmnode2.ready_time > rmiface2.ready_time)
 250
 251         # - Interface needs to wait until Node is provisioned to be ready
 252         self.assertTrue(rmnode1.provision_time < rmiface1.ready_time)
 253         self.assertTrue(rmnode2.provision_time < rmiface2.ready_time)
 254
 255         # - Interface needs to wait until Channel is ready to be ready
 256         self.assertTrue(rmchan.ready_time < rmiface1.ready_time)
 257         self.assertTrue(rmchan.ready_time < rmiface2.ready_time)
 258
 259     def test_concurrency(self):
 260         from nepi.execution.resource import ResourceFactory
 261
 262         ResourceFactory.register_type(Application)
 263         ResourceFactory.register_type(Node)
 264         ResourceFactory.register_type(Interface)
 265         ResourceFactory.register_type(Channel)
 266
 267         ec = ExperimentController()
 268
 269         node = ec.register_resource("Node")
 270
 271         apps = list()
 272         for i in xrange(1000):
 273             app = ec.register_resource("Application")
 274             ec.register_connection(app, node)
 275             apps.append(app)
 276
 277         ec.deploy()
 278
 279         ec.wait_finished(apps)
 280
 281         self.assertTrue(ec.state(node) == ResourceState.STARTED)
 282         self.assertTrue(
 283                all([ec.state(guid) == ResourceState.STOPPED \
 284                 for guid in apps])
 285                 )
 286
 287         ec.shutdown()
 288
 289     def test_exception(self):
 290         from nepi.execution.resource import ResourceFactory
 291
 292         ResourceFactory.register_type(ErrorApplication)
 293         ResourceFactory.register_type(Node)
 294         ResourceFactory.register_type(Interface)
 295
 296         ec = ExperimentController()
 297
 298         node = ec.register_resource("Node")
 299
 300         app = ec.register_resource("ErrorApplication")
 301         ec.register_connection(app, node)
 302
 303         ec.deploy()
 304
 305         ec.wait_finished(app)
 306
 307         ec.shutdown()
 308
 309         self.assertEquals(ec._fm._failure_level, FailureLevel.RM_FAILURE)
 310
 311     def test_critical(self):
 312         from nepi.execution.resource import ResourceFactory
 313
 314         ResourceFactory.register_type(ErrorApplication)
 315         ResourceFactory.register_type(Application)
 316         ResourceFactory.register_type(Node)
 317         ResourceFactory.register_type(Interface)
 318
 319         ec = ExperimentController()
 320
 321         node = ec.register_resource("Node")
 322
 323         apps = list()
 324
 325         eapp = ec.register_resource("ErrorApplication")
 326         ec.set(eapp, "critical", False)
 327         ec.register_connection(eapp, node)
 328         apps.append(eapp)
 329
 330         for i in xrange(10):
 331             app = ec.register_resource("Application")
 332             ec.register_connection(app, node)
 333             apps.append(app)
 334
 335         ec.deploy()
 336
 337         ec.wait_finished(apps)
 338
 339         state = ec.state(eapp)
 340         self.assertEquals(state, ResourceState.FAILED)
 341
 342         apps.remove(eapp)
 343
 344         for app in apps:
 345             state = ec.state(app)
 346             self.assertEquals(state, ResourceState.STOPPED)
 347
 348         ec.shutdown()
 349
 350         self.assertEquals(ec._fm._failure_level, FailureLevel.OK)
 351
 352     def test_start_with_condition(self):
 353         from nepi.execution.resource import ResourceFactory
 354
 355         ResourceFactory.register_type(Application)
 356         ResourceFactory.register_type(Node)
 357         ResourceFactory.register_type(Interface)
 358
 359         ec = ExperimentController()
 360
 361         node = ec.register_resource("Node")
 362
 363         app1 = ec.register_resource("Application")
 364         ec.register_connection(app1, node)
 365
 366         app2 = ec.register_resource("Application")
 367         ec.register_connection(app2, node)
 368
 369         ec.register_condition(app2, ResourceAction.START, app1,
 370                 ResourceState.STARTED, time = "5s")
 371
 372         ec.deploy()
 373
 374         ec.wait_finished([app1, app2])
 375
 376         rmapp1 = ec.get_resource(app1)
 377         rmapp2 = ec.get_resource(app2)
 378
 379         self.assertTrue(rmapp2.start_time > rmapp1.start_time)
 380
 381         ec.shutdown()
 382
 383     def test_stop_with_condition(self):
 384         from nepi.execution.resource import ResourceFactory
 385
 386         ResourceFactory.register_type(Application)
 387         ResourceFactory.register_type(Node)
 388         ResourceFactory.register_type(Interface)
 389
 390         ec = ExperimentController()
 391
 392         node = ec.register_resource("Node")
 393
 394         app1 = ec.register_resource("Application")
 395         ec.register_connection(app1, node)
 396
 397         app2 = ec.register_resource("Application")
 398         ec.register_connection(app2, node)
 399
 400         ec.register_condition(app2, ResourceAction.START, app1,
 401                 ResourceState.STOPPED)
 402
 403         ec.deploy()
 404
 405         ec.wait_finished([app1, app2])
 406
 407         rmapp1 = ec.get_resource(app1)
 408         rmapp2 = ec.get_resource(app2)
 409
 410         self.assertTrue(rmapp2.start_time > rmapp1.stop_time)
 411
 412         ec.shutdown()
 413
 414     def ztest_set_with_condition(self):
 415         # TODO!!!
 416         pass
 417
 418
 419 if __name__ == '__main__':
 420     unittest.main()
 421