Improving CMD-exec err msgs/handling.

Issue-ID: CCSDK-2049

rev1. initial commit
rev2. added better logging/passing errors.
rev3. bugfixes
rev4. bugfix for CommandExecutorServer.prepareEnv return status was still returning bool.
(was caught only during a defective mariadb pip package install!)

Signed-off-by: Oleg Mitsura <oleg.mitsura@amdocs.com>
Change-Id: Ifbddaeb0e15857f42b121af10e30e29dec5c20b2

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