azure/aria/aria-extension-cloudify/adapters/extension.py

   1 #
   2 # Copyright (c) 2017 GigaSpaces Technologies Ltd. All rights reserved.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 # not use this file except in compliance with the License. You may obtain
   6 # 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, WITHOUT
  12 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 # License for the specific language governing permissions and limitations
  14 # under the License.
  15 #
  16
  17 from functools import wraps
  18 from contextlib import contextmanager
  19
  20 from aria import extension as aria_extension
  21
  22 from .context_adapter import CloudifyContextAdapter
  23
  24
  25 @aria_extension.process_executor
  26 class CloudifyExecutorExtension(object):
  27
  28     def decorate(self):
  29         def decorator(function):
  30             @wraps(function)
  31             def wrapper(ctx, **operation_inputs):
  32                 # We assume that any Cloudify-based plugin would use the plugins-common, thus two
  33                 # different paths are created
  34                 is_cloudify_dependent = ctx.task.plugin and any(
  35                     'cloudify_plugins_common' in w for w in ctx.task.plugin.wheels)
  36
  37                 if is_cloudify_dependent:
  38                     from cloudify import context
  39                     from cloudify.exceptions import (NonRecoverableError, RecoverableError)
  40
  41                     with ctx.model.instrument(*ctx.INSTRUMENTATION_FIELDS):
  42                         # We need to create a new class dynamically, since CloudifyContextAdapter
  43                         # doesn't exist at runtime
  44                         ctx_adapter = type('_CloudifyContextAdapter',
  45                                            (CloudifyContextAdapter, context.CloudifyContext),
  46                                            {}, )(ctx)
  47
  48                         exception = None
  49                         with _push_cfy_ctx(ctx_adapter, operation_inputs):
  50                             try:
  51                                 function(ctx=ctx_adapter, **operation_inputs)
  52                             except NonRecoverableError as e:
  53                                 ctx.task.abort(str(e))
  54                             except RecoverableError as e:
  55                                 ctx.task.retry(str(e), retry_interval=e.retry_after)
  56                             except BaseException as e:
  57                                 # Keep exception and raise it outside of "with", because
  58                                 # contextmanager does not allow raising exceptions
  59                                 exception = e
  60                         if exception is not None:
  61                             raise exception
  62                 else:
  63                     function(ctx=ctx, **operation_inputs)
  64             return wrapper
  65         return decorator
  66
  67
  68 @contextmanager
  69 def _push_cfy_ctx(ctx, params):
  70     from cloudify import state
  71
  72     try:
  73         # Support for Cloudify > 4.0
  74         with state.current_ctx.push(ctx, params) as current_ctx:
  75             yield current_ctx
  76
  77     except AttributeError:
  78         # Support for Cloudify < 4.0
  79         try:
  80             original_ctx = state.current_ctx.get_ctx()
  81         except RuntimeError:
  82             original_ctx = None
  83         try:
  84             original_params = state.current_ctx.get_parameters()
  85         except RuntimeError:
  86             original_params = None
  87
  88         state.current_ctx.set(ctx, params)
  89         try:
  90             yield state.current_ctx.get_ctx()
  91         finally:
  92             state.current_ctx.set(original_ctx, original_params)