From 791875cce39cf10a971ddd7984fdb3ebc681a64c Mon Sep 17 00:00:00 2001 From: Hongwei Date: Wed, 23 Sep 2026 15:15:44 +0200 Subject: [PATCH 01/14] fix: production CPU/GC incident - unconditional log masking, missing schema cache, and default DEBUG logging Root-caused a production incident (sustained 600% CPU, Full-GC thrashing, eventual service timeout) on a sandbox instance. Three independent fixes, bundled together since they trace back to the same investigation and are small enough to review as a set. 1. MdcLoggable (code/util/Helper.scala) ran SecureLogging.maskSensitive (~19 regex passes) and Redis-shipping JSON serialization unconditionally on every info/warn/error/debug/trace call, before checking whether the underlying logger or Redis would even consume the message - and, once gated, still ran that work inline on the calling thread. On the http4s/cats-effect request path that's often a fiber-managed worker; non-yielding CPU-bound work there looks identical, to the runtime's own starvation detector, to genuine blocking I/O, and its response (spinning up compensating worker/blocker threads that are never reclaimed) links directly to a heap dump showing hundreds of threads each holding a private Scala-reflection symbol-table cache. Fixed by gating on isXEnabled/RedisLogger.shouldShip (new redis_logging_min_level prop, default INFO) and dispatching the actual work onto a small dedicated ExecutionContext instead of the calling thread. 2. JsonSchemaGenerator.messageDocsToJsonSchema had no cache of its own, walking every message type's field tree via Scala runtime reflection on every call; its caller's Redis-backed cache silently falls through to a full recompute on any Redis failure. Added an in-process memoization (Guava-backed, no Redis dependency) alongside the existing one, using the same pattern already proven correct by Helper.getRequiredFieldInfo. Independent hardening for a real gap in this file - not confirmed as this incident's cause; JsonSchemaGenerator doesn't appear anywhere in this incident's JFR execution samples. 3. logback.xml.example sat inert (Logback only auto-loads files named exactly logback.xml) with root level hard-coded to DEBUG; deployments that never manually copied it over fell back to Logback's own built-in DEBUG-to-console default. Renamed to logback.xml so it's actually loaded, and changed the level to ${LOG_LEVEL:-INFO}: verbose logging should be an explicit per-environment opt-in, not the default every deployment silently pays for. Test plan: see individual test additions (RedisLoggerShouldShipTest, MdcLoggableDispatchTest, JsonSchemaGeneratorCacheTest, LogbackDefaultLevelTest) plus full local suite and CI. --- README.md | 7 +- .../docs/brief_system_documentation.md | 2 +- obp-api/src/main/resources/logback.xml | 17 ++ .../src/main/resources/logback.xml.example | 12 -- .../resources/props/sample.props.template | 22 +++ .../scala/code/api/cache/RedisLogger.scala | 29 +++- .../code/api/util/JsonSchemaGenerator.scala | 29 +++- obp-api/src/main/scala/code/util/Helper.scala | 149 ++++++++++++++---- .../http4s/LogbackDefaultLevelTest.scala | 55 +++++++ .../api/cache/RedisLoggerShouldShipTest.scala | 61 +++++++ .../util/JsonSchemaGeneratorCacheTest.scala | 116 ++++++++++++++ .../code/util/MdcLoggableDispatchTest.scala | 67 ++++++++ 12 files changed, 522 insertions(+), 44 deletions(-) create mode 100644 obp-api/src/main/resources/logback.xml delete mode 100644 obp-api/src/main/resources/logback.xml.example create mode 100644 obp-api/src/test/scala/bootstrap/http4s/LogbackDefaultLevelTest.scala create mode 100644 obp-api/src/test/scala/code/api/cache/RedisLoggerShouldShipTest.scala create mode 100644 obp-api/src/test/scala/code/api/util/JsonSchemaGeneratorCacheTest.scala create mode 100644 obp-api/src/test/scala/code/util/MdcLoggableDispatchTest.scala 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..8044beb0bf --- /dev/null +++ b/obp-api/src/main/resources/logback.xml @@ -0,0 +1,17 @@ + + + + + %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/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..950de02420 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,21 @@ 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) was a contributing factor in a real incident. Keep this small; it's meant +## to decouple from the calling thread, not to parallelize heavy logging. +mdc_logging_dispatch_thread_pool_size = 2 + ## 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/code/api/cache/RedisLogger.scala b/obp-api/src/main/scala/code/api/cache/RedisLogger.scala index 7c22956d5a..5342d11a23 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,33 @@ 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 = + LogLevel.valueOf(APIUtil.getPropsValue("redis_logging_min_level", "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 +202,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..c465a8bb33 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 is + * what drove a production old-gen heap to exhaustion. 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/util/Helper.scala b/obp-api/src/main/scala/code/util/Helper.scala index b45b10414d..0b3cfdf817 100644 --- a/obp-api/src/main/scala/code/util/Helper.scala +++ b/obp-api/src/main/scala/code/util/Helper.scala @@ -286,6 +286,50 @@ 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). + private lazy val mdcLoggingExecutor: java.util.concurrent.ExecutorService = { + val threadCount = new java.util.concurrent.atomic.AtomicInteger(0) + java.util.concurrent.Executors.newFixedThreadPool( + APIUtil.getPropsAsIntValue("mdc_logging_dispatch_thread_pool_size", 2), + (r: Runnable) => { + val t = new Thread(r, s"mdc-log-dispatch-${threadCount.incrementAndGet()}") + t.setDaemon(true) + t + } + ) + } + private lazy val mdcLoggingExecutionContext: scala.concurrent.ExecutionContext = + scala.concurrent.ExecutionContext.fromExecutor(mdcLoggingExecutor) + trait MdcLoggable extends Loggable { // Capture the class name of the component mixing in this trait @@ -311,63 +355,114 @@ object Helper extends Loggable { 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 mdcLoggingExecutionContext (see its own doc comment) + // rather than run inline on whatever thread called into this logger. + private def dispatch(body: => Unit): Unit = { + val future = scala.concurrent.Future(body)(mdcLoggingExecutionContext) + future.failed.foreach { e => + System.err.println(s"[$clazzName] background log dispatch failed: ${e.getMessage}") + }(mdcLoggingExecutionContext) + } + // 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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..ca2b057bb8 --- /dev/null +++ b/obp-api/src/test/scala/code/api/cache/RedisLoggerShouldShipTest.scala @@ -0,0 +1,61 @@ +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 + } +} 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/util/MdcLoggableDispatchTest.scala b/obp-api/src/test/scala/code/util/MdcLoggableDispatchTest.scala new file mode 100644 index 0000000000..56de27238f --- /dev/null +++ b/obp-api/src/test/scala/code/util/MdcLoggableDispatchTest.scala @@ -0,0 +1,67 @@ +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, off 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) + event.getThreadName should not equal callingThreadName + event.getThreadName should startWith("mdc-log-dispatch-") + } finally { + logbackLogger.detachAppender(appender) + logbackLogger.setLevel(originalLevel) + } + } +} From 1ddd86db94b695b97ad01213d4192c4817eaff9c Mon Sep 17 00:00:00 2001 From: Hongwei Date: Thu, 24 Sep 2026 11:58:16 +0200 Subject: [PATCH 02/14] fix: cache the v2.2.0 message-docs response per connector The endpoint rebuilt the whole response on every request, running Scala runtime reflection over every message doc example and growing the reflection universe. Cache the built response per resolved connector name with single-flight loading, a hard size bound and no caching of failures. --- .../scala/code/api/v2_2_0/Http4s220.scala | 4 +- .../api/v2_2_0/MessageDocsJsonCache.scala | 43 ++++++++++++ .../api/v2_2_0/MessageDocsJsonCacheTest.scala | 66 +++++++++++++++++++ 3 files changed, 112 insertions(+), 1 deletion(-) create mode 100644 obp-api/src/main/scala/code/api/v2_2_0/MessageDocsJsonCache.scala create mode 100644 obp-api/src/test/scala/code/api/v2_2_0/MessageDocsJsonCacheTest.scala 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..be771936d9 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) { + 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..d5743a5918 --- /dev/null +++ b/obp-api/src/main/scala/code/api/v2_2_0/MessageDocsJsonCache.scala @@ -0,0 +1,43 @@ +package code.api.v2_2_0 + +import java.util.concurrent.Callable + +import com.google.common.cache.{Cache, CacheBuilder} + +/** + * In-process 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, so it is + * safe to keep per connector name for the life of the process. + * + * 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. + * - Single flight: concurrent cold requests for one connector run the generator once. + * - A generator failure is not cached; the next request retries. + * - The cached value is an immutable case class tree and is shared between requests. + * - Invalidation: `invalidateAll()`. Nothing in production mutates a connector's + * message docs after start-up, so nothing calls it there. + */ +object MessageDocsJsonCache { + private val MaxEntries = 64L + + private val cache: Cache[String, JSONFactory220.MessageDocsJson] = + CacheBuilder.newBuilder().maximumSize(MaxEntries).build[String, JSONFactory220.MessageDocsJson]() + + def getOrCompute(connectorName: String)(generate: => JSONFactory220.MessageDocsJson): JSONFactory220.MessageDocsJson = + try cache.get(connectorName, new Callable[JSONFactory220.MessageDocsJson] { def call() = generate }) + catch { + // Surface the generator'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/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..9a89f4b478 --- /dev/null +++ b/obp-api/src/test/scala/code/api/v2_2_0/MessageDocsJsonCacheTest.scala @@ -0,0 +1,66 @@ +package code.api.v2_2_0 + +import java.util.concurrent.{CountDownLatch, Executors, TimeUnit} +import java.util.concurrent.atomic.AtomicInteger + +import org.scalatest.{BeforeAndAfterEach, FlatSpec, Matchers} + +class MessageDocsJsonCacheTest extends FlatSpec with Matchers with BeforeAndAfterEach { + + override def beforeEach(): Unit = MessageDocsJsonCache.invalidateAll() + + "MessageDocsJsonCache" should "run the generator once for repeated requests to one connector" in { + val calls = new AtomicInteger(0) + val results = (1 to 5).map(_ => MessageDocsJsonCache.getOrCompute("c1") { calls.incrementAndGet(); JSONFactory220.MessageDocsJson(Nil) }) + calls.get shouldBe 1 + results.distinct.size shouldBe 1 + (results.head eq results.last) shouldBe true + } + + it should "keep one entry per connector" in { + val calls = new AtomicInteger(0) + MessageDocsJsonCache.getOrCompute("c1") { calls.incrementAndGet(); JSONFactory220.MessageDocsJson(Nil) } + MessageDocsJsonCache.getOrCompute("c2") { calls.incrementAndGet(); JSONFactory220.MessageDocsJson(Nil) } + MessageDocsJsonCache.getOrCompute("c1") { calls.incrementAndGet(); JSONFactory220.MessageDocsJson(Nil) } + calls.get shouldBe 2 + MessageDocsJsonCache.size shouldBe 2 + } + + it should "run the generator once for a concurrent cold burst" in { + 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") { calls.incrementAndGet(); Thread.sleep(200); JSONFactory220.MessageDocsJson(Nil) } + done.countDown() + } + }) + } + start.countDown() + done.await(30, TimeUnit.SECONDS) shouldBe true + pool.shutdownNow() + calls.get shouldBe 1 + } + + it should "not cache a failure and should rethrow the original exception" in { + val calls = new AtomicInteger(0) + val boom = new RuntimeException("boom") + val thrown = the[RuntimeException] thrownBy MessageDocsJsonCache.getOrCompute("bad") { calls.incrementAndGet(); throw boom } + (thrown eq boom) shouldBe true + MessageDocsJsonCache.getOrCompute("bad") { calls.incrementAndGet(); JSONFactory220.MessageDocsJson(Nil) } + calls.get shouldBe 2 + } + + it should "regenerate after invalidateAll" in { + val calls = new AtomicInteger(0) + MessageDocsJsonCache.getOrCompute("c1") { calls.incrementAndGet(); JSONFactory220.MessageDocsJson(Nil) } + MessageDocsJsonCache.invalidateAll() + MessageDocsJsonCache.getOrCompute("c1") { calls.incrementAndGet(); JSONFactory220.MessageDocsJson(Nil) } + calls.get shouldBe 2 + } +} From a0627451ae1dd05459c212451916b9786ace53cf Mon Sep 17 00:00:00 2001 From: Hongwei Date: Thu, 24 Sep 2026 13:40:05 +0200 Subject: [PATCH 03/14] feat: add a shared Redis level behind the message-docs cache Reuse the store the resource-doc and swagger endpoints already use, with the same key prefix, TTL and fail-safe behaviour, so replicas and restarts share one instance's generation work. The in-process level stays in front so an unreachable Redis cannot send requests back into reflection. --- .../scala/code/api/v2_2_0/Http4s220.scala | 2 +- .../api/v2_2_0/MessageDocsJsonCache.scala | 71 ++++++++++++++---- .../api/v2_2_0/MessageDocsJsonCacheTest.scala | 74 ++++++++++++++++--- 3 files changed, 121 insertions(+), 26 deletions(-) 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 be771936d9..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 @@ -426,7 +426,7 @@ object Http4s220 { s"$InvalidConnector Current Input is $connector. It should be eg: rest_vMar2019..." ) MessageDocsJsonCache.getOrCompute(connector) { - JSONFactory220.createMessageDocsJson(connectorObject.messageDocs.toList) + 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 index d5743a5918..999dd45049 100644 --- 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 @@ -2,37 +2,78 @@ 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 /** - * In-process cache for the fully built `GET /message-docs/CONNECTOR` response. + * 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, so it is - * safe to keep per connector name for the life of the process. + * 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. - * - Single flight: concurrent cold requests for one connector run the generator once. - * - A generator failure is not cached; the next request retries. - * - The cached value is an immutable case class tree and is shared between requests. - * - Invalidation: `invalidateAll()`. Nothing in production mutates a connector's - * message docs after start-up, so nothing calls it there. + * `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 { +object MessageDocsJsonCache extends Loggable { private val MaxEntries = 64L - private val cache: Cache[String, JSONFactory220.MessageDocsJson] = - CacheBuilder.newBuilder().maximumSize(MaxEntries).build[String, JSONFactory220.MessageDocsJson]() + /** 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)(generate: => JSONFactory220.MessageDocsJson): JSONFactory220.MessageDocsJson = - try cache.get(connectorName, new Callable[JSONFactory220.MessageDocsJson] { def call() = generate }) + 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 { + val generated = generate + store.set(key, compactRender(generated)) + generated + } + } + }) catch { - // Surface the generator's own exception, not Guava's wrapper. + // 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 } 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 index 9a89f4b478..55932d59c8 100644 --- 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 @@ -3,30 +3,52 @@ 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") { calls.incrementAndGet(); JSONFactory220.MessageDocsJson(Nil) }) + 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") { calls.incrementAndGet(); JSONFactory220.MessageDocsJson(Nil) } - MessageDocsJsonCache.getOrCompute("c2") { calls.incrementAndGet(); JSONFactory220.MessageDocsJson(Nil) } - MessageDocsJsonCache.getOrCompute("c1") { calls.incrementAndGet(); JSONFactory220.MessageDocsJson(Nil) } + 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) @@ -36,7 +58,7 @@ class MessageDocsJsonCacheTest extends FlatSpec with Matchers with BeforeAndAfte pool.execute(new Runnable { def run(): Unit = { start.await() - MessageDocsJsonCache.getOrCompute("burst") { calls.incrementAndGet(); Thread.sleep(200); JSONFactory220.MessageDocsJson(Nil) } + MessageDocsJsonCache.getOrCompute("burst", store) { calls.incrementAndGet(); Thread.sleep(200); doc("burst") } done.countDown() } }) @@ -45,22 +67,54 @@ class MessageDocsJsonCacheTest extends FlatSpec with Matchers with BeforeAndAfte 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") { calls.incrementAndGet(); throw boom } + val thrown = the[RuntimeException] thrownBy MessageDocsJsonCache.getOrCompute("bad", store) { calls.incrementAndGet(); throw boom } (thrown eq boom) shouldBe true - MessageDocsJsonCache.getOrCompute("bad") { calls.incrementAndGet(); JSONFactory220.MessageDocsJson(Nil) } + store.sets.get shouldBe 0 + MessageDocsJsonCache.getOrCompute("bad", store) { calls.incrementAndGet(); doc("bad") } calls.get shouldBe 2 } - it should "regenerate after invalidateAll" in { + 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 "regenerate after invalidateAll when the shared level is empty" in { val calls = new AtomicInteger(0) - MessageDocsJsonCache.getOrCompute("c1") { calls.incrementAndGet(); JSONFactory220.MessageDocsJson(Nil) } + val a = new FakeStore + MessageDocsJsonCache.getOrCompute("c1", a) { calls.incrementAndGet(); doc("c1") } MessageDocsJsonCache.invalidateAll() - MessageDocsJsonCache.getOrCompute("c1") { calls.incrementAndGet(); JSONFactory220.MessageDocsJson(Nil) } + MessageDocsJsonCache.getOrCompute("c1", new FakeStore) { calls.incrementAndGet(); doc("c1") } calls.get shouldBe 2 } } From bc226dda8cfd504d2fa6368ed5fe460afced4db6 Mon Sep 17 00:00:00 2001 From: Hongwei Date: Thu, 24 Sep 2026 15:09:17 +0200 Subject: [PATCH 04/14] fix: bound the log dispatch queue and define its overflow behaviour The pool that runs log masking and shipping used an unbounded queue, so a burst of log calls could grow the heap. Use a bounded queue (mdc_logging_dispatch_queue_size, default 10000). When it is full, DEBUG/TRACE/INFO entries are dropped and counted while WARN/ERROR run on the calling thread. Expose the drop count and queue depth, and give the queue a short drain window on shutdown. --- .../resources/props/sample.props.template | 6 ++ obp-api/src/main/scala/code/util/Helper.scala | 92 ++++++++++++++----- .../util/MdcLoggableBoundedQueueTest.scala | 66 +++++++++++++ 3 files changed, 142 insertions(+), 22 deletions(-) create mode 100644 obp-api/src/test/scala/code/util/MdcLoggableBoundedQueueTest.scala diff --git a/obp-api/src/main/resources/props/sample.props.template b/obp-api/src/main/resources/props/sample.props.template index 950de02420..608c9a9a27 100644 --- a/obp-api/src/main/resources/props/sample.props.template +++ b/obp-api/src/main/resources/props/sample.props.template @@ -2000,6 +2000,12 @@ redis_logging_all_queue_max_entries = 1000 # Max ALL messages ## 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/code/util/Helper.scala b/obp-api/src/main/scala/code/util/Helper.scala index 0b3cfdf817..9a16b9d984 100644 --- a/obp-api/src/main/scala/code/util/Helper.scala +++ b/obp-api/src/main/scala/code/util/Helper.scala @@ -316,19 +316,72 @@ object Helper extends Loggable { // 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). - private lazy val mdcLoggingExecutor: java.util.concurrent.ExecutorService = { + // 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) + + private lazy val mdcLoggingExecutor: java.util.concurrent.ThreadPoolExecutor = { val threadCount = new java.util.concurrent.atomic.AtomicInteger(0) - java.util.concurrent.Executors.newFixedThreadPool( - APIUtil.getPropsAsIntValue("mdc_logging_dispatch_thread_pool_size", 2), + val poolSize = APIUtil.getPropsAsIntValue("mdc_logging_dispatch_thread_pool_size", 2) + val queueSize = 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. + Runtime.getRuntime.addShutdownHook(new Thread(() => { + executor.shutdown() + try executor.awaitTermination(2, java.util.concurrent.TimeUnit.SECONDS) + catch { case _: InterruptedException => () } + }, "mdc-log-dispatch-shutdown")) + 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) + + // 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 = { + val task: Runnable = () => + try body + catch { case e: Throwable => System.err.println(s"[$clazzName] background log dispatch failed: ${e.getMessage}") } + try executor.execute(task) + catch { + case _: java.util.concurrent.RejectedExecutionException => + if (critical) task.run() + else { + 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") + } + } } - private lazy val mdcLoggingExecutionContext: scala.concurrent.ExecutionContext = - scala.concurrent.ExecutionContext.fromExecutor(mdcLoggingExecutor) trait MdcLoggable extends Loggable { @@ -360,19 +413,14 @@ object Helper extends Loggable { // 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 mdcLoggingExecutionContext (see its own doc comment) + // 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(body: => Unit): Unit = { - val future = scala.concurrent.Future(body)(mdcLoggingExecutionContext) - future.failed.foreach { e => - System.err.println(s"[$clazzName] background log dispatch failed: ${e.getMessage}") - }(mdcLoggingExecutionContext) - } + private def dispatch(critical: Boolean)(body: => Unit): Unit = dispatchLog(clazzName, critical)(body) // INFO override def info(msg: => AnyRef): Unit = { if (underlyingLogger.isInfoEnabled || RedisLogger.shouldShip(RedisLogger.LogLevel.INFO)) { - dispatch { + dispatch(critical = false) { val maskedMsg = SecureLogging.maskSensitive(msg) if (underlyingLogger.isInfoEnabled) underlyingLogger.info(maskedMsg) RedisLogger.logAsync(RedisLogger.LogLevel.INFO, toRedisFormat(maskedMsg)) @@ -382,7 +430,7 @@ object Helper extends Loggable { override def info(msg: => AnyRef, t: => Throwable): Unit = { if (underlyingLogger.isInfoEnabled || RedisLogger.shouldShip(RedisLogger.LogLevel.INFO)) { - dispatch { + dispatch(critical = false) { val maskedMsg = SecureLogging.maskSensitive(msg) val capturedT = t if (underlyingLogger.isInfoEnabled) underlyingLogger.info(maskedMsg, capturedT) @@ -394,7 +442,7 @@ object Helper extends Loggable { // WARN override def warn(msg: => AnyRef): Unit = { if (underlyingLogger.isWarnEnabled || RedisLogger.shouldShip(RedisLogger.LogLevel.WARNING)) { - dispatch { + dispatch(critical = true) { val maskedMsg = SecureLogging.maskSensitive(msg) if (underlyingLogger.isWarnEnabled) underlyingLogger.warn(maskedMsg) RedisLogger.logAsync(RedisLogger.LogLevel.WARNING, toRedisFormat(maskedMsg)) @@ -404,7 +452,7 @@ object Helper extends Loggable { override def warn(msg: => AnyRef, t: Throwable): Unit = { if (underlyingLogger.isWarnEnabled || RedisLogger.shouldShip(RedisLogger.LogLevel.WARNING)) { - dispatch { + 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) @@ -415,7 +463,7 @@ object Helper extends Loggable { // ERROR override def error(msg: => AnyRef): Unit = { if (underlyingLogger.isErrorEnabled || RedisLogger.shouldShip(RedisLogger.LogLevel.ERROR)) { - dispatch { + dispatch(critical = true) { val maskedMsg = SecureLogging.maskSensitive(msg) if (underlyingLogger.isErrorEnabled) underlyingLogger.error(maskedMsg) RedisLogger.logAsync(RedisLogger.LogLevel.ERROR, toRedisFormat(maskedMsg)) @@ -425,7 +473,7 @@ object Helper extends Loggable { override def error(msg: => AnyRef, t: Throwable): Unit = { if (underlyingLogger.isErrorEnabled || RedisLogger.shouldShip(RedisLogger.LogLevel.ERROR)) { - dispatch { + 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) @@ -436,7 +484,7 @@ object Helper extends Loggable { // DEBUG override def debug(msg: => AnyRef): Unit = { if (underlyingLogger.isDebugEnabled || RedisLogger.shouldShip(RedisLogger.LogLevel.DEBUG)) { - dispatch { + dispatch(critical = false) { val maskedMsg = SecureLogging.maskSensitive(msg) if (underlyingLogger.isDebugEnabled) underlyingLogger.debug(maskedMsg) RedisLogger.logAsync(RedisLogger.LogLevel.DEBUG, toRedisFormat(maskedMsg)) @@ -446,7 +494,7 @@ object Helper extends Loggable { override def debug(msg: => AnyRef, t: Throwable): Unit = { if (underlyingLogger.isDebugEnabled || RedisLogger.shouldShip(RedisLogger.LogLevel.DEBUG)) { - dispatch { + 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) @@ -457,7 +505,7 @@ object Helper extends Loggable { // TRACE override def trace(msg: => AnyRef): Unit = { if (underlyingLogger.isTraceEnabled || RedisLogger.shouldShip(RedisLogger.LogLevel.TRACE)) { - dispatch { + dispatch(critical = false) { val maskedMsg = SecureLogging.maskSensitive(msg) if (underlyingLogger.isTraceEnabled) underlyingLogger.trace(maskedMsg) RedisLogger.logAsync(RedisLogger.LogLevel.TRACE, toRedisFormat(maskedMsg)) 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..9dddfc63fa --- /dev/null +++ b/obp-api/src/test/scala/code/util/MdcLoggableBoundedQueueTest.scala @@ -0,0 +1,66 @@ +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 "keep the production pool's queue bounded and observable" in { + Helper.mdcLogQueueDepth should be >= 0 + Helper.mdcLogDroppedCount should be >= 0L + } +} From 1ef767477f5c23fe70bca512ddd135f03aa6afb2 Mon Sep 17 00:00:00 2001 From: Hongwei Date: Thu, 24 Sep 2026 15:09:17 +0200 Subject: [PATCH 05/14] feat: log the effective root log level at start-up Warn when it is DEBUG or TRACE, since that is expensive and should only be enabled deliberately through LOG_LEVEL. --- obp-api/src/main/scala/bootstrap/liftweb/Boot.scala | 12 ++++++++++++ 1 file changed, 12 insertions(+) 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) From 7e1a7f13c44b680c8d3a3316e395ddcd1cbbb9af Mon Sep 17 00:00:00 2001 From: Hongwei Date: Thu, 24 Sep 2026 16:00:18 +0200 Subject: [PATCH 06/14] fix: keep the originating thread name on asynchronously dispatched log entries Log writes run on a dedicated pool, so the %t pattern and the Redis line format reported the pool thread instead of the request, actor or compute thread that logged. Record the caller's thread name before dispatch, lend it to the pool thread for the write, and restore the pool thread's name after. --- obp-api/src/main/scala/code/util/Helper.scala | 12 ++++++++++- .../util/MdcLoggableBoundedQueueTest.scala | 21 +++++++++++++++++++ .../code/util/MdcLoggableDispatchTest.scala | 6 +++--- 3 files changed, 35 insertions(+), 4 deletions(-) diff --git a/obp-api/src/main/scala/code/util/Helper.scala b/obp-api/src/main/scala/code/util/Helper.scala index 9a16b9d984..c24efca7db 100644 --- a/obp-api/src/main/scala/code/util/Helper.scala +++ b/obp-api/src/main/scala/code/util/Helper.scala @@ -368,9 +368,19 @@ object Helper extends Loggable { // 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 = { - val task: Runnable = () => + // Logback's %t and the Redis line format both read the current thread's name. Once the write + // moves to a pool thread that name would always be "mdc-log-dispatch-N", hiding which + // request/actor/compute thread logged. Record the caller's name now and lend it to the pool + // thread for the duration of the write, then restore it. + val callerThreadName = Thread.currentThread().getName + val task: Runnable = () => { + val current = Thread.currentThread() + val poolThreadName = current.getName + if (poolThreadName != callerThreadName) current.setName(callerThreadName) try body catch { case e: Throwable => System.err.println(s"[$clazzName] background log dispatch failed: ${e.getMessage}") } + finally if (current.getName != poolThreadName) current.setName(poolThreadName) + } try executor.execute(task) catch { case _: java.util.concurrent.RejectedExecutionException => diff --git a/obp-api/src/test/scala/code/util/MdcLoggableBoundedQueueTest.scala b/obp-api/src/test/scala/code/util/MdcLoggableBoundedQueueTest.scala index 9dddfc63fa..2790c7969e 100644 --- a/obp-api/src/test/scala/code/util/MdcLoggableBoundedQueueTest.scala +++ b/obp-api/src/test/scala/code/util/MdcLoggableBoundedQueueTest.scala @@ -59,6 +59,27 @@ class MdcLoggableBoundedQueueTest extends FlatSpec with Matchers { } finally ex.shutdownNow() } + it should "run on the pool thread under the caller's thread name, then restore the pool thread's name" 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)]() + val done = new CountDownLatch(1) + Helper.dispatchOn(ex, "test", critical = false) { + seen.set((Thread.currentThread(), Thread.currentThread().getName)); 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 Thread.currentThread().getName // but attributed to it + + val restored = new java.util.concurrent.atomic.AtomicReference[String]() + val done2 = new CountDownLatch(1) + ex.execute(() => { restored.set(Thread.currentThread().getName); done2.countDown() }) + done2.await(10, TimeUnit.SECONDS) shouldBe true + restored.get shouldBe poolThreadName + } 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 index 56de27238f..b9a1ed65fe 100644 --- a/obp-api/src/test/scala/code/util/MdcLoggableDispatchTest.scala +++ b/obp-api/src/test/scala/code/util/MdcLoggableDispatchTest.scala @@ -24,7 +24,7 @@ class MdcLoggableDispatchTest extends FlatSpec with Matchers { def probeDebug(msg: String): Unit = logger.debug(msg) } - "MdcLoggable" should "deliver an enabled log call to the underlying logger asynchronously, off the calling thread" in { + "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) @@ -57,8 +57,8 @@ class MdcLoggableDispatchTest extends FlatSpec with Matchers { captured should not be empty val event = captured.get(0) event.getFormattedMessage should include(marker) - event.getThreadName should not equal callingThreadName - event.getThreadName should startWith("mdc-log-dispatch-") + // The write runs on the pool, but the record must still name the thread that logged. + event.getThreadName shouldBe callingThreadName } finally { logbackLogger.detachAppender(appender) logbackLogger.setLevel(originalLevel) From 910fcc8c7d5d4f52b95a65eae1930d068cee723f Mon Sep 17 00:00:00 2001 From: Hongwei Date: Thu, 24 Sep 2026 16:11:18 +0200 Subject: [PATCH 07/14] fix: fall back to INFO for an unrecognised redis_logging_min_level The floor was parsed with a throwing valueOf while RedisLogger initialises, and MdcLoggable consults RedisLogger on every log call, so a mistyped value turned into failures at ordinary logging statements. Log the problem on stderr and use INFO instead. --- .../main/scala/code/api/cache/RedisLogger.scala | 17 ++++++++++++++++- .../api/cache/RedisLoggerShouldShipTest.scala | 16 ++++++++++++++++ 2 files changed, 32 insertions(+), 1 deletion(-) 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 5342d11a23..723cc58c5c 100644 --- a/obp-api/src/main/scala/code/api/cache/RedisLogger.scala +++ b/obp-api/src/main/scala/code/api/cache/RedisLogger.scala @@ -76,7 +76,22 @@ object RedisLogger { * restore the old "ship everything" behaviour. */ private val redisLoggingMinLevel: LogLevel.LogLevel = - LogLevel.valueOf(APIUtil.getPropsValue("redis_logging_min_level", "INFO")) + 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 diff --git a/obp-api/src/test/scala/code/api/cache/RedisLoggerShouldShipTest.scala b/obp-api/src/test/scala/code/api/cache/RedisLoggerShouldShipTest.scala index ca2b057bb8..782af6f960 100644 --- a/obp-api/src/test/scala/code/api/cache/RedisLoggerShouldShipTest.scala +++ b/obp-api/src/test/scala/code/api/cache/RedisLoggerShouldShipTest.scala @@ -58,4 +58,20 @@ class RedisLoggerShouldShipTest extends FlatSpec with Matchers { 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) + } } From ecfcd6f5b360c59a5407f61b6226a5296fdb0b0e Mon Sep 17 00:00:00 2001 From: Hongwei Date: Thu, 24 Sep 2026 16:11:37 +0200 Subject: [PATCH 08/14] fix: keep log entries written during shutdown and survive late pool creation After the shutdown hook closed the dispatch pool, INFO/DEBUG/TRACE entries from other shutdown hooks were counted as dropped instead of written; write them on the calling thread since nothing is overloaded at that point. Registering the hook also threw when the first log call of the process happened during shutdown; ignore that refusal. --- obp-api/src/main/scala/code/util/Helper.scala | 19 +++++++++++++------ .../util/MdcLoggableBoundedQueueTest.scala | 10 ++++++++++ 2 files changed, 23 insertions(+), 6 deletions(-) diff --git a/obp-api/src/main/scala/code/util/Helper.scala b/obp-api/src/main/scala/code/util/Helper.scala index c24efca7db..3f0101f131 100644 --- a/obp-api/src/main/scala/code/util/Helper.scala +++ b/obp-api/src/main/scala/code/util/Helper.scala @@ -345,11 +345,14 @@ object Helper extends Loggable { ) // 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. - Runtime.getRuntime.addShutdownHook(new Thread(() => { + // 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 } @@ -384,11 +387,15 @@ object Helper extends Loggable { try executor.execute(task) catch { case _: java.util.concurrent.RejectedExecutionException => - if (critical) task.run() - else { - 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") + 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") } } } diff --git a/obp-api/src/test/scala/code/util/MdcLoggableBoundedQueueTest.scala b/obp-api/src/test/scala/code/util/MdcLoggableBoundedQueueTest.scala index 2790c7969e..7e827a02d4 100644 --- a/obp-api/src/test/scala/code/util/MdcLoggableBoundedQueueTest.scala +++ b/obp-api/src/test/scala/code/util/MdcLoggableBoundedQueueTest.scala @@ -80,6 +80,16 @@ class MdcLoggableBoundedQueueTest extends FlatSpec with Matchers { } 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 "keep the production pool's queue bounded and observable" in { Helper.mdcLogQueueDepth should be >= 0 Helper.mdcLogDroppedCount should be >= 0L From 3308e02829d6dbb404e121ec99fd9d601a46c55c Mon Sep 17 00:00:00 2001 From: Hongwei Date: Thu, 24 Sep 2026 16:11:37 +0200 Subject: [PATCH 09/14] fix: carry the logging thread in the MDC instead of renaming the pool thread Renaming the pool thread around every write cost two native calls per entry and made thread dumps show pool threads under request thread names. Put the caller's name in the callerThread MDC key for the duration of the write, print it after %t in the default logback pattern, and read it in the Redis line format. --- obp-api/src/main/resources/logback.xml | 4 +++- obp-api/src/main/scala/code/util/Helper.scala | 22 +++++++++++-------- .../util/MdcLoggableBoundedQueueTest.scala | 17 ++++++++------ .../code/util/MdcLoggableDispatchTest.scala | 3 ++- 4 files changed, 28 insertions(+), 18 deletions(-) diff --git a/obp-api/src/main/resources/logback.xml b/obp-api/src/main/resources/logback.xml index 8044beb0bf..e7256d79ac 100644 --- a/obp-api/src/main/resources/logback.xml +++ b/obp-api/src/main/resources/logback.xml @@ -2,10 +2,12 @@ - %d{yyyy-MM-dd HH:mm:ss} %t %c{0} [%p] %m%n + %d{yyyy-MM-dd HH:mm:ss} %t %X{callerThread:-} %c{0} [%p] %m%n +