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.BluePrintPathConfiguration
  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.utils.BluePrintMetadataUtils
  34 import org.slf4j.LoggerFactory
  35 import org.springframework.http.codec.multipart.FilePart
  36 import org.springframework.stereotype.Service
  37 import java.io.File
  38 import java.io.IOException
  39 import java.util.*
  40 import java.util.stream.Collectors
  41
  42 @Service
  43 class ExecutionServiceHandler(private val bluePrintPathConfiguration: BluePrintPathConfiguration,
  44                               private val blueprintsProcessorCatalogService: BluePrintCatalogService,
  45                               private val bluePrintWorkflowExecutionService
  46                               : BluePrintWorkflowExecutionService<ExecutionServiceInput, ExecutionServiceOutput>) {
  47
  48     private val log = LoggerFactory.getLogger(ExecutionServiceHandler::class.toString())
  49
  50     suspend fun upload(filePart: FilePart): String {
  51         val saveId = UUID.randomUUID().toString()
  52         val blueprintArchive = normalizedPathName(bluePrintPathConfiguration.blueprintArchivePath, saveId)
  53         val blueprintWorking = normalizedPathName(bluePrintPathConfiguration.blueprintWorkingPath, saveId)
  54         try {
  55
  56             val compressedFile = normalizedFile(blueprintArchive, "cba.zip")
  57             compressedFile.parentFile.reCreateNBDirs()
  58             // Copy the File Part to Local File
  59             copyFromFilePart(filePart, compressedFile)
  60             // Save the Copied file to Database
  61             return blueprintsProcessorCatalogService.saveToDatabase(saveId, compressedFile, true)
  62         } catch (e: IOException) {
  63             throw BluePrintException(ErrorCode.IO_FILE_INTERRUPT.value,
  64                 "Error in Upload CBA: ${e.message}", e)
  65         } finally {
  66             deleteNBDir(blueprintArchive)
  67             deleteNBDir(blueprintWorking)
  68         }
  69     }
  70
  71     suspend fun remove(name: String, version: String) {
  72         blueprintsProcessorCatalogService.deleteFromDatabase(name, version)
  73     }
  74
  75     suspend fun process(executionServiceInput: ExecutionServiceInput,
  76                         responseObserver: StreamObserver<org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceOutput>) {
  77         when {
  78             executionServiceInput.actionIdentifiers.mode == ACTION_MODE_ASYNC -> {
  79                 GlobalScope.launch(Dispatchers.Default) {
  80                     val executionServiceOutput = doProcess(executionServiceInput)
  81                     responseObserver.onNext(executionServiceOutput.toProto())
  82                     responseObserver.onCompleted()
  83                 }
  84                 responseObserver.onNext(response(executionServiceInput).toProto())
  85             }
  86             executionServiceInput.actionIdentifiers.mode == ACTION_MODE_SYNC -> {
  87                 val executionServiceOutput = doProcess(executionServiceInput)
  88                 responseObserver.onNext(executionServiceOutput.toProto())
  89                 responseObserver.onCompleted()
  90             }
  91             else -> responseObserver.onNext(response(executionServiceInput,
  92                 "Failed to process request, 'actionIdentifiers.mode' not specified. Valid value are: 'sync' or 'async'.",
  93                 true).toProto());
  94         }
  95     }
  96
  97     suspend fun doProcess(executionServiceInput: ExecutionServiceInput): ExecutionServiceOutput {
  98         val requestId = executionServiceInput.commonHeader.requestId
  99         log.info("processing request id $requestId")
 100         val actionIdentifiers = executionServiceInput.actionIdentifiers
 101         val blueprintName = actionIdentifiers.blueprintName
 102         val blueprintVersion = actionIdentifiers.blueprintVersion
 103         try {
 104             val basePath = blueprintsProcessorCatalogService.getFromDatabase(blueprintName, blueprintVersion)
 105             log.info("blueprint base path $basePath")
 106
 107             val blueprintRuntimeService = BluePrintMetadataUtils.getBluePrintRuntime(requestId, basePath.toString())
 108
 109             val output = bluePrintWorkflowExecutionService.executeBluePrintWorkflow(blueprintRuntimeService,
 110                 executionServiceInput, hashMapOf())
 111
 112             val errors = blueprintRuntimeService.getBluePrintError().errors
 113             if (errors.isNotEmpty()) {
 114                 val errorMessage = errors.stream().map { it.toString() }.collect(Collectors.joining(", "))
 115                 setErrorStatus(errorMessage, output.status)
 116             }
 117             return output
 118         } catch (e: Exception) {
 119             log.error("fail processing request id $requestId", e)
 120             return response(executionServiceInput, e.localizedMessage ?: e.message ?: e.toString(), true)
 121         }
 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 = executionServiceInput.payload
 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 }