Implement nodetemplate locking feature
[ccsdk/cds.git] / ms / blueprintsprocessor / modules / services / execution-service / src / main / kotlin / org / onap / ccsdk / cds / blueprintsprocessor / services / execution / AbstractComponentFunction.kt
1 /*
2  *  Copyright © 2017-2018 AT&T Intellectual Property.
3  *  Modifications Copyright © 2019 IBM.
4  *
5  *  Licensed under the Apache License, Version 2.0 (the "License");
6  *  you may not use this file except in compliance with the License.
7  *  You may obtain a copy of the License at
8  *
9  *      http://www.apache.org/licenses/LICENSE-2.0
10  *
11  *  Unless required by applicable law or agreed to in writing, software
12  *  distributed under the License is distributed on an "AS IS" BASIS,
13  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14  *  See the License for the specific language governing permissions and
15  *  limitations under the License.
16  */
17
18 package org.onap.ccsdk.cds.blueprintsprocessor.services.execution
19
20 import com.fasterxml.jackson.databind.JsonNode
21 import kotlinx.coroutines.withTimeout
22 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceInput
23 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceOutput
24 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.Status
25 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.StepData
26 import org.onap.ccsdk.cds.blueprintsprocessor.core.cluster.executeWithLock
27 import org.onap.ccsdk.cds.blueprintsprocessor.core.service.BluePrintClusterService
28 import org.onap.ccsdk.cds.blueprintsprocessor.core.service.CDS_LOCK_GROUP
29 import org.onap.ccsdk.cds.controllerblueprints.common.api.EventType
30 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants
31 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
32 import org.onap.ccsdk.cds.controllerblueprints.core.asJsonType
33 import org.onap.ccsdk.cds.controllerblueprints.core.checkNotBlank
34 import org.onap.ccsdk.cds.controllerblueprints.core.checkNotEmpty
35 import org.onap.ccsdk.cds.controllerblueprints.core.data.Implementation
36 import org.onap.ccsdk.cds.controllerblueprints.core.getAsString
37 import org.onap.ccsdk.cds.controllerblueprints.core.interfaces.BlueprintFunctionNode
38 import org.onap.ccsdk.cds.controllerblueprints.core.jsonPathParse
39 import org.onap.ccsdk.cds.controllerblueprints.core.normalizedFile
40 import org.onap.ccsdk.cds.controllerblueprints.core.readNBLines
41 import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintRuntimeService
42 import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintVelocityTemplateService
43 import org.slf4j.LoggerFactory
44
45 /**
46  * AbstractComponentFunction
47  * @author Brinda Santh
48  */
49 abstract class AbstractComponentFunction : BlueprintFunctionNode<ExecutionServiceInput, ExecutionServiceOutput> {
50
51     @Transient
52     private val log = LoggerFactory.getLogger(AbstractComponentFunction::class.java)
53
54     lateinit var executionServiceInput: ExecutionServiceInput
55     var executionServiceOutput = ExecutionServiceOutput()
56     lateinit var bluePrintRuntimeService: BluePrintRuntimeService<*>
57     lateinit var bluePrintClusterService: BluePrintClusterService
58     lateinit var implementation: Implementation
59     lateinit var processId: String
60     lateinit var workflowName: String
61     lateinit var stepName: String
62     lateinit var interfaceName: String
63     lateinit var operationName: String
64     lateinit var nodeTemplateName: String
65     var operationInputs: MutableMap<String, JsonNode> = hashMapOf()
66
67     override fun getName(): String {
68         return stepName
69     }
70
71     override suspend fun prepareRequestNB(executionRequest: ExecutionServiceInput): ExecutionServiceInput {
72         check(this::bluePrintRuntimeService.isInitialized) { "failed to prepare blueprint runtime" }
73         checkNotNull(executionRequest.stepData) { "failed to get step info" }
74
75         // Get the Step Name and Step Inputs
76         this.stepName = executionRequest.stepData!!.name
77         this.operationInputs = executionRequest.stepData!!.properties
78
79         checkNotEmpty(stepName) { "failed to get step name from step data" }
80
81         this.executionServiceInput = executionRequest
82
83         processId = executionRequest.commonHeader.requestId
84         check(processId.isNotEmpty()) { "couldn't get process id for step($stepName)" }
85
86         workflowName = executionRequest.actionIdentifiers.actionName
87         check(workflowName.isNotEmpty()) { "couldn't get action name for step($stepName)" }
88
89         log.info("preparing request id($processId) for workflow($workflowName) step($stepName)")
90
91         nodeTemplateName = this.operationInputs.getAsString(BluePrintConstants.PROPERTY_CURRENT_NODE_TEMPLATE)
92         check(nodeTemplateName.isNotEmpty()) { "couldn't get NodeTemplate name for step($stepName)" }
93
94         interfaceName = this.operationInputs.getAsString(BluePrintConstants.PROPERTY_CURRENT_INTERFACE)
95         check(interfaceName.isNotEmpty()) { "couldn't get Interface name for step($stepName)" }
96
97         operationName = this.operationInputs.getAsString(BluePrintConstants.PROPERTY_CURRENT_OPERATION)
98         check(operationName.isNotEmpty()) { "couldn't get Operation name for step($stepName)" }
99
100         /** Get the Implementation Details */
101         implementation = bluePrintRuntimeService.bluePrintContext()
102             .nodeTemplateOperationImplementation(nodeTemplateName, interfaceName, operationName)
103             ?: Implementation()
104
105         /** Resolve and validate lock properties */
106         implementation.lock?.apply {
107             val resolvedValues = bluePrintRuntimeService.resolvePropertyAssignments(
108                     nodeTemplateName,
109                     interfaceName,
110                     mutableMapOf("key" to this.key, "acquireTimeout" to this.acquireTimeout))
111             this.key = resolvedValues["key"] ?: "".asJsonType()
112             this.acquireTimeout = resolvedValues["acquireTimeout"] ?: "".asJsonType()
113
114             checkNotBlank(this.key.textValue()) { "Failed to resolve lock key" }
115             check(this.acquireTimeout.isInt && this.acquireTimeout.intValue() >= 0) {
116                 "Failed to resolve lock acquireTimeout - must be a positive integer"
117             }
118         }
119
120         check(this::implementation.isInitialized) { "failed to prepare implementation" }
121
122         val operationResolvedProperties = bluePrintRuntimeService
123             .resolveNodeTemplateInterfaceOperationInputs(nodeTemplateName, interfaceName, operationName)
124
125         this.operationInputs.putAll(operationResolvedProperties)
126
127         return executionRequest
128     }
129
130     override suspend fun prepareResponseNB(): ExecutionServiceOutput {
131         log.info("Preparing Response...")
132         executionServiceOutput.commonHeader = executionServiceInput.commonHeader
133         executionServiceOutput.actionIdentifiers = executionServiceInput.actionIdentifiers
134         val status = Status()
135         try {
136             // Resolve the Output Expression
137             val stepOutputs = bluePrintRuntimeService
138                 .resolveNodeTemplateInterfaceOperationOutputs(nodeTemplateName, interfaceName, operationName)
139
140             val stepOutputData = StepData().apply {
141                 name = stepName
142                 properties = stepOutputs
143             }
144             executionServiceOutput.stepData = stepOutputData
145             // Set the Default Step Status
146             status.eventType = EventType.EVENT_COMPONENT_EXECUTED.name
147         } catch (e: Exception) {
148             status.message = BluePrintConstants.STATUS_FAILURE
149             status.eventType = EventType.EVENT_COMPONENT_FAILURE.name
150         }
151         executionServiceOutput.status = status
152         return this.executionServiceOutput
153     }
154
155     override suspend fun applyNB(executionServiceInput: ExecutionServiceInput): ExecutionServiceOutput {
156         prepareRequestNB(executionServiceInput)
157         return implementation.lock?.let {
158             bluePrintClusterService.clusterLock("${it.key.textValue()}@$CDS_LOCK_GROUP")
159                     .executeWithLock(it.acquireTimeout.intValue().times(1000).toLong()) {
160                         applyNBWithTimeout(executionServiceInput)
161                     }
162         } ?: applyNBWithTimeout(executionServiceInput)
163     }
164
165     private suspend fun applyNBWithTimeout(executionServiceInput: ExecutionServiceInput): ExecutionServiceOutput {
166         try {
167             withTimeout((implementation.timeout * 1000).toLong()) {
168                 log.debug("DEBUG::: AbstractComponentFunction.withTimeout section ${implementation.timeout} seconds")
169                 processNB(executionServiceInput)
170             }
171         } catch (runtimeException: RuntimeException) {
172             log.error("failed in ${getName()} : ${runtimeException.message}", runtimeException)
173             recoverNB(runtimeException, executionServiceInput)
174         }
175         return prepareResponseNB()
176     }
177
178     fun getOperationInput(key: String): JsonNode {
179         return operationInputs[key]
180             ?: throw BluePrintProcessorException("couldn't get the operation input($key) value.")
181     }
182
183     fun getOptionalOperationInput(key: String): JsonNode? {
184         return operationInputs[key]
185     }
186
187     fun setAttribute(key: String, value: JsonNode) {
188         bluePrintRuntimeService.setNodeTemplateAttributeValue(nodeTemplateName, key, value)
189     }
190
191     fun addError(type: String, name: String, error: String) {
192         bluePrintRuntimeService.getBluePrintError().addError(type, name, error)
193     }
194
195     fun addError(error: String) {
196         bluePrintRuntimeService.getBluePrintError().addError(error)
197     }
198
199     /**
200      * Get Execution Input Payload data
201      */
202     fun requestPayload(): JsonNode? {
203         return executionServiceInput.payload
204     }
205
206     /**
207      * Get Execution Input payload action property with [expression]
208      * ex: requestPayloadActionProperty("data") will look for path "payload/<action-name>-request/data"
209      */
210     fun requestPayloadActionProperty(expression: String?): JsonNode? {
211         val requestExpression = if (expression.isNullOrBlank()) {
212             "$workflowName-request"
213         } else {
214             "$workflowName-request.$expression"
215         }
216         return executionServiceInput.payload.jsonPathParse(".$requestExpression")
217     }
218
219     suspend fun artifactContent(artifactName: String): String {
220         return bluePrintRuntimeService.resolveNodeTemplateArtifact(nodeTemplateName, artifactName)
221     }
222
223     suspend fun relationshipProperty(relationshipName: String, propertyName: String): JsonNode {
224         return bluePrintRuntimeService.resolveRelationshipTemplateProperties(relationshipName).get(propertyName)
225             ?: throw BluePrintProcessorException("failed to get relationship($relationshipName) property($propertyName)")
226     }
227
228     suspend fun mashTemplateNData(artifactName: String, json: String): String {
229         val content = artifactContent(artifactName)
230         return BluePrintVelocityTemplateService.generateContent(content, json)
231     }
232
233     suspend fun readLinesFromArtifact(artifactName: String): List<String> {
234         val artifactDefinition =
235             bluePrintRuntimeService.resolveNodeTemplateArtifactDefinition(nodeTemplateName, artifactName)
236         val file = normalizedFile(bluePrintRuntimeService.bluePrintContext().rootPath, artifactDefinition.file)
237         return file.readNBLines()
238     }
239 }