ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/selfservice/api/ExecutionServiceHandler.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.selfservice.api
  19
  20 import io.grpc.stub.StreamObserver
  21 import io.micrometer.core.instrument.MeterRegistry
  22 import io.micrometer.core.instrument.Timer
  23 import kotlinx.coroutines.Dispatchers
  24 import kotlinx.coroutines.GlobalScope
  25 import kotlinx.coroutines.launch
  26 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ACTION_MODE_ASYNC
  27 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ACTION_MODE_SYNC
  28 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceInput
  29 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceOutput
  30 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.Status
  31 import org.onap.ccsdk.cds.blueprintsprocessor.core.utils.toProto
  32 import org.onap.ccsdk.cds.blueprintsprocessor.selfservice.api.SelfServiceMetricConstants.COUNTER_PROCESS
  33 import org.onap.ccsdk.cds.blueprintsprocessor.selfservice.api.SelfServiceMetricConstants.TIMER_PROCESS
  34 import org.onap.ccsdk.cds.blueprintsprocessor.selfservice.api.utils.cbaMetricTags
  35 import org.onap.ccsdk.cds.blueprintsprocessor.services.execution.AbstractServiceFunction
  36 import org.onap.ccsdk.cds.controllerblueprints.common.api.EventType
  37 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants
  38 import org.onap.ccsdk.cds.controllerblueprints.core.config.BluePrintLoadConfiguration
  39 import org.onap.ccsdk.cds.controllerblueprints.core.interfaces.BluePrintCatalogService
  40 import org.onap.ccsdk.cds.controllerblueprints.core.interfaces.BluePrintWorkflowExecutionService
  41 import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService
  42 import org.onap.ccsdk.cds.controllerblueprints.core.utils.BluePrintMetadataUtils
  43 import org.slf4j.LoggerFactory
  44 import org.springframework.stereotype.Service
  45
  46 @Service
  47 class ExecutionServiceHandler(
  48     private val bluePrintLoadConfiguration: BluePrintLoadConfiguration,
  49     private val blueprintsProcessorCatalogService: BluePrintCatalogService,
  50     private val bluePrintWorkflowExecutionService:
  51         BluePrintWorkflowExecutionService<ExecutionServiceInput, ExecutionServiceOutput>,
  52     private val publishAuditService: PublishAuditService,
  53     private val meterRegistry: MeterRegistry
  54 ) {
  55
  56     private val log = LoggerFactory.getLogger(ExecutionServiceHandler::class.toString())
  57
  58     suspend fun process(
  59         executionServiceInput: ExecutionServiceInput,
  60         responseObserver: StreamObserver<org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceOutput>
  61     ) {
  62         when {
  63             executionServiceInput.actionIdentifiers.mode == ACTION_MODE_ASYNC -> {
  64                 GlobalScope.launch(Dispatchers.Default) {
  65                     val executionServiceOutput = doProcess(executionServiceInput)
  66                     responseObserver.onNext(executionServiceOutput.toProto())
  67                     responseObserver.onCompleted()
  68                 }
  69                 responseObserver.onNext(response(executionServiceInput).toProto())
  70             }
  71             executionServiceInput.actionIdentifiers.mode == ACTION_MODE_SYNC -> {
  72                 val executionServiceOutput = doProcess(executionServiceInput)
  73                 responseObserver.onNext(executionServiceOutput.toProto())
  74                 responseObserver.onCompleted()
  75             }
  76             else -> {
  77                 publishAuditService.publishExecutionInput(executionServiceInput)
  78                 val executionServiceOutput = response(
  79                     executionServiceInput,
  80                     "Failed to process request, 'actionIdentifiers.mode' not specified. Valid value are: 'sync' or 'async'.",
  81                     true
  82                 )
  83                 meterRegistry.counter(COUNTER_PROCESS, cbaMetricTags(executionServiceOutput)).increment()
  84                 publishAuditService.publishExecutionOutput(executionServiceInput.correlationUUID, executionServiceOutput)
  85                 responseObserver.onNext(
  86                     executionServiceOutput.toProto()
  87                 )
  88             }
  89         }
  90     }
  91
  92     suspend fun doProcess(executionServiceInput: ExecutionServiceInput): ExecutionServiceOutput {
  93         val requestId = executionServiceInput.commonHeader.requestId
  94         val actionIdentifiers = executionServiceInput.actionIdentifiers
  95         val blueprintName = actionIdentifiers.blueprintName
  96         val blueprintVersion = actionIdentifiers.blueprintVersion
  97
  98         lateinit var executionServiceOutput: ExecutionServiceOutput
  99
 100         log.info("processing request id $requestId")
 101
 102         // Audit input
 103         publishAuditService.publishExecutionInput(executionServiceInput)
 104
 105         val sample = Timer.start()
 106         try {
 107             /** Check Blueprint is needed for this request */
 108             if (checkServiceFunction(executionServiceInput)) {
 109                 executionServiceOutput = executeServiceFunction(executionServiceInput)
 110             } else {
 111                 val basePath = blueprintsProcessorCatalogService.getFromDatabase(blueprintName, blueprintVersion)
 112                 log.info("blueprint base path $basePath")
 113
 114                 val blueprintRuntimeService = BluePrintMetadataUtils.getBluePrintRuntime(requestId, basePath.toString())
 115
 116                 executionServiceOutput = bluePrintWorkflowExecutionService.executeBluePrintWorkflow(
 117                     blueprintRuntimeService,
 118                     executionServiceInput, hashMapOf()
 119                 )
 120
 121                 val errors = blueprintRuntimeService.getBluePrintError().allErrors()
 122                 if (errors.isNotEmpty()) {
 123                     setErrorStatus(errors.joinToString(", "), executionServiceOutput.status)
 124                 }
 125             }
 126         } catch (e: Exception) {
 127             log.error("fail processing request id $requestId", e)
 128             executionServiceOutput = response(executionServiceInput, e.localizedMessage ?: e.message ?: e.toString(), true)
 129         }
 130         // Update process metrics
 131         sample.stop(meterRegistry.timer(TIMER_PROCESS, cbaMetricTags(executionServiceInput)))
 132         meterRegistry.counter(COUNTER_PROCESS, cbaMetricTags(executionServiceOutput)).increment()
 133
 134         // Audit output
 135         publishAuditService.publishExecutionOutput(executionServiceInput.correlationUUID, executionServiceOutput)
 136         return executionServiceOutput
 137     }
 138
 139     /** If the blueprint name is default, It means no blueprint is needed for the execution */
 140     fun checkServiceFunction(executionServiceInput: ExecutionServiceInput): Boolean {
 141         return executionServiceInput.actionIdentifiers.blueprintName == "default"
 142     }
 143
 144     /** If no blueprint is needed, then get the Service function instance mapping to the action name and execute it */
 145     suspend fun executeServiceFunction(executionServiceInput: ExecutionServiceInput): ExecutionServiceOutput {
 146         val actionName = executionServiceInput.actionIdentifiers.actionName
 147         val instance = BluePrintDependencyService.instance<AbstractServiceFunction>(actionName)
 148         checkNotNull(instance) { "failed to initialize service function($actionName)" }
 149         instance.actionName = actionName
 150         return instance.applyNB(executionServiceInput)
 151     }
 152
 153     private fun setErrorStatus(errorMessage: String, status: Status) {
 154         status.errorMessage = errorMessage
 155         status.eventType = EventType.EVENT_COMPONENT_FAILURE.name
 156         status.code = 500
 157         status.message = BluePrintConstants.STATUS_FAILURE
 158     }
 159
 160     private fun response(
 161         executionServiceInput: ExecutionServiceInput,
 162         errorMessage: String = "",
 163         failure: Boolean = false
 164     ): ExecutionServiceOutput {
 165         val executionServiceOutput = ExecutionServiceOutput()
 166         executionServiceOutput.commonHeader = executionServiceInput.commonHeader
 167         executionServiceOutput.actionIdentifiers = executionServiceInput.actionIdentifiers
 168         executionServiceOutput.payload = executionServiceInput.payload
 169
 170         val status = Status()
 171         if (failure) {
 172             setErrorStatus(errorMessage, status)
 173         } else {
 174             status.eventType = EventType.EVENT_COMPONENT_PROCESSING.name
 175             status.code = 200
 176             status.message = BluePrintConstants.STATUS_PROCESSING
 177         }
 178
 179         executionServiceOutput.status = status
 180
 181         return executionServiceOutput
 182     }
 183 }