From 136e3e19a1821a694d799e7c8d670407720c9e66 Mon Sep 17 00:00:00 2001 From: Julien Fontaine Date: Tue, 26 Jan 2021 14:47:32 -0500 Subject: Improve logging for CDS Kafka workers Modified CDS Kafka consumersand producers logs to provide more details about the topic of the consumer record being consumed or published. Refactored the publish callback to make it more readable. Refactored audit service log error messages. Issue-ID: CCSDK-3154 Signed-off-by: Julien Fontaine Change-Id: I7b42930e926bc15ce175974a78d3bfe2f219b0a8 --- .../message/service/KafkaMessageConsumerService.kt | 9 +++++- .../message/service/KafkaMessageProducerService.kt | 32 ++++++++++++++-------- 2 files changed, 29 insertions(+), 12 deletions(-) (limited to 'ms/blueprintsprocessor/modules/commons/message-lib/src/main') diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaMessageConsumerService.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaMessageConsumerService.kt index 83cc0e022..af689a1f2 100644 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaMessageConsumerService.kt +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaMessageConsumerService.kt @@ -1,6 +1,6 @@ /* * Copyright © 2019 IBM. - * Modifications Copyright © 2018-2019 AT&T Intellectual Property. + * Modifications Copyright © 2018-2021 AT&T, Bell Canada Intellectual Property. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -81,6 +81,13 @@ open class KafkaMessageConsumerService( /** execute the command block */ if (!channel.isClosedForSend) { channel.send(consumerRecord) + log.info( + "Channel sent Consumer Record : topic(${consumerRecord.topic()}) " + + "partition(${consumerRecord.partition()}) " + + "leaderEpoch(${consumerRecord.leaderEpoch().get()}) " + + "offset(${consumerRecord.offset()}) " + + "key(${consumerRecord.key()})" + ) } else { log.error("Channel is closed to receive message") } 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 eccf75301..88b0dfaae 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 @@ -1,6 +1,6 @@ /* * Copyright © 2019 IBM. - * Modifications Copyright © 2018-2019 AT&T Intellectual Property. + * Modifications Copyright © 2018-2021 AT&T, Bell Canada Intellectual Property. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -77,16 +77,12 @@ class KafkaMessageProducerService( } val callback = Callback { metadata, exception -> if (exception != null) - log.error("ERROR : ${exception.message}") + log.error("Couldn't publish ${clonedMessage::class.simpleName} ${getMessageLogData(clonedMessage)}.", exception) else { - var logMessage = when (clonedMessage) { - is ExecutionServiceInput -> - "Request published to ${metadata.topic()} for CBA: ${clonedMessage.actionIdentifiers.blueprintName} version: ${clonedMessage.actionIdentifiers.blueprintVersion}" - is ExecutionServiceOutput -> - "Response published to ${metadata.topic()} for CBA: ${clonedMessage.actionIdentifiers.blueprintName} version: ${clonedMessage.actionIdentifiers.blueprintVersion}" - else -> "Message published to(${metadata.topic()}), offset(${metadata.offset()}), headers :$headers" - } - log.info(logMessage) + val message = "${clonedMessage::class.simpleName} published : topic(${metadata.topic()}) " + + "partition(${metadata.partition()}) " + + "offset(${metadata.offset()}) ${getMessageLogData(clonedMessage)}." + log.info(message) } } messageTemplate().send(record, callback) @@ -114,7 +110,7 @@ class KafkaMessageProducerService( /** 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)}" + + truncErrMsg = truncErrMsg.substring(0, MAX_ERR_MSG_LEN) + " [...]. Check Blueprint Processor logs for more information." } /** Truncation for Command Executor responses */ @@ -139,4 +135,18 @@ class KafkaMessageProducerService( stepData = executionServiceOutput.stepData } } + + private fun getMessageLogData(message: Any): String { + return when (message) { + is ExecutionServiceInput -> { + val actionIdentifiers = message.actionIdentifiers + "CBA(${actionIdentifiers.blueprintName}/${actionIdentifiers.blueprintVersion}/${actionIdentifiers.actionName})" + } + is ExecutionServiceOutput -> { + val actionIdentifiers = message.actionIdentifiers + "CBA(${actionIdentifiers.blueprintName}/${actionIdentifiers.blueprintVersion}/${actionIdentifiers.actionName})" + } + else -> "message($message)" + } + } } -- cgit 1.2.3-korg