azure/aria/aria-extension-cloudify/src/aria/aria/modeling/orchestration.py

   1 # Licensed to the Apache Software Foundation (ASF) under one or more
   2 # contributor license agreements.  See the NOTICE file distributed with
   3 # this work for additional information regarding copyright ownership.
   4 # The ASF licenses this file to You under the Apache License, Version 2.0
   5 # (the "License"); you may not use this file except in compliance with
   6 # the License.  You may obtain a copy of the License at
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 """
  17 ARIA modeling orchestration module
  18 """
  19
  20 # pylint: disable=no-self-argument, no-member, abstract-method
  21 from datetime import datetime
  22
  23 from sqlalchemy import (
  24     Column,
  25     Integer,
  26     Text,
  27     DateTime,
  28     Boolean,
  29     Enum,
  30     String,
  31     Float,
  32     orm,
  33     PickleType)
  34 from sqlalchemy.ext.declarative import declared_attr
  35
  36 from ..orchestrator.exceptions import (TaskAbortException, TaskRetryException)
  37 from . import mixins
  38 from . import (
  39     relationship,
  40     types as modeling_types
  41 )
  42
  43
  44 class ExecutionBase(mixins.ModelMixin):
  45     """
  46     Workflow execution.
  47     """
  48
  49     __tablename__ = 'execution'
  50
  51     __private_fields__ = ('service_fk',
  52                           'service_template')
  53
  54     SUCCEEDED = 'succeeded'
  55     FAILED = 'failed'
  56     CANCELLED = 'cancelled'
  57     PENDING = 'pending'
  58     STARTED = 'started'
  59     CANCELLING = 'cancelling'
  60
  61     STATES = (SUCCEEDED, FAILED, CANCELLED, PENDING, STARTED, CANCELLING)
  62     END_STATES = (SUCCEEDED, FAILED, CANCELLED)
  63
  64     VALID_TRANSITIONS = {
  65         PENDING: (STARTED, CANCELLED),
  66         STARTED: END_STATES + (CANCELLING,),
  67         CANCELLING: END_STATES,
  68         # Retrying
  69         CANCELLED: PENDING,
  70         FAILED: PENDING
  71     }
  72
  73     # region one_to_many relationships
  74
  75     @declared_attr
  76     def inputs(cls):
  77         """
  78         Execution parameters.
  79
  80         :type: {:obj:`basestring`: :class:`Input`}
  81         """
  82         return relationship.one_to_many(cls, 'input', dict_key='name')
  83
  84     @declared_attr
  85     def tasks(cls):
  86         """
  87         Tasks.
  88
  89         :type: [:class:`Task`]
  90         """
  91         return relationship.one_to_many(cls, 'task')
  92
  93     @declared_attr
  94     def logs(cls):
  95         """
  96         Log messages for the execution (including log messages for its tasks).
  97
  98         :type: [:class:`Log`]
  99         """
 100         return relationship.one_to_many(cls, 'log')
 101
 102     # endregion
 103
 104     # region many_to_one relationships
 105
 106     @declared_attr
 107     def service(cls):
 108         """
 109         Associated service.
 110
 111         :type: :class:`Service`
 112         """
 113         return relationship.many_to_one(cls, 'service')
 114
 115     # endregion
 116
 117     # region association proxies
 118
 119     @declared_attr
 120     def service_name(cls):
 121         return relationship.association_proxy('service', cls.name_column_name())
 122
 123     @declared_attr
 124     def service_template(cls):
 125         return relationship.association_proxy('service', 'service_template')
 126
 127     @declared_attr
 128     def service_template_name(cls):
 129         return relationship.association_proxy('service', 'service_template_name')
 130
 131     # endregion
 132
 133     # region foreign keys
 134
 135     @declared_attr
 136     def service_fk(cls):
 137         return relationship.foreign_key('service')
 138
 139     # endregion
 140
 141     created_at = Column(DateTime, index=True, doc="""
 142     Creation timestamp.
 143
 144     :type: :class:`~datetime.datetime`
 145     """)
 146
 147     started_at = Column(DateTime, nullable=True, index=True, doc="""
 148     Started timestamp.
 149
 150     :type: :class:`~datetime.datetime`
 151     """)
 152
 153     ended_at = Column(DateTime, nullable=True, index=True, doc="""
 154     Ended timestamp.
 155
 156     :type: :class:`~datetime.datetime`
 157     """)
 158
 159     error = Column(Text, nullable=True, doc="""
 160     Error message.
 161
 162     :type: :obj:`basestring`
 163     """)
 164
 165     status = Column(Enum(*STATES, name='execution_status'), default=PENDING, doc="""
 166     Status.
 167
 168     :type: :obj:`basestring`
 169     """)
 170
 171     workflow_name = Column(Text, doc="""
 172     Workflow name.
 173
 174     :type: :obj:`basestring`
 175     """)
 176
 177     @orm.validates('status')
 178     def validate_status(self, key, value):
 179         """Validation function that verifies execution status transitions are OK"""
 180         try:
 181             current_status = getattr(self, key)
 182         except AttributeError:
 183             return
 184         valid_transitions = self.VALID_TRANSITIONS.get(current_status, [])
 185         if all([current_status is not None,
 186                 current_status != value,
 187                 value not in valid_transitions]):
 188             raise ValueError('Cannot change execution status from {current} to {new}'.format(
 189                 current=current_status,
 190                 new=value))
 191         return value
 192
 193     def has_ended(self):
 194         return self.status in self.END_STATES
 195
 196     def is_active(self):
 197         return not self.has_ended() and self.status != self.PENDING
 198
 199     def __str__(self):
 200         return '<{0} id=`{1}` (status={2})>'.format(
 201             self.__class__.__name__,
 202             getattr(self, self.name_column_name()),
 203             self.status
 204         )
 205
 206
 207 class TaskBase(mixins.ModelMixin):
 208     """
 209     Represents the smallest unit of stateful execution in ARIA. The task state includes inputs,
 210     outputs, as well as an atomic status, ensuring that the task can only be running once at any
 211     given time.
 212
 213     The Python :attr:`function` is usually provided by an associated :class:`Plugin`. The
 214     :attr:`arguments` of the function should be set according to the specific signature of the
 215     function.
 216
 217     Tasks may be "one shot" or may be configured to run repeatedly in the case of failure.
 218
 219     Tasks are often based on :class:`Operation`, and thus act on either a :class:`Node` or a
 220     :class:`Relationship`, however this is not required.
 221     """
 222
 223     __tablename__ = 'task'
 224
 225     __private_fields__ = ('node_fk',
 226                           'relationship_fk',
 227                           'plugin_fk',
 228                           'execution_fk')
 229
 230     START_WORKFLOW = 'start_workflow'
 231     END_WORKFLOW = 'end_workflow'
 232     START_SUBWROFKLOW = 'start_subworkflow'
 233     END_SUBWORKFLOW = 'end_subworkflow'
 234     STUB = 'stub'
 235     CONDITIONAL = 'conditional'
 236
 237     STUB_TYPES = (
 238         START_WORKFLOW,
 239         START_SUBWROFKLOW,
 240         END_WORKFLOW,
 241         END_SUBWORKFLOW,
 242         STUB,
 243         CONDITIONAL,
 244     )
 245
 246     PENDING = 'pending'
 247     RETRYING = 'retrying'
 248     SENT = 'sent'
 249     STARTED = 'started'
 250     SUCCESS = 'success'
 251     FAILED = 'failed'
 252     STATES = (
 253         PENDING,
 254         RETRYING,
 255         SENT,
 256         STARTED,
 257         SUCCESS,
 258         FAILED,
 259     )
 260     INFINITE_RETRIES = -1
 261
 262     # region one_to_many relationships
 263
 264     @declared_attr
 265     def logs(cls):
 266         """
 267         Log messages.
 268
 269         :type: [:class:`Log`]
 270         """
 271         return relationship.one_to_many(cls, 'log')
 272
 273     @declared_attr
 274     def arguments(cls):
 275         """
 276         Arguments sent to the Python :attr:`function``.
 277
 278         :type: {:obj:`basestring`: :class:`Argument`}
 279         """
 280         return relationship.one_to_many(cls, 'argument', dict_key='name')
 281
 282     # endregion
 283
 284     # region many_one relationships
 285
 286     @declared_attr
 287     def execution(cls):
 288         """
 289         Containing execution.
 290
 291         :type: :class:`Execution`
 292         """
 293         return relationship.many_to_one(cls, 'execution')
 294
 295     @declared_attr
 296     def node(cls):
 297         """
 298         Node actor (can be ``None``).
 299
 300         :type: :class:`Node`
 301         """
 302         return relationship.many_to_one(cls, 'node')
 303
 304     @declared_attr
 305     def relationship(cls):
 306         """
 307         Relationship actor (can be ``None``).
 308
 309         :type: :class:`Relationship`
 310         """
 311         return relationship.many_to_one(cls, 'relationship')
 312
 313     @declared_attr
 314     def plugin(cls):
 315         """
 316         Associated plugin.
 317
 318         :type: :class:`Plugin`
 319         """
 320         return relationship.many_to_one(cls, 'plugin')
 321
 322     # endregion
 323
 324     # region association proxies
 325
 326     @declared_attr
 327     def node_name(cls):
 328         return relationship.association_proxy('node', cls.name_column_name())
 329
 330     @declared_attr
 331     def relationship_name(cls):
 332         return relationship.association_proxy('relationship', cls.name_column_name())
 333
 334     @declared_attr
 335     def execution_name(cls):
 336         return relationship.association_proxy('execution', cls.name_column_name())
 337
 338     # endregion
 339
 340     # region foreign keys
 341
 342     @declared_attr
 343     def execution_fk(cls):
 344         return relationship.foreign_key('execution', nullable=True)
 345
 346     @declared_attr
 347     def node_fk(cls):
 348         return relationship.foreign_key('node', nullable=True)
 349
 350     @declared_attr
 351     def relationship_fk(cls):
 352         return relationship.foreign_key('relationship', nullable=True)
 353
 354     @declared_attr
 355     def plugin_fk(cls):
 356         return relationship.foreign_key('plugin', nullable=True)
 357
 358     # endregion
 359
 360     status = Column(Enum(*STATES, name='status'), default=PENDING, doc="""
 361     Current atomic status ('pending', 'retrying', 'sent', 'started', 'success', 'failed').
 362
 363     :type: :obj:`basestring`
 364     """)
 365
 366     due_at = Column(DateTime, nullable=False, index=True, default=datetime.utcnow(), doc="""
 367     Timestamp to start the task.
 368
 369     :type: :class:`~datetime.datetime`
 370     """)
 371
 372     started_at = Column(DateTime, default=None, doc="""
 373     Started timestamp.
 374
 375     :type: :class:`~datetime.datetime`
 376     """)
 377
 378     ended_at = Column(DateTime, default=None, doc="""
 379     Ended timestamp.
 380
 381     :type: :class:`~datetime.datetime`
 382     """)
 383
 384     attempts_count = Column(Integer, default=1, doc="""
 385     How many attempts occurred.
 386
 387     :type: :class:`~datetime.datetime`
 388     """)
 389
 390     function = Column(String, doc="""
 391     Full path to Python function.
 392
 393     :type: :obj:`basestring`
 394     """)
 395
 396     max_attempts = Column(Integer, default=1, doc="""
 397     Maximum number of attempts allowed in case of task failure.
 398
 399     :type: :obj:`int`
 400     """)
 401
 402     retry_interval = Column(Float, default=0, doc="""
 403     Interval between task retry attemps (in seconds).
 404
 405     :type: :obj:`float`
 406     """)
 407
 408     ignore_failure = Column(Boolean, default=False, doc="""
 409     Set to ``True`` to ignore failures.
 410
 411     :type: :obj:`bool`
 412     """)
 413
 414     interface_name = Column(String, doc="""
 415     Name of interface on node or relationship.
 416
 417     :type: :obj:`basestring`
 418     """)
 419
 420     operation_name = Column(String, doc="""
 421     Name of operation in interface on node or relationship.
 422
 423     :type: :obj:`basestring`
 424     """)
 425
 426     _api_id = Column(String)
 427     _executor = Column(PickleType)
 428     _context_cls = Column(PickleType)
 429     _stub_type = Column(Enum(*STUB_TYPES))
 430
 431     @property
 432     def actor(self):
 433         """
 434         Actor of the task (node or relationship).
 435         """
 436         return self.node or self.relationship
 437
 438     @orm.validates('max_attempts')
 439     def validate_max_attempts(self, _, value):                                  # pylint: disable=no-self-use
 440         """
 441         Validates that max attempts is either -1 or a positive number.
 442         """
 443         if value < 1 and value != TaskBase.INFINITE_RETRIES:
 444             raise ValueError('Max attempts can be either -1 (infinite) or any positive number. '
 445                              'Got {value}'.format(value=value))
 446         return value
 447
 448     @staticmethod
 449     def abort(message=None):
 450         raise TaskAbortException(message)
 451
 452     @staticmethod
 453     def retry(message=None, retry_interval=None):
 454         raise TaskRetryException(message, retry_interval=retry_interval)
 455
 456     @declared_attr
 457     def dependencies(cls):
 458         return relationship.many_to_many(cls, self=True)
 459
 460     def has_ended(self):
 461         return self.status in (self.SUCCESS, self.FAILED)
 462
 463     def is_waiting(self):
 464         if self._stub_type:
 465             return not self.has_ended()
 466         else:
 467             return self.status in (self.PENDING, self.RETRYING)
 468
 469     @classmethod
 470     def from_api_task(cls, api_task, executor, **kwargs):
 471         instantiation_kwargs = {}
 472
 473         if hasattr(api_task.actor, 'outbound_relationships'):
 474             instantiation_kwargs['node'] = api_task.actor
 475         elif hasattr(api_task.actor, 'source_node'):
 476             instantiation_kwargs['relationship'] = api_task.actor
 477         else:
 478             raise RuntimeError('No operation context could be created for {actor.model_cls}'
 479                                .format(actor=api_task.actor))
 480
 481         instantiation_kwargs.update(
 482             {
 483                 'name': api_task.name,
 484                 'status': cls.PENDING,
 485                 'max_attempts': api_task.max_attempts,
 486                 'retry_interval': api_task.retry_interval,
 487                 'ignore_failure': api_task.ignore_failure,
 488                 'execution': api_task._workflow_context.execution,
 489                 'interface_name': api_task.interface_name,
 490                 'operation_name': api_task.operation_name,
 491
 492                 # Only non-stub tasks have these fields
 493                 'plugin': api_task.plugin,
 494                 'function': api_task.function,
 495                 'arguments': api_task.arguments,
 496                 '_context_cls': api_task._context_cls,
 497                 '_executor': executor,
 498             }
 499         )
 500
 501         instantiation_kwargs.update(**kwargs)
 502
 503         return cls(**instantiation_kwargs)
 504
 505
 506 class LogBase(mixins.ModelMixin):
 507     """
 508     Single log message.
 509     """
 510
 511     __tablename__ = 'log'
 512
 513     __private_fields__ = ('execution_fk',
 514                           'task_fk')
 515
 516     # region many_to_one relationships
 517
 518     @declared_attr
 519     def execution(cls):
 520         """
 521         Containing execution.
 522
 523         :type: :class:`Execution`
 524         """
 525         return relationship.many_to_one(cls, 'execution')
 526
 527     @declared_attr
 528     def task(cls):
 529         """
 530         Containing task (can be ``None``).
 531
 532         :type: :class:`Task`
 533         """
 534         return relationship.many_to_one(cls, 'task')
 535
 536     # endregion
 537
 538     # region foreign keys
 539
 540     @declared_attr
 541     def execution_fk(cls):
 542         return relationship.foreign_key('execution')
 543
 544     @declared_attr
 545     def task_fk(cls):
 546         return relationship.foreign_key('task', nullable=True)
 547
 548     # endregion
 549
 550     level = Column(String, doc="""
 551     Log level.
 552
 553     :type: :obj:`basestring`
 554     """)
 555
 556     msg = Column(String, doc="""
 557     Log message.
 558
 559     :type: :obj:`basestring`
 560     """)
 561
 562     created_at = Column(DateTime, index=True, doc="""
 563     Creation timestamp.
 564
 565     :type: :class:`~datetime.datetime`
 566     """)
 567
 568     traceback = Column(Text, doc="""
 569     Error traceback in case of failure.
 570
 571     :type: :class:`~datetime.datetime`
 572     """)
 573
 574     def __str__(self):
 575         return self.msg
 576
 577     def __repr__(self):
 578         name = (self.task.actor if self.task else self.execution).name
 579         return '{name}: {self.msg}'.format(name=name, self=self)
 580
 581
 582 class PluginBase(mixins.ModelMixin):
 583     """
 584     Installed plugin.
 585
 586     Plugins are usually packaged as `wagons <https://github.com/cloudify-cosmo/wagon>`__, which
 587     are archives of one or more `wheels <https://packaging.python.org/distributing/#wheels>`__.
 588     Most of these fields are indeed extracted from the installed wagon's metadata.
 589     """
 590
 591     __tablename__ = 'plugin'
 592
 593     # region one_to_many relationships
 594
 595     @declared_attr
 596     def tasks(cls):
 597         """
 598         Associated Tasks.
 599
 600         :type: [:class:`Task`]
 601         """
 602         return relationship.one_to_many(cls, 'task')
 603
 604     # endregion
 605
 606     archive_name = Column(Text, nullable=False, index=True, doc="""
 607     Filename (not the full path) of the wagon's archive, often with a ``.wgn`` extension.
 608
 609     :type: :obj:`basestring`
 610     """)
 611
 612     distribution = Column(Text, doc="""
 613     Name of the operating system on which the wagon was installed (e.g. ``ubuntu``).
 614
 615     :type: :obj:`basestring`
 616     """)
 617
 618     distribution_release = Column(Text, doc="""
 619     Release of the operating system on which the wagon was installed (e.g. ``trusty``).
 620
 621     :type: :obj:`basestring`
 622     """)
 623
 624     distribution_version = Column(Text, doc="""
 625     Version of the operating system on which the wagon was installed (e.g. ``14.04``).
 626
 627     :type: :obj:`basestring`
 628     """)
 629
 630     package_name = Column(Text, nullable=False, index=True, doc="""
 631     Primary Python package name used when the wagon was installed, which is one of the wheels in the
 632     wagon (e.g. ``cloudify-script-plugin``).
 633
 634     :type: :obj:`basestring`
 635     """)
 636
 637     package_source = Column(Text, doc="""
 638     Full install string for the primary Python package name used when the wagon was installed (e.g.
 639     ``cloudify-script-plugin==1.2``).
 640
 641     :type: :obj:`basestring`
 642     """)
 643
 644     package_version = Column(Text, doc="""
 645     Version for the primary Python package name used when the wagon was installed (e.g. ``1.2``).
 646
 647     :type: :obj:`basestring`
 648     """)
 649
 650     supported_platform = Column(Text, doc="""
 651     If the wheels are *all* pure Python then this would be "any", otherwise it would be the
 652     installed platform name (e.g. ``linux_x86_64``).
 653
 654     :type: :obj:`basestring`
 655     """)
 656
 657     supported_py_versions = Column(modeling_types.StrictList(basestring), doc="""
 658     Python versions supported by all the wheels (e.g. ``["py26", "py27"]``)
 659
 660     :type: [:obj:`basestring`]
 661     """)
 662
 663     wheels = Column(modeling_types.StrictList(basestring), nullable=False, doc="""
 664     Filenames of the wheels archived in the wagon, often with a ``.whl`` extension.
 665
 666     :type: [:obj:`basestring`]
 667     """)
 668
 669     uploaded_at = Column(DateTime, nullable=False, index=True, doc="""
 670     Timestamp for when the wagon was installed.
 671
 672     :type: :class:`~datetime.datetime`
 673     """)
 674
 675
 676 class ArgumentBase(mixins.ParameterMixin):
 677     """
 678     Python function argument parameter.
 679     """
 680
 681     __tablename__ = 'argument'
 682
 683     # region many_to_one relationships
 684
 685     @declared_attr
 686     def task(cls):
 687         """
 688         Containing task (can be ``None``);
 689
 690         :type: :class:`Task`
 691         """
 692         return relationship.many_to_one(cls, 'task')
 693
 694     @declared_attr
 695     def operation(cls):
 696         """
 697         Containing operation (can be ``None``);
 698
 699         :type: :class:`Operation`
 700         """
 701         return relationship.many_to_one(cls, 'operation')
 702
 703     # endregion
 704
 705     # region foreign keys
 706
 707     @declared_attr
 708     def task_fk(cls):
 709         return relationship.foreign_key('task', nullable=True)
 710
 711     @declared_attr
 712     def operation_fk(cls):
 713         return relationship.foreign_key('operation', nullable=True)
 714
 715     # endregion