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
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
|
/*
* Copyright © 2017-2018 AT&T Intellectual Property.
* Modifications 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.execution
import com.fasterxml.jackson.databind.JsonNode
import kotlinx.coroutines.withTimeout
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.blueprintsprocessor.core.api.data.StepData
import org.onap.ccsdk.cds.blueprintsprocessor.core.cluster.executeWithLock
import org.onap.ccsdk.cds.blueprintsprocessor.core.service.BlueprintClusterService
import org.onap.ccsdk.cds.blueprintsprocessor.core.service.CDS_LOCK_GROUP
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.asJsonType
import org.onap.ccsdk.cds.controllerblueprints.core.checkNotBlank
import org.onap.ccsdk.cds.controllerblueprints.core.checkNotEmpty
import org.onap.ccsdk.cds.controllerblueprints.core.data.Implementation
import org.onap.ccsdk.cds.controllerblueprints.core.getAsString
import org.onap.ccsdk.cds.controllerblueprints.core.interfaces.BlueprintFunctionNode
import org.onap.ccsdk.cds.controllerblueprints.core.jsonPathParse
import org.onap.ccsdk.cds.controllerblueprints.core.normalizedFile
import org.onap.ccsdk.cds.controllerblueprints.core.readNBLines
import org.onap.ccsdk.cds.controllerblueprints.core.service.BlueprintRuntimeService
import org.onap.ccsdk.cds.controllerblueprints.core.service.BlueprintVelocityTemplateService
import org.slf4j.LoggerFactory
/**
* AbstractComponentFunction
* @author Brinda Santh
*/
abstract class AbstractComponentFunction : BlueprintFunctionNode<ExecutionServiceInput, ExecutionServiceOutput> {
@Transient
private val log = LoggerFactory.getLogger(AbstractComponentFunction::class.java)
lateinit var executionServiceInput: ExecutionServiceInput
var executionServiceOutput = ExecutionServiceOutput()
lateinit var bluePrintRuntimeService: BlueprintRuntimeService<*>
lateinit var bluePrintClusterService: BlueprintClusterService
lateinit var implementation: Implementation
lateinit var processId: String
lateinit var workflowName: String
lateinit var stepName: String
lateinit var interfaceName: String
lateinit var operationName: String
lateinit var nodeTemplateName: String
var operationInputs: MutableMap<String, JsonNode> = hashMapOf()
override fun getName(): String {
return "$stepName - $nodeTemplateName"
}
override suspend fun prepareRequestNB(executionRequest: ExecutionServiceInput): ExecutionServiceInput {
check(this::bluePrintRuntimeService.isInitialized) { "failed to prepare blueprint runtime" }
checkNotNull(executionRequest.stepData) { "failed to get step info" }
// Get the Step Name and Step Inputs
this.stepName = executionRequest.stepData!!.name
this.operationInputs = executionRequest.stepData!!.properties
checkNotEmpty(stepName) { "failed to get step name from step data" }
this.executionServiceInput = executionRequest
processId = executionRequest.commonHeader.requestId
check(processId.isNotEmpty()) { "couldn't get process id for step($stepName)" }
workflowName = executionRequest.actionIdentifiers.actionName
check(workflowName.isNotEmpty()) { "couldn't get action name for step($stepName)" }
log.info("preparing request id($processId) for workflow($workflowName) step($stepName)")
nodeTemplateName = this.operationInputs.getAsString(BlueprintConstants.PROPERTY_CURRENT_NODE_TEMPLATE)
check(nodeTemplateName.isNotEmpty()) { "couldn't get NodeTemplate name for step($stepName)" }
interfaceName = this.operationInputs.getAsString(BlueprintConstants.PROPERTY_CURRENT_INTERFACE)
check(interfaceName.isNotEmpty()) { "couldn't get Interface name for step($stepName)" }
operationName = this.operationInputs.getAsString(BlueprintConstants.PROPERTY_CURRENT_OPERATION)
check(operationName.isNotEmpty()) { "couldn't get Operation name for step($stepName)" }
/** Get the Implementation Details */
implementation = bluePrintRuntimeService.bluePrintContext()
.nodeTemplateOperationImplementation(nodeTemplateName, interfaceName, operationName)
?: Implementation()
/** Resolve and validate lock properties */
implementation.lock?.apply {
val resolvedValues = bluePrintRuntimeService.resolvePropertyAssignments(
BlueprintConstants.MODEL_DEFINITION_TYPE_NODE_TEMPLATE,
interfaceName,
mutableMapOf("key" to this.key, "acquireTimeout" to this.acquireTimeout)
)
this.key = resolvedValues["key"] ?: "".asJsonType()
this.acquireTimeout = resolvedValues["acquireTimeout"] ?: "".asJsonType()
checkNotBlank(this.key.textValue()) { "Failed to resolve lock key" }
check(this.acquireTimeout.isInt && this.acquireTimeout.intValue() >= 0) {
"Failed to resolve lock acquireTimeout - must be a positive integer"
}
}
check(this::implementation.isInitialized) { "failed to prepare implementation" }
val operationResolvedProperties = bluePrintRuntimeService
.resolveNodeTemplateInterfaceOperationInputs(nodeTemplateName, interfaceName, operationName)
this.operationInputs.putAll(operationResolvedProperties)
return executionRequest
}
override suspend fun prepareResponseNB(): ExecutionServiceOutput {
log.info("Preparing Response...")
executionServiceOutput.commonHeader = executionServiceInput.commonHeader
executionServiceOutput.actionIdentifiers = executionServiceInput.actionIdentifiers
val status = Status()
try {
// Resolve the Output Expression
val stepOutputs = bluePrintRuntimeService
.resolveNodeTemplateInterfaceOperationOutputs(nodeTemplateName, interfaceName, operationName)
val stepOutputData = StepData().apply {
name = stepName
properties = stepOutputs
}
executionServiceOutput.stepData = stepOutputData
status.eventType = EventType.EVENT_COMPONENT_EXECUTED.name
bluePrintRuntimeService.getBlueprintError().stepErrors(stepName)?.let {
status.message = BlueprintConstants.STATUS_FAILURE
}
} catch (e: Exception) {
status.message = BlueprintConstants.STATUS_FAILURE
status.eventType = EventType.EVENT_COMPONENT_FAILURE.name
}
executionServiceOutput.status = status
return this.executionServiceOutput
}
override suspend fun applyNB(executionServiceInput: ExecutionServiceInput): ExecutionServiceOutput {
try {
prepareRequestNB(executionServiceInput)
implementation.lock?.let {
bluePrintClusterService.clusterLock("${it.key.textValue()}@$CDS_LOCK_GROUP")
.executeWithLock(it.acquireTimeout.intValue().times(1000).toLong()) {
applyNBWithTimeout(executionServiceInput)
}
} ?: applyNBWithTimeout(executionServiceInput)
} catch (runtimeException: RuntimeException) {
log.error("failed in ${getName()} : ${runtimeException.message}", runtimeException)
recoverNB(runtimeException, executionServiceInput)
}
return prepareResponseNB()
}
private suspend fun applyNBWithTimeout(executionServiceInput: ExecutionServiceInput) =
withTimeout((implementation.timeout * 1000).toLong()) {
log.debug(
"DEBUG::: AbstractComponentFunction.withTimeout " +
"section ${implementation.timeout} seconds"
)
processNB(executionServiceInput)
}
fun getOperationInput(key: String): JsonNode {
return operationInputs[key]
?: throw BlueprintProcessorException("couldn't get the operation input($key) value.")
}
fun getOptionalOperationInput(key: String): JsonNode? {
return operationInputs[key]
}
fun setAttribute(key: String, value: JsonNode) {
bluePrintRuntimeService.setNodeTemplateAttributeValue(nodeTemplateName, key, value)
}
fun addError(type: String, name: String, error: String) {
bluePrintRuntimeService.getBlueprintError().addError(type, name, error, stepName)
}
fun addError(error: String) {
bluePrintRuntimeService.getBlueprintError().addError(error, stepName)
}
/**
* Get Execution Input Payload data
*/
fun requestPayload(): JsonNode? {
return executionServiceInput.payload
}
/**
* Get Execution Input payload action property with [expression]
* ex: requestPayloadActionProperty("data") will look for path "payload/<action-name>-request/data"
*/
fun requestPayloadActionProperty(expression: String?): JsonNode? {
val requestExpression = if (expression.isNullOrBlank()) {
"$workflowName-request"
} else {
"$workflowName-request.$expression"
}
return executionServiceInput.payload.jsonPathParse(".$requestExpression")
}
suspend fun artifactContent(artifactName: String): String {
return bluePrintRuntimeService.resolveNodeTemplateArtifact(nodeTemplateName, artifactName)
}
suspend fun relationshipProperty(relationshipName: String, propertyName: String): JsonNode {
return bluePrintRuntimeService.resolveRelationshipTemplateProperties(relationshipName).get(propertyName)
?: throw BlueprintProcessorException("failed to get relationship($relationshipName) property($propertyName)")
}
suspend fun mashTemplateNData(artifactName: String, json: String): String {
val content = artifactContent(artifactName)
return BlueprintVelocityTemplateService.generateContent(content, json)
}
suspend fun readLinesFromArtifact(artifactName: String): List<String> {
val artifactDefinition =
bluePrintRuntimeService.resolveNodeTemplateArtifactDefinition(nodeTemplateName, artifactName)
val file = normalizedFile(bluePrintRuntimeService.bluePrintContext().rootPath, artifactDefinition.file)
return file.readNBLines()
}
}
|
