aboutsummaryrefslogtreecommitdiffstats
path: root/ms/blueprintsprocessor
diff options
context:
space:
mode:
authorJulien Fontaine <julien.fontaine@bell.ca>2020-04-20 11:53:41 -0400
committerJulien Fontaine <julien.fontaine@bell.ca>2020-05-05 18:17:52 -0400
commit888f2f78a580a3deccb66bef50f2375a04b39eb6 (patch)
tree2b76202c7a7f9f73cb36a1bc7d38496d939de9e3 /ms/blueprintsprocessor
parentc86d7262d03b299dc635b3ba68e4a7c2c0fd6a6e (diff)
Truncate message published on Kafka / Spike: Define solution for logs separation
Refactoring of cmd-exec component - Improve display of error messages within the response - Fix gRPC buffer limitation (4Mb) by truncating error messages and cmd-exec logs if too heavy (>3Mb) Truncation of BP responses (<4Kb) before sending them in kafka audit topics. - Truncation if needed of error messages for every response - Truncation of cmd-exec logs in cmd-exec responses (Spike) Add a flag in the application.properties to enable/disable the display of cmd-exec responses on the BP side (Fix) Correction of BP processing with kafka regression (Fix) Changed default SSL Endpoint Algo Issue-ID: CCSDK-2326 Change-Id: If4d0e661117d1dd156cf19c95774824e754d870a Signed-off-by: Julien Fontaine <julien.fontaine@bell.ca>
Diffstat (limited to 'ms/blueprintsprocessor')
-rwxr-xr-xms/blueprintsprocessor/application/src/main/resources/application-dev.properties16
-rw-r--r--ms/blueprintsprocessor/functions/python-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutor.kt97
-rw-r--r--ms/blueprintsprocessor/functions/python-executor/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutorTest.kt11
-rw-r--r--ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageLibData.kt6
-rw-r--r--ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaMessageProducerService.kt52
-rw-r--r--ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageConsumerServiceTest.kt2
-rw-r--r--ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageProducerServiceTest.kt2
-rw-r--r--ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/selfservice/api/BluePrintProcessingKafkaConsumer.kt25
-rw-r--r--ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/selfservice/api/KafkaPublishAuditService.kt10
9 files changed, 174 insertions, 47 deletions
diff --git a/ms/blueprintsprocessor/application/src/main/resources/application-dev.properties b/ms/blueprintsprocessor/application/src/main/resources/application-dev.properties
index bf5e23bc9..fb32d9afc 100755
--- a/ms/blueprintsprocessor/application/src/main/resources/application-dev.properties
+++ b/ms/blueprintsprocessor/application/src/main/resources/application-dev.properties
@@ -69,9 +69,9 @@ blueprintsprocessor.grpcclient.py-executor.trustCertCollection=src/main/resource
# Blueprint Processor File Execution and Handling Properties
### use absolute paths if testing inside docker
-### blueprintsprocessor.blueprintDeployPath=/opt/app/onap/blueprints/deploy
-### blueprintsprocessor.blueprintArchivePath=/opt/app/onap/blueprints/archive
-### blueprintsprocessor.blueprintWorkingPath=/opt/app/onap/blueprints/working
+#blueprintsprocessor.blueprintDeployPath=/opt/app/onap/blueprints/deploy
+#blueprintsprocessor.blueprintArchivePath=/opt/app/onap/blueprints/archive
+#blueprintsprocessor.blueprintWorkingPath=/opt/app/onap/blueprints/working
# db
@@ -120,10 +120,12 @@ blueprintsprocessor.netconfExecutor.enabled=true
blueprintsprocessor.restConfExecutor.enabled=true
blueprintsprocessor.cliExecutor.enabled=true
### If enabling remote python executor, set this value to true
-### blueprintsprocessor.remoteScriptCommand.enabled=true
+#blueprintsprocessor.remoteScriptCommand.enabled=true
blueprintsprocessor.remoteScriptCommand.enabled=false
+blueprintsprocessor.remote-script-command.response.log.enabled=false
# Kafka-message-lib Configurations
+## Request consumer
blueprintsprocessor.messageconsumer.self-service-api.kafkaEnable=false
blueprintsprocessor.messageconsumer.self-service-api.type=kafka-basic-auth
blueprintsprocessor.messageconsumer.self-service-api.bootstrapServers=127.0.0.1:9092
@@ -141,6 +143,12 @@ blueprintsprocessor.messageconsumer.self-service-api.pollMillSec=1000
#blueprintsprocessor.messageconsumer.self-service-api.scramUsername=test-user
#blueprintsprocessor.messageconsumer.self-service-api.scramPassword=testUserPassword
+## Response producer
+blueprintsprocessor.messageproducer.self-service-api.type=kafka-basic-auth
+blueprintsprocessor.messageproducer.self-service-api.bootstrapServers=127.0.0.1:9092
+blueprintsprocessor.messageproducer.self-service-api.clientId=producer-id
+blueprintsprocessor.messageproducer.self-service-api.topic=producer.t
+
# Kafka audit service Configurations
## Audit request
blueprintsprocessor.messageproducer.self-service-api.audit.kafkaEnable=false
diff --git a/ms/blueprintsprocessor/functions/python-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutor.kt b/ms/blueprintsprocessor/functions/python-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutor.kt
index d66e8b374..d4c8841a8 100644
--- a/ms/blueprintsprocessor/functions/python-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutor.kt
+++ b/ms/blueprintsprocessor/functions/python-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutor.kt
@@ -21,6 +21,7 @@ import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.TimeoutCancellationException
import kotlinx.coroutines.async
import kotlinx.coroutines.withTimeout
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceInput
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.PrepareRemoteEnvInput
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.RemoteIdentifier
@@ -47,11 +48,15 @@ import org.springframework.stereotype.Component
@ConditionalOnBean(name = [ExecutionServiceConstant.SERVICE_GRPC_REMOTE_SCRIPT_EXECUTION])
@Component("component-remote-python-executor")
@Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE)
-open class ComponentRemotePythonExecutor(private val remoteScriptExecutionService: RemoteScriptExecutionService) : AbstractComponentFunction() {
+open class ComponentRemotePythonExecutor(
+ private val remoteScriptExecutionService: RemoteScriptExecutionService,
+ private var bluePrintPropertiesService: BluePrintPropertiesService
+) : AbstractComponentFunction() {
private val log = LoggerFactory.getLogger(ComponentRemotePythonExecutor::class.java)!!
companion object {
+ const val SELECTOR_CMD_EXEC = "blueprintsprocessor.remote-script-command"
const val INPUT_ENDPOINT_SELECTOR = "endpoint-selector"
const val INPUT_DYNAMIC_PROPERTIES = "dynamic-properties"
const val INPUT_ARGUMENT_PROPERTIES = "argument-properties"
@@ -62,6 +67,8 @@ open class ComponentRemotePythonExecutor(private val remoteScriptExecutionServic
const val INPUT_ENV_PREPARE_TIMEOUT = "env-prepare-timeout"
const val INPUT_EXECUTE_TIMEOUT = "execution-timeout"
+ const val STEP_PREPARE_ENV = "prepare-env"
+ const val STEP_EXEC_CMD = "execute-command"
const val ATTRIBUTE_EXEC_CMD_STATUS = "status"
const val ATTRIBUTE_PREPARE_ENV_LOG = "prepare-environment-logs"
const val ATTRIBUTE_EXEC_CMD_LOG = "execute-command-logs"
@@ -74,6 +81,8 @@ open class ComponentRemotePythonExecutor(private val remoteScriptExecutionServic
log.debug("Processing : $operationInputs")
+ val isLogResponseEnabled = bluePrintPropertiesService.propertyBeanType("$SELECTOR_CMD_EXEC.response.log.enabled", Boolean::class.java)
+
val bluePrintContext = bluePrintRuntimeService.bluePrintContext()
val blueprintName = bluePrintContext.name()
val blueprintVersion = bluePrintContext.version()
@@ -142,15 +151,25 @@ open class ComponentRemotePythonExecutor(private val remoteScriptExecutionServic
)
val prepareEnvOutput = remoteScriptExecutionService.prepareEnv(prepareEnvInput)
log.info("$ATTRIBUTE_PREPARE_ENV_LOG - ${prepareEnvOutput.response}")
- val logs = prepareEnvOutput.response
+ val logs = JacksonUtils.jsonNodeFromObject(prepareEnvOutput.response)
val logsEnv = logs.toString().asJsonPrimitive()
setAttribute(ATTRIBUTE_PREPARE_ENV_LOG, logsEnv)
if (prepareEnvOutput.status != StatusType.SUCCESS) {
- setAttribute(ATTRIBUTE_EXEC_CMD_LOG, "N/A".asJsonPrimitive())
- setNodeOutputErrors(prepareEnvOutput.status.name, logsEnv)
+ val errorMessage = prepareEnvOutput.payload
+ setNodeOutputErrors(prepareEnvOutput.status.name,
+ STEP_PREPARE_ENV,
+ logs,
+ errorMessage,
+ isLogResponseEnabled
+ )
} else {
- setNodeOutputProperties(prepareEnvOutput.status.name.asJsonPrimitive(), logsEnv, "".asJsonPrimitive())
+ setNodeOutputProperties(prepareEnvOutput.status.name.asJsonPrimitive(),
+ STEP_PREPARE_ENV,
+ logsEnv,
+ "".asJsonPrimitive(),
+ isLogResponseEnabled
+ )
}
} else {
// set env preparation log to empty...
@@ -159,13 +178,13 @@ open class ComponentRemotePythonExecutor(private val remoteScriptExecutionServic
} catch (grpcEx: io.grpc.StatusRuntimeException) {
val grpcErrMsg = "Command failed during env. preparation... timeout($envPrepTimeout) requestId ($processId)."
setAttribute(ATTRIBUTE_PREPARE_ENV_LOG, grpcErrMsg.asJsonPrimitive())
- setNodeOutputErrors(status = grpcErrMsg, message = "${grpcEx.status}".asJsonPrimitive())
+ setNodeOutputErrors(status = grpcErrMsg, step = STEP_PREPARE_ENV, error = "${grpcEx.status}".asJsonPrimitive(), logging = isLogResponseEnabled)
log.error(grpcErrMsg, grpcEx)
addError(grpcErrMsg)
} catch (e: Exception) {
val timeoutErrMsg = "Command executor failed during env. preparation.. timeout($envPrepTimeout) requestId ($processId)."
setAttribute(ATTRIBUTE_PREPARE_ENV_LOG, e.message.asJsonPrimitive())
- setNodeOutputErrors(status = timeoutErrMsg, message = "${e.message}".asJsonPrimitive())
+ setNodeOutputErrors(status = timeoutErrMsg, step = STEP_PREPARE_ENV, error = "${e.message}".asJsonPrimitive(), logging = isLogResponseEnabled)
log.error("Failed to process on remote executor requestId ($processId)", e)
addError(timeoutErrMsg)
}
@@ -195,18 +214,37 @@ open class ComponentRemotePythonExecutor(private val remoteScriptExecutionServic
}
val logs = JacksonUtils.jsonNodeFromObject(remoteExecutionOutput.response)
if (remoteExecutionOutput.status != StatusType.SUCCESS) {
- setNodeOutputErrors(remoteExecutionOutput.status.name, logs, remoteExecutionOutput.payload)
+ setNodeOutputErrors(remoteExecutionOutput.status.name,
+ STEP_EXEC_CMD,
+ logs,
+ remoteExecutionOutput.payload,
+ isLogResponseEnabled
+ )
} else {
- setNodeOutputProperties(remoteExecutionOutput.status.name.asJsonPrimitive(), logs,
- remoteExecutionOutput.payload)
+ setNodeOutputProperties(remoteExecutionOutput.status.name.asJsonPrimitive(),
+ STEP_EXEC_CMD,
+ logs,
+ remoteExecutionOutput.payload,
+ isLogResponseEnabled
+ )
}
} catch (timeoutEx: TimeoutCancellationException) {
val timeoutErrMsg = "Command executor timed out executing after $executionTimeout seconds requestId ($processId)"
- setNodeOutputErrors(status = timeoutErrMsg, message = "".asJsonPrimitive())
+ setNodeOutputErrors(status = timeoutErrMsg,
+ step = STEP_EXEC_CMD,
+ logs = "".asJsonPrimitive(),
+ error = "".asJsonPrimitive(),
+ logging = isLogResponseEnabled
+ )
log.error(timeoutErrMsg, timeoutEx)
} catch (grpcEx: io.grpc.StatusRuntimeException) {
val timeoutErrMsg = "Command executor timed out executing after $executionTimeout seconds requestId ($processId)"
- setNodeOutputErrors(status = timeoutErrMsg, message = "".asJsonPrimitive())
+ setNodeOutputErrors(status = timeoutErrMsg,
+ step = STEP_EXEC_CMD,
+ logs = "".asJsonPrimitive(),
+ error = "".asJsonPrimitive(),
+ logging = isLogResponseEnabled
+ )
log.error("Command executor time out during GRPC call", grpcEx)
} catch (e: Exception) {
log.error("Failed to process on remote executor requestId ($processId)", e)
@@ -234,25 +272,38 @@ open class ComponentRemotePythonExecutor(private val remoteScriptExecutionServic
/**
* Utility function to set the output properties of the executor node
*/
- private fun setNodeOutputProperties(status: JsonNode, message: JsonNode, artifacts: JsonNode) {
+ private fun setNodeOutputProperties(status: JsonNode, step: String, message: JsonNode, artifacts: JsonNode, logging: Boolean = true) {
setAttribute(ATTRIBUTE_EXEC_CMD_STATUS, status)
- log.info("Executor status : $status")
setAttribute(ATTRIBUTE_RESPONSE_DATA, artifacts)
- log.info("Executor artifacts: $artifacts")
setAttribute(ATTRIBUTE_EXEC_CMD_LOG, message)
- log.info("Executor message : $message")
+
+ if (logging) {
+ log.info("Executor status : $step : $status")
+ log.info("Executor artifacts: $step : $artifacts")
+ log.info("Executor message : $step : $message")
+ }
}
/**
* Utility function to set the output properties and errors of the executor node, in cas of errors
*/
- private fun setNodeOutputErrors(status: String, message: JsonNode, artifacts: JsonNode = "".asJsonPrimitive()) {
+ private fun setNodeOutputErrors(
+ status: String,
+ step: String,
+ logs: JsonNode = "N/A".asJsonPrimitive(),
+ error: JsonNode,
+ logging: Boolean = true
+ ) {
setAttribute(ATTRIBUTE_EXEC_CMD_STATUS, status.asJsonPrimitive())
- log.info("Executor status : $status")
- setAttribute(ATTRIBUTE_EXEC_CMD_LOG, message)
- log.info("Executor message : $message")
- setAttribute(ATTRIBUTE_RESPONSE_DATA, artifacts)
- log.info("Executor artifacts: $artifacts")
- addError(status, ATTRIBUTE_EXEC_CMD_LOG, message.toString())
+ setAttribute(ATTRIBUTE_EXEC_CMD_LOG, logs)
+ setAttribute(ATTRIBUTE_RESPONSE_DATA, "N/A".asJsonPrimitive())
+
+ if (logging) {
+ log.info("Executor status : $step : $status")
+ log.info("Executor message : $step : $error")
+ log.info("Executor logs : $step : $logs")
+ }
+
+ addError(status, step, error.toString())
}
}
diff --git a/ms/blueprintsprocessor/functions/python-executor/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutorTest.kt b/ms/blueprintsprocessor/functions/python-executor/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutorTest.kt
index 5e57b9eb7..d4edf4bb2 100644
--- a/ms/blueprintsprocessor/functions/python-executor/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutorTest.kt
+++ b/ms/blueprintsprocessor/functions/python-executor/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/python/executor/ComponentRemotePythonExecutorTest.kt
@@ -22,6 +22,7 @@ import io.mockk.every
import io.mockk.mockk
import kotlinx.coroutines.runBlocking
import org.junit.Test
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceInput
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.PrepareRemoteEnvInput
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.RemoteScriptExecutionInput
@@ -47,7 +48,10 @@ class ComponentRemotePythonExecutorTest {
runBlocking {
val remoteScriptExecutionService = MockRemoteScriptExecutionService()
- val componentRemotePythonExecutor = ComponentRemotePythonExecutor(remoteScriptExecutionService)
+ val componentRemotePythonExecutor = ComponentRemotePythonExecutor(
+ remoteScriptExecutionService,
+ mockk<BluePrintPropertiesService>()
+ )
val executionServiceInput =
JacksonUtils.readValueFromClassPathFile(
@@ -88,7 +92,10 @@ class ComponentRemotePythonExecutorTest {
fun testComponentRemotePythonExecutorProcessNB() {
runBlocking {
val remoteScriptExecutionService = MockRemoteScriptExecutionService()
- val componentRemotePythonExecutor = ComponentRemotePythonExecutor(remoteScriptExecutionService)
+ val componentRemotePythonExecutor = ComponentRemotePythonExecutor(
+ remoteScriptExecutionService,
+ mockk<BluePrintPropertiesService>()
+ )
val bluePrintRuntime = mockk<DefaultBluePrintRuntimeService>("123456-1000")
every { bluePrintRuntime.getBluePrintError() } answers { BluePrintError() } // successful case.
diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageLibData.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageLibData.kt
index ac35fbf2c..b07d64388 100644
--- a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageLibData.kt
+++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageLibData.kt
@@ -78,7 +78,7 @@ open class KafkaSslAuthMessageProducerProperties : KafkaBasicAuthMessageProducer
var keystore: String? = null
var keystorePassword: String? = null
var keystoreType: String = SslConfigs.DEFAULT_SSL_KEYSTORE_TYPE
- var sslEndpointIdentificationAlgorithm: String = ""
+ var sslEndpointIdentificationAlgorithm: String = SslConfigs.DEFAULT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM
override fun getConfig(): HashMap<String, Any> {
val configProps = super.getConfig()
@@ -142,7 +142,7 @@ open class KafkaStreamsSslAuthConsumerProperties : KafkaStreamsBasicAuthConsumer
var keystore: String? = null
var keystorePassword: String? = null
var keystoreType: String = SslConfigs.DEFAULT_SSL_KEYSTORE_TYPE
- var sslEndpointIdentificationAlgorithm: String = ""
+ var sslEndpointIdentificationAlgorithm: String = SslConfigs.DEFAULT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM
override fun getConfig(): HashMap<String, Any> {
val configProps = super.getConfig()
@@ -218,7 +218,7 @@ open class KafkaSslAuthMessageConsumerProperties : KafkaBasicAuthMessageConsumer
var keystore: String? = null
var keystorePassword: String? = null
var keystoreType: String = SslConfigs.DEFAULT_SSL_KEYSTORE_TYPE
- var sslEndpointIdentificationAlgorithm: String = ""
+ var sslEndpointIdentificationAlgorithm: String = SslConfigs.DEFAULT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM
override fun getConfig(): HashMap<String, Any> {
val configProps = super.getConfig()
diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaMessageProducerService.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaMessageProducerService.kt
index 931f052ed..e4991d2d8 100644
--- a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaMessageProducerService.kt
+++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaMessageProducerService.kt
@@ -17,12 +17,16 @@
package org.onap.ccsdk.cds.blueprintsprocessor.message.service
+import com.fasterxml.jackson.databind.node.ObjectNode
import org.apache.commons.lang.builder.ToStringBuilder
import org.apache.kafka.clients.producer.Callback
import org.apache.kafka.clients.producer.KafkaProducer
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.header.internals.RecordHeader
+import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceOutput
+import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.Status
import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageProducerProperties
+import org.onap.ccsdk.cds.controllerblueprints.core.asJsonPrimitive
import org.onap.ccsdk.cds.controllerblueprints.core.asJsonString
import org.onap.ccsdk.cds.controllerblueprints.core.defaultToUUID
import org.slf4j.LoggerFactory
@@ -39,6 +43,10 @@ class KafkaMessageProducerService(
private val messageLoggerService = MessageLoggerService()
+ companion object {
+ const val MAX_ERR_MSG_LEN = 128
+ }
+
override suspend fun sendMessageNB(message: Any): Boolean {
checkNotNull(messageProducerProperties.topic) { "default topic is not configured" }
return sendMessageNB(messageProducerProperties.topic!!, message)
@@ -54,9 +62,14 @@ class KafkaMessageProducerService(
message: Any,
headers: MutableMap<String, String>?
): Boolean {
- val byteArrayMessage = when (message) {
- is String -> message.toByteArray(Charset.defaultCharset())
- else -> message.asJsonString().toByteArray(Charset.defaultCharset())
+ var clonedMessage = message
+ if (clonedMessage is ExecutionServiceOutput) {
+ clonedMessage = truncateResponse(clonedMessage)
+ }
+
+ val byteArrayMessage = when (clonedMessage) {
+ is String -> clonedMessage.toByteArray(Charset.defaultCharset())
+ else -> clonedMessage.asJsonString().toByteArray(Charset.defaultCharset())
}
val record = ProducerRecord<String, ByteArray>(topic, defaultToUUID(), byteArrayMessage)
@@ -85,4 +98,37 @@ class KafkaMessageProducerService(
return kafkaProducer!!
}
+
+ /**
+ * Truncation of BP responses
+ */
+ private fun truncateResponse(executionServiceOutput: ExecutionServiceOutput): ExecutionServiceOutput {
+ /** Truncation of error messages */
+ var truncErrMsg = executionServiceOutput.status.errorMessage
+ if (truncErrMsg != null && truncErrMsg.length > MAX_ERR_MSG_LEN) {
+ truncErrMsg = "${truncErrMsg.substring(0,MAX_ERR_MSG_LEN)}" +
+ " [...]. Check Blueprint Processor logs for more information."
+ }
+ /** Truncation for Command Executor responses */
+ var truncPayload = executionServiceOutput.payload.deepCopy()
+ val workflowName = executionServiceOutput.actionIdentifiers.actionName
+ if (truncPayload.path("$workflowName-response").has("execute-command-logs")) {
+ var cmdExecLogNode = truncPayload.path("$workflowName-response") as ObjectNode
+ cmdExecLogNode.replace("execute-command-logs", "Check Command Executor logs for more information.".asJsonPrimitive())
+ }
+ return ExecutionServiceOutput().apply {
+ correlationUUID = executionServiceOutput.correlationUUID
+ commonHeader = executionServiceOutput.commonHeader
+ actionIdentifiers = executionServiceOutput.actionIdentifiers
+ status = Status().apply {
+ code = executionServiceOutput.status.code
+ eventType = executionServiceOutput.status.eventType
+ timestamp = executionServiceOutput.status.timestamp
+ errorMessage = truncErrMsg
+ message = executionServiceOutput.status.message
+ }
+ payload = truncPayload
+ stepData = executionServiceOutput.stepData
+ }
+ }
}
diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageConsumerServiceTest.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageConsumerServiceTest.kt
index ac08dc7b7..fdf6e48e7 100644
--- a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageConsumerServiceTest.kt
+++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageConsumerServiceTest.kt
@@ -214,7 +214,7 @@ open class BlueprintMessageConsumerServiceTest {
SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG to "/path/to/keystore.jks",
SslConfigs.SSL_KEYSTORE_TYPE_CONFIG to "JKS",
SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG to "secretpassword",
- SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG to "",
+ SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG to SslConfigs.DEFAULT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM,
SaslConfigs.SASL_MECHANISM to "SCRAM-SHA-512",
SaslConfigs.SASL_JAAS_CONFIG to "${ScramLoginModule::class.java.canonicalName} required " +
"username=\"sample-user\" " +
diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageProducerServiceTest.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageProducerServiceTest.kt
index 72a47ed56..da7394998 100644
--- a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageProducerServiceTest.kt
+++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageProducerServiceTest.kt
@@ -109,7 +109,7 @@ open class BlueprintMessageProducerServiceTest {
SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG to "/path/to/keystore.jks",
SslConfigs.SSL_KEYSTORE_TYPE_CONFIG to "JKS",
SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG to "secretpassword",
- SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG to "",
+ SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG to SslConfigs.DEFAULT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM,
SaslConfigs.SASL_MECHANISM to "SCRAM-SHA-512",
SaslConfigs.SASL_JAAS_CONFIG to "${ScramLoginModule::class.java.canonicalName} required " +
"username=\"sample-user\" " +
diff --git a/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/selfservice/api/BluePrintProcessingKafkaConsumer.kt b/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/selfservice/api/BluePrintProcessingKafkaConsumer.kt
index 49f2a50d5..a95af8123 100644
--- a/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/selfservice/api/BluePrintProcessingKafkaConsumer.kt
+++ b/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/selfservice/api/BluePrintProcessingKafkaConsumer.kt
@@ -51,13 +51,15 @@ open class BluePrintProcessingKafkaConsumer(
companion object {
const val CONSUMER_SELECTOR = "self-service-api"
+ const val PRODUCER_SELECTOR = "self-service-api"
}
@EventListener(ApplicationReadyEvent::class)
fun setupMessageListener() = runBlocking {
try {
log.info(
- "Setting up message consumer($CONSUMER_SELECTOR)"
+ "Setting up message consumer($CONSUMER_SELECTOR)" +
+ "message producer($PRODUCER_SELECTOR)..."
)
/** Get the Message Consumer Service **/
@@ -72,6 +74,18 @@ open class BluePrintProcessingKafkaConsumer(
throw BluePrintProcessorException("failed to create consumer service ${e.message}")
}
+ /** Get the Message Producer Service **/
+ val blueprintMessageProducerService = try {
+ bluePrintMessageLibPropertyService
+ .blueprintMessageProducerService(PRODUCER_SELECTOR)
+ } catch (e: BluePrintProcessorException) {
+ val errorMsg = "Failed creating Kafka producer message service."
+ throw e.updateErrorMessage(SelfServiceApiDomains.SELF_SERVICE_API, errorMsg,
+ "Wrong Kafka selector provided or internal error in Kafka service.")
+ } catch (e: Exception) {
+ throw BluePrintProcessorException("failed to create producer service ${e.message}")
+ }
+
launch {
/** Subscribe to the consumer topics */
val additionalConfig: MutableMap<String, Any> = hashMapOf()
@@ -82,7 +96,8 @@ open class BluePrintProcessingKafkaConsumer(
ph.register()
log.trace("Consumed Message : $message")
val executionServiceInput = message.jsonAsType<ExecutionServiceInput>()
- executionServiceHandler.doProcess(executionServiceInput)
+ val executionServiceOutput = executionServiceHandler.doProcess(executionServiceInput)
+ blueprintMessageProducerService.sendMessage(executionServiceOutput)
} catch (e: Exception) {
log.error("failed in processing the consumed message : $message", e)
} finally {
@@ -93,7 +108,8 @@ open class BluePrintProcessingKafkaConsumer(
}
} catch (e: Exception) {
log.error(
- "failed to start message consumer($CONSUMER_SELECTOR) ", e
+ "failed to start message consumer($CONSUMER_SELECTOR) " +
+ "message producer($PRODUCER_SELECTOR) ", e
)
}
}
@@ -102,7 +118,8 @@ open class BluePrintProcessingKafkaConsumer(
fun shutdownMessageListener() = runBlocking {
try {
log.info(
- "Shutting down message consumer($CONSUMER_SELECTOR)"
+ "Shutting down message consumer($CONSUMER_SELECTOR)" +
+ "message producer($PRODUCER_SELECTOR)..."
)
blueprintMessageConsumerService.shutDown()
ph.arriveAndAwaitAdvance()
diff --git a/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/selfservice/api/KafkaPublishAuditService.kt b/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/selfservice/api/KafkaPublishAuditService.kt
index 129e7a54d..1c5d47c27 100644
--- a/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/selfservice/api/KafkaPublishAuditService.kt
+++ b/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/selfservice/api/KafkaPublishAuditService.kt
@@ -24,6 +24,7 @@ import org.onap.ccsdk.cds.blueprintsprocessor.functions.resource.resolution.Reso
import org.onap.ccsdk.cds.blueprintsprocessor.message.service.BluePrintMessageLibPropertyService
import org.onap.ccsdk.cds.blueprintsprocessor.message.service.BlueprintMessageProducerService
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
+import org.onap.ccsdk.cds.controllerblueprints.core.asJsonPrimitive
import org.onap.ccsdk.cds.controllerblueprints.core.common.ApplicationConstants
import org.onap.ccsdk.cds.controllerblueprints.core.interfaces.BluePrintCatalogService
import org.onap.ccsdk.cds.controllerblueprints.core.utils.BluePrintMetadataUtils
@@ -48,12 +49,9 @@ class KafkaPublishAuditService(
private val bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService,
private val blueprintsProcessorCatalogService: BluePrintCatalogService
) : PublishAuditService {
-
private var inputInstance: BlueprintMessageProducerService? = null
private var outputInstance: BlueprintMessageProducerService? = null
-
private lateinit var correlationUUID: String
-
private val log = LoggerFactory.getLogger(KafkaPublishAuditService::class.toString())
companion object {
@@ -127,7 +125,8 @@ class KafkaPublishAuditService(
correlationUUID = executionServiceInput.correlationUUID
commonHeader = executionServiceInput.commonHeader
actionIdentifiers = executionServiceInput.actionIdentifiers
- payload = executionServiceInput.payload
+ payload = executionServiceInput.payload.deepCopy()
+ stepData = executionServiceInput.stepData
}
val blueprintName = clonedExecutionServiceInput.actionIdentifiers.blueprintName
@@ -173,8 +172,7 @@ class KafkaPublishAuditService(
sensitiveParameters.forEach { sensitiveParameter ->
if (workflowProperties.has(sensitiveParameter)) {
- workflowProperties.remove(sensitiveParameter)
- workflowProperties.put(sensitiveParameter, ApplicationConstants.LOG_REDACTED)
+ workflowProperties.replace(sensitiveParameter, ApplicationConstants.LOG_REDACTED.asJsonPrimitive())
}
}
}