/*
- * 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.
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) {
?.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)
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}")
} 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()
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) {
log.info("Executor message : $message")
setAttribute(ATTRIBUTE_RESPONSE_DATA, artifacts)
log.info("Executor artifacts: $artifacts")
-
addError(status, ATTRIBUTE_EXEC_CMD_LOG, message.toString())
}
}
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
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):
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())
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:
# 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))
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))
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)
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
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
Code Review / ccsdk / cds.git / commitdiff