azure/aria/aria-extension-cloudify/src/aria/aria/orchestrator/execution_plugin/common.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 Execution plugin utilities.
  18 """
  19
  20 import json
  21 import os
  22 import tempfile
  23
  24 import requests
  25
  26 from . import constants
  27 from . import exceptions
  28
  29
  30 def is_windows():
  31     return os.name == 'nt'
  32
  33
  34 def download_script(ctx, script_path):
  35     split = script_path.split('://')
  36     schema = split[0]
  37     suffix = script_path.split('/')[-1]
  38     file_descriptor, dest_script_path = tempfile.mkstemp(suffix='-{0}'.format(suffix))
  39     os.close(file_descriptor)
  40     try:
  41         if schema in ('http', 'https'):
  42             response = requests.get(script_path)
  43             if response.status_code == 404:
  44                 ctx.task.abort('Failed to download script: {0} (status code: {1})'
  45                                .format(script_path, response.status_code))
  46             content = response.text
  47             with open(dest_script_path, 'wb') as f:
  48                 f.write(content)
  49         else:
  50             ctx.download_resource(destination=dest_script_path, path=script_path)
  51     except:
  52         os.remove(dest_script_path)
  53         raise
  54     return dest_script_path
  55
  56
  57 def create_process_config(script_path, process, operation_kwargs, quote_json_env_vars=False):
  58     """
  59     Updates a process with its environment variables, and return it.
  60
  61     Gets a dict representing a process and a dict representing the environment variables. Converts
  62     each environment variable to a format of::
  63
  64         <string representing the name of the variable>:
  65         <json formatted string representing the value of the variable>.
  66
  67     Finally, updates the process with the newly formatted environment variables, and return the
  68     process.
  69
  70     :param process: dict representing a process
  71     :type process: dict
  72     :param operation_kwargs: dict representing environment variables that should exist in the
  73      process's running environment.
  74     :type operation_kwargs: dict
  75     :return: process updated with its environment variables
  76     :rtype: dict
  77     """
  78     process = process or {}
  79     env_vars = operation_kwargs.copy()
  80     if 'ctx' in env_vars:
  81         del env_vars['ctx']
  82     env_vars.update(process.get('env', {}))
  83     for k, v in env_vars.items():
  84         if isinstance(v, (dict, list, tuple, bool, int, float)):
  85             v = json.dumps(v)
  86             if quote_json_env_vars:
  87                 v = "'{0}'".format(v)
  88         if is_windows():
  89             # These <k,v> environment variables will subsequently
  90             # be used in a subprocess.Popen() call, as the `env` parameter.
  91             # In some windows python versions, if an environment variable
  92             # name is not of type str (e.g. unicode), the Popen call will
  93             # fail.
  94             k = str(k)
  95             # The windows shell removes all double quotes - escape them
  96             # to still be able to pass JSON in env vars to the shell.
  97             v = v.replace('"', '\\"')
  98         del env_vars[k]
  99         env_vars[k] = str(v)
 100     process['env'] = env_vars
 101     args = process.get('args')
 102     command = script_path
 103     command_prefix = process.get('command_prefix')
 104     if command_prefix:
 105         command = '{0} {1}'.format(command_prefix, command)
 106     if args:
 107         command = ' '.join([command] + [str(a) for a in args])
 108     process['command'] = command
 109     return process
 110
 111
 112 def patch_ctx(ctx):
 113     ctx._error = None
 114     task = ctx.task
 115
 116     def _validate_legal_action():
 117         if ctx._error is not None:
 118             ctx._error = RuntimeError(constants.ILLEGAL_CTX_OPERATION_MESSAGE)
 119             raise ctx._error
 120
 121     def abort_operation(message=None):
 122         _validate_legal_action()
 123         ctx._error = exceptions.ScriptException(message=message, retry=False)
 124         return ctx._error
 125     task.abort = abort_operation
 126
 127     def retry_operation(message=None, retry_interval=None):
 128         _validate_legal_action()
 129         ctx._error = exceptions.ScriptException(message=message,
 130                                                 retry=True,
 131                                                 retry_interval=retry_interval)
 132         return ctx._error
 133     task.retry = retry_operation
 134
 135
 136 def check_error(ctx, error_check_func=None, reraise=False):
 137     _error = ctx._error
 138     # this happens when a script calls task.abort/task.retry more than once
 139     if isinstance(_error, RuntimeError):
 140         ctx.task.abort(str(_error))
 141     # ScriptException is populated by the ctx proxy server when task.abort or task.retry
 142     # are called
 143     elif isinstance(_error, exceptions.ScriptException):
 144         if _error.retry:
 145             ctx.task.retry(_error.message, _error.retry_interval)
 146         else:
 147             ctx.task.abort(_error.message)
 148     # local and ssh operations may pass an additional logic check for errors here
 149     if error_check_func:
 150         error_check_func()
 151     # if this function is called from within an ``except`` clause, a re-raise maybe required
 152     if reraise:
 153         raise  # pylint: disable=misplaced-bare-raise
 154     return _error