gerrit.onap Code Review - ccsdk/cds.git/blob

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