How to log request and response bodies in Spring WebFlux

This is what I came up with for java.

public class RequestResponseLoggingFilter implements WebFilter {

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        ServerHttpRequest httpRequest = exchange.getRequest();
        final String httpUrl = httpRequest.getURI().toString();

        ServerHttpRequestDecorator loggingServerHttpRequestDecorator = new ServerHttpRequestDecorator(exchange.getRequest()) {
            String requestBody = "";

            @Override
            public Flux<DataBuffer> getBody() {
                return super.getBody().doOnNext(dataBuffer -> {
                    try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream()) {
                        Channels.newChannel(byteArrayOutputStream).write(dataBuffer.asByteBuffer().asReadOnlyBuffer());
                        requestBody = IOUtils.toString(byteArrayOutputStream.toByteArray(), "UTF-8");
                        commonLogger.info(LogMessage.builder()
                                .step(httpUrl)
                                .message("log incoming http request")
                                .stringPayload(requestBody)
                                .build());
                    } catch (IOException e) {
                        commonLogger.error(LogMessage.builder()
                                .step("log incoming request for " + httpUrl)
                                .message("fail to log incoming http request")
                                .errorType("IO exception")
                                .stringPayload(requestBody)
                                .build(), e);
                    }
                });
            }
        };

        ServerHttpResponseDecorator loggingServerHttpResponseDecorator = new ServerHttpResponseDecorator(exchange.getResponse()) {
            String responseBody = "";
            @Override
            public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
                Mono<DataBuffer> buffer = Mono.from(body);
                return super.writeWith(buffer.doOnNext(dataBuffer -> {
                    try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream()) {
                        Channels.newChannel(byteArrayOutputStream).write(dataBuffer.asByteBuffer().asReadOnlyBuffer());
                        responseBody = IOUtils.toString(byteArrayOutputStream.toByteArray(), "UTF-8");
                        commonLogger.info(LogMessage.builder()
                                .step("log outgoing response for " + httpUrl)
                                .message("incoming http request")
                                .stringPayload(responseBody)
                                .build());
                    } catch (Exception e) {
                        commonLogger.error(LogMessage.builder()
                                .step("log outgoing response for " + httpUrl)
                                .message("fail to log http response")
                                .errorType("IO exception")
                                .stringPayload(responseBody)
                                .build(), e);
                    }
                }));
            }
        };
        return chain.filter(exchange.mutate().request(loggingServerHttpRequestDecorator).response(loggingServerHttpResponseDecorator).build());
    }

}

Since Spring Boot 2.2.x, Spring Webflux supports Kotlin coroutines. With coroutines, you can have the advantages of non-blocking calls without having to handle Mono and Flux wrapped objects. It adds extensions to ServerRequest and ServerResponse, adding methods like ServerRequest#awaitBody() and ServerResponse.BodyBuilder.bodyValueAndAwait(body: Any). So you could rewrite you code like this:

@Bean
fun apiRouter() = coRouter {
    (accept(MediaType.APPLICATION_JSON) and "/api").nest {
        "/user".nest {
            /* the handler methods now use ServerRequest and ServerResponse directly
             you just need to add suspend before your function declaration:
             suspend fun listUsers(ServerRequest req, ServerResponse res) */ 
            GET("/", userHandler::listUsers)
            POST("/{userId}", userHandler::updateUser)
        }
    }

    // this filter will be applied to all routes built by this coRouter
    filter { request, next ->
      // using non-blocking request.awayBody<T>()
      logger.info("Processing $request with body ${request.awaitBody<String>()}")
        val res = next(request)
        logger.info("Handling with Content-Type ${res.headers().contentType} and status code ${res.rawStatusCode()}")
        res 
    }
}

In order to create a WebFilter Bean with coRoutines, I think you can use this CoroutineWebFilter interface (I haven't tested it, I don't know if it works).

Here is the GitHub Repo with complete implementation to log both request and response body along with http headers for webflux/java based application...

Assuming we are dealing with a simple JSON or XML response, if debug level for corresponding loggers is not sufficient for some reason, one can use string representation before transforming it to object:

Mono<Response> mono = WebClient.create()
                               .post()
                               .body(Mono.just(request), Request.class)
                               .retrieve()
                               .bodyToMono(String.class)
                               .doOnNext(this::sideEffectWithResponseAsString)
                               .map(this::transformToResponse);

the following are the side-effect and transformation methods:

private void sideEffectWithResponseAsString(String response) { ... }
private Response transformToResponse(String response) { /*use Jackson or JAXB*/ }    

If your using controller instead of handler best way is aop with annotating you controller class with @Log annotation.And FYI this takes plain json object as request not mono.

@Target(AnnotationTarget.FUNCTION)
@Retention(AnnotationRetention.RUNTIME)
annotation class Log

@Aspect
@Component
class LogAspect {
    companion object {
        val log = KLogging().logger
    }

    @Around("@annotation(Log)")
    @Throws(Throwable::class)
    fun logAround(joinPoint: ProceedingJoinPoint): Any? {
        val start = System.currentTimeMillis()
        val result = joinPoint.proceed()
        return if (result is Mono<*>) result.doOnSuccess(getConsumer(joinPoint, start)) else result
    }

    fun getConsumer(joinPoint: ProceedingJoinPoint, start: Long): Consumer<Any>? {
        return Consumer {
            var response = ""
            if (Objects.nonNull(it)) response = it.toString()
            log.info(
                "Enter: {}.{}() with argument[s] = {}",
                joinPoint.signature.declaringTypeName, joinPoint.signature.name,
                joinPoint.args
            )
            log.info(
                "Exit: {}.{}() had arguments = {}, with result = {}, Execution time = {} ms",
                joinPoint.signature.declaringTypeName, joinPoint.signature.name,
                joinPoint.args[0],
                response, System.currentTimeMillis() - start
            )
        }
    }
}

I think the appropriate thing to do here is to write the contents of each request to a file in an asynchronous manner (java.nio) and set up an interval that reads those request body files asynchrolusly and writes them to the log in a memory usage aware manner (atleast one file at a time but up too 100 mb at a time) and after logging them removes the files from disk.

Ivan Lymar's answer but in Kotlin:

import org.apache.commons.io.IOUtils
import org.reactivestreams.Publisher
import org.springframework.core.io.buffer.DataBuffer
import org.springframework.http.server.reactive.ServerHttpRequestDecorator
import org.springframework.http.server.reactive.ServerHttpResponseDecorator
import org.springframework.stereotype.Component
import org.springframework.web.server.ServerWebExchange
import org.springframework.web.server.WebFilter
import org.springframework.web.server.WebFilterChain
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.io.ByteArrayOutputStream
import java.io.IOException
import java.nio.channels.Channels

@Component
class LoggingWebFilter : WebFilter {

    override fun filter(exchange: ServerWebExchange, chain: WebFilterChain): Mono<Void> {
        val httpRequest = exchange.request
        val httpUrl = httpRequest.uri.toString()
        val loggingServerHttpRequestDecorator: ServerHttpRequestDecorator =
            object : ServerHttpRequestDecorator(exchange.request) {
                var requestBody = ""
                override fun getBody(): Flux<DataBuffer> {
                    return super.getBody().doOnNext { dataBuffer: DataBuffer ->
                        try {
                            ByteArrayOutputStream().use { byteArrayOutputStream ->
                                Channels.newChannel(byteArrayOutputStream)
                                    .write(dataBuffer.asByteBuffer().asReadOnlyBuffer())
                                requestBody =
                                    IOUtils.toString(
                                        byteArrayOutputStream.toByteArray(),
                                        "UTF-8"
                                    )
                                log.info(
                                    "Logging Request Filter: {} {}",
                                    httpUrl,
                                    requestBody
                                )
                            }
                        } catch (e: IOException) {
                            log.error(
                                "Logging Request Filter Error: {} {}",
                                httpUrl,
                                requestBody,
                                e
                            )
                        }
                    }
                }
            }

        val loggingServerHttpResponseDecorator: ServerHttpResponseDecorator =
            object : ServerHttpResponseDecorator(exchange.response) {
                var responseBody = ""
                override fun writeWith(body: Publisher<out DataBuffer>): Mono<Void> {
                    val buffer: Mono<DataBuffer> = Mono.from(body)
                    return super.writeWith(
                        buffer.doOnNext { dataBuffer: DataBuffer ->
                            try {
                                ByteArrayOutputStream().use { byteArrayOutputStream ->
                                    Channels.newChannel(byteArrayOutputStream)
                                        .write(
                                            dataBuffer
                                                .asByteBuffer()
                                                .asReadOnlyBuffer()
                                        )
                                    responseBody = IOUtils.toString(
                                        byteArrayOutputStream.toByteArray(),
                                        "UTF-8"
                                    )
                                    log.info(
                                        "Logging Response Filter: {} {}",
                                        httpUrl,
                                        responseBody
                                    )
                                }
                            } catch (e: Exception) {
                                log.error(
                                    "Logging Response Filter Error: {} {}",
                                    httpUrl,
                                    responseBody,
                                    e
                                )
                            }
                        }
                    )
                }
            }
        return chain.filter(
            exchange.mutate().request(loggingServerHttpRequestDecorator)
                .response(loggingServerHttpResponseDecorator)
                .build()
        )
    }
}