Improve log format consistency

Logs related to request processing should include: requestId,
subrequestId and originatorId. Each API (rest,grpc,kafka) would
produce different outputs, due to MDC context not being passed
properly between coroutine contexts or values not being populated.

Issue-ID: CCSDK-3686
Signed-off-by: Jozsef Csongvai <jozsef.csongvai@bell.ca>
Change-Id: Ibafdffd3409b9724ad91633ca5840070f7e287f5

4 files changed, 44 insertions, 0 deletions

diff --git a/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/service/GrpcLoggerService.kt b/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/service/GrpcLoggerService.kt
index 732f4fecf..987ac447a 100644
--- a/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/service/GrpcLoggerService.kt
+++ b/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/service/GrpcLoggerService.kt
@@ -20,6 +20,10 @@ import io.grpc.Grpc
 import io.grpc.Metadata
 import io.grpc.ServerCall
 import io.grpc.ServerCallHandler
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.newCoroutineContext
+import kotlinx.coroutines.withContext
 import org.onap.ccsdk.cds.blueprintsprocessor.grpc.getStringKey
 import org.onap.ccsdk.cds.blueprintsprocessor.grpc.putStringKeyValue
 import org.onap.ccsdk.cds.controllerblueprints.common.api.CommonHeader
@@ -28,9 +32,11 @@ import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants.ONAP_INVO
 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants.ONAP_PARTNER_NAME
 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants.ONAP_REQUEST_ID
 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
+import org.onap.ccsdk.cds.controllerblueprints.core.MDCContext
 import org.onap.ccsdk.cds.controllerblueprints.core.defaultToEmpty
 import org.onap.ccsdk.cds.controllerblueprints.core.defaultToUUID
 import org.onap.ccsdk.cds.controllerblueprints.core.logger
+import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput
 import org.slf4j.MDC
 import java.net.InetAddress
 import java.net.InetSocketAddress
@@ -108,3 +114,16 @@ class GrpcLoggerService {
         }
     }
 }
+suspend fun <T> mdcGrpcCoroutineScope(
+    executionServiceInput: ExecutionServiceInput,
+    block: suspend CoroutineScope.() -> T
+) = coroutineScope {
+
+    MDC.put("RequestID", executionServiceInput.commonHeader.requestId)
+    MDC.put("SubRequestID", executionServiceInput.commonHeader.subRequestId)
+    MDC.put("OriginatorID", executionServiceInput.commonHeader.originatorId)
+
+    withContext(newCoroutineContext(this.coroutineContext + MDCContext(MDC.getCopyOfContextMap()))) {
+        block()
+    }
+}
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 cccc61f40..e5064957d 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
@@ -32,6 +32,7 @@ import org.onap.ccsdk.cds.blueprintsprocessor.message.utils.BlueprintMessageUtil
 import org.onap.ccsdk.cds.controllerblueprints.core.asJsonPrimitive
 import org.onap.ccsdk.cds.controllerblueprints.core.asJsonString
 import org.slf4j.LoggerFactory
+import org.slf4j.MDC
 import java.nio.charset.Charset
 
 class KafkaMessageProducerService(
@@ -78,7 +79,9 @@ class KafkaMessageProducerService(
         headers?.let {
             headers.forEach { (key, value) -> recordHeaders.add(RecordHeader(key, value.toByteArray())) }
         }
+        val context = MDC.getCopyOfContextMap()
         val callback = Callback { metadata, exception ->
+            MDC.setContextMap(context)
             meterRegistry.counter(
                 BlueprintMessageMetricConstants.KAFKA_PRODUCED_MESSAGES_COUNTER,
                 BlueprintMessageUtils.kafkaMetricTag(topic)
diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/MessageLoggerService.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/MessageLoggerService.kt
index 90b850017..ff2d43fcd 100644
--- a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/MessageLoggerService.kt
+++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/MessageLoggerService.kt
@@ -16,12 +16,18 @@
 
 package org.onap.ccsdk.cds.blueprintsprocessor.message.service
 
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.newCoroutineContext
+import kotlinx.coroutines.withContext
 import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.apache.kafka.common.header.Headers
 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.CommonHeader
+import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceInput
 import org.onap.ccsdk.cds.blueprintsprocessor.message.addHeader
 import org.onap.ccsdk.cds.blueprintsprocessor.message.toMap
 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants
+import org.onap.ccsdk.cds.controllerblueprints.core.MDCContext
 import org.onap.ccsdk.cds.controllerblueprints.core.defaultToEmpty
 import org.onap.ccsdk.cds.controllerblueprints.core.defaultToUUID
 import org.onap.ccsdk.cds.controllerblueprints.core.logger
@@ -93,3 +99,17 @@ class MessageLoggerService {
         MDC.clear()
     }
 }
+
+suspend fun <T> mdcKafkaCoroutineScope(
+    executionServiceInput: ExecutionServiceInput,
+    block: suspend CoroutineScope.() -> T
+) = coroutineScope {
+
+    MDC.put("RequestID", executionServiceInput.commonHeader.requestId)
+    MDC.put("SubRequestID", executionServiceInput.commonHeader.subRequestId)
+    MDC.put("OriginatorID", executionServiceInput.commonHeader.originatorId)
+
+    withContext(newCoroutineContext(this.coroutineContext + MDCContext(MDC.getCopyOfContextMap()))) {
+        block()
+    }
+}
diff --git a/ms/blueprintsprocessor/modules/commons/rest-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/rest/service/RestLoggerService.kt b/ms/blueprintsprocessor/modules/commons/rest-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/rest/service/RestLoggerService.kt
index b1d8abd16..4b9bbb19f 100644
--- a/ms/blueprintsprocessor/modules/commons/rest-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/rest/service/RestLoggerService.kt
+++ b/ms/blueprintsprocessor/modules/commons/rest-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/rest/service/RestLoggerService.kt
@@ -126,6 +126,8 @@ suspend fun <T> mdcWebCoroutineScope(
         val mdcContext = if (executionServiceInput != null) {
             // MDC Context with overridden request ID
             MDC.put("RequestID", executionServiceInput.commonHeader.requestId)
+            MDC.put("SubRequestID", executionServiceInput.commonHeader.subRequestId)
+            MDC.put("OriginatorID", executionServiceInput.commonHeader.originatorId)
             MDCContext(MDC.getCopyOfContextMap())
         } else {
             // Default MDC Context