ExchangeLogEmitter.kt
package eu.inqudium.limesium.reactive.logging
import eu.inqudium.limesium.common.EndpointLogField
import eu.inqudium.limesium.common.EndpointLoggingMetrics
import eu.inqudium.limesium.common.ExchangeLine
import eu.inqudium.limesium.common.HeaderValueMasker
import eu.inqudium.limesium.common.MdcScope
import eu.inqudium.limesium.common.NanoTimeSource
import eu.inqudium.limesium.common.addKeyValue
import eu.inqudium.limesium.common.addKeyValueIfPresent
import eu.inqudium.limesium.common.failOpen
import eu.inqudium.limesium.common.setCauseIfPresent
import org.slf4j.LoggerFactory
import org.slf4j.event.Level
import java.time.Duration
/**
* Builds and emits the log events of an exchange - the arrival line and the completion event - with the
* IDENTICAL message and field format of the limesium-servlet-logging emitter. The stack-neutral core
* (message texts, header rendering, the arrival line, the body measurements) is the shared
* [ExchangeLine]; what this class owns is what differs on this stack: the disposition vocabulary
* (`cancelled` where the servlet twin has `timeout`, no `endpoint_async` - everything is asynchronous
* here, the flag would carry no information), the status that may be absent, and the response reads
* against the reactive response.
*
* ## Levels
*
* The level carries severity only, `endpoint_outcome` the semantic: ERROR when the chain errored, WARN
* for a 5xx, a cancellation, or an exchange that reached
* [RequestLoggingProperties.slowRequestThreshold], INFO otherwise. Severity and outcome are resolved
* BEFORE the event is built, so a disabled level costs no assembly.
*
* ## Fail-open
*
* The guard covers everything after the exactly-once CAS: a failure inside the emission is reported on
* this class's own logger, counted on the [EndpointLoggingMetrics] fail-open counter, and an interrupt
* is re-raised as a flag. Requests are never affected.
*/
internal class ExchangeLogEmitter(
private val properties: RequestLoggingProperties,
private val nanoTime: NanoTimeSource,
private val metrics: EndpointLoggingMetrics,
private val masker: HeaderValueMasker,
) {
private val exchangeLog = LoggerFactory.getLogger(properties.loggerName)
/** The optional arrival line ([RequestLoggingProperties.logRequestStart]) - the shared [ExchangeLine.logRequestStart]. */
fun logRequestStart(exchange: Exchange) = ExchangeLine.logRequestStart(exchangeLog, internalLog, metrics, exchange)
/**
* The single emission point, called exactly once per exchange from `ExchangeLifecycle.complete` -
* the one place that wins the [Exchange.state] transition to `COMPLETED` (terminal signal, or the
* commit callback for the deferred error case - see the filter).
*/
fun logExchange(exchange: Exchange) {
failOpen(
onInterrupted = { e ->
metrics.emissionFailure()
internalLog.debug("Interrupted while logging an exchange; the event is dropped", e)
},
onFailure = { e ->
metrics.emissionFailure()
internalLog.error(
"Exception while logging exchange {} {}: {}",
exchange.method,
exchange.path,
e.toString(),
e,
)
},
) {
emitExchange(exchange)
}
}
private fun emitExchange(exchange: Exchange) {
// Freeze FIRST: from here on a late body chunk (an onNext still in flight after a cancellation)
// can no longer move the captures - body text and size sample are one consistent snapshot.
exchange.requestCapture?.freeze()
exchange.responseCapture?.freeze()
val elapsedNanos = nanoTime.nanoTime() - exchange.startNanos
val durationMs = elapsedNanos / ExchangeLine.NANOS_PER_MS
val failure = exchange.failure
val cancelled = exchange.cancelled
// The commit-time status is authoritative (the error path defers emission until the upstream
// handler rendered); a cancelled exchange may never have committed - status then stays null, the
// message shows "-" and the status field is omitted rather than invented.
val status: Int? = exchange.committedStatus ?: exchange.response.statusCode?.value()
// Compared at full precision and overflow-free (Duration comparison, no toMillis/toNanos
// truncation): a 1.5 ms threshold must not flag a 1 ms exchange. The logged duration field keeps
// its millisecond resolution, which is why the properties reject thresholds below 1 ms.
val slow = Duration.ofNanos(elapsedNanos) >= properties.slowRequestThreshold
ExchangeLine.recordBodySizesQuietly(
internalLog,
metrics,
exchange,
exchange.pathTemplate,
exchange.requestCapture,
exchange.responseCapture,
properties.measureRequestBodySize,
properties.measureResponseBodySize,
)
// Severity and semantic decoupled, exactly like the servlet twin - with `cancelled` where the
// servlet stack has `timeout`: an error signal is ERROR, a 5xx without one is WARN (the
// application already handled it), a client disconnect is WARN; slow escalates INFO -> WARN
// without changing the outcome.
val (baseLevel, outcome) =
when {
failure != null -> Level.ERROR to EndpointLoggingMetrics.OUTCOME_FAILURE
cancelled -> Level.WARN to EndpointLoggingMetrics.OUTCOME_CANCELLED
(status ?: 0) >= 500 -> Level.WARN to EndpointLoggingMetrics.OUTCOME_FAILURE
else -> Level.INFO to EndpointLoggingMetrics.OUTCOME_SUCCESS
}
val level = if (slow && baseLevel == Level.INFO) Level.WARN else baseLevel
if (!exchangeLog.isEnabledForLevel(level)) {
return
}
// The emission scope carries the exchange identity and the traceparent-derived trace context into
// the MDC, so a structured encoder emits them as fields; the message repeats the gist inline for
// plain-text appenders - identical to the servlet twin. The scope OWNS the trace keys (a bridge's
// spanId included): under Boot's default `limited` propagation the emitting event-loop or commit
// thread carries no bridge MDC, but under `spring.reactor.context-propagation=auto` - the mode the
// handler-MDC parity asks for - Micrometer's ObservationThreadLocalAccessor restores the server
// span's traceId/spanId around this operator, and a traceless exchange would otherwise inherit a
// trace the event must not carry (ADR-0002). `use` restores the scope and records a close-time
// failure as suppressed instead of masking an emission failure (both land in logExchange's guard
// either way).
MdcScope(
exchange.requestId,
exchange.method,
exchange.path,
exchange.traceId,
exchange.parentSpanId,
ownsTraceKeys = true,
).use {
// Multi-value resolution, natively from the reactive HttpHeaders.
val responseHeaders =
properties.responseHeaders.select(exchange.response.headers.headerNames(), masker) { name ->
exchange.response.headers[name]
?.takeIf { it.isNotEmpty() }
?.joinToString(", ")
}
// Body fields only when the direction's mode admits THIS outcome: `on-failure` captured the
// bytes (the outcome is unknown while they flow) and discards them here for a clean
// exchange (success outcome, no 4xx). A capture may also exist in count-only mode for the size metrics, and its empty buffer
// must not surface as a truncated-looking field.
val failed = outcome != EndpointLoggingMetrics.OUTCOME_SUCCESS || (status ?: 0) in 400..499
val requestBody =
if (properties.logRequestBody.logs(failed)) exchange.requestCapture?.loggedValue(exchange.requestCharset) else null
val responseBody =
if (properties.logResponseBody.logs(failed)) {
exchange.responseCapture?.loggedValue(exchange.response.headers.declaredCharsetOrUtf8())
} else {
null
}
// One immutable builder chain; optional fields are left off by the *IfPresent helpers. Both
// halves of the path pair, query as its own field - identical to the servlet twin.
exchangeLog
.atLevel(level)
.setMessage(ExchangeLine.exchangeMessage(exchange, status?.toString() ?: "-"))
.addKeyValue(EndpointLogField.OUTCOME, outcome)
.addKeyValue(EndpointLogField.DURATION_MS, durationMs)
.addKeyValue(EndpointLogField.REQUEST_METHOD, exchange.method)
.addKeyValue(EndpointLogField.URL_PATH, exchange.path)
.addKeyValueIfPresent(EndpointLogField.RESPONSE_STATUS_CODE, status)
.setCauseIfPresent(failure)
.addKeyValueIfPresent(EndpointLogField.SLOW, true.takeIf { slow })
.addKeyValueIfPresent(EndpointLogField.URL_TEMPLATE, exchange.pathTemplate)
.addKeyValueIfPresent(EndpointLogField.URL_QUERY, exchange.query)
.addKeyValueIfPresent(EndpointLogField.REQUEST_HEADERS, ExchangeLine.renderHeaders(exchange.requestHeaders))
.addKeyValueIfPresent(EndpointLogField.RESPONSE_HEADERS, ExchangeLine.renderHeaders(responseHeaders))
.addKeyValueIfPresent(EndpointLogField.REQUEST_BODY, requestBody)
.addKeyValueIfPresent(EndpointLogField.RESPONSE_BODY, responseBody)
.log()
// Guarded inside the metrics: a throwing host counter after a successful log() must not be
// reported as a lost emission.
metrics.eventEmitted(outcome)
}
}
companion object {
// Failures of the logging itself go to the module's own logger, never onto the exchange logger -
// the exchange log stream stays parseable.
private val internalLog = LoggerFactory.getLogger(ExchangeLogEmitter::class.java)
}
}