Add log diagnostic context

As it's not trivial to use MDC directly from logging framework in
reactive application, we need to do some work manually. The approach
proposed is an explicit MDC handling, which means that context is
kept as an object created after establishing client connection. Next,
new instance of HvVesCollector (and its dependencies) is created. Every
object is propagated with ClientContext so it can use it when calling
logger methods.

In the future ClientContext might be used to support other use-cases,
ie. per-topic access control.

As a by-product I had to refactor our Logger wrapper, too. It already
had too many functions and after adding MDC number would be doubled.

Change-Id: I9c5d3f5e1d1be1db66d28d292eb0e1c38d8d0ffe
Issue-ID: DCAEGEN2-671
Signed-off-by: Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

5 files changed, 173 insertions, 144 deletions

diff --git a/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/http/ratpack.kt b/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/http/ratpack.kt
index 5a733f24..a25b2912 100644
--- a/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/http/ratpack.kt
+++ b/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/http/ratpack.kt
@@ -51,7 +51,7 @@ fun <A> ratpack.http.Response.sendEitherErrorOrResponse(response: Either<A, Resp
 fun ratpack.http.Response.sendAndHandleErrors(response: IO<Response>) {
     response.attempt().unsafeRunSync().fold(
             { err ->
-                logger.warn("Error occurred. Sending .", err)
+                logger.withWarn { log("Error occurred. Sending .", err) }
                 val message = err.message
                 send(errorResponse(message))
             },
diff --git a/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/logging/Logger.kt b/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/logging/Logger.kt
index 033dd5e5..2fb48803 100644
--- a/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/logging/Logger.kt
+++ b/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/logging/Logger.kt
@@ -21,117 +21,171 @@ package org.onap.dcae.collectors.veshv.utils.logging
 
 import kotlin.reflect.KClass
 import org.slf4j.LoggerFactory
+import org.slf4j.MDC
+
+typealias MappedDiagnosticContext = () -> Map<String, String>
 
 @Suppress("TooManyFunctions", "SuboptimalLoggerUsage")
-class Logger(val logger: org.slf4j.Logger) {
+class Logger(logger: org.slf4j.Logger) {
     constructor(clazz: KClass<out Any>) : this(LoggerFactory.getLogger(clazz.java))
     constructor(name: String) : this(LoggerFactory.getLogger(name))
 
-    //
-    // TRACE
-    //
+    private val errorLogger = if (logger.isErrorEnabled) ErrorLevelLogger(logger) else OffLevelLogger
+    private val warnLogger = if (logger.isWarnEnabled) WarnLevelLogger(logger) else OffLevelLogger
+    private val infoLogger = if (logger.isInfoEnabled) InfoLevelLogger(logger) else OffLevelLogger
+    private val debugLogger = if (logger.isDebugEnabled) DebugLevelLogger(logger) else OffLevelLogger
+    private val traceLogger = if (logger.isTraceEnabled) TraceLevelLogger(logger) else OffLevelLogger
 
-    val traceEnabled: Boolean
-        get() = logger.isTraceEnabled
+    // ERROR
 
-    fun trace(messageProvider: () -> String) {
-        if (logger.isTraceEnabled) {
-            logger.trace(messageProvider())
-        }
-    }
+    fun withError(block: AtLevelLogger.() -> Unit) = errorLogger.block()
 
-    //
-    // DEBUG
-    //
+    fun withError(mdc: MappedDiagnosticContext, block: AtLevelLogger.() -> Unit) =
+            errorLogger.withMdc(mdc, block)
 
-    fun debug(message: String) {
-        logger.debug(message)
+    fun error(message: () -> String) = errorLogger.run {
+        log(message())
     }
 
-    fun debug(message: String, t: Throwable) {
-        logger.debug(message, t)
+    fun error(mdc: MappedDiagnosticContext, message: () -> String) =
+            errorLogger.withMdc(mdc) { log(message()) }
+
+    // WARN
+
+    fun withWarn(block: AtLevelLogger.() -> Unit) = warnLogger.block()
+
+    fun withWarn(mdc: MappedDiagnosticContext, block: AtLevelLogger.() -> Unit) =
+            warnLogger.withMdc(mdc, block)
+
+    fun warn(message: () -> String) = warnLogger.run {
+        log(message())
     }
 
-    fun debug(messageProvider: () -> String) {
-        if (logger.isDebugEnabled) {
-            logger.debug(messageProvider())
-        }
+    fun warn(mdc: MappedDiagnosticContext, message: () -> String) =
+            warnLogger.withMdc(mdc) { log(message()) }
+
+
+    // INFO
+
+    fun withInfo(block: AtLevelLogger.() -> Unit) = infoLogger.block()
+
+    fun withInfo(mdc: MappedDiagnosticContext, block: AtLevelLogger.() -> Unit) =
+            infoLogger.withMdc(mdc, block)
+
+    fun info(message: () -> String) = infoLogger.run {
+        log(message())
     }
 
-    fun debug(t: Throwable, messageProvider: () -> String) {
-        if (logger.isDebugEnabled) {
-            logger.debug(messageProvider(), t)
-        }
+    fun info(mdc: MappedDiagnosticContext, message: () -> String) =
+            infoLogger.withMdc(mdc) { log(message()) }
+
+    // DEBUG
+
+    fun withDebug(block: AtLevelLogger.() -> Unit) = debugLogger.block()
+
+    fun withDebug(mdc: MappedDiagnosticContext, block: AtLevelLogger.() -> Unit) =
+            debugLogger.withMdc(mdc, block)
+
+    fun debug(message: () -> String) = debugLogger.run {
+        log(message())
     }
 
-    //
-    // INFO
-    //
-    fun info(message: String) {
-        logger.info(message)
+    fun debug(mdc: MappedDiagnosticContext, message: () -> String) =
+            debugLogger.withMdc(mdc) { log(message()) }
+
+
+    // TRACE
+
+    fun withTrace(block: AtLevelLogger.() -> Unit) = traceLogger.block()
+
+    fun withTrace(mdc: MappedDiagnosticContext, block: AtLevelLogger.() -> Unit) =
+            traceLogger.withMdc(mdc, block)
+
+    fun trace(message: () -> String) = traceLogger.run {
+        log(message())
     }
 
-    fun info(messageProvider: () -> String) {
-        if (logger.isInfoEnabled) {
-            logger.info(messageProvider())
+    fun trace(mdc: MappedDiagnosticContext, message: () -> String) =
+            traceLogger.withMdc(mdc) { log(message()) }
+
+}
+
+abstract class AtLevelLogger {
+    abstract fun log(message: String)
+    abstract fun log(message: String, t: Throwable)
+    open val enabled: Boolean
+        get() = true
+
+    inline fun withMdc(mdc: MappedDiagnosticContext, block: AtLevelLogger.() -> Unit) {
+        if (enabled) {
+            try {
+                MDC.setContextMap(mdc())
+                block()
+            } finally {
+                MDC.clear()
+            }
         }
     }
+}
 
-    fun info(message: String, t: Throwable) {
-        logger.info(message, t)
+object OffLevelLogger : AtLevelLogger() {
+    override val enabled = false
+
+    override fun log(message: String) {
+        // do not log anything
     }
 
-    fun info(t: Throwable, messageProvider: () -> String) {
-        if (logger.isInfoEnabled) {
-            logger.info(messageProvider(), t)
-        }
+    override fun log(message: String, t: Throwable) {
+        // do not log anything
     }
+}
 
-    //
-    // WARN
-    //
+class ErrorLevelLogger(private val logger: org.slf4j.Logger) : AtLevelLogger() {
+    override fun log(message: String) {
+        logger.error(message)
+    }
+
+    override fun log(message: String, t: Throwable) {
+        logger.error(message, t)
+    }
+}
 
-    fun warn(message: String) {
+class WarnLevelLogger(private val logger: org.slf4j.Logger) : AtLevelLogger() {
+    override fun log(message: String) {
         logger.warn(message)
     }
 
-    fun warn(message: String, t: Throwable) {
+    override fun log(message: String, t: Throwable) {
         logger.warn(message, t)
     }
+}
 
-    fun warn(messageProvider: () -> String) {
-        if (logger.isWarnEnabled) {
-            logger.warn(messageProvider())
-        }
+class InfoLevelLogger(private val logger: org.slf4j.Logger) : AtLevelLogger() {
+    override fun log(message: String) {
+        logger.info(message)
     }
 
-    fun warn(t: Throwable, messageProvider: () -> String) {
-        if (logger.isWarnEnabled) {
-            logger.warn(messageProvider(), t)
-        }
+    override fun log(message: String, t: Throwable) {
+        logger.info(message, t)
     }
+}
 
-    //
-    // ERROR
-    //
-
-    fun error(message: String) {
-        logger.error(message)
+class DebugLevelLogger(private val logger: org.slf4j.Logger) : AtLevelLogger() {
+    override fun log(message: String) {
+        logger.debug(message)
     }
 
-    fun error(message: String, t: Throwable) {
-        logger.error(message, t)
+    override fun log(message: String, t: Throwable) {
+        logger.debug(message, t)
     }
+}
 
-    fun error(messageProvider: () -> String) {
-        if (logger.isErrorEnabled) {
-            logger.error(messageProvider())
-        }
+class TraceLevelLogger(private val logger: org.slf4j.Logger) : AtLevelLogger() {
+    override fun log(message: String) {
+        logger.trace(message)
     }
 
-    fun error(t: Throwable, messageProvider: () -> String) {
-        if (logger.isErrorEnabled) {
-            logger.error(messageProvider(), t)
-        }
+    override fun log(message: String, t: Throwable) {
+        logger.trace(message, t)
     }
 }
diff --git a/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/logging/reactive_logging.kt b/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/logging/reactive_logging.kt
index e8ec2549..1e98f2fc 100644
--- a/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/logging/reactive_logging.kt
+++ b/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/logging/reactive_logging.kt
@@ -25,42 +25,49 @@ import arrow.core.Try
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 
-fun <T> Logger.handleReactiveStreamError(ex: Throwable, returnFlux: Flux<T> = Flux.empty()): Flux<T> {
-    logger.warn("Error while handling message stream: ${ex::class.qualifiedName} (${ex.localizedMessage})")
-    logger.debug("Detailed stack trace", ex)
+fun <T> Logger.handleReactiveStreamError(
+        context: MappedDiagnosticContext,
+        ex: Throwable,
+        returnFlux: Flux<T> = Flux.empty()): Flux<T> {
+    warn(context) { "Error while handling message stream: ${ex::class.qualifiedName} (${ex.localizedMessage})" }
+    withDebug(context) { log("Detailed stack trace", ex) }
     return returnFlux
 }
 
 
 fun <T> Try<T>.filterFailedWithLog(logger: Logger,
+                                   context: MappedDiagnosticContext,
                                    acceptedMsg: (T) -> String,
                                    rejectedMsg: (Throwable) -> String): Flux<T> =
-        fold({
-            logger.warn(rejectedMsg(it))
+        fold({ ex ->
+            logger.withWarn(context) { log(rejectedMsg(ex)) }
             Flux.empty<T>()
-        }, {
-            logger.trace { acceptedMsg(it) }
-            Flux.just(it)
+        }, { obj ->
+            logger.trace(context) { acceptedMsg(obj) }
+            Flux.just(obj)
         })
 
 fun <T> Option<T>.filterEmptyWithLog(logger: Logger,
+                                     context: MappedDiagnosticContext,
                                      acceptedMsg: (T) -> String,
                                      rejectedMsg: () -> String): Flux<T> =
         fold({
-            logger.warn(rejectedMsg)
+            logger.warn(context, rejectedMsg)
             Flux.empty<T>()
         }, {
-            logger.trace { acceptedMsg(it) }
+            logger.trace(context) { acceptedMsg(it) }
             Flux.just(it)
         })
 
-fun <T> Flux<T>.filterFailedWithLog(logger: Logger, predicate: (T) -> Either<() -> String, () -> String>) =
+fun <T> Flux<T>.filterFailedWithLog(logger: Logger,
+                                    context: MappedDiagnosticContext,
+                                    predicate: (T) -> Either<() -> String, () -> String>) =
         flatMap { t ->
             predicate(t).fold({
-                logger.warn(it)
+                logger.warn(context, it)
                 Mono.empty<T>()
             }, {
-                logger.trace(it)
+                logger.trace(context, it)
                 Mono.just<T>(t)
             })
         }
diff --git a/sources/hv-collector-utils/src/test/kotlin/org/onap/dcae/collectors/veshv/utils/logging/LoggerTest.kt b/sources/hv-collector-utils/src/test/kotlin/org/onap/dcae/collectors/veshv/utils/logging/LoggerTest.kt
index c27fb8c8..10fc8d8f 100644
--- a/sources/hv-collector-utils/src/test/kotlin/org/onap/dcae/collectors/veshv/utils/logging/LoggerTest.kt
+++ b/sources/hv-collector-utils/src/test/kotlin/org/onap/dcae/collectors/veshv/utils/logging/LoggerTest.kt
@@ -34,11 +34,16 @@ import org.jetbrains.spek.api.dsl.it
 object LoggerTest : Spek({
 
     lateinit var slf4jLogger: org.slf4j.Logger
-    lateinit var cut: Logger
+    fun cut() = Logger(slf4jLogger).also {
+        verify(slf4jLogger).isTraceEnabled
+        verify(slf4jLogger).isDebugEnabled
+        verify(slf4jLogger).isInfoEnabled
+        verify(slf4jLogger).isWarnEnabled
+        verify(slf4jLogger).isErrorEnabled
+    }
 
     beforeEachTest {
         slf4jLogger = mock()
-        cut = Logger(slf4jLogger)
     }
 
     afterEachTest {
@@ -50,28 +55,19 @@ object LoggerTest : Spek({
         val exception = Exception("fail")
 
         describe("debug levels") {
-            it("should log message") {
-                cut.debug(message)
-                verify(slf4jLogger).debug(message)
-            }
-
-            it("should log message with exception") {
-                cut.debug(message, exception)
-                verify(slf4jLogger).debug(message, exception)
-            }
 
             describe("lazy logging message") {
 
                 it("should log when debug is ON") {
                     whenever(slf4jLogger.isDebugEnabled).thenReturn(true)
-                    cut.debug { message }
+                    cut().debug { message }
                     verify(slf4jLogger).isDebugEnabled
                     verify(slf4jLogger).debug(message)
                 }
 
                 it("should not log when debug is OFF") {
                     whenever(slf4jLogger.isDebugEnabled).thenReturn(false)
-                    cut.debug { message }
+                    cut().debug { message }
                     verify(slf4jLogger).isDebugEnabled
                 }
             }
@@ -80,42 +76,33 @@ object LoggerTest : Spek({
 
                 it("should log when debug is ON") {
                     whenever(slf4jLogger.isDebugEnabled).thenReturn(true)
-                    cut.debug(exception) { message }
+                    cut().withDebug { log(message, exception) }
                     verify(slf4jLogger).isDebugEnabled
                     verify(slf4jLogger).debug(message, exception)
                 }
 
                 it("should not log when debug is OFF") {
                     whenever(slf4jLogger.isDebugEnabled).thenReturn(false)
-                    cut.debug(exception) { message }
+                    cut().withDebug { log(message, exception) }
                     verify(slf4jLogger).isDebugEnabled
                 }
             }
         }
 
         describe("info levels") {
-            it("should log message") {
-                cut.info(message)
-                verify(slf4jLogger).info(message)
-            }
-
-            it("should log message with exception") {
-                cut.info(message, exception)
-                verify(slf4jLogger).info(message, exception)
-            }
 
             describe("lazy logging message") {
 
                 it("should log when debug is ON") {
                     whenever(slf4jLogger.isInfoEnabled).thenReturn(true)
-                    cut.info { message }
+                    cut().info { message }
                     verify(slf4jLogger).isInfoEnabled
                     verify(slf4jLogger).info(message)
                 }
 
                 it("should not log when debug is OFF") {
                     whenever(slf4jLogger.isInfoEnabled).thenReturn(false)
-                    cut.info { message }
+                    cut().info { message }
                     verify(slf4jLogger).isInfoEnabled
                 }
             }
@@ -124,42 +111,32 @@ object LoggerTest : Spek({
 
                 it("should log when debug is ON") {
                     whenever(slf4jLogger.isInfoEnabled).thenReturn(true)
-                    cut.info(exception) { message }
+                    cut().withInfo { log(message, exception) }
                     verify(slf4jLogger).isInfoEnabled
                     verify(slf4jLogger).info(message, exception)
                 }
 
                 it("should not log when debug is OFF") {
                     whenever(slf4jLogger.isInfoEnabled).thenReturn(false)
-                    cut.info(exception) { message }
+                    cut().withInfo { log(message, exception) }
                     verify(slf4jLogger).isInfoEnabled
                 }
             }
         }
 
         describe("warning levels") {
-            it("should log message") {
-                cut.warn(message)
-                verify(slf4jLogger).warn(message)
-            }
-
-            it("should log message with exception") {
-                cut.warn(message, exception)
-                verify(slf4jLogger).warn(message, exception)
-            }
-
             describe("lazy logging message") {
 
                 it("should log when debug is ON") {
                     whenever(slf4jLogger.isWarnEnabled).thenReturn(true)
-                    cut.warn { message }
+                    cut().warn { message }
                     verify(slf4jLogger).isWarnEnabled
                     verify(slf4jLogger).warn(message)
                 }
 
                 it("should not log when debug is OFF") {
                     whenever(slf4jLogger.isWarnEnabled).thenReturn(false)
-                    cut.warn { message }
+                    cut().warn { message }
                     verify(slf4jLogger).isWarnEnabled
                 }
             }
@@ -168,42 +145,33 @@ object LoggerTest : Spek({
 
                 it("should log when debug is ON") {
                     whenever(slf4jLogger.isWarnEnabled).thenReturn(true)
-                    cut.warn(exception) { message }
+                    cut().withWarn { log(message, exception) }
                     verify(slf4jLogger).isWarnEnabled
                     verify(slf4jLogger).warn(message, exception)
                 }
 
                 it("should not log when debug is OFF") {
                     whenever(slf4jLogger.isWarnEnabled).thenReturn(false)
-                    cut.warn(exception) { message }
+                    cut().withWarn { log(message, exception) }
                     verify(slf4jLogger).isWarnEnabled
                 }
             }
         }
 
         describe("error levels") {
-            it("should log message") {
-                cut.error(message)
-                verify(slf4jLogger).error(message)
-            }
-
-            it("should log message with exception") {
-                cut.error(message, exception)
-                verify(slf4jLogger).error(message, exception)
-            }
 
             describe("lazy logging message") {
 
                 it("should log when debug is ON") {
                     whenever(slf4jLogger.isErrorEnabled).thenReturn(true)
-                    cut.error { message }
+                    cut().error { message }
                     verify(slf4jLogger).isErrorEnabled
                     verify(slf4jLogger).error(message)
                 }
 
                 it("should not log when debug is OFF") {
                     whenever(slf4jLogger.isErrorEnabled).thenReturn(false)
-                    cut.error { message }
+                    cut().error { message }
                     verify(slf4jLogger).isErrorEnabled
                 }
             }
@@ -212,14 +180,14 @@ object LoggerTest : Spek({
 
                 it("should log when debug is ON") {
                     whenever(slf4jLogger.isErrorEnabled).thenReturn(true)
-                    cut.error(exception) { message }
+                    cut().withError { log(message, exception) }
                     verify(slf4jLogger).isErrorEnabled
                     verify(slf4jLogger).error(message, exception)
                 }
 
                 it("should not log when debug is OFF") {
                     whenever(slf4jLogger.isErrorEnabled).thenReturn(false)
-                    cut.error(exception) { message }
+                    cut().withError { log(message, exception) }
                     verify(slf4jLogger).isErrorEnabled
                 }
             }
diff --git a/sources/hv-collector-utils/src/test/kotlin/org/onap/dcae/collectors/veshv/utils/logging/ReactiveLoggingTest.kt b/sources/hv-collector-utils/src/test/kotlin/org/onap/dcae/collectors/veshv/utils/logging/ReactiveLoggingTest.kt
index 0f359df3..da956bec 100644
--- a/sources/hv-collector-utils/src/test/kotlin/org/onap/dcae/collectors/veshv/utils/logging/ReactiveLoggingTest.kt
+++ b/sources/hv-collector-utils/src/test/kotlin/org/onap/dcae/collectors/veshv/utils/logging/ReactiveLoggingTest.kt
@@ -42,7 +42,7 @@ class ReactiveLoggingTest : Spek({
                 val cut = Try.just(event)
 
                 it("should not filter stream event and log accepted message") {
-                    cut.filterFailedWithLog(logger, ACCEPTED_MESSAGE, FAILED_WITH_EXCEPTION_MESSAGE)
+                    cut.filterFailedWithLog(logger, ::emptyMap, ACCEPTED_MESSAGE, FAILED_WITH_EXCEPTION_MESSAGE)
                             .test()
                             .expectNext(event)
                             .verifyComplete()
@@ -53,7 +53,7 @@ class ReactiveLoggingTest : Spek({
                 val e = Exception()
                 val cut = Failure(e)
                 it("should filter stream event and log rejected message") {
-                    cut.filterFailedWithLog(logger, ACCEPTED_MESSAGE, FAILED_WITH_EXCEPTION_MESSAGE)
+                    cut.filterFailedWithLog(logger, ::emptyMap, ACCEPTED_MESSAGE, FAILED_WITH_EXCEPTION_MESSAGE)
                             .test()
                             .verifyComplete()
                 }
@@ -65,7 +65,7 @@ class ReactiveLoggingTest : Spek({
                 val cut = Option.just(event)
 
                 it("should not filter stream event and log accepted message") {
-                    cut.filterEmptyWithLog(logger, ACCEPTED_MESSAGE, FAILED_MESSAGE)
+                    cut.filterEmptyWithLog(logger, ::emptyMap, ACCEPTED_MESSAGE, FAILED_MESSAGE)
                             .test()
                             .expectNext(event)
                             .verifyComplete()
@@ -75,7 +75,7 @@ class ReactiveLoggingTest : Spek({
             given("empty Option") {
                 val cut = Option.empty<Int>()
                 it("should filter stream event and log rejected message") {
-                    cut.filterEmptyWithLog(logger, ACCEPTED_MESSAGE, FAILED_MESSAGE)
+                    cut.filterEmptyWithLog(logger,::emptyMap, ACCEPTED_MESSAGE, FAILED_MESSAGE)
                             .test()
                             .verifyComplete()
                 }
@@ -88,7 +88,7 @@ class ReactiveLoggingTest : Spek({
                 val cut = Flux.just(event)
 
                 it("should not filter stream event and log accepted message") {
-                    cut.filterFailedWithLog(logger, right())
+                    cut.filterFailedWithLog(logger,::emptyMap, right())
                             .test()
                             .expectNext(event)
                             .verifyComplete()
@@ -99,7 +99,7 @@ class ReactiveLoggingTest : Spek({
                 val cut = Flux.just(event)
 
                 it("should filter stream event and log rejected message") {
-                    cut.filterFailedWithLog(logger, left())
+                    cut.filterFailedWithLog(logger,::emptyMap, left())
                             .test()
                             .verifyComplete()
                 }