aboutsummaryrefslogtreecommitdiffstats
path: root/ms
diff options
context:
space:
mode:
Diffstat (limited to 'ms')
-rw-r--r--ms/blueprintsprocessor/functions/python-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutor.kt34
-rw-r--r--ms/blueprintsprocessor/functions/python-executor/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutorTest.kt2
-rw-r--r--ms/blueprintsprocessor/modules/services/execution-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/AbstractComponentFunction.kt1
-rw-r--r--ms/blueprintsprocessor/modules/services/execution-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/RemoteScriptExecutionService.kt29
4 files changed, 43 insertions, 23 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 ddfaf9ac2..26661fd41 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
@@ -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 3d58afad8..5e57b9eb7 100644
--- 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 3e329d7f5..aa39a1dd0 100644
--- 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 3b83261e5..861a95507 100644
--- 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() {