Improve bulk processing

Change-Id: Ia8bd447563072284aa71e00eab0ef9b93e1776b1
Issue-ID: CCSDK-1127
Signed-off-by: Muthuramalingam, Brinda Santh <brindasanth@in.ibm.com>

4 files changed, 98 insertions, 84 deletions

diff --git a/ms/blueprintsprocessor/functions/resource-resolution/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/resource/resolution/ResourceResolutionService.kt b/ms/blueprintsprocessor/functions/resource-resolution/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/resource/resolution/ResourceResolutionService.kt
index dad804692..8a14ded61 100644
--- a/ms/blueprintsprocessor/functions/resource-resolution/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/resource/resolution/ResourceResolutionService.kt
+++ b/ms/blueprintsprocessor/functions/resource-resolution/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/resource/resolution/ResourceResolutionService.kt
@@ -1,6 +1,6 @@
 /*
  *  Copyright © 2017-2018 AT&T Intellectual Property.
- *  Modifications Copyright © 2018 IBM.
+ *  Modifications Copyright © 2018-2019 IBM.
  *
  *  Licensed under the Apache License, Version 2.0 (the "License");
  *  you may not use this file except in compliance with the License.
@@ -17,7 +17,9 @@
 
 package org.onap.ccsdk.cds.blueprintsprocessor.functions.resource.resolution
 
-import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.coroutineScope
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.resource.resolution.db.ResourceResolutionResultService
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.resource.resolution.processor.ResourceAssignmentProcessor
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.resource.resolution.utils.ResourceAssignmentUtils
@@ -39,22 +41,22 @@ interface ResourceResolutionService {
 
     fun registeredResourceSources(): List<String>
 
-    fun resolveFromDatabase(bluePrintRuntimeService: BluePrintRuntimeService<*>, artifactTemplate: String,
-                            resolutionKey: String): String
+    suspend fun resolveFromDatabase(bluePrintRuntimeService: BluePrintRuntimeService<*>, artifactTemplate: String,
+                                    resolutionKey: String): String
 
-    fun resolveResources(bluePrintRuntimeService: BluePrintRuntimeService<*>, nodeTemplateName: String,
-                         artifactNames: List<String>, properties: Map<String, Any>): MutableMap<String, String>
+    suspend fun resolveResources(bluePrintRuntimeService: BluePrintRuntimeService<*>, nodeTemplateName: String,
+                                 artifactNames: List<String>, properties: Map<String, Any>): MutableMap<String, String>
 
-    fun resolveResources(bluePrintRuntimeService: BluePrintRuntimeService<*>, nodeTemplateName: String,
-                         artifactPrefix: String, properties: Map<String, Any>): String
+    suspend fun resolveResources(bluePrintRuntimeService: BluePrintRuntimeService<*>, nodeTemplateName: String,
+                                 artifactPrefix: String, properties: Map<String, Any>): String
 
-    fun resolveResources(bluePrintRuntimeService: BluePrintRuntimeService<*>, nodeTemplateName: String,
-                         artifactMapping: String, artifactTemplate: String?): String
+    suspend fun resolveResources(bluePrintRuntimeService: BluePrintRuntimeService<*>, nodeTemplateName: String,
+                                 artifactMapping: String, artifactTemplate: String?): String
 
-    fun resolveResourceAssignments(blueprintRuntimeService: BluePrintRuntimeService<*>,
-                                   resourceDictionaries: MutableMap<String, ResourceDefinition>,
-                                   resourceAssignments: MutableList<ResourceAssignment>,
-                                   identifierName: String)
+    suspend fun resolveResourceAssignments(blueprintRuntimeService: BluePrintRuntimeService<*>,
+                                           resourceDictionaries: MutableMap<String, ResourceDefinition>,
+                                           resourceAssignments: MutableList<ResourceAssignment>,
+                                           identifierName: String)
 }
 
 @Service(ResourceResolutionConstants.SERVICE_RESOURCE_RESOLUTION)
@@ -70,9 +72,15 @@ open class ResourceResolutionServiceImpl(private var applicationContext: Applica
                 .map { it.substringAfter(ResourceResolutionConstants.PREFIX_RESOURCE_RESOLUTION_PROCESSOR) }
     }
 
-    override fun resolveResources(bluePrintRuntimeService: BluePrintRuntimeService<*>, nodeTemplateName: String,
-                                  artifactNames: List<String>,
-                                  properties: Map<String, Any>): MutableMap<String, String> {
+    override suspend fun resolveFromDatabase(bluePrintRuntimeService: BluePrintRuntimeService<*>,
+                                             artifactTemplate: String,
+                                             resolutionKey: String): String {
+        return resolutionResultService.read(bluePrintRuntimeService, artifactTemplate, resolutionKey)
+    }
+
+    override suspend fun resolveResources(bluePrintRuntimeService: BluePrintRuntimeService<*>, nodeTemplateName: String,
+                                          artifactNames: List<String>,
+                                          properties: Map<String, Any>): MutableMap<String, String> {
 
         val resolvedParams: MutableMap<String, String> = hashMapOf()
         artifactNames.forEach { artifactName ->
@@ -82,8 +90,8 @@ open class ResourceResolutionServiceImpl(private var applicationContext: Applica
         return resolvedParams
     }
 
-    override fun resolveResources(bluePrintRuntimeService: BluePrintRuntimeService<*>, nodeTemplateName: String,
-                                  artifactPrefix: String, properties: Map<String, Any>): String {
+    override suspend fun resolveResources(bluePrintRuntimeService: BluePrintRuntimeService<*>, nodeTemplateName: String,
+                                          artifactPrefix: String, properties: Map<String, Any>): String {
 
         // Velocity Artifact Definition Name
         val artifactTemplate = "$artifactPrefix-template"
@@ -95,20 +103,17 @@ open class ResourceResolutionServiceImpl(private var applicationContext: Applica
         if (properties.containsKey(ResourceResolutionConstants.RESOURCE_RESOLUTION_INPUT_STORE_RESULT)
                 && properties[ResourceResolutionConstants.RESOURCE_RESOLUTION_INPUT_STORE_RESULT] as Boolean) {
             resolutionResultService.write(properties, result, bluePrintRuntimeService, artifactPrefix)
+            log.info("resolution saved into database successfully : ($properties)")
         }
 
         return result
     }
 
-    override fun resolveFromDatabase(bluePrintRuntimeService: BluePrintRuntimeService<*>, artifactTemplate: String,
-                                     resolutionKey: String): String {
-        return resolutionResultService.read(bluePrintRuntimeService, artifactTemplate, resolutionKey)
-    }
 
-    override fun resolveResources(bluePrintRuntimeService: BluePrintRuntimeService<*>, nodeTemplateName: String,
-                                  artifactMapping: String, artifactTemplate: String?): String {
+    override suspend fun resolveResources(bluePrintRuntimeService: BluePrintRuntimeService<*>, nodeTemplateName: String,
+                                          artifactMapping: String, artifactTemplate: String?): String {
 
-        var resolvedContent = ""
+        val resolvedContent: String
         log.info("Resolving resource for template artifact($artifactTemplate) with resource assignment artifact($artifactMapping)")
 
         val identifierName = artifactTemplate ?: "no-template"
@@ -128,7 +133,6 @@ open class ResourceResolutionServiceImpl(private var applicationContext: Applica
 
         val resourceDictionaries: MutableMap<String, ResourceDefinition> =
                 JacksonUtils.getMapFromFile(dictionaryFile, ResourceDefinition::class.java)
-                        ?: throw BluePrintProcessorException("couldn't get Dictionary Definitions")
 
         // Resolve resources
         resolveResourceAssignments(bluePrintRuntimeService, resourceDictionaries, resourceAssignments, identifierName)
@@ -153,46 +157,51 @@ open class ResourceResolutionServiceImpl(private var applicationContext: Applica
      * name, then get the type of the Resource Definition, Get the instance for the Resource Type and process the
      * request.
      */
-    override fun resolveResourceAssignments(blueprintRuntimeService: BluePrintRuntimeService<*>,
-                                            resourceDictionaries: MutableMap<String, ResourceDefinition>,
-                                            resourceAssignments: MutableList<ResourceAssignment>,
-                                            identifierName: String) {
+    override suspend fun resolveResourceAssignments(blueprintRuntimeService: BluePrintRuntimeService<*>,
+                                                    resourceDictionaries: MutableMap<String, ResourceDefinition>,
+                                                    resourceAssignments: MutableList<ResourceAssignment>,
+                                                    identifierName: String) {
 
         val bulkSequenced = BulkResourceSequencingUtils.process(resourceAssignments)
         val resourceAssignmentRuntimeService =
                 ResourceAssignmentUtils.transformToRARuntimeService(blueprintRuntimeService, identifierName)
 
-        bulkSequenced.map { batchResourceAssignments ->
-            batchResourceAssignments.filter { it.name != "*" && it.name != "start" }
-                    .forEach { resourceAssignment ->
-                        val dictionaryName = resourceAssignment.dictionaryName
-                        val dictionarySource = resourceAssignment.dictionarySource
-                        /**
-                         * Get the Processor name
-                         */
-                        val processorName = processorName(dictionaryName!!, dictionarySource!!, resourceDictionaries)
-
-                        val resourceAssignmentProcessor =
-                                applicationContext.getBean(processorName) as? ResourceAssignmentProcessor
-                                        ?: throw BluePrintProcessorException("failed to get resource processor for name($processorName) " +
-                                                "for resource assignment(${resourceAssignment.name})")
-                        try {
-                            // Set BluePrint Runtime Service
-                            resourceAssignmentProcessor.raRuntimeService = resourceAssignmentRuntimeService
-                            // Set Resource Dictionaries
-                            resourceAssignmentProcessor.resourceDictionaries = resourceDictionaries
-                            // TODO ("Implement suspend function")
-                            runBlocking {
-                                // Invoke Apply Method
-                                resourceAssignmentProcessor.applyNB(resourceAssignment)
+        coroutineScope {
+            bulkSequenced.forEach { batchResourceAssignments ->
+                // Execute Non Dependent Assignments in parallel ( ie asynchronously )
+                val deferred = batchResourceAssignments.filter { it.name != "*" && it.name != "start" }
+                        .map { resourceAssignment ->
+                            async {
+                                val dictionaryName = resourceAssignment.dictionaryName
+                                val dictionarySource = resourceAssignment.dictionarySource
+                                /**
+                                 * Get the Processor name
+                                 */
+                                val processorName = processorName(dictionaryName!!, dictionarySource!!, resourceDictionaries)
+
+                                val resourceAssignmentProcessor =
+                                        applicationContext.getBean(processorName) as? ResourceAssignmentProcessor
+                                                ?: throw BluePrintProcessorException("failed to get resource processor ($processorName) " +
+                                                        "for resource assignment(${resourceAssignment.name})")
+                                try {
+                                    // Set BluePrint Runtime Service
+                                    resourceAssignmentProcessor.raRuntimeService = resourceAssignmentRuntimeService
+                                    // Set Resource Dictionaries
+                                    resourceAssignmentProcessor.resourceDictionaries = resourceDictionaries
+                                    // Invoke Apply Method
+                                    resourceAssignmentProcessor.applyNB(resourceAssignment)
+                                    // Set errors from RA
+                                    blueprintRuntimeService.setBluePrintError(resourceAssignmentRuntimeService.getBluePrintError())
+                                } catch (e: RuntimeException) {
+                                    throw BluePrintProcessorException(e)
+                                }
                             }
-                            // Set errors from RA
-                            blueprintRuntimeService.setBluePrintError(resourceAssignmentRuntimeService.getBluePrintError())
-                        } catch (e: RuntimeException) {
-                            throw BluePrintProcessorException(e)
                         }
-                    }
+                log.debug("Resolving (${deferred.size})resources parallel.")
+                deferred.awaitAll()
+            }
         }
+
     }
 
 
@@ -202,13 +211,12 @@ open class ResourceResolutionServiceImpl(private var applicationContext: Applica
      */
     private fun processorName(dictionaryName: String, dictionarySource: String,
                               resourceDictionaries: MutableMap<String, ResourceDefinition>): String {
-        var processorName: String? = null
-        when (dictionarySource) {
+        val processorName: String = when (dictionarySource) {
             "input" -> {
-                processorName = "${ResourceResolutionConstants.PREFIX_RESOURCE_RESOLUTION_PROCESSOR}source-input"
+                "${ResourceResolutionConstants.PREFIX_RESOURCE_RESOLUTION_PROCESSOR}source-input"
             }
             "default" -> {
-                processorName = "${ResourceResolutionConstants.PREFIX_RESOURCE_RESOLUTION_PROCESSOR}source-default"
+                "${ResourceResolutionConstants.PREFIX_RESOURCE_RESOLUTION_PROCESSOR}source-default"
             }
             else -> {
                 val resourceDefinition = resourceDictionaries[dictionaryName]
@@ -217,8 +225,7 @@ open class ResourceResolutionServiceImpl(private var applicationContext: Applica
                 val resourceSource = resourceDefinition.sources[dictionarySource]
                         ?: throw BluePrintProcessorException("couldn't get resource definition $dictionaryName source($dictionarySource)")
 
-                processorName = ResourceResolutionConstants.PREFIX_RESOURCE_RESOLUTION_PROCESSOR
-                        .plus(resourceSource.type)
+                ResourceResolutionConstants.PREFIX_RESOURCE_RESOLUTION_PROCESSOR.plus(resourceSource.type)
             }
         }
         checkNotEmptyOrThrow(processorName,
diff --git a/ms/blueprintsprocessor/functions/resource-resolution/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/resource/resolution/db/ResourceResolutionResultService.kt b/ms/blueprintsprocessor/functions/resource-resolution/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/resource/resolution/db/ResourceResolutionResultService.kt
index 225251fce..cbc68bbc9 100644
--- a/ms/blueprintsprocessor/functions/resource-resolution/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/resource/resolution/db/ResourceResolutionResultService.kt
+++ b/ms/blueprintsprocessor/functions/resource-resolution/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/resource/resolution/db/ResourceResolutionResultService.kt
@@ -1,5 +1,6 @@
 /*
  * Copyright (C) 2019 Bell Canada.
+ * Modifications Copyright © 2019 IBM.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -15,6 +16,8 @@
  */
 package org.onap.ccsdk.cds.blueprintsprocessor.functions.resource.resolution.db
 
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.resource.resolution.ResourceResolutionConstants
 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants
 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintException
@@ -26,8 +29,8 @@ import java.util.*
 @Service
 class ResourceResolutionResultService(private val resourceResolutionRepository: ResourceResolutionRepository) {
 
-    fun write(properties: Map<String, Any>, result: String, bluePrintRuntimeService: BluePrintRuntimeService<*>,
-              artifactPrefix: String) {
+    suspend fun write(properties: Map<String, Any>, result: String, bluePrintRuntimeService: BluePrintRuntimeService<*>,
+                      artifactPrefix: String) = withContext(Dispatchers.IO) {
 
         val metadata = bluePrintRuntimeService.bluePrintContext().metadata!!
 
@@ -37,7 +40,7 @@ class ResourceResolutionResultService(private val resourceResolutionRepository:
         resourceResolutionResult.blueprintVersion = metadata[BluePrintConstants.METADATA_TEMPLATE_VERSION]
         resourceResolutionResult.blueprintName = metadata[BluePrintConstants.METADATA_TEMPLATE_NAME]
         resourceResolutionResult.resolutionKey =
-            properties[ResourceResolutionConstants.RESOURCE_RESOLUTION_INPUT_KEY].toString()
+                properties[ResourceResolutionConstants.RESOURCE_RESOLUTION_INPUT_KEY].toString()
         resourceResolutionResult.result = result
 
         try {
@@ -47,18 +50,18 @@ class ResourceResolutionResultService(private val resourceResolutionRepository:
         }
     }
 
-    fun read(bluePrintRuntimeService: BluePrintRuntimeService<*>, artifactPrefix: String,
-             resolutionKey: String): String {
+    suspend fun read(bluePrintRuntimeService: BluePrintRuntimeService<*>, artifactPrefix: String,
+                     resolutionKey: String): String = withContext(Dispatchers.IO) {
 
         val metadata = bluePrintRuntimeService.bluePrintContext().metadata!!
 
         val blueprintVersion = metadata[BluePrintConstants.METADATA_TEMPLATE_VERSION]
         val blueprintName = metadata[BluePrintConstants.METADATA_TEMPLATE_NAME]
 
-        return resourceResolutionRepository.findByResolutionKeyAndBlueprintNameAndBlueprintVersionAndArtifactName(
-            resolutionKey,
-            blueprintName,
-            blueprintVersion,
-            artifactPrefix).result!!
+        resourceResolutionRepository.findByResolutionKeyAndBlueprintNameAndBlueprintVersionAndArtifactName(
+                resolutionKey,
+                blueprintName,
+                blueprintVersion,
+                artifactPrefix).result!!
     }
 }
\ No newline at end of file
diff --git a/ms/blueprintsprocessor/functions/resource-resolution/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/resource/resolution/ResourceResolutionServiceTest.kt b/ms/blueprintsprocessor/functions/resource-resolution/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/resource/resolution/ResourceResolutionServiceTest.kt
index 433ee1d85..5ad9b3ace 100644
--- a/ms/blueprintsprocessor/functions/resource-resolution/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/resource/resolution/ResourceResolutionServiceTest.kt
+++ b/ms/blueprintsprocessor/functions/resource-resolution/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/resource/resolution/ResourceResolutionServiceTest.kt
@@ -17,6 +17,7 @@
 
 package org.onap.ccsdk.cds.blueprintsprocessor.functions.resource.resolution
 
+import kotlinx.coroutines.runBlocking
 import org.junit.Assert
 import org.junit.Test
 import org.junit.runner.RunWith
@@ -72,20 +73,22 @@ class ResourceResolutionServiceTest {
     @Test
     @Throws(Exception::class)
     fun testResolveResource() {
+        runBlocking {
 
-        Assert.assertNotNull("failed to create ResourceResolutionService", resourceResolutionService)
+            Assert.assertNotNull("failed to create ResourceResolutionService", resourceResolutionService)
 
-        val bluePrintRuntimeService = BluePrintMetadataUtils.getBluePrintRuntime("1234",
-                "./../../../../components/model-catalog/blueprint-model/test-blueprint/baseconfiguration")
+            val bluePrintRuntimeService = BluePrintMetadataUtils.getBluePrintRuntime("1234",
+                    "./../../../../components/model-catalog/blueprint-model/test-blueprint/baseconfiguration")
 
-        val executionServiceInput = JacksonUtils.readValueFromClassPathFile("payload/requests/sample-resourceresolution-request.json",
-                ExecutionServiceInput::class.java)!!
+            val executionServiceInput = JacksonUtils.readValueFromClassPathFile("payload/requests/sample-resourceresolution-request.json",
+                    ExecutionServiceInput::class.java)!!
 
-        // Prepare Inputs
-        PayloadUtils.prepareInputsFromWorkflowPayload(bluePrintRuntimeService, executionServiceInput.payload, "resource-assignment")
+            // Prepare Inputs
+            PayloadUtils.prepareInputsFromWorkflowPayload(bluePrintRuntimeService, executionServiceInput.payload, "resource-assignment")
 
-        resourceResolutionService.resolveResources(bluePrintRuntimeService, "resource-assignment", "baseconfig", mapOf())
+            resourceResolutionService.resolveResources(bluePrintRuntimeService, "resource-assignment", "baseconfig", mapOf())
 
+        }
     }
 
 }
diff --git a/ms/blueprintsprocessor/functions/resource-resolution/src/test/resources/logback-test.xml b/ms/blueprintsprocessor/functions/resource-resolution/src/test/resources/logback-test.xml
index 0c8d93bf0..c37e0b5a0 100644
--- a/ms/blueprintsprocessor/functions/resource-resolution/src/test/resources/logback-test.xml
+++ b/ms/blueprintsprocessor/functions/resource-resolution/src/test/resources/logback-test.xml
@@ -19,13 +19,14 @@
         <!-- encoders are assigned the type

              ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->

         <encoder>

-            <pattern>%d{HH:mm:ss.SSS} %-5level %logger{100} - %msg%n</pattern>

+            <pattern>%d{HH:mm:ss.SSS} %-5level [%thread] %logger{50} - %msg%n</pattern>

         </encoder>

     </appender>

 

 

     <logger name="org.springframework" level="warn"/>

     <logger name="org.hibernate" level="info"/>

+    <logger name="org.onap.ccsdk.cds.controllerblueprints" level="warn"/>

     <logger name="org.onap.ccsdk.cds.blueprintsprocessor" level="info"/>

 

     <root level="warn">