execution timeout not respected...

Issue-ID: CCSDK-2012

rev1: initial commit
rev2: reworked + added grpc deadline
rev3: wrong err msg..
rev4: timeouts in millis
rev5: timeout defaulting was after logging...
Signed-off-by: Oleg Mitsura <oleg.mitsura@amdocs.com>
Change-Id: I02eb414904cc5dfcf51ff7b7552dafe54857ed1e
(cherry picked from commit 3f56fcfec7d14e7351280b9e2a1d220638626693)

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 ddfaf9a..26661fd 100644 (file)
--- 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
@@ -17,6 +17,10 @@
 package org.onap.ccsdk.cds.blueprintsprocessor.functions.python.executor
 
 import com.fasterxml.jackson.databind.JsonNode
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.TimeoutCancellationException
+import kotlinx.coroutines.async
+import kotlinx.coroutines.withTimeout
 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceInput
 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.PrepareRemoteEnvInput
 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.RemoteIdentifier
@@ -143,22 +147,34 @@ open class ComponentRemotePythonExecutor(private val remoteScriptExecutionServic
                     requestId = processId,
                     remoteIdentifier = RemoteIdentifier(blueprintName = blueprintName, blueprintVersion = blueprintVersion),
                     command = scriptCommand,
-                    properties = properties
-                )
-                val remoteExecutionOutput = remoteScriptExecutionService.executeCommand(remoteExecutionInput)
+                    properties = properties,
+                    timeOut = implementation.timeout.toLong())
+
+                val remoteExecutionOutputDeferred = GlobalScope.async {
+                    remoteScriptExecutionService.executeCommand(remoteExecutionInput)
+                }
 
+                val remoteExecutionOutput = withTimeout(implementation.timeout * 1000L) {
+                    remoteExecutionOutputDeferred.await()
+                }
+
+                checkNotNull(remoteExecutionOutput) {
+                    "Error: Request-id $processId did not return a restul from remote command execution."
+                }
                 val logs = JacksonUtils.jsonNodeFromObject(remoteExecutionOutput.response)
                 if (remoteExecutionOutput.status != StatusType.SUCCESS) {
                     setNodeOutputErrors(remoteExecutionOutput.status.name, logs, remoteExecutionOutput.payload)
                 } else {
-                    setNodeOutputProperties(
-                        remoteExecutionOutput.status.name.asJsonPrimitive(), logs,
-                        remoteExecutionOutput.payload
-                    )
+                    setNodeOutputProperties(remoteExecutionOutput.status.name.asJsonPrimitive(), logs,
+                        remoteExecutionOutput.payload)
                 }
             }
-        } catch (e: Exception) {
-            log.error("Failed to process on remote executor", 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()
         }

diff --git a/ms/blueprintsprocessor/functions/python-executor/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutorTest.kt b/ms/blueprintsprocessor/functions/python-executor/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutorTest.kt
index 3d58afa..5e57b9e 100644 (file)
--- a/ms/blueprintsprocessor/functions/python-executor/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutorTest.kt
+++ b/ms/blueprintsprocessor/functions/python-executor/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutorTest.kt
@@ -30,6 +30,7 @@ import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.StatusType
 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.StepData
 import org.onap.ccsdk.cds.blueprintsprocessor.services.execution.RemoteScriptExecutionService
 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintError
 import org.onap.ccsdk.cds.controllerblueprints.core.asJsonPrimitive
 import org.onap.ccsdk.cds.controllerblueprints.core.asJsonType
 import org.onap.ccsdk.cds.controllerblueprints.core.putJsonElement
@@ -90,6 +91,7 @@ class ComponentRemotePythonExecutorTest {
             val componentRemotePythonExecutor = ComponentRemotePythonExecutor(remoteScriptExecutionService)
             val bluePrintRuntime = mockk<DefaultBluePrintRuntimeService>("123456-1000")
 
+            every { bluePrintRuntime.getBluePrintError() } answers { BluePrintError() } // successful case.
             every { bluePrintRuntime.setNodeTemplateAttributeValue(any(), any(), any()) } answers {}
 
             val input = getMockedOutput(bluePrintRuntime)

diff --git a/ms/blueprintsprocessor/modules/services/execution-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/AbstractComponentFunction.kt b/ms/blueprintsprocessor/modules/services/execution-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/AbstractComponentFunction.kt
index 3e329d7..aa39a1d 100644 (file)
--- a/ms/blueprintsprocessor/modules/services/execution-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/AbstractComponentFunction.kt
+++ b/ms/blueprintsprocessor/modules/services/execution-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/AbstractComponentFunction.kt
@@ -134,6 +134,7 @@ abstract class AbstractComponentFunction : BlueprintFunctionNode<ExecutionServic
         try {
             prepareRequestNB(executionServiceInput)
             withTimeout((implementation.timeout * 1000).toLong()) {
+                log.debug("DEBUG::: AbstractComponentFunction.withTimeout section ${implementation.timeout} seconds")
                 processNB(executionServiceInput)
             }
         } catch (runtimeException: RuntimeException) {

diff --git a/ms/blueprintsprocessor/modules/services/execution-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/RemoteScriptExecutionService.kt b/ms/blueprintsprocessor/modules/services/execution-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/RemoteScriptExecutionService.kt
index 3b83261..861a955 100644 (file)
--- a/ms/blueprintsprocessor/modules/services/execution-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/RemoteScriptExecutionService.kt
+++ b/ms/blueprintsprocessor/modules/services/execution-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/RemoteScriptExecutionService.kt
@@ -41,6 +41,7 @@ import org.springframework.beans.factory.config.ConfigurableBeanFactory
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.context.annotation.Scope
 import org.springframework.stereotype.Service
+import java.util.concurrent.TimeUnit
 
 interface RemoteScriptExecutionService {
     suspend fun init(selector: Any)
@@ -65,12 +66,12 @@ class GrpcRemoteScriptExecutionService(private val bluePrintGrpcLibPropertyServi
 
     override suspend fun init(selector: Any) {
         // Get the GRPC Client Service based on selector
-        val grpcClientService: BluePrintGrpcClientService
-        if (selector is JsonNode) {
-            grpcClientService = bluePrintGrpcLibPropertyService.blueprintGrpcClientService(selector)
+        val grpcClientService: BluePrintGrpcClientService = if (selector is JsonNode) {
+            bluePrintGrpcLibPropertyService.blueprintGrpcClientService(selector)
         } else {
-            grpcClientService = bluePrintGrpcLibPropertyService.blueprintGrpcClientService(selector.toString())
+            bluePrintGrpcLibPropertyService.blueprintGrpcClientService(selector.toString())
         }
+
         // Get the GRPC Channel
         channel = grpcClientService.channel()
         // Create Non Blocking Stub
@@ -81,9 +82,10 @@ class GrpcRemoteScriptExecutionService(private val bluePrintGrpcLibPropertyServi
         }
     }
 
-    override suspend fun prepareEnv(prepareEnvInput: PrepareRemoteEnvInput):
-            RemoteScriptExecutionOutput {
-        val grpResponse = commandExecutorServiceGrpc.prepareEnv(prepareEnvInput.asGrpcData())
+    override suspend fun prepareEnv(prepareEnvInput: PrepareRemoteEnvInput): RemoteScriptExecutionOutput {
+        val grpResponse = commandExecutorServiceGrpc
+            .withDeadlineAfter(prepareEnvInput.timeOut * 1000, TimeUnit.MILLISECONDS)
+            .prepareEnv(prepareEnvInput.asGrpcData())
 
         checkNotNull(grpResponse.status) {
             "failed to get GRPC prepare env response status for requestId(${prepareEnvInput.requestId})"
@@ -95,19 +97,18 @@ class GrpcRemoteScriptExecutionService(private val bluePrintGrpcLibPropertyServi
         return remoteScriptExecutionOutput
     }
 
-    override suspend fun executeCommand(remoteExecutionInput: RemoteScriptExecutionInput):
-            RemoteScriptExecutionOutput {
-
-        val grpResponse = commandExecutorServiceGrpc.executeCommand(remoteExecutionInput.asGrpcData())
+    override suspend fun executeCommand(remoteExecutionInput: RemoteScriptExecutionInput): RemoteScriptExecutionOutput {
+        val grpResponse =
+            commandExecutorServiceGrpc
+                .withDeadlineAfter(remoteExecutionInput.timeOut * 1000, TimeUnit.MILLISECONDS)
+                .executeCommand(remoteExecutionInput.asGrpcData())
 
         checkNotNull(grpResponse.status) {
             "failed to get GRPC response status for requestId(${remoteExecutionInput.requestId})"
         }
 
-        val remoteScriptExecutionOutput = grpResponse.asJavaData()
         log.debug("Received response from command server for requestId(${remoteExecutionInput.requestId})")
-
-        return remoteScriptExecutionOutput
+        return grpResponse.asJavaData()
     }
 
     override suspend fun close() {