aboutsummaryrefslogtreecommitdiffstats
path: root/ms/blueprintsprocessor/modules/services/workflow-service
diff options
context:
space:
mode:
authorBrinda Santh <brindasanth@in.ibm.com>2019-08-16 10:14:25 -0400
committerBrinda Santh <brindasanth@in.ibm.com>2019-08-16 19:04:17 -0400
commit43957fac4ead89a43ef97bf749b1bddc1adf0d6c (patch)
tree0e7be8c853f5830f75a42877137bb9fee2b2115b /ms/blueprintsprocessor/modules/services/workflow-service
parentad9b4a41a7be5ed8c579a2e96bbb4d2da629c036 (diff)
Component node timeout implementation
Change-Id: I99fc9efba76595693d95772e409f0f982aeae9b1 Issue-ID: CCSDK-1619 Signed-off-by: Brinda Santh <brindasanth@in.ibm.com>
Diffstat (limited to 'ms/blueprintsprocessor/modules/services/workflow-service')
-rw-r--r--ms/blueprintsprocessor/modules/services/workflow-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/workflow/ImperativeWorkflowExecutionService.kt42
-rw-r--r--ms/blueprintsprocessor/modules/services/workflow-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/workflow/NodeTemplateExecutionService.kt20
2 files changed, 37 insertions, 25 deletions
diff --git a/ms/blueprintsprocessor/modules/services/workflow-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/workflow/ImperativeWorkflowExecutionService.kt b/ms/blueprintsprocessor/modules/services/workflow-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/workflow/ImperativeWorkflowExecutionService.kt
index 2a14be216..6bee17f4b 100644
--- a/ms/blueprintsprocessor/modules/services/workflow-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/workflow/ImperativeWorkflowExecutionService.kt
+++ b/ms/blueprintsprocessor/modules/services/workflow-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/workflow/ImperativeWorkflowExecutionService.kt
@@ -44,10 +44,8 @@ class ImperativeWorkflowExecutionService(
val graph = bluePrintContext.workflowByName(workflowName).asGraph()
- val deferredOutput = CompletableDeferred<ExecutionServiceOutput>()
- imperativeBluePrintWorkflowService.executeWorkflow(graph, bluePrintRuntimeService,
- executionServiceInput, deferredOutput)
- return deferredOutput.await()
+ return imperativeBluePrintWorkflowService.executeWorkflow(graph, bluePrintRuntimeService,
+ executionServiceInput)
}
}
@@ -60,35 +58,41 @@ open class ImperativeBluePrintWorkflowService(private val nodeTemplateExecutionS
lateinit var bluePrintRuntimeService: BluePrintRuntimeService<*>
lateinit var executionServiceInput: ExecutionServiceInput
lateinit var workflowName: String
- lateinit var deferredExecutionServiceOutput: CompletableDeferred<ExecutionServiceOutput>
override suspend fun executeWorkflow(graph: Graph, bluePrintRuntimeService: BluePrintRuntimeService<*>,
- input: ExecutionServiceInput,
- output: CompletableDeferred<ExecutionServiceOutput>) {
+ input: ExecutionServiceInput): ExecutionServiceOutput {
this.graph = graph
this.bluePrintRuntimeService = bluePrintRuntimeService
this.executionServiceInput = input
this.workflowName = this.executionServiceInput.actionIdentifiers.actionName
- this.deferredExecutionServiceOutput = output
this.workflowId = bluePrintRuntimeService.id()
+ val output = CompletableDeferred<ExecutionServiceOutput>()
val startMessage = WorkflowExecuteMessage(input, output)
- workflowActor().send(startMessage)
+ val workflowActor = workflowActor()
+ if (!workflowActor.isClosedForSend) {
+ workflowActor.send(startMessage)
+ } else {
+ throw BluePrintProcessorException("workflow($workflowActor) actor is closed")
+ }
+ return output.await()
}
override suspend fun initializeWorkflow(input: ExecutionServiceInput): EdgeLabel {
return EdgeLabel.SUCCESS
}
- override suspend fun prepareWorkflowOutput(exception: BluePrintProcessorException?): ExecutionServiceOutput {
- val wfStatus = if (exception != null) {
- val status = Status()
- status.message = BluePrintConstants.STATUS_FAILURE
- status.errorMessage = exception.message
- status
- } else {
- val status = Status()
- status.message = BluePrintConstants.STATUS_SUCCESS
- status
+ override suspend fun prepareWorkflowOutput(): ExecutionServiceOutput {
+ val wfStatus = Status().apply {
+ if (exceptions.isNotEmpty()) {
+ exceptions.forEach {
+ val errorMessage = it.message ?: ""
+ bluePrintRuntimeService.getBluePrintError().addError(errorMessage)
+ log.error("workflow($workflowId) exception :", it)
+ }
+ message = BluePrintConstants.STATUS_FAILURE
+ } else {
+ message = BluePrintConstants.STATUS_SUCCESS
+ }
}
return ExecutionServiceOutput().apply {
commonHeader = executionServiceInput.commonHeader
diff --git a/ms/blueprintsprocessor/modules/services/workflow-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/workflow/NodeTemplateExecutionService.kt b/ms/blueprintsprocessor/modules/services/workflow-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/workflow/NodeTemplateExecutionService.kt
index 89732e300..b64177aab 100644
--- a/ms/blueprintsprocessor/modules/services/workflow-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/workflow/NodeTemplateExecutionService.kt
+++ b/ms/blueprintsprocessor/modules/services/workflow-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/workflow/NodeTemplateExecutionService.kt
@@ -22,7 +22,7 @@ import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceOutp
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.StepData
import org.onap.ccsdk.cds.blueprintsprocessor.services.execution.AbstractComponentFunction
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants
-import org.onap.ccsdk.cds.controllerblueprints.core.putJsonElement
+import org.onap.ccsdk.cds.controllerblueprints.core.asJsonPrimitive
import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService
import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintRuntimeService
import org.slf4j.LoggerFactory
@@ -37,15 +37,22 @@ open class NodeTemplateExecutionService {
executionServiceInput: ExecutionServiceInput): ExecutionServiceOutput {
// Get the Blueprint Context
val blueprintContext = bluePrintRuntimeService.bluePrintContext()
+
+ val nodeTemplate = blueprintContext.nodeTemplateByName(nodeTemplateName)
// Get the Component Name, NodeTemplate type is mapped to Component Name
- val componentName = blueprintContext.nodeTemplateByName(nodeTemplateName).type
+ val componentName = nodeTemplate.type
val interfaceName = blueprintContext.nodeTemplateFirstInterfaceName(nodeTemplateName)
val operationName = blueprintContext.nodeTemplateFirstInterfaceFirstOperationName(nodeTemplateName)
+ val nodeTemplateImplementation = blueprintContext
+ .nodeTemplateOperationImplementation(nodeTemplateName, interfaceName, operationName)
+
+ val timeout: Int = nodeTemplateImplementation?.timeout ?: 180
+
log.info("executing node template($nodeTemplateName) component($componentName) " +
- "interface($interfaceName) operation($operationName)")
+ "interface($interfaceName) operation($operationName) with timeout($timeout) sec.")
// Get the Component Instance
val plugin = BluePrintDependencyService.instance<AbstractComponentFunction>(componentName)
@@ -62,9 +69,10 @@ open class NodeTemplateExecutionService {
// Populate Step Meta Data
val stepInputs: MutableMap<String, JsonNode> = hashMapOf()
- stepInputs.putJsonElement(BluePrintConstants.PROPERTY_CURRENT_NODE_TEMPLATE, nodeTemplateName)
- stepInputs.putJsonElement(BluePrintConstants.PROPERTY_CURRENT_INTERFACE, interfaceName)
- stepInputs.putJsonElement(BluePrintConstants.PROPERTY_CURRENT_OPERATION, operationName)
+ stepInputs[BluePrintConstants.PROPERTY_CURRENT_NODE_TEMPLATE] = nodeTemplateName.asJsonPrimitive()
+ stepInputs[BluePrintConstants.PROPERTY_CURRENT_INTERFACE] = interfaceName.asJsonPrimitive()
+ stepInputs[BluePrintConstants.PROPERTY_CURRENT_OPERATION] = operationName.asJsonPrimitive()
+ stepInputs[BluePrintConstants.PROPERTY_CURRENT_TIMEOUT] = timeout.asJsonPrimitive()
val stepInputData = StepData().apply {
name = nodeTemplateName
properties = stepInputs