1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
|
/*
* 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onap.ccsdk.cds.blueprintsprocessor.services.workflow
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.coroutineScope
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceInput
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceOutput
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.Status
import org.onap.ccsdk.cds.controllerblueprints.common.api.EventType
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintException
import org.onap.ccsdk.cds.controllerblueprints.core.MDCContext
import org.onap.ccsdk.cds.controllerblueprints.core.asGraph
import org.onap.ccsdk.cds.controllerblueprints.core.checkNotEmpty
import org.onap.ccsdk.cds.controllerblueprints.core.data.EdgeLabel
import org.onap.ccsdk.cds.controllerblueprints.core.data.Graph
import org.onap.ccsdk.cds.controllerblueprints.core.interfaces.BluePrintWorkflowExecutionService
import org.onap.ccsdk.cds.controllerblueprints.core.isAcyclic
import org.onap.ccsdk.cds.controllerblueprints.core.logger
import org.onap.ccsdk.cds.controllerblueprints.core.service.AbstractBluePrintWorkFlowService
import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintRuntimeService
import org.onap.ccsdk.cds.controllerblueprints.core.service.NodeExecuteMessage
import org.onap.ccsdk.cds.controllerblueprints.core.service.NodeSkipMessage
import org.onap.ccsdk.cds.controllerblueprints.core.service.WorkflowExecuteMessage
import org.springframework.stereotype.Service
import kotlin.coroutines.CoroutineContext
@Service("imperativeWorkflowExecutionService")
class ImperativeWorkflowExecutionService(
private val nodeTemplateExecutionService: NodeTemplateExecutionService
) :
BluePrintWorkflowExecutionService<ExecutionServiceInput, ExecutionServiceOutput> {
override suspend fun executeBluePrintWorkflow(
bluePrintRuntimeService: BluePrintRuntimeService<*>,
executionServiceInput: ExecutionServiceInput,
properties: MutableMap<String, Any>
): ExecutionServiceOutput {
val bluePrintContext = bluePrintRuntimeService.bluePrintContext()
val workflowName = executionServiceInput.actionIdentifiers.actionName
val graph = bluePrintContext.workflowByName(workflowName).asGraph()
if (!graph.isAcyclic()) {
throw BluePrintException("Imperative workflow must be acyclic. Check on_success/on_failure for circular references")
}
return coroutineScope {
ImperativeBluePrintWorkflowService(
nodeTemplateExecutionService,
this.coroutineContext[MDCContext]
)
}.let { it.executeWorkflow(graph, bluePrintRuntimeService, executionServiceInput) }
}
}
open class ImperativeBluePrintWorkflowService(private val nodeTemplateExecutionService: NodeTemplateExecutionService, private val mdcContext: CoroutineContext?) :
AbstractBluePrintWorkFlowService<ExecutionServiceInput, ExecutionServiceOutput>() {
final override val coroutineContext: CoroutineContext
get() = mdcContext?.let { super.coroutineContext + it } ?: super.coroutineContext
val log = logger(ImperativeBluePrintWorkflowService::class)
lateinit var bluePrintRuntimeService: BluePrintRuntimeService<*>
lateinit var executionServiceInput: ExecutionServiceInput
lateinit var workflowName: String
override suspend fun executeWorkflow(
graph: Graph,
bluePrintRuntimeService: BluePrintRuntimeService<*>,
input: ExecutionServiceInput
): ExecutionServiceOutput {
this.graph = graph
this.bluePrintRuntimeService = bluePrintRuntimeService
this.executionServiceInput = input
this.workflowName = this.executionServiceInput.actionIdentifiers.actionName
this.workflowId = bluePrintRuntimeService.id()
val output = CompletableDeferred<ExecutionServiceOutput>()
val startMessage = WorkflowExecuteMessage(input, output)
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(): ExecutionServiceOutput {
val wfStatus = Status().apply {
if (exceptions.isNotEmpty()) {
exceptions.forEach {
val errorMessage = it.message ?: ""
bluePrintRuntimeService.getBluePrintError().addError(errorMessage, "workflow")
log.error("workflow($workflowId) exception :", it)
}
message = BluePrintConstants.STATUS_FAILURE
} else {
message = BluePrintConstants.STATUS_SUCCESS
}
eventType = EventType.EVENT_COMPONENT_EXECUTED.name
}
return ExecutionServiceOutput().apply {
commonHeader = executionServiceInput.commonHeader
actionIdentifiers = executionServiceInput.actionIdentifiers
status = wfStatus
}
}
override suspend fun prepareNodeExecutionMessage(node: Graph.Node):
NodeExecuteMessage<ExecutionServiceInput, ExecutionServiceOutput> {
val nodeOutput = ExecutionServiceOutput().apply {
commonHeader = executionServiceInput.commonHeader
actionIdentifiers = executionServiceInput.actionIdentifiers
}
return NodeExecuteMessage(node, executionServiceInput, nodeOutput)
}
override suspend fun prepareNodeSkipMessage(node: Graph.Node):
NodeSkipMessage<ExecutionServiceInput, ExecutionServiceOutput> {
val nodeOutput = ExecutionServiceOutput().apply {
commonHeader = executionServiceInput.commonHeader
actionIdentifiers = executionServiceInput.actionIdentifiers
}
return NodeSkipMessage(node, executionServiceInput, nodeOutput)
}
override suspend fun executeNode(
node: Graph.Node,
nodeInput: ExecutionServiceInput,
nodeOutput: ExecutionServiceOutput
): EdgeLabel {
log.info("Executing workflow($workflowName[${this.workflowId}])'s step(${node.id})")
val step = bluePrintRuntimeService.bluePrintContext().workflowStepByName(this.workflowName, node.id)
checkNotEmpty(step.target) { "couldn't get step target for workflow(${this.workflowName})'s step(${node.id})" }
val nodeTemplateName = step.target!!
/** execute node template */
val executionServiceOutput = nodeTemplateExecutionService
.executeNodeTemplate(bluePrintRuntimeService, node.id, nodeTemplateName, nodeInput)
if (executionServiceOutput.status.message == BluePrintConstants.STATUS_FAILURE) {
// Clear step errors so that the workflow does not fail
bluePrintRuntimeService.getBluePrintError().stepErrors(node.id)?.clear()
return EdgeLabel.FAILURE
}
return EdgeLabel.SUCCESS
}
override suspend fun skipNode(
node: Graph.Node,
nodeInput: ExecutionServiceInput,
nodeOutput: ExecutionServiceOutput
): EdgeLabel {
return EdgeLabel.SUCCESS
}
override suspend fun cancelNode(
node: Graph.Node,
nodeInput: ExecutionServiceInput,
nodeOutput: ExecutionServiceOutput
): EdgeLabel {
TODO("not implemented")
}
override suspend fun restartNode(
node: Graph.Node,
nodeInput: ExecutionServiceInput,
nodeOutput: ExecutionServiceOutput
): EdgeLabel {
TODO("not implemented")
}
}
|
