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 com.fasterxml.jackson.databind.node.JsonNodeFactory
  21 import io.grpc.stub.StreamObserver
  22 import kotlinx.coroutines.Dispatchers
  23 import kotlinx.coroutines.GlobalScope
  24 import kotlinx.coroutines.launch
  25 import kotlinx.coroutines.reactive.awaitSingle
  26 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.*
  27 import org.onap.ccsdk.cds.blueprintsprocessor.selfservice.api.utils.toProto
  28 import org.onap.ccsdk.cds.controllerblueprints.common.api.EventType
  29 import org.onap.ccsdk.cds.controllerblueprints.core.*
  30 import org.onap.ccsdk.cds.controllerblueprints.core.config.BluePrintPathConfiguration
  31 import org.onap.ccsdk.cds.controllerblueprints.core.data.ErrorCode
  32 import org.onap.ccsdk.cds.controllerblueprints.core.interfaces.BluePrintCatalogService
  33 import org.onap.ccsdk.cds.controllerblueprints.core.interfaces.BluePrintWorkflowExecutionService
  34 import org.onap.ccsdk.cds.controllerblueprints.core.utils.BluePrintMetadataUtils
  35 import org.slf4j.LoggerFactory
  36 import org.springframework.http.codec.multipart.FilePart
  37 import org.springframework.stereotype.Service
  38 import java.io.File
  39 import java.io.IOException
  40 import java.util.*
  41 import java.util.stream.Collectors
  42
  43 @Service
  44 class ExecutionServiceHandler(private val bluePrintPathConfiguration: BluePrintPathConfiguration,
  45                               private val bluePrintCatalogService: BluePrintCatalogService,
  46                               private val bluePrintWorkflowExecutionService
  47                               : BluePrintWorkflowExecutionService<ExecutionServiceInput, ExecutionServiceOutput>) {
  48
  49     private val log = LoggerFactory.getLogger(ExecutionServiceHandler::class.toString())
  50
  51     suspend fun upload(filePart: FilePart): String {
  52         val saveId = UUID.randomUUID().toString()
  53         val blueprintArchive = normalizedPathName(bluePrintPathConfiguration.blueprintArchivePath, saveId)
  54         val blueprintWorking = normalizedPathName(bluePrintPathConfiguration.blueprintWorkingPath, saveId)
  55         try {
  56
  57             val compressedFile = normalizedFile(blueprintArchive, "cba.zip")
  58             compressedFile.parentFile.reCreateNBDirs()
  59             // Copy the File Part to Local File
  60             copyFromFilePart(filePart, compressedFile)
  61             // Save the Copied file to Database
  62             return bluePrintCatalogService.saveToDatabase(saveId, compressedFile, true)
  63         } catch (e: IOException) {
  64             throw BluePrintException(ErrorCode.IO_FILE_INTERRUPT.value,
  65                     "Error in Upload CBA: ${e.message}", e)
  66         } finally {
  67             deleteNBDir(blueprintArchive)
  68             deleteNBDir(blueprintWorking)
  69         }
  70     }
  71
  72     suspend fun remove(name: String, version: String) {
  73         bluePrintCatalogService.deleteFromDatabase(name, version)
  74     }
  75
  76     suspend fun process(executionServiceInput: ExecutionServiceInput,
  77                         responseObserver: StreamObserver<org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceOutput>) {
  78         when {
  79             executionServiceInput.actionIdentifiers.mode == ACTION_MODE_ASYNC -> {
  80                 GlobalScope.launch(Dispatchers.Default) {
  81                     val executionServiceOutput = doProcess(executionServiceInput)
  82                     responseObserver.onNext(executionServiceOutput.toProto())
  83                     responseObserver.onCompleted()
  84                 }
  85                 responseObserver.onNext(response(executionServiceInput).toProto())
  86             }
  87             executionServiceInput.actionIdentifiers.mode == ACTION_MODE_SYNC -> {
  88                 val executionServiceOutput = doProcess(executionServiceInput)
  89                 responseObserver.onNext(executionServiceOutput.toProto())
  90                 responseObserver.onCompleted()
  91             }
  92             else -> responseObserver.onNext(response(executionServiceInput,
  93                     "Failed to process request, 'actionIdentifiers.mode' not specified. Valid value are: 'sync' or 'async'.",
  94                     true).toProto());
  95         }
  96     }
  97
  98     suspend fun doProcess(executionServiceInput: ExecutionServiceInput): ExecutionServiceOutput {
  99         val requestId = executionServiceInput.commonHeader.requestId
 100         log.info("processing request id $requestId")
 101
 102         val actionIdentifiers = executionServiceInput.actionIdentifiers
 103
 104         val blueprintName = actionIdentifiers.blueprintName
 105         val blueprintVersion = actionIdentifiers.blueprintVersion
 106
 107         val basePath = bluePrintCatalogService.getFromDatabase(blueprintName, blueprintVersion)
 108         log.info("blueprint base path $basePath")
 109
 110         val blueprintRuntimeService = BluePrintMetadataUtils.getBluePrintRuntime(requestId, basePath.toString())
 111
 112         val output = bluePrintWorkflowExecutionService.executeBluePrintWorkflow(blueprintRuntimeService,
 113                 executionServiceInput, hashMapOf())
 114
 115         val errors = blueprintRuntimeService.getBluePrintError().errors
 116         if (errors.isNotEmpty()) {
 117             val errorMessage = errors.stream().map { it.toString() }.collect(Collectors.joining(", "))
 118             setErrorStatus(errorMessage, output.status)
 119         }
 120
 121         return output
 122     }
 123
 124     private suspend fun copyFromFilePart(filePart: FilePart, targetFile: File): File {
 125         return filePart.transferTo(targetFile)
 126                 .thenReturn(targetFile)
 127                 .awaitSingle()
 128     }
 129
 130     private fun setErrorStatus(errorMessage: String, status: Status) {
 131         status.errorMessage = errorMessage
 132         status.eventType = EventType.EVENT_COMPONENT_FAILURE.name
 133         status.code = 500
 134         status.message = BluePrintConstants.STATUS_FAILURE
 135     }
 136
 137     private fun response(executionServiceInput: ExecutionServiceInput, errorMessage: String = "",
 138                          failure: Boolean = false): ExecutionServiceOutput {
 139         val executionServiceOutput = ExecutionServiceOutput()
 140         executionServiceOutput.commonHeader = executionServiceInput.commonHeader
 141         executionServiceOutput.actionIdentifiers = executionServiceInput.actionIdentifiers
 142         executionServiceOutput.payload = JsonNodeFactory.instance.objectNode()
 143
 144         val status = Status()
 145         if (failure) {
 146             setErrorStatus(errorMessage, status)
 147         } else {
 148             status.eventType = EventType.EVENT_COMPONENT_PROCESSING.name
 149             status.code = 200
 150             status.message = BluePrintConstants.STATUS_PROCESSING
 151         }
 152
 153         executionServiceOutput.status = status
 154
 155         return executionServiceOutput
 156     }
 157
 158 }