diff --git a/README.md b/README.md index 0863230ffa..1cec985add 100644 --- a/README.md +++ b/README.md @@ -304,9 +304,12 @@ Note: Your Java environment may need to be set up correctly to use SSL. Restart OBP-API, if you get an error, check your Java environment can connect to the host over SSL. -Note: You can copy the following example files to prepare your own configurations: +Note: `/obp-api/src/main/resources/logback.xml` ships ready to use, defaulting to INFO. Set the +`LOG_LEVEL` environment variable (e.g. `LOG_LEVEL=DEBUG` or `LOG_LEVEL=TRACE`) to override it per +environment without editing the file. + +You can also copy the following example file to prepare your own test-run configuration: -- `/obp-api/src/main/resources/logback.xml.example` -> `/obp-api/src/main/resources/logback.xml` (try TRACE or DEBUG). - `/obp-api/src/main/resources/logback-test.xml.example` -> `/obp-api/src/main/resources/logback-test.xml` (try TRACE or DEBUG). There is a gist/tool which is useful for this. Search the web for SSLPoke. Note this is an external repository. diff --git a/obp-api/src/main/resources/docs/brief_system_documentation.md b/obp-api/src/main/resources/docs/brief_system_documentation.md index 86d568b7cd..e93a6ca464 100644 --- a/obp-api/src/main/resources/docs/brief_system_documentation.md +++ b/obp-api/src/main/resources/docs/brief_system_documentation.md @@ -137,7 +137,7 @@ User ──(has roles/entitlements)──► Bank/System actions **Logging** -- Copy `logback.xml.example` to `logback.xml`; adjust levels (TRACE/DEBUG/INFO) per environment. +- `logback.xml` ships ready to use, defaulting to INFO; set `LOG_LEVEL` (TRACE/DEBUG/INFO/...) to override per environment. - In Docker/K8s, logs go to stdout/stderr → aggregate with your stack (e.g., Loki/Promtail, EFK). **Health & metrics** diff --git a/obp-api/src/main/resources/logback.xml b/obp-api/src/main/resources/logback.xml new file mode 100644 index 0000000000..e7256d79ac --- /dev/null +++ b/obp-api/src/main/resources/logback.xml @@ -0,0 +1,19 @@ + + + + + %d{yyyy-MM-dd HH:mm:ss} %t %X{callerThread:-} %c{0} [%p] %m%n + + + + + + + + + \ No newline at end of file diff --git a/obp-api/src/main/resources/logback.xml.example b/obp-api/src/main/resources/logback.xml.example deleted file mode 100644 index 871aea062a..0000000000 --- a/obp-api/src/main/resources/logback.xml.example +++ /dev/null @@ -1,12 +0,0 @@ - - - - - %d{yyyy-MM-dd HH:mm:ss} %t %c{0} [%p] %m%n - - - - - - - \ No newline at end of file diff --git a/obp-api/src/main/resources/props/sample.props.template b/obp-api/src/main/resources/props/sample.props.template index 0882711cb1..93a4e5a2d4 100644 --- a/obp-api/src/main/resources/props/sample.props.template +++ b/obp-api/src/main/resources/props/sample.props.template @@ -1948,6 +1948,13 @@ ethereum.rpc.url=http://127.0.0.1:8545 ## Enable Redis logging (true/false) redis_logging_enabled = false +## Lowest level shipped to Redis when redis_logging_enabled is true (TRACE/DEBUG/INFO/WARNING/ERROR). +## Defaults to INFO so enabling Redis log shipping doesn't also force every DEBUG/TRACE call site +## (which can be extremely high-volume) to pay the cost of formatting and masking a message that +## would otherwise be skipped locally. Set to DEBUG or TRACE only when you actually want that volume +## shipped, e.g. temporarily while investigating an issue. +redis_logging_min_level = INFO + ## Batch size for sending logs to Redis ## Smaller batch size reduces latency for logging critical messages. redis_logging_batch_size = 50 @@ -1978,6 +1985,27 @@ redis_logging_warning_queue_max_entries = 1000 # Max WARNING messages redis_logging_error_queue_max_entries = 1000 # Max ERROR messages redis_logging_all_queue_max_entries = 1000 # Max ALL messages + + +########################################################## +# MDC Logging Dispatch # +########################################################## +## Number of threads used to run secret-masking (SecureLogging.maskSensitive) and the local/ +## Redis log write itself, once a call to logger.debug/info/warn/error/trace has determined +## the message will actually be consumed. This work used to run inline on whatever thread +## called the logger, including http4s/cats-effect fiber-managed threads -- CPU-bound work +## that doesn't yield can trip the runtime's own fiber-starvation detector the same way real +## blocking I/O would, and its response (spinning up more worker/blocker threads that are +## never reclaimed) only adds overhead. Keep this small; it's meant to decouple from the +## calling thread, not to parallelize heavy logging. +mdc_logging_dispatch_thread_pool_size = 2 + +## Maximum number of log entries that may wait for a dispatch thread. The queue is bounded so a +## burst of log calls cannot become heap growth. When it is full, DEBUG/TRACE/INFO entries are +## dropped (and counted) while WARN/ERROR entries run on the calling thread, so a warning or an +## error is never silently lost. Default 10000. +mdc_logging_dispatch_queue_size = 10000 + ## Optional: Circuit breaker reset interval (ms) ## How long before retrying after circuit breaker opens. Default 60s. redis_logging_circuit_breaker_reset_ms = 60000 diff --git a/obp-api/src/main/scala/bootstrap/liftweb/Boot.scala b/obp-api/src/main/scala/bootstrap/liftweb/Boot.scala index 94e6049506..fa832b93f4 100644 --- a/obp-api/src/main/scala/bootstrap/liftweb/Boot.scala +++ b/obp-api/src/main/scala/bootstrap/liftweb/Boot.scala @@ -234,6 +234,18 @@ class Boot extends MdcLoggable { } if (Props.mode == Props.RunModes.Development) logger.info("OBP-API Props all fields : \n" + Props.props.mkString("\n")) + // Make the effective root log level visible at start-up. DEBUG/TRACE is expensive on this + // code base and is only meant to be enabled deliberately (LOG_LEVEL), so say so loudly. + try { + org.slf4j.LoggerFactory.getLogger(org.slf4j.Logger.ROOT_LOGGER_NAME) match { + case l: ch.qos.logback.classic.Logger => + val level = l.getEffectiveLevel + if (level.levelInt <= ch.qos.logback.classic.Level.DEBUG_INT) + logger.warn(s"Effective root log level is $level (LOG_LEVEL override); expect higher CPU and log volume") + else logger.info(s"Effective root log level is $level") + case _ => () + } + } catch { case _: Throwable => () } logger.info("external props folder: " + propsPath) TimeZone.setDefault(TimeZone.getTimeZone("UTC")) logger.info("Current Project TimeZone: " + TimeZone.getDefault) diff --git a/obp-api/src/main/scala/code/api/cache/RedisLogger.scala b/obp-api/src/main/scala/code/api/cache/RedisLogger.scala index 7c22956d5a..723cc58c5c 100644 --- a/obp-api/src/main/scala/code/api/cache/RedisLogger.scala +++ b/obp-api/src/main/scala/code/api/cache/RedisLogger.scala @@ -60,6 +60,48 @@ object RedisLogger { // Performance and reliability improvements private val redisLoggingEnabled = APIUtil.getPropsAsBoolValue("redis_logging_enabled", false) + + /** + * Whether Redis log shipping is turned on for this instance. Callers that want to skip + * building a log message when neither the local logger nor Redis will consume it (e.g. + * MdcLoggable) should check this alongside the local logger's isXEnabled. + */ + def isEnabled: Boolean = redisLoggingEnabled + + /** + * Floor below which levels are not shipped to Redis even when redis_logging_enabled is + * true. Defaults to INFO so that turning Redis log shipping on for a deployment doesn't + * silently force every DEBUG/TRACE call site (which can be extremely high-volume) to pay + * the cost of formatting and masking a message it would otherwise skip. Set to TRACE to + * restore the old "ship everything" behaviour. + */ + private val redisLoggingMinLevel: LogLevel.LogLevel = + parseMinLevel(APIUtil.getPropsValue("redis_logging_min_level", "INFO")) + + /** + * Parse the configured floor. An unrecognised value must not throw: this runs while the + * RedisLogger object is being initialised, and MdcLoggable consults RedisLogger on every log + * call, so a throwing initialiser would turn a config typo into failures at ordinary logging + * statements across the application. Fall back to INFO and say so on stderr (not through a + * logger, which would re-enter this initialiser). + */ + private[cache] def parseMinLevel(configured: String): LogLevel.LogLevel = + try LogLevel.valueOf(configured.trim) + catch { + case _: IllegalArgumentException => + System.err.println(s"Invalid redis_logging_min_level '$configured'; using INFO. Valid values: TRACE, DEBUG, INFO, WARN, WARNING, ERROR, ALL") + LogLevel.INFO + } + + /** + * Whether a message at this level should be shipped to Redis right now. Combines the + * on/off switch with the minimum-level floor above. Callers that only want to know "is + * Redis shipping on at all" (e.g. to decide whether to enable a config UI) should keep + * using `isEnabled`; callers deciding whether to build+ship a specific message should use + * this instead. + */ + def shouldShip(level: LogLevel.LogLevel): Boolean = + redisLoggingEnabled && level != LogLevel.ALL && level.id >= redisLoggingMinLevel.id private val batchSize = APIUtil.getPropsAsIntValue("redis_logging_batch_size", 100) private val flushIntervalMs = APIUtil.getPropsAsIntValue("redis_logging_flush_interval_ms", 1000) private val maxRetries = APIUtil.getPropsAsIntValue("redis_logging_max_retries", 3) @@ -175,7 +217,7 @@ object RedisLogger { * Returns a Future[Unit], failures are handled gracefully. */ def logAsync(level: LogLevel.LogLevel, message: String): Future[Unit] = { - if (!redisLoggingEnabled) { + if (!shouldShip(level)) { return Future.successful(()) } diff --git a/obp-api/src/main/scala/code/api/util/JsonSchemaGenerator.scala b/obp-api/src/main/scala/code/api/util/JsonSchemaGenerator.scala index 55d41bcb76..043a6eee50 100644 --- a/obp-api/src/main/scala/code/api/util/JsonSchemaGenerator.scala +++ b/obp-api/src/main/scala/code/api/util/JsonSchemaGenerator.scala @@ -30,8 +30,10 @@ package code.api.util import org.json4s._ import code.api.util.APIUtil.MessageDoc import com.openbankproject.commons.util.ReflectUtils +import com.tesobe.CacheKeyFromArguments import org.json4s.JsonDSL._ +import scala.concurrent.duration._ import scala.reflect.runtime.universe._ /** @@ -42,9 +44,34 @@ import scala.reflect.runtime.universe._ object JsonSchemaGenerator { /** - * Convert a list of MessageDoc to a complete JSON Schema document + * Convert a list of MessageDoc to a complete JSON Schema document. + * + * Memoized in-process, keyed by connectorName: for a given connector the message docs + * (and therefore the schema) are static for the lifetime of the JVM, but building it + * walks every message type's full field tree via Scala runtime reflection (`<:<`/`=:=` + * subtype checks), which is expensive and -- unlike a plain field lookup -- leaves behind + * long-lived reflection bookkeeping objects (TypeConstraint/UndoPair/Symbol) that don't + * get reclaimed promptly. Recomputing this on every request under sustained polling keeps + * adding them and grows old-gen heap usage. The caller (Http4s600) also has a + * Redis-backed cache in front of this, but that one silently falls through to a full + * recompute if Redis is unreachable or slow -- this in-memory layer doesn't depend on + * Redis at all, so it stays a working safety net even when Redis is the one struggling. */ def messageDocsToJsonSchema(messageDocs: List[MessageDoc], connectorName: String): JObject = { + // This 3-tuple of random UUIDs is a placeholder only -- CacheKeyFromArguments is a macro + // that replaces it at compile time with a real key derived from this method's owner, + // name and arguments (regardless of this method's own arity; the convention throughout + // this codebase is always a 3-tuple here). See: + // https://github.com/OpenBankProject/scala-macros/blob/master/macros/src/main/scala/com/tesobe/CacheKeyFromArgumentsMacro.scala#L49 + var cacheKey = (java.util.UUID.randomUUID().toString, java.util.UUID.randomUUID().toString, java.util.UUID.randomUUID().toString) + CacheKeyFromArguments.buildCacheKey { + code.api.cache.Caching.memoizeSyncWithImMemory(Some(cacheKey.toString()))(100000.days) { + messageDocsToJsonSchemaUncached(messageDocs, connectorName) + } + } + } + + private def messageDocsToJsonSchemaUncached(messageDocs: List[MessageDoc], connectorName: String): JObject = { val allDefinitions = scala.collection.mutable.Map[String, JObject]() val messages = messageDocs.map { messageDoc => diff --git a/obp-api/src/main/scala/code/api/v2_2_0/Http4s220.scala b/obp-api/src/main/scala/code/api/v2_2_0/Http4s220.scala index f86cb79e12..71dc927940 100644 --- a/obp-api/src/main/scala/code/api/v2_2_0/Http4s220.scala +++ b/obp-api/src/main/scala/code/api/v2_2_0/Http4s220.scala @@ -425,7 +425,9 @@ object Http4s220 { Some(cc), s"$InvalidConnector Current Input is $connector. It should be eg: rest_vMar2019..." ) - JSONFactory220.createMessageDocsJson(connectorObject.messageDocs.toList) + MessageDocsJsonCache.getOrCompute(connector) { + Extraction.decompose(JSONFactory220.createMessageDocsJson(connectorObject.messageDocs.toList)) + } } } } diff --git a/obp-api/src/main/scala/code/api/v2_2_0/MessageDocsJsonCache.scala b/obp-api/src/main/scala/code/api/v2_2_0/MessageDocsJsonCache.scala new file mode 100644 index 0000000000..0ab393ddc3 --- /dev/null +++ b/obp-api/src/main/scala/code/api/v2_2_0/MessageDocsJsonCache.scala @@ -0,0 +1,88 @@ +package code.api.v2_2_0 + +import java.util.concurrent.Callable + +import code.api.cache.Caching +import com.google.common.cache.{Cache, CacheBuilder} +import com.openbankproject.commons.util.JsonAliases.{compactRender, parse} +import net.liftweb.common.Loggable +import org.json4s.JValue + +/** + * Two-level cache for the fully built `GET /message-docs/CONNECTOR` response. + * + * Building the response runs Scala runtime reflection over every message doc's example + * inbound and outbound message, which is expensive in CPU and grows the reflection + * universe. The result only depends on the connector's `messageDocs`, which a connector + * fills once while its singleton is initialised and never changes afterwards. + * + * Levels, checked in this order: + * 1. In-process: a bounded Guava cache holding the immutable JValue. It keeps working when + * Redis is down, so an unreachable Redis can never send every request back into + * reflection. + * 2. Shared: the same Redis-backed store the resource-doc and swagger endpoints use + * (`Caching.getStaticSwaggerDocCache`, same key prefix and TTL, same fail-safe behaviour: + * an unreachable Redis is a miss, never an error). It lets replicas and restarts reuse + * one instance's work. + * + * Contract: + * - Key: the connector name, and only after it has been resolved to a real connector. + * Unknown names fail before reaching the cache, so request input cannot grow it; + * `MaxEntries` is a second, hard bound on the in-process level. + * - Single flight: concurrent cold requests for one connector run the loader once, and + * therefore touch Redis and the generator once. + * - A failure is never cached; the next request retries. + * - Redis is written only after a successful generation. An unparsable Redis value is + * treated as a miss and regenerated. + * - Invalidation: `invalidateAll()` clears the in-process level. The Redis level expires by + * `staticResourceDocsObp.cache.ttl.seconds`. Nothing in production mutates a connector's + * message docs after start-up, so nothing calls `invalidateAll()` there. + */ +object MessageDocsJsonCache extends Loggable { + private val MaxEntries = 64L + + /** The shared level. Abstracted so tests can count reads and writes without a Redis. */ + trait SharedStore { + def get(key: String): Option[String] + def set(key: String, value: String): Unit + } + + object RedisStore extends SharedStore { + def get(key: String): Option[String] = Caching.getStaticSwaggerDocCache(key) + def set(key: String, value: String): Unit = Caching.setStaticSwaggerDocCache(key, value) + } + + private def sharedKey(connectorName: String) = s"message-docs-v2.2.0-$connectorName" + + private val cache: Cache[String, JValue] = + CacheBuilder.newBuilder().maximumSize(MaxEntries).build[String, JValue]() + + def getOrCompute(connectorName: String, store: SharedStore = RedisStore)(generate: => JValue): JValue = + try cache.get(connectorName, new Callable[JValue] { + def call(): JValue = { + val key = sharedKey(connectorName) + val fromShared = store.get(key).flatMap { s => + try Some(parse(s)) + catch { case e: Exception => logger.warn(s"Ignoring unparsable shared message-docs entry $key: ${e.getMessage}"); None } + } + fromShared.getOrElse { + // Serve the round-tripped form even on the instance that generated it. Otherwise this + // instance would return the JValue it built while every other replica (and this one + // after a restart) returns parse(compactRender(...)), and number formatting could differ + // between them. + val rendered = compactRender(generate) + store.set(key, rendered) + parse(rendered) + } + } + }) + catch { + // Surface the loader's own exception, not Guava's wrapper. + case e: java.util.concurrent.ExecutionException if e.getCause != null => throw e.getCause + case e: com.google.common.util.concurrent.UncheckedExecutionException if e.getCause != null => throw e.getCause + } + + def invalidateAll(): Unit = cache.invalidateAll() + + def size: Long = cache.size() +} diff --git a/obp-api/src/main/scala/code/util/Helper.scala b/obp-api/src/main/scala/code/util/Helper.scala index b45b10414d..0d23280b29 100644 --- a/obp-api/src/main/scala/code/util/Helper.scala +++ b/obp-api/src/main/scala/code/util/Helper.scala @@ -286,6 +286,133 @@ object Helper extends Loggable { + // Shared across every class mixing in MdcLoggable below -- created once at object-init, + // not per mixing instance. Masking (SecureLogging.maskSensitive, ~19 regex passes) and + // Redis-shipping serialization are CPU-bound work that used to run inline on whatever + // thread called logger.debug/info/etc. On this codebase's http4s/cats-effect request path + // that thread is often a fiber-managed worker; running non-yielding CPU work on it directly + // is indistinguishable, from the runtime's own fairness/starvation detector, from genuine + // blocking I/O -- cats-effect's "your CPU is probably starving" warning fires the same way + // either way, and its response is to compensate by spinning up additional worker/blocker + // threads that are never reclaimed. Dispatching this work onto a small dedicated pool + // instead means the calling thread (fiber or otherwise) returns immediately, regardless of + // what kind of thread it happens to be -- this trait is mixed into ~260 classes, many of + // them called from contexts (actors, scheduled jobs, Lift-era code) that have no IO runtime + // at all, so wrapping in cats.effect.IO.blocking isn't an option here: nothing would ever + // run it in those contexts. A plain background ExecutionContext works everywhere the + // trait itself is used. Same pattern RedisLogger already uses for its own async shipping. + // + // Trade-off worth being explicit about: log output for a given logger is no longer + // strictly write-ordered relative to other concurrent callers (each dispatched entry lands + // whenever its turn on this small pool comes up). That's the same trade-off any async + // logging setup makes (Logback's own AsyncAppender, Log4j2's AsyncLogger); this was never a + // strict global ordering guarantee to begin with once Redis shipping (already async) was in + // the picture. + // lazy, not val: Helper's own static initializer transitively touches other objects + // (APIUtil/Constant among them) that log during THEIR initialization, which can re-enter + // here before a plain val declared at this point in the object body would have run yet -- + // observed as a NullPointerException on this executor during Helper's own , caught + // by MdcLoggableDispatchTest. `lazy val` computes on first real use instead of at a fixed + // point in top-to-bottom initialization order, which is what this needs given how + // entangled this codebase's early object initialization already is (not something + // introduced here). + // Bounded on purpose. An unbounded queue in front of this pool would turn a burst of log + // calls -- each one holding the message closure and everything it captured -- into heap + // growth, i.e. the failure this pool exists to avoid. Failure mode when the queue is full, + // documented here because it is a deliberate choice: + // * DEBUG/TRACE/INFO entries are dropped and counted (`mdcLogDroppedCount`); + // * WARN/ERROR entries are run inline on the calling thread, so a warning or error is + // never silently lost. That only happens while the pool is saturated, which bounds the + // extra work on request threads to the overload window itself. + // A drop is reported on stderr for the first occurrence and then once per + // `MdcLogDropReportEvery`, so a sustained overload cannot turn into a stderr flood either. + private val MdcLogDropReportEvery = 10000L + private val mdcLogDropped = new java.util.concurrent.atomic.AtomicLong(0) + + /** + * ThreadPoolExecutor and ArrayBlockingQueue reject a size below 1 with IllegalArgumentException. + * The pool is created lazily on the first log call, so a 0 or negative prop would fail that call + * and every later one. Clamp both values to at least 1 instead. + */ + private[util] def dispatchPoolSettings(configuredPoolSize: Int, configuredQueueSize: Int): (Int, Int) = + (math.max(1, configuredPoolSize), math.max(1, configuredQueueSize)) + + private lazy val mdcLoggingExecutor: java.util.concurrent.ThreadPoolExecutor = { + val threadCount = new java.util.concurrent.atomic.AtomicInteger(0) + val (poolSize, queueSize) = dispatchPoolSettings( + APIUtil.getPropsAsIntValue("mdc_logging_dispatch_thread_pool_size", 2), + APIUtil.getPropsAsIntValue("mdc_logging_dispatch_queue_size", 10000)) + val executor = new java.util.concurrent.ThreadPoolExecutor( + poolSize, poolSize, 0L, java.util.concurrent.TimeUnit.MILLISECONDS, + new java.util.concurrent.ArrayBlockingQueue[Runnable](queueSize), + (r: Runnable) => { + val t = new Thread(r, s"mdc-log-dispatch-${threadCount.incrementAndGet()}") + t.setDaemon(true) + t + }, + new java.util.concurrent.ThreadPoolExecutor.AbortPolicy() + ) + // The threads are daemons so they never hold the JVM open, which also means anything + // still queued at exit would be lost. Give the queue a short, bounded chance to drain. + // Registration is refused once the JVM is already shutting down. That must not make the + // first log call of a dying process throw, so the pool simply runs without a hook then. + try Runtime.getRuntime.addShutdownHook(new Thread(() => { + executor.shutdown() + try executor.awaitTermination(2, java.util.concurrent.TimeUnit.SECONDS) + catch { case _: InterruptedException => () } + }, "mdc-log-dispatch-shutdown")) + catch { case _: IllegalStateException => () } + executor + } + + /** Entries dropped because the dispatch queue was full since start-up. */ + def mdcLogDroppedCount: Long = mdcLogDropped.get() + + /** Entries currently waiting for a dispatch thread. */ + def mdcLogQueueDepth: Int = mdcLoggingExecutor.getQueue.size() + + /** + * Run `body` on the dispatch pool. When the queue is full, `critical` work runs inline on the + * caller and anything else is dropped and counted. `body` never throws to the caller. + */ + private[util] def dispatchLog(clazzName: String, critical: Boolean)(body: => Unit): Unit = + dispatchOn(mdcLoggingExecutor, clazzName, critical)(body) + + /** MDC key carrying the name of the thread that logged, set while a dispatched entry is written. */ + val MdcCallerThreadKey = "callerThread" + + // The executor is a parameter so a test can drive a tiny queue to saturation. + private[util] def dispatchOn(executor: java.util.concurrent.Executor, clazzName: String, critical: Boolean)(body: => Unit): Unit = { + // The write happens on a pool thread, so the thread name Logback and the Redis line format + // would report is always "mdc-log-dispatch-N". Carry the caller's name in the MDC instead of + // renaming the pool thread: it costs no native calls, and thread dumps still show what each + // pool thread really is. The default logback.xml pattern prints it after %t. + val callerThreadName = Thread.currentThread().getName + val task: Runnable = () => { + val previous = org.slf4j.MDC.get(MdcCallerThreadKey) + org.slf4j.MDC.put(MdcCallerThreadKey, callerThreadName) + try body + catch { case e: Throwable => System.err.println(s"[$clazzName] background log dispatch failed: ${e.getMessage}") } + finally { + if (previous == null) org.slf4j.MDC.remove(MdcCallerThreadKey) else org.slf4j.MDC.put(MdcCallerThreadKey, previous) + } + } + try executor.execute(task) + catch { + case _: java.util.concurrent.RejectedExecutionException => + executor match { + // The pool has been shut down (JVM exit): nothing is overloaded, so write the entry + // on the caller instead of losing what other shutdown hooks log. + case s: java.util.concurrent.ExecutorService if s.isShutdown => task.run() + case _ if critical => task.run() + case _ => + val dropped = mdcLogDropped.incrementAndGet() + if (dropped == 1L || dropped % MdcLogDropReportEvery == 0L) + System.err.println(s"[$clazzName] log dispatch queue is full; $dropped non-critical log entries dropped so far") + } + } + } + trait MdcLoggable extends Loggable { // Capture the class name of the component mixing in this trait @@ -307,67 +434,113 @@ object Helper extends Loggable { private def toRedisFormat(msg: AnyRef): String = { val ts = dateFormatTL.get().format(new Date()) - val thread = Thread.currentThread().getName + val thread = Option(org.slf4j.MDC.get(MdcCallerThreadKey)).getOrElse(Thread.currentThread().getName) s"[$ts] [$thread] [$clazzName] ${msg.toString}" } + // Every level below builds `maskedMsg` by running the message through ~19 regex + // passes in SecureLogging.maskSensitive. That cost must only be paid when the + // result is actually going to be consumed (by the local logger and/or Redis + // shipping) -- not unconditionally on every call site, which is what made this a + // hot path under high request volume with DEBUG enabled. And once it is going to be + // paid, it's dispatched onto mdcLoggingExecutor (see the failure-mode note above it) + // rather than run inline on whatever thread called into this logger. + private def dispatch(critical: Boolean)(body: => Unit): Unit = dispatchLog(clazzName, critical)(body) + // INFO override def info(msg: => AnyRef): Unit = { - val maskedMsg = SecureLogging.maskSensitive(msg) - underlyingLogger.info(maskedMsg) - RedisLogger.logAsync(RedisLogger.LogLevel.INFO, toRedisFormat(maskedMsg)) + if (underlyingLogger.isInfoEnabled || RedisLogger.shouldShip(RedisLogger.LogLevel.INFO)) { + dispatch(critical = false) { + val maskedMsg = SecureLogging.maskSensitive(msg) + if (underlyingLogger.isInfoEnabled) underlyingLogger.info(maskedMsg) + RedisLogger.logAsync(RedisLogger.LogLevel.INFO, toRedisFormat(maskedMsg)) + } + } } override def info(msg: => AnyRef, t: => Throwable): Unit = { - val maskedMsg = SecureLogging.maskSensitive(msg) - underlyingLogger.info(maskedMsg, t) - RedisLogger.logAsync(RedisLogger.LogLevel.INFO, toRedisFormat(maskedMsg) + "\n" + t.toString) + if (underlyingLogger.isInfoEnabled || RedisLogger.shouldShip(RedisLogger.LogLevel.INFO)) { + dispatch(critical = false) { + val maskedMsg = SecureLogging.maskSensitive(msg) + val capturedT = t + if (underlyingLogger.isInfoEnabled) underlyingLogger.info(maskedMsg, capturedT) + RedisLogger.logAsync(RedisLogger.LogLevel.INFO, toRedisFormat(maskedMsg) + "\n" + capturedT.toString) + } + } } // WARN override def warn(msg: => AnyRef): Unit = { - val maskedMsg = SecureLogging.maskSensitive(msg) - underlyingLogger.warn(maskedMsg) - RedisLogger.logAsync(RedisLogger.LogLevel.WARNING, toRedisFormat(maskedMsg)) + if (underlyingLogger.isWarnEnabled || RedisLogger.shouldShip(RedisLogger.LogLevel.WARNING)) { + dispatch(critical = true) { + val maskedMsg = SecureLogging.maskSensitive(msg) + if (underlyingLogger.isWarnEnabled) underlyingLogger.warn(maskedMsg) + RedisLogger.logAsync(RedisLogger.LogLevel.WARNING, toRedisFormat(maskedMsg)) + } + } } override def warn(msg: => AnyRef, t: Throwable): Unit = { - val maskedMsg = SecureLogging.maskSensitive(msg) - underlyingLogger.warn(maskedMsg, t) - RedisLogger.logAsync(RedisLogger.LogLevel.WARNING, toRedisFormat(maskedMsg) + "\n" + t.toString) + if (underlyingLogger.isWarnEnabled || RedisLogger.shouldShip(RedisLogger.LogLevel.WARNING)) { + dispatch(critical = true) { + val maskedMsg = SecureLogging.maskSensitive(msg) + if (underlyingLogger.isWarnEnabled) underlyingLogger.warn(maskedMsg, t) + RedisLogger.logAsync(RedisLogger.LogLevel.WARNING, toRedisFormat(maskedMsg) + "\n" + t.toString) + } + } } // ERROR override def error(msg: => AnyRef): Unit = { - val maskedMsg = SecureLogging.maskSensitive(msg) - underlyingLogger.error(maskedMsg) - RedisLogger.logAsync(RedisLogger.LogLevel.ERROR, toRedisFormat(maskedMsg)) + if (underlyingLogger.isErrorEnabled || RedisLogger.shouldShip(RedisLogger.LogLevel.ERROR)) { + dispatch(critical = true) { + val maskedMsg = SecureLogging.maskSensitive(msg) + if (underlyingLogger.isErrorEnabled) underlyingLogger.error(maskedMsg) + RedisLogger.logAsync(RedisLogger.LogLevel.ERROR, toRedisFormat(maskedMsg)) + } + } } override def error(msg: => AnyRef, t: Throwable): Unit = { - val maskedMsg = SecureLogging.maskSensitive(msg) - underlyingLogger.error(maskedMsg, t) - RedisLogger.logAsync(RedisLogger.LogLevel.ERROR, toRedisFormat(maskedMsg) + "\n" + t.toString) + if (underlyingLogger.isErrorEnabled || RedisLogger.shouldShip(RedisLogger.LogLevel.ERROR)) { + dispatch(critical = true) { + val maskedMsg = SecureLogging.maskSensitive(msg) + if (underlyingLogger.isErrorEnabled) underlyingLogger.error(maskedMsg, t) + RedisLogger.logAsync(RedisLogger.LogLevel.ERROR, toRedisFormat(maskedMsg) + "\n" + t.toString) + } + } } // DEBUG override def debug(msg: => AnyRef): Unit = { - val maskedMsg = SecureLogging.maskSensitive(msg) - underlyingLogger.debug(maskedMsg) - RedisLogger.logAsync(RedisLogger.LogLevel.DEBUG, toRedisFormat(maskedMsg)) + if (underlyingLogger.isDebugEnabled || RedisLogger.shouldShip(RedisLogger.LogLevel.DEBUG)) { + dispatch(critical = false) { + val maskedMsg = SecureLogging.maskSensitive(msg) + if (underlyingLogger.isDebugEnabled) underlyingLogger.debug(maskedMsg) + RedisLogger.logAsync(RedisLogger.LogLevel.DEBUG, toRedisFormat(maskedMsg)) + } + } } override def debug(msg: => AnyRef, t: Throwable): Unit = { - val maskedMsg = SecureLogging.maskSensitive(msg) - underlyingLogger.debug(maskedMsg, t) - RedisLogger.logAsync(RedisLogger.LogLevel.DEBUG, toRedisFormat(maskedMsg) + "\n" + t.toString) + if (underlyingLogger.isDebugEnabled || RedisLogger.shouldShip(RedisLogger.LogLevel.DEBUG)) { + dispatch(critical = false) { + val maskedMsg = SecureLogging.maskSensitive(msg) + if (underlyingLogger.isDebugEnabled) underlyingLogger.debug(maskedMsg, t) + RedisLogger.logAsync(RedisLogger.LogLevel.DEBUG, toRedisFormat(maskedMsg) + "\n" + t.toString) + } + } } // TRACE override def trace(msg: => AnyRef): Unit = { - val maskedMsg = SecureLogging.maskSensitive(msg) - underlyingLogger.trace(maskedMsg) - RedisLogger.logAsync(RedisLogger.LogLevel.TRACE, toRedisFormat(maskedMsg)) + if (underlyingLogger.isTraceEnabled || RedisLogger.shouldShip(RedisLogger.LogLevel.TRACE)) { + dispatch(critical = false) { + val maskedMsg = SecureLogging.maskSensitive(msg) + if (underlyingLogger.isTraceEnabled) underlyingLogger.trace(maskedMsg) + RedisLogger.logAsync(RedisLogger.LogLevel.TRACE, toRedisFormat(maskedMsg)) + } + } } // Delegate enabled checks diff --git a/obp-api/src/test/scala/bootstrap/http4s/LogbackDefaultLevelTest.scala b/obp-api/src/test/scala/bootstrap/http4s/LogbackDefaultLevelTest.scala new file mode 100644 index 0000000000..527c81bfa5 --- /dev/null +++ b/obp-api/src/test/scala/bootstrap/http4s/LogbackDefaultLevelTest.scala @@ -0,0 +1,55 @@ +package bootstrap.http4s + +import ch.qos.logback.classic.{Level, Logger, LoggerContext} +import ch.qos.logback.classic.joran.JoranConfigurator +import ch.qos.logback.classic.spi.LoggerContextListener +import org.scalatest.{BeforeAndAfterEach, FlatSpec, Matchers} +import org.slf4j.Logger.ROOT_LOGGER_NAME + +/** + * Directly verifies the shipped src/main/resources/logback.xml resolves + * the way this fix depends on: + * + * - LOG_LEVEL unset -> root defaults to INFO, not the old hard-coded DEBUG. + * - LOG_LEVEL set -> root honours it. + * + * The application's actual SLF4J LoggerContext is configured once, at JVM start, from whatever + * LOG_LEVEL happened to be set at that moment -- so asserting against + * LoggerFactory.getILoggerFactory() here wouldn't exercise the default-value resolution at all, + * only whatever this test JVM's own environment happened to be. Instead this parses the actual + * classpath resource into a fresh, throwaway LoggerContext per scenario (via Joran, the same + * parser Logback uses internally), with the system property set immediately beforehand -- this is + * the standard way to unit-test a logback.xml's own logic rather than the ambient environment. + */ +class LogbackDefaultLevelTest extends FlatSpec with Matchers with BeforeAndAfterEach { + + private val propertyName = "LOG_LEVEL" + + override def afterEach(): Unit = { + System.clearProperty(propertyName) + } + + private def rootLevelFromShippedConfig(): Level = { + val context = new LoggerContext() + context.reset() + val configurator = new JoranConfigurator() + configurator.setContext(context) + configurator.doConfigure(getClass.getClassLoader.getResource("logback.xml")) + context.getLogger(ROOT_LOGGER_NAME).getLevel + } + + "the shipped logback.xml" should "default the root level to INFO when LOG_LEVEL is unset" in { + System.clearProperty(propertyName) + rootLevelFromShippedConfig() should equal(Level.INFO) + } + + it should "honour LOG_LEVEL=DEBUG when set" in { + System.setProperty(propertyName, "DEBUG") + rootLevelFromShippedConfig() should equal(Level.DEBUG) + } + + it should "honour LOG_LEVEL=WARN when set" in { + System.setProperty(propertyName, "WARN") + rootLevelFromShippedConfig() should equal(Level.WARN) + } +} diff --git a/obp-api/src/test/scala/code/api/cache/RedisLoggerShouldShipTest.scala b/obp-api/src/test/scala/code/api/cache/RedisLoggerShouldShipTest.scala new file mode 100644 index 0000000000..782af6f960 --- /dev/null +++ b/obp-api/src/test/scala/code/api/cache/RedisLoggerShouldShipTest.scala @@ -0,0 +1,77 @@ +package code.api.cache + +import org.scalatest.{FlatSpec, Matchers} + +/** + * Guards the two invariants MdcLoggable's level gate (code.util.Helper) depends on to skip + * SecureLogging.maskSensitive when nothing will consume the message: + * + * 1. LogLevel's declaration order is severity-ascending (TRACE < DEBUG < INFO < WARNING < ERROR), + * since RedisLogger.shouldShip compares levels by `.id`. Scala Enumeration assigns ids by + * declaration order, not by name, so this is a silent footgun if the `val TRACE, DEBUG, ... = + * Value` line is ever reordered or a new level inserted in the wrong place - nothing about the + * Enumeration itself would complain, shouldShip would just start comparing the wrong things. + * + * 2. With Redis log shipping disabled - the default, and what this test suite runs with - + * shouldShip is false for every level. This is the fix's core contract: SecureLogging. + * maskSensitive must only run when the local logger or Redis will actually consume the + * message. + * + * What this does NOT cover: shouldShip's own enabled/min-level reading of props + * (redis_logging_enabled / redis_logging_min_level). RedisLogger is a singleton object whose + * config vals are read once, on first JVM access to the object - by the time any test in this + * suite runs, something else has almost certainly already touched it during boot, so changing + * props at test time (even via PropsReset) has no effect on what RedisLogger already cached. + * That half of the behaviour was verified by hand against the actual props read at JVM start + * (see the redis_logging_min_level default in sample.props.template) rather than by a test here. + */ +class RedisLoggerShouldShipTest extends FlatSpec with Matchers { + + "LogLevel" should "be declared in ascending severity order" in { + RedisLogger.LogLevel.TRACE.id should be < RedisLogger.LogLevel.DEBUG.id + RedisLogger.LogLevel.DEBUG.id should be < RedisLogger.LogLevel.INFO.id + RedisLogger.LogLevel.INFO.id should be < RedisLogger.LogLevel.WARNING.id + RedisLogger.LogLevel.WARNING.id should be < RedisLogger.LogLevel.ERROR.id + } + + it should "round-trip through valueOf for every level MdcLoggable actually calls shouldShip with" in { + RedisLogger.LogLevel.valueOf("TRACE") should equal(RedisLogger.LogLevel.TRACE) + RedisLogger.LogLevel.valueOf("DEBUG") should equal(RedisLogger.LogLevel.DEBUG) + RedisLogger.LogLevel.valueOf("INFO") should equal(RedisLogger.LogLevel.INFO) + RedisLogger.LogLevel.valueOf("WARNING") should equal(RedisLogger.LogLevel.WARNING) + RedisLogger.LogLevel.valueOf("ERROR") should equal(RedisLogger.LogLevel.ERROR) + } + + "RedisLogger.shouldShip" should "be false for every level when Redis log shipping is disabled" in { + // This suite's props leave redis_logging_enabled at its default (false), which is also the + // production-safe default - so this exercises the same configuration MdcLoggable runs under + // wherever nobody has opted into Redis log shipping. + RedisLogger.isEnabled shouldBe false + + RedisLogger.shouldShip(RedisLogger.LogLevel.TRACE) shouldBe false + RedisLogger.shouldShip(RedisLogger.LogLevel.DEBUG) shouldBe false + RedisLogger.shouldShip(RedisLogger.LogLevel.INFO) shouldBe false + RedisLogger.shouldShip(RedisLogger.LogLevel.WARNING) shouldBe false + RedisLogger.shouldShip(RedisLogger.LogLevel.ERROR) shouldBe false + } + + it should "never ship LogLevel.ALL, which is a read-side aggregate queue, not a real message level" in { + RedisLogger.shouldShip(RedisLogger.LogLevel.ALL) shouldBe false + } + "RedisLogger.parseMinLevel" should "accept every documented level, ignoring case and surrounding spaces" in { + RedisLogger.parseMinLevel("TRACE") should equal(RedisLogger.LogLevel.TRACE) + RedisLogger.parseMinLevel(" debug ") should equal(RedisLogger.LogLevel.DEBUG) + RedisLogger.parseMinLevel("warn") should equal(RedisLogger.LogLevel.WARNING) + RedisLogger.parseMinLevel("Warning") should equal(RedisLogger.LogLevel.WARNING) + RedisLogger.parseMinLevel("ERROR") should equal(RedisLogger.LogLevel.ERROR) + } + + it should "fall back to INFO instead of throwing for an unrecognised value" in { + // A throwing parse would fail the RedisLogger initialiser, and MdcLoggable consults RedisLogger + // on every log call, so a typo in props would break ordinary logging statements. + noException should be thrownBy RedisLogger.parseMinLevel("OFF") + RedisLogger.parseMinLevel("OFF") should equal(RedisLogger.LogLevel.INFO) + RedisLogger.parseMinLevel("") should equal(RedisLogger.LogLevel.INFO) + RedisLogger.parseMinLevel("FATAL") should equal(RedisLogger.LogLevel.INFO) + } +} diff --git a/obp-api/src/test/scala/code/api/util/JsonSchemaGeneratorCacheTest.scala b/obp-api/src/test/scala/code/api/util/JsonSchemaGeneratorCacheTest.scala new file mode 100644 index 0000000000..eef09f6d1e --- /dev/null +++ b/obp-api/src/test/scala/code/api/util/JsonSchemaGeneratorCacheTest.scala @@ -0,0 +1,116 @@ +package code.api.util + +import code.api.util.APIUtil.MessageDoc +import org.scalatest.{FlatSpec, Matchers} + +/** + * Guards the in-process cache added to JsonSchemaGenerator.messageDocsToJsonSchema. + * + * Building the schema walks the full field tree of every message type via Scala runtime + * reflection (recursive <: "v")) + SchemaCacheTestPayload("id-1", BigDecimal(1), List("a", "b"), address, batch) + } + + private def sampleMessageDocs(connectorName: String): List[MessageDoc] = List( + MessageDoc( + process = s"obp.$connectorName.getThing", + messageFormat = "JSON", + description = "test message doc for cache verification", + exampleOutboundMessage = samplePayload(), + exampleInboundMessage = samplePayload() + ) + ) + + "JsonSchemaGenerator.messageDocsToJsonSchema" should "return identical output on a cache hit as on the initial cold call" in { + val connectorName = freshConnectorName("identical-output") + val docs = sampleMessageDocs(connectorName) + + val first = JsonSchemaGenerator.messageDocsToJsonSchema(docs, connectorName) + val second = JsonSchemaGenerator.messageDocsToJsonSchema(docs, connectorName) + + second should equal(first) + } + + it should "be dramatically faster on a cache hit than on the initial cold call" in { + val connectorName = freshConnectorName("timing") + val docs = sampleMessageDocs(connectorName) + + val coldStart = System.nanoTime() + JsonSchemaGenerator.messageDocsToJsonSchema(docs, connectorName) + val coldNanos = System.nanoTime() - coldStart + + // A handful of warm calls, not just one: a single fast call could be a lucky JIT/scheduling + // blip rather than an actual cache hit. Comparing the median (not every sample, and not the + // fastest) against the cold call keeps this robust to a single GC pause or JIT hiccup landing + // on one warm sample while still requiring a real, consistent improvement. + val warmNanosSamples = (1 to 7).map { _ => + val start = System.nanoTime() + JsonSchemaGenerator.messageDocsToJsonSchema(docs, connectorName) + System.nanoTime() - start + } + val medianWarmNanos = warmNanosSamples.sorted.apply(warmNanosSamples.size / 2) + + withClue(s"cold=${coldNanos}ns warm=${warmNanosSamples.sorted.mkString(",")}ns median=${medianWarmNanos}ns: ") { + medianWarmNanos should be < (coldNanos / 2) + } + } + + it should "isolate different connector names as independent cache entries" in { + val connectorA = freshConnectorName("connector-a") + val connectorB = freshConnectorName("connector-b") + + val schemaA = JsonSchemaGenerator.messageDocsToJsonSchema(sampleMessageDocs(connectorA), connectorA) + val schemaB = JsonSchemaGenerator.messageDocsToJsonSchema(sampleMessageDocs(connectorB), connectorB) + + (schemaA \ "title") should equal(org.json4s.JString(s"$connectorA Message Schemas")) + (schemaB \ "title") should equal(org.json4s.JString(s"$connectorB Message Schemas")) + } +} diff --git a/obp-api/src/test/scala/code/api/v2_2_0/MessageDocsCacheEndToEndTest.scala b/obp-api/src/test/scala/code/api/v2_2_0/MessageDocsCacheEndToEndTest.scala new file mode 100644 index 0000000000..6a6bc28d27 --- /dev/null +++ b/obp-api/src/test/scala/code/api/v2_2_0/MessageDocsCacheEndToEndTest.scala @@ -0,0 +1,29 @@ +package code.api.v2_2_0 + +import code.setup.DefaultUsers +import com.openbankproject.commons.util.JsonAliases.compactRender + +/** + * The message-docs response is cached in process and in the shared store. A second replica, or + * this one after a restart, serves the shared copy, so that copy must be identical to what the + * first request returned. + */ +class MessageDocsCacheEndToEndTest extends V220ServerSetup with DefaultUsers { + + feature("GET /obp/v2.2.0/message-docs/CONNECTOR is stable across the cache levels") { + scenario("the response is identical whether it was generated or read back from the shared level") { + MessageDocsJsonCache.invalidateAll() + val request = (v2_2Request / "message-docs" / "rest_vMar2019").GET + + val generated = makeGetRequest(request) + generated.code should equal(200) + + // Drop the in-process level so the next request goes through the shared level. + MessageDocsJsonCache.invalidateAll() + val again = makeGetRequest(request) + again.code should equal(200) + + compactRender(again.body) should equal(compactRender(generated.body)) + } + } +} diff --git a/obp-api/src/test/scala/code/api/v2_2_0/MessageDocsJsonCacheTest.scala b/obp-api/src/test/scala/code/api/v2_2_0/MessageDocsJsonCacheTest.scala new file mode 100644 index 0000000000..4076250e4b --- /dev/null +++ b/obp-api/src/test/scala/code/api/v2_2_0/MessageDocsJsonCacheTest.scala @@ -0,0 +1,136 @@ +package code.api.v2_2_0 + +import java.util.concurrent.{CountDownLatch, Executors, TimeUnit} +import java.util.concurrent.atomic.AtomicInteger + +import code.api.v2_2_0.MessageDocsJsonCache.SharedStore +import org.json4s.JValue +import org.json4s.JsonDSL._ +import org.scalatest.{BeforeAndAfterEach, FlatSpec, Matchers} + +import scala.collection.concurrent.TrieMap + +class MessageDocsJsonCacheTest extends FlatSpec with Matchers with BeforeAndAfterEach { + + /** A shared level that counts traffic; `down` makes it behave like an unreachable Redis. */ + private class FakeStore(down: Boolean = false) extends SharedStore { + val data = TrieMap.empty[String, String] + val gets = new AtomicInteger(0) + val sets = new AtomicInteger(0) + def get(key: String): Option[String] = { gets.incrementAndGet(); if (down) None else data.get(key) } + def set(key: String, value: String): Unit = { sets.incrementAndGet(); if (!down) data.put(key, value) } + } + + private def doc(tag: String): JValue = ("message_docs" -> List(("process" -> tag): JValue)) + + override def beforeEach(): Unit = MessageDocsJsonCache.invalidateAll() + + "MessageDocsJsonCache" should "run the generator once for repeated requests to one connector" in { + val store = new FakeStore + val calls = new AtomicInteger(0) + val results = (1 to 5).map(_ => MessageDocsJsonCache.getOrCompute("c1", store) { calls.incrementAndGet(); doc("c1") }) + calls.get shouldBe 1 + results.distinct.size shouldBe 1 + (results.head eq results.last) shouldBe true + store.gets.get shouldBe 1 // the in-process level answers every request after the first + store.sets.get shouldBe 1 + } + + it should "keep one entry per connector" in { + val store = new FakeStore + val calls = new AtomicInteger(0) + MessageDocsJsonCache.getOrCompute("c1", store) { calls.incrementAndGet(); doc("c1") } + MessageDocsJsonCache.getOrCompute("c2", store) { calls.incrementAndGet(); doc("c2") } + MessageDocsJsonCache.getOrCompute("c1", store) { calls.incrementAndGet(); doc("c1") } + calls.get shouldBe 2 + MessageDocsJsonCache.size shouldBe 2 + store.data.size shouldBe 2 + } + + it should "run the generator once for a concurrent cold burst" in { + val store = new FakeStore + val calls = new AtomicInteger(0) + val n = 16 + val pool = Executors.newFixedThreadPool(n) + val start = new CountDownLatch(1) + val done = new CountDownLatch(n) + (1 to n).foreach { _ => + pool.execute(new Runnable { + def run(): Unit = { + start.await() + MessageDocsJsonCache.getOrCompute("burst", store) { calls.incrementAndGet(); Thread.sleep(200); doc("burst") } + done.countDown() + } + }) + } + start.countDown() + done.await(30, TimeUnit.SECONDS) shouldBe true + pool.shutdownNow() + calls.get shouldBe 1 + store.gets.get shouldBe 1 + store.sets.get shouldBe 1 + } + + it should "not cache a failure and should rethrow the original exception" in { + val store = new FakeStore + val calls = new AtomicInteger(0) + val boom = new RuntimeException("boom") + val thrown = the[RuntimeException] thrownBy MessageDocsJsonCache.getOrCompute("bad", store) { calls.incrementAndGet(); throw boom } + (thrown eq boom) shouldBe true + store.sets.get shouldBe 0 + MessageDocsJsonCache.getOrCompute("bad", store) { calls.incrementAndGet(); doc("bad") } + calls.get shouldBe 2 + } + + it should "serve from the shared level without running the generator, e.g. after a restart" in { + val store = new FakeStore + MessageDocsJsonCache.getOrCompute("c1", store) { doc("c1") } + MessageDocsJsonCache.invalidateAll() // a fresh process: empty in-process level, warm Redis + val calls = new AtomicInteger(0) + val again = MessageDocsJsonCache.getOrCompute("c1", store) { calls.incrementAndGet(); doc("other") } + calls.get shouldBe 0 + again shouldBe doc("c1") + store.sets.get shouldBe 1 + } + + it should "still cache in-process when the shared level is unreachable" in { + val store = new FakeStore(down = true) + val calls = new AtomicInteger(0) + (1 to 5).foreach(_ => MessageDocsJsonCache.getOrCompute("c1", store) { calls.incrementAndGet(); doc("c1") }) + calls.get shouldBe 1 + } + + it should "treat an unparsable shared entry as a miss and regenerate" in { + val store = new FakeStore + store.data.put("message-docs-v2.2.0-c1", "{not json") + val calls = new AtomicInteger(0) + val r = MessageDocsJsonCache.getOrCompute("c1", store) { calls.incrementAndGet(); doc("c1") } + calls.get shouldBe 1 + r shouldBe doc("c1") + } + + it should "serve the same JSON from the generating instance as from another instance reading the shared level" in { + // Numbers are where a render/parse round trip can change the value or its formatting. + val awkward: JValue = ("message_docs" -> List[JValue]( + ("decimal" -> BigDecimal("10.10")): JValue, + ("double" -> 0.1): JValue, + ("big" -> BigInt("12345678901234567890")): JValue, + ("neg" -> -5): JValue)) + val store = new FakeStore + val generated = MessageDocsJsonCache.getOrCompute("c1", store) { awkward } + MessageDocsJsonCache.invalidateAll() // another replica: empty in-process level, same shared level + val fromShared = MessageDocsJsonCache.getOrCompute("c1", store) { fail("must be served from the shared level") } + generated shouldBe fromShared + org.json4s.native.JsonMethods.compact(org.json4s.native.JsonMethods.render(generated)) shouldBe + org.json4s.native.JsonMethods.compact(org.json4s.native.JsonMethods.render(fromShared)) + } + + it should "regenerate after invalidateAll when the shared level is empty" in { + val calls = new AtomicInteger(0) + val a = new FakeStore + MessageDocsJsonCache.getOrCompute("c1", a) { calls.incrementAndGet(); doc("c1") } + MessageDocsJsonCache.invalidateAll() + MessageDocsJsonCache.getOrCompute("c1", new FakeStore) { calls.incrementAndGet(); doc("c1") } + calls.get shouldBe 2 + } +} diff --git a/obp-api/src/test/scala/code/util/MdcLoggableBoundedQueueTest.scala b/obp-api/src/test/scala/code/util/MdcLoggableBoundedQueueTest.scala new file mode 100644 index 0000000000..bb7f305f64 --- /dev/null +++ b/obp-api/src/test/scala/code/util/MdcLoggableBoundedQueueTest.scala @@ -0,0 +1,111 @@ +package code.util + +import java.util.concurrent.{ArrayBlockingQueue, CountDownLatch, ThreadPoolExecutor, TimeUnit} +import java.util.concurrent.atomic.AtomicInteger + +import org.scalatest.{FlatSpec, Matchers} + +/** + * The log dispatch pool must have a bounded, documented failure mode: when its queue is full, + * non-critical entries are dropped and counted, critical (warn/error) entries still run on the + * caller, and nothing throws to the caller. + */ +class MdcLoggableBoundedQueueTest extends FlatSpec with Matchers { + + /** One worker, queue of one, AbortPolicy: the third submission while the worker is busy is rejected. */ + private def tinyExecutor(): ThreadPoolExecutor = + new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue[Runnable](1), + new ThreadPoolExecutor.AbortPolicy()) + + /** Occupy the worker and fill the queue; returns the latch that releases the worker. */ + private def saturate(ex: ThreadPoolExecutor): CountDownLatch = { + val release = new CountDownLatch(1) + val started = new CountDownLatch(1) + Helper.dispatchOn(ex, "test", critical = true) { started.countDown(); release.await(30, TimeUnit.SECONDS) } + started.await(10, TimeUnit.SECONDS) shouldBe true + Helper.dispatchOn(ex, "test", critical = true) { () } // sits in the queue + ex.getQueue.size() shouldBe 1 + release + } + + "the log dispatch" should "drop and count non-critical entries when the queue is full, without throwing" in { + val ex = tinyExecutor() + val release = saturate(ex) + try { + val ran = new AtomicInteger(0) + val before = Helper.mdcLogDroppedCount + (1 to 5).foreach(_ => Helper.dispatchOn(ex, "test", critical = false) { ran.incrementAndGet() }) + Helper.mdcLogDroppedCount - before shouldBe 5 + ran.get shouldBe 0 + } finally { release.countDown(); ex.shutdownNow() } + } + + it should "run critical entries inline on the caller when the queue is full" in { + val ex = tinyExecutor() + val release = saturate(ex) + try { + val ranOn = new java.util.concurrent.atomic.AtomicReference[String]() + val before = Helper.mdcLogDroppedCount + Helper.dispatchOn(ex, "test", critical = true) { ranOn.set(Thread.currentThread().getName) } + ranOn.get shouldBe Thread.currentThread().getName + Helper.mdcLogDroppedCount shouldBe before // a critical entry is never counted as dropped + } finally { release.countDown(); ex.shutdownNow() } + } + + it should "not let a failing log body escape to the caller" in { + val ex = tinyExecutor() + try { + noException should be thrownBy Helper.dispatchOn(ex, "test", critical = true) { throw new RuntimeException("boom") } + } finally ex.shutdownNow() + } + + it should "run on the pool thread with the caller's name in the MDC, and leave the pool thread untouched" in { + val poolThreadName = "test-pool-thread" + val ex = java.util.concurrent.Executors.newSingleThreadExecutor((r: Runnable) => new Thread(r, poolThreadName)) + try { + val seen = new java.util.concurrent.atomic.AtomicReference[(Thread, String, String)]() + val done = new CountDownLatch(1) + Helper.dispatchOn(ex, "test", critical = false) { + seen.set((Thread.currentThread(), Thread.currentThread().getName, org.slf4j.MDC.get(Helper.MdcCallerThreadKey))) + done.countDown() + } + done.await(10, TimeUnit.SECONDS) shouldBe true + seen.get._1 should not be theSameInstanceAs(Thread.currentThread()) // still off the calling thread + seen.get._2 shouldBe poolThreadName // the pool thread keeps its own name + seen.get._3 shouldBe Thread.currentThread().getName // the caller travels in the MDC + + // The key must not leak into the next task on the same pool thread. + val leaked = new java.util.concurrent.atomic.AtomicReference[String]("unset") + val done2 = new CountDownLatch(1) + ex.execute(() => { leaked.set(org.slf4j.MDC.get(Helper.MdcCallerThreadKey)); done2.countDown() }) + done2.await(10, TimeUnit.SECONDS) shouldBe true + leaked.get shouldBe null + } finally ex.shutdownNow() + } + + it should "write on the caller instead of dropping when the pool has been shut down" in { + val ex = tinyExecutor() + ex.shutdown() + val ranOn = new java.util.concurrent.atomic.AtomicReference[String]() + val before = Helper.mdcLogDroppedCount + Helper.dispatchOn(ex, "test", critical = false) { ranOn.set(Thread.currentThread().getName) } + ranOn.get shouldBe Thread.currentThread().getName + Helper.mdcLogDroppedCount shouldBe before // shutdown is not overload, so nothing is counted as dropped + } + + it should "clamp a zero or negative pool or queue size so the pool can still be created" in { + Helper.dispatchPoolSettings(0, 0) shouldBe ((1, 1)) + Helper.dispatchPoolSettings(-3, -10) shouldBe ((1, 1)) + Helper.dispatchPoolSettings(2, 10000) shouldBe ((2, 10000)) + // The clamped values are accepted by the real constructors, the raw ones are not. + an[IllegalArgumentException] should be thrownBy new ArrayBlockingQueue[Runnable](0) + val (pool, queue) = Helper.dispatchPoolSettings(0, 0) + val ex = new ThreadPoolExecutor(pool, pool, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue[Runnable](queue)) + try ex.getCorePoolSize shouldBe 1 finally ex.shutdownNow() + } + + it should "keep the production pool's queue bounded and observable" in { + Helper.mdcLogQueueDepth should be >= 0 + Helper.mdcLogDroppedCount should be >= 0L + } +} diff --git a/obp-api/src/test/scala/code/util/MdcLoggableDispatchTest.scala b/obp-api/src/test/scala/code/util/MdcLoggableDispatchTest.scala new file mode 100644 index 0000000000..151b9eaaea --- /dev/null +++ b/obp-api/src/test/scala/code/util/MdcLoggableDispatchTest.scala @@ -0,0 +1,68 @@ +package code.util + +import ch.qos.logback.classic.spi.ILoggingEvent +import ch.qos.logback.classic.{Level, Logger => LogbackLogger} +import ch.qos.logback.core.AppenderBase +import org.scalatest.{FlatSpec, Matchers} +import org.slf4j.LoggerFactory + +import java.util.concurrent.{CopyOnWriteArrayList, TimeUnit} + +/** + * Verifies the actual behavioural change from the fix in Helper.scala: masking + the local + * write now happen on a dedicated background pool (mdcLoggingExecutionContext) instead of + * inline on the calling thread. RedisLoggerShouldShipTest covers the gating logic + * (shouldShip); this covers the dispatch itself -- that a call which passes the gate still + * reliably reaches the underlying logger (just asynchronously), and that it does so on a + * thread other than the caller's. + */ +class MdcLoggableDispatchTest extends FlatSpec with Matchers { + + private object ProbeLogger extends Helper.MdcLoggable { + // `logger` is protected on the MdcLoggable/Loggable trait -- only reachable from within a + // mixing class's own body, not from the test. Expose the one call the test needs. + def probeDebug(msg: String): Unit = logger.debug(msg) + } + + "MdcLoggable" should "deliver an enabled log call to the underlying logger asynchronously, attributed to the calling thread" in { + val logbackLogger = LoggerFactory.getLogger(ProbeLogger.getClass.getName).asInstanceOf[LogbackLogger] + val originalLevel = logbackLogger.getLevel + logbackLogger.setLevel(Level.DEBUG) + + val captured = new CopyOnWriteArrayList[ILoggingEvent]() + val appender = new AppenderBase[ILoggingEvent] { + override def append(event: ILoggingEvent): Unit = captured.add(event) + } + appender.setContext(logbackLogger.getLoggerContext) + appender.start() + logbackLogger.addAppender(appender) + + val callingThreadName = Thread.currentThread().getName + // Not a bare digit run: a 16-19 digit nanoTime() collides with SecureLogging's + // credit-card-shaped masking pattern and gets partially masked -- correct behaviour from + // the masking regex, but it broke this test's own substring check the first time around. + val marker = s"MdcLoggableDispatchTest-probe-id${System.nanoTime()}" + + try { + ProbeLogger.probeDebug(marker) + + // The dispatch is asynchronous, so the event won't necessarily be there immediately -- + // poll briefly rather than asserting right away (which would just be testing timing, + // not the actual delivery guarantee). + val deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5) + while (captured.isEmpty && System.nanoTime() < deadline) { + Thread.sleep(20) + } + + captured should not be empty + val event = captured.get(0) + event.getFormattedMessage should include(marker) + // The write runs on the pool, but the record must still name the thread that logged. + event.getThreadName should startWith("mdc-log-dispatch-") + event.getMDCPropertyMap.get(Helper.MdcCallerThreadKey) shouldBe callingThreadName + } finally { + logbackLogger.detachAppender(appender) + logbackLogger.setLevel(originalLevel) + } + } +}