diff options
Diffstat (limited to 'ms')
4 files changed, 163 insertions, 76 deletions
diff --git a/ms/blueprintsprocessor/functions/python-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutor.kt b/ms/blueprintsprocessor/functions/python-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutor.kt index 26661fd41..d66e8b374 100644 --- a/ms/blueprintsprocessor/functions/python-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutor.kt +++ b/ms/blueprintsprocessor/functions/python-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutor.kt @@ -1,5 +1,5 @@ /* - * Copyright © 2019 IBM. + * Copyright © 2019 IBM. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -55,14 +55,19 @@ open class ComponentRemotePythonExecutor(private val remoteScriptExecutionServic const val INPUT_ENDPOINT_SELECTOR = "endpoint-selector" const val INPUT_DYNAMIC_PROPERTIES = "dynamic-properties" const val INPUT_ARGUMENT_PROPERTIES = "argument-properties" + const val INPUT_COMMAND = "command" const val INPUT_PACKAGES = "packages" const val DEFAULT_SELECTOR = "remote-python" + const val INPUT_ENV_PREPARE_TIMEOUT = "env-prepare-timeout" + const val INPUT_EXECUTE_TIMEOUT = "execution-timeout" const val ATTRIBUTE_EXEC_CMD_STATUS = "status" const val ATTRIBUTE_PREPARE_ENV_LOG = "prepare-environment-logs" const val ATTRIBUTE_EXEC_CMD_LOG = "execute-command-logs" const val ATTRIBUTE_RESPONSE_DATA = "response-data" + const val DEFAULT_ENV_PREPARE_TIMEOUT_IN_SEC = 120 + const val DEFAULT_EXECUTE_TIMEOUT_IN_SEC = 180 } override suspend fun processNB(executionRequest: ExecutionServiceInput) { @@ -99,6 +104,16 @@ open class ComponentRemotePythonExecutor(private val remoteScriptExecutionServic ?.rootFieldsToMap()?.toSortedMap()?.values?.joinToString(" ") { formatNestedJsonNode(it) } val command = getOperationInput(INPUT_COMMAND).asText() + + /** + * Timeouts that are specific to the command executor. + * Note: the interface->input->timeout is the component level timeout. + */ + val envPrepTimeout = getOptionalOperationInput(INPUT_ENV_PREPARE_TIMEOUT)?.asInt() + ?: DEFAULT_ENV_PREPARE_TIMEOUT_IN_SEC + val executionTimeout = getOptionalOperationInput(INPUT_EXECUTE_TIMEOUT)?.asInt() + ?: DEFAULT_EXECUTE_TIMEOUT_IN_SEC + var scriptCommand = command.replace(pythonScript.name, pythonScript.absolutePath) if (args != null && args.isNotEmpty()) { scriptCommand = scriptCommand.plus(" ").plus(args) @@ -120,9 +135,10 @@ open class ComponentRemotePythonExecutor(private val remoteScriptExecutionServic requestId = processId, remoteIdentifier = RemoteIdentifier( blueprintName = blueprintName, - blueprintVersion = blueprintVersion - ), - packages = packages + blueprintVersion = blueprintVersion), + packages = packages, + timeOut = envPrepTimeout.toLong() + ) val prepareEnvOutput = remoteScriptExecutionService.prepareEnv(prepareEnvInput) log.info("$ATTRIBUTE_PREPARE_ENV_LOG - ${prepareEnvOutput.response}") @@ -136,10 +152,26 @@ open class ComponentRemotePythonExecutor(private val remoteScriptExecutionServic } else { setNodeOutputProperties(prepareEnvOutput.status.name.asJsonPrimitive(), logsEnv, "".asJsonPrimitive()) } + } else { + // set env preparation log to empty... + setAttribute(ATTRIBUTE_PREPARE_ENV_LOG, "".asJsonPrimitive()) } - - // if Env preparation was successful, then proceed with command execution in this Env - if (bluePrintRuntimeService.getBluePrintError().errors.isEmpty()) { + } catch (grpcEx: io.grpc.StatusRuntimeException) { + val grpcErrMsg = "Command failed during env. preparation... timeout($envPrepTimeout) requestId ($processId)." + setAttribute(ATTRIBUTE_PREPARE_ENV_LOG, grpcErrMsg.asJsonPrimitive()) + setNodeOutputErrors(status = grpcErrMsg, message = "${grpcEx.status}".asJsonPrimitive()) + log.error(grpcErrMsg, grpcEx) + addError(grpcErrMsg) + } catch (e: Exception) { + val timeoutErrMsg = "Command executor failed during env. preparation.. timeout($envPrepTimeout) requestId ($processId)." + setAttribute(ATTRIBUTE_PREPARE_ENV_LOG, e.message.asJsonPrimitive()) + setNodeOutputErrors(status = timeoutErrMsg, message = "${e.message}".asJsonPrimitive()) + log.error("Failed to process on remote executor requestId ($processId)", e) + addError(timeoutErrMsg) + } + // if Env preparation was successful, then proceed with command execution in this Env + if (bluePrintRuntimeService.getBluePrintError().errors.isEmpty()) { + try { // Populate command execution properties and pass it to the remote server val properties = dynamicProperties?.returnNullIfMissing()?.rootFieldsToMap() ?: hashMapOf() @@ -168,16 +200,20 @@ open class ComponentRemotePythonExecutor(private val remoteScriptExecutionServic setNodeOutputProperties(remoteExecutionOutput.status.name.asJsonPrimitive(), logs, remoteExecutionOutput.payload) } + } catch (timeoutEx: TimeoutCancellationException) { + val timeoutErrMsg = "Command executor timed out executing after $executionTimeout seconds requestId ($processId)" + setNodeOutputErrors(status = timeoutErrMsg, message = "".asJsonPrimitive()) + log.error(timeoutErrMsg, timeoutEx) + } catch (grpcEx: io.grpc.StatusRuntimeException) { + val timeoutErrMsg = "Command executor timed out executing after $executionTimeout seconds requestId ($processId)" + setNodeOutputErrors(status = timeoutErrMsg, message = "".asJsonPrimitive()) + log.error("Command executor time out during GRPC call", grpcEx) + } catch (e: Exception) { + log.error("Failed to process on remote executor requestId ($processId)", e) } - } catch (timeoutEx: TimeoutCancellationException) { - setNodeOutputErrors(status = "Command executor timed out after ${implementation.timeout} seconds", message = "".asJsonPrimitive()) - log.error("Command executor timed out after ${implementation.timeout} seconds", timeoutEx) - } catch (grpcEx: io.grpc.StatusRuntimeException) { - setNodeOutputErrors(status = "Command executor timed out in GRPC call", message = "${grpcEx.status}".asJsonPrimitive()) - log.error("Command executor time out during GRPC call", grpcEx) - } finally { - remoteScriptExecutionService.close() } + log.debug("Trying to close GRPC channel. request ($processId)") + remoteScriptExecutionService.close() } override suspend fun recoverNB(runtimeException: RuntimeException, executionRequest: ExecutionServiceInput) { @@ -217,7 +253,6 @@ open class ComponentRemotePythonExecutor(private val remoteScriptExecutionServic log.info("Executor message : $message") setAttribute(ATTRIBUTE_RESPONSE_DATA, artifacts) log.info("Executor artifacts: $artifacts") - addError(status, ATTRIBUTE_EXEC_CMD_LOG, message.toString()) } } diff --git a/ms/command-executor/src/main/python/command_executor_handler.py b/ms/command-executor/src/main/python/command_executor_handler.py index 3219a974b..9d41b2c7b 100644 --- a/ms/command-executor/src/main/python/command_executor_handler.py +++ b/ms/command-executor/src/main/python/command_executor_handler.py @@ -45,41 +45,68 @@ class CommandExecutorHandler: def prepare_env(self, request, results): if not self.is_installed(): - self.create_venv() - if not self.activate_venv(): - return False - - f = open(self.installed, "w+") - if not self.install_packages(request, CommandExecutor_pb2.pip, f, results): - return False - f.write("\r\n") - results.append("\n") - if not self.install_packages(request, CommandExecutor_pb2.ansible_galaxy, f, results): - return False - f.close() + create_venv_status = self.create_venv() + if not create_venv_status["cds_is_successful"]: + err_msg = "ERROR: failed to prepare environment for request {} due to error in creating virtual Python env. Original error {}".format(self.blueprint_id, create_venv_status["err_msg"]) + self.logger.error(err_msg) + return utils.build_ret_data(False, err_msg) + + activate_venv_status = self.activate_venv() + if not activate_venv_status["cds_is_successful"]: + err_msg = "ERROR: failed to prepare environment for request {} due Python venv_activation. Original error {}".format(self.blueprint_id, activate_venv_status["err_msg"]) + self.logger.error(err_msg) + return utils.build_ret_data(False, err_msg) + try: + with open(self.installed, "w+") as f: + if not self.install_packages(request, CommandExecutor_pb2.pip, f, results): + return utils.build_ret_data(False, "ERROR: failed to prepare environment for request {} during pip package install.".format(self.blueprint_id)) + f.write("\r\n") # TODO: is \r needed? + results.append("\n") + if not self.install_packages(request, CommandExecutor_pb2.ansible_galaxy, f, results): + return utils.build_ret_data(False, "ERROR: failed to prepare environment for request {} during Ansible install.".format(self.blueprint_id)) + except Exception as ex: + err_msg = "ERROR: failed to prepare environment for request {} during installing packages. Exception: {}".format(self.blueprint_id, ex) + self.logger.error(err_msg) + return utils.build_ret_data(False, err_msg) else: - f = open(self.installed, "r") - results.append(f.read()) - f.close() + try: + with open(self.installed, "r") as f: + results.append(f.read()) + except Exception as ex: + return utils.build_ret_data(False, "ERROR: failed to prepare environment during reading 'installed' file {}. Exception: {}".format(self.installed, ex)) # deactivate_venv(blueprint_id) - return True + return utils.build_ret_data(True, "") def execute_command(self, request, results): - - if not self.activate_venv(): - return False - - cmd = "cd " + self.venv_home - - if "ansible-playbook" in request.command: - cmd = cmd + "; " + request.command + " -e 'ansible_python_interpreter=" + self.venv_home + "/bin/python'" - else: - cmd = cmd + "; " + request.command + " " + re.escape(MessageToJson(request.properties)) - payload_result = {} - payload_section = [] - is_payload_section = False + # workaround for when packages are not specified, we may not want to go through the install step + # can just call create_venv from here. + if not self.is_installed(): + self.create_venv() + try: + if not self.is_installed(): + create_venv_status = self.create_venv + if not create_venv_status["cds_is_successful"]: + err_msg = "{} - Failed to execute command during venv creation. Original error: {}".format(self.blueprint_id, create_venv_status["err_msg"]) + results.append(err_msg) + return utils.build_ret_data(False, err_msg) + activate_response = self.activate_venv() + if not activate_response["cds_is_successful"]: + orig_error = activate_response["err_msg"] + err_msg = "{} - Failed to execute command during environment activation. Original error: {}".format(self.blueprint_id, orig_error) + results.append(err_msg) #TODO: get rid of results and just rely on the return data struct. + return utils.build_ret_data(False, err_msg) + + cmd = "cd " + self.venv_home + + ### TODO: replace with os.environ['VIRTUAL_ENV']? + if "ansible-playbook" in request.command: + cmd = cmd + "; " + request.command + " -e 'ansible_python_interpreter=" + self.venv_home + "/bin/python'" + else: + cmd = cmd + "; " + request.command + " " + re.escape(MessageToJson(request.properties)) + payload_section = [] + is_payload_section = False ### extract the original header request into sys-env variables ### RequestID @@ -113,14 +140,17 @@ class CommandExecutorHandler: payload_section.append(output.strip()) rc = newProcess.poll() except Exception as e: - self.logger.info("{} - Failed to execute command. Error: {}".format(self.blueprint_id, e)) + err_msg = "{} - Failed to execute command. Error: {}".format(self.blueprint_id, e) + self.logger.info(err_msg) results.append(e) - payload_result["cds_return_code"] = False + payload_result.update(utils.build_ret_data(False, err_msg)) return payload_result # deactivate_venv(blueprint_id) - - payload_result["cds_return_code"] = rc + #Since return code is only used to check if it's zero (success), we can just return success flag instead. + self.logger.debug("python return_code : {}".format(rc)) + is_execution_successful = rc == 0 + payload_result.update(utils.build_ret_data(is_execution_successful, "")) return payload_result def install_packages(self, request, type, f, results): @@ -155,6 +185,7 @@ class CommandExecutorHandler: env = dict(os.environ) if "https_proxy" in os.environ: env['https_proxy'] = os.environ['https_proxy'] + self.logger.info("Using https_proxy: ", env['https_proxy']) try: results.append(subprocess.run(command, check=True, stdout=PIPE, stderr=PIPE, env=env).stdout.decode()) @@ -182,6 +213,10 @@ class CommandExecutorHandler: results.append(e.stderr.decode()) return False + # Returns a map with 'status' and 'err_msg'. + # 'status' True indicates success. + # 'err_msg' indicates an error occurred. The presence of err_msg may not be fatal, + # status should be set to False for fatal errors. def create_venv(self): self.logger.info("{} - Create Python Virtual Environment".format(self.blueprint_id)) try: @@ -189,10 +224,14 @@ class CommandExecutorHandler: # venv doesn't populate the activate_this.py script, hence we use from virtualenv venv.create(self.venv_home, with_pip=True, system_site_packages=True) virtualenv.writefile(os.path.join(bin_dir, "activate_this.py"), virtualenv.ACTIVATE_THIS) + return utils.build_ret_data(True, "") except Exception as err: - self.logger.info( - "{} - Failed to provision Python Virtual Environment. Error: {}".format(self.blueprint_id, err)) + err_msg = "{} - Failed to provision Python Virtual Environment. Error: {}".format(self.blueprint_id, err) + self.logger.info(err_msg) + return utils.build_ret_data(False, err_msg) + # return map cds_is_successful and err_msg. Status is True on success. err_msg may existence doesn't necessarily indicate fatal condition. + # the 'status' should be set to False to indicate error. def activate_venv(self): self.logger.info("{} - Activate Python Virtual Environment".format(self.blueprint_id)) @@ -203,14 +242,15 @@ class CommandExecutorHandler: path = "%s/bin/activate_this.py" % self.venv_home try: - exec (open(path).read(), {'__file__': path}) + with open(path) as activate_this_script: + exec (activate_this_script.read(), {'__file__': path}) exec (fixpathenvvar) self.logger.info("Running with PATH : {}".format(os.environ['PATH'])) - return True + return utils.build_ret_data(True, "") except Exception as err: - self.logger.info( - "{} - Failed to activate Python Virtual Environment. Error: {}".format(self.blueprint_id, err)) - return False + err_msg ="{} - Failed to activate Python Virtual Environment. Error: {}".format(self.blueprint_id, err) + self.logger.info( err_msg) + return utils.build_ret_data(False, err_msg) def deactivate_venv(self): self.logger.info("{} - Deactivate Python Virtual Environment".format(self.blueprint_id)) diff --git a/ms/command-executor/src/main/python/command_executor_server.py b/ms/command-executor/src/main/python/command_executor_server.py index 39cd1e6da..3435e2272 100644 --- a/ms/command-executor/src/main/python/command_executor_server.py +++ b/ms/command-executor/src/main/python/command_executor_server.py @@ -37,11 +37,12 @@ class CommandExecutorServer(CommandExecutor_pb2_grpc.CommandExecutorServiceServi results = [] handler = CommandExecutorHandler(request) - if not handler.prepare_env(request, results): + prepare_env_response = handler.prepare_env(request, results) + if not prepare_env_response["cds_is_successful"]: self.logger.info("{} - Failed to prepare python environment. {}".format(blueprint_id, results)) - return utils.build_response(request, results, {}, False) + return utils.build_grpc_response(request, results, {}, False) self.logger.info("{} - Package installation logs {}".format(blueprint_id, results)) - return utils.build_response(request, results, {}, True) + return utils.build_grpc_response(request, results, {}, True) def executeCommand(self, request, context): blueprint_id = utils.get_blueprint_id(request) @@ -53,12 +54,12 @@ class CommandExecutorServer(CommandExecutor_pb2_grpc.CommandExecutorServiceServi payload_result = {} handler = CommandExecutorHandler(request) payload_result = handler.execute_command(request, log_results) - if payload_result["cds_return_code"] != 0: + if not payload_result["cds_is_successful"]: self.logger.info("{} - Failed to executeCommand. {}".format(blueprint_id, log_results)) else: self.logger.info("{} - Execution finished successfully.".format(blueprint_id)) - ret = utils.build_response(request, log_results, payload_result, payload_result["cds_return_code"] == 0) + ret = utils.build_grpc_response(request, log_results, payload_result, payload_result["cds_is_successful"]) self.logger.info("Payload returned %s" % payload_result) return ret
\ No newline at end of file diff --git a/ms/command-executor/src/main/python/utils.py b/ms/command-executor/src/main/python/utils.py index a3748eb17..574be51db 100644 --- a/ms/command-executor/src/main/python/utils.py +++ b/ms/command-executor/src/main/python/utils.py @@ -18,23 +18,34 @@ from google.protobuf.timestamp_pb2 import Timestamp import proto.CommandExecutor_pb2 as CommandExecutor_pb2 import json + def get_blueprint_id(request): - blueprint_name = request.identifiers.blueprintName - blueprint_version = request.identifiers.blueprintVersion - return blueprint_name + '/' + blueprint_version + blueprint_name = request.identifiers.blueprintName + blueprint_version = request.identifiers.blueprintVersion + return blueprint_name + '/' + blueprint_version +# Create a response for grpc. Fills in the timestamp as well as removes cds_is_successful element +def build_grpc_response(request, log_results, payload_return, is_success=False): + if is_success: + status = CommandExecutor_pb2.SUCCESS + else: + status = CommandExecutor_pb2.FAILURE -def build_response(request, log_results, payload_return, is_success=False): - if is_success: - status = CommandExecutor_pb2.SUCCESS - else: - status = CommandExecutor_pb2.FAILURE + timestamp = Timestamp() + timestamp.GetCurrentTime() - timestamp = Timestamp() - timestamp.GetCurrentTime() + if "cds_is_successful" in payload_return: + payload_return.pop('cds_is_successful') + payload_str = json.dumps(payload_return) + return CommandExecutor_pb2.ExecutionOutput(requestId=request.requestId, + response=log_results, + status=status, + payload=payload_str, + timestamp=timestamp) - if 'cds_return_code' in payload_return: - payload_return.pop('cds_return_code') - payload_str = json.dumps(payload_return) - return CommandExecutor_pb2.ExecutionOutput(requestId=request.requestId, response=log_results, status=status, - payload=payload_str, timestamp=timestamp) +# build a return data structure which may contain an error message +def build_ret_data(cds_is_successful, err_msg): + ret_data = {"cds_is_successful": cds_is_successful } + if err_msg != "": + ret_data["err_msg"] = err_msg + return ret_data |