Logging Request Entity Body with Quarkus Resteasy Reactive

[KyriacosP]

Describe the bug

I'm trying to implement a logging filter to log the request and response to the API endpoints of my Quarkus application. I'm using Quarkus 1.13.3.Final and quarkus-resteasy-reactive. I have a problem trying to log the request body when calling a non blocking endpoint. This is the code I'm using to log the request:

    @ServerRequestFilter(priority = 0)
    public void getFilter(UriInfo info,HttpServerRequest request,ContainerRequestContext ctx) {
        String uuid = UUID.randomUUID().toString();
        ctx.setProperty("log_id", uuid);
        String body = new BufferedReader(new InputStreamReader(ctx.getEntityStream())).lines().collect(Collectors.joining("\n"));
        ctx.setEntityStream(new ByteArrayInputStream(body.getBytes()));
        logger.info("Request: " + uuid + " Method: "+ ctx.getMethod() + " Path: " + info.getPath() + " Remote Address: " +  request.remoteAddress().toString() + " Body: " + body);
    }

Expected behavior

Log the request body for both blocking and non blocking endpoints

Actual behavior

This works fine when I'm calling an API endpoint that has the @Blocking annotations but when I call a non blocking API I get the following error:

2021-05-17 10:15:29,159 ERROR [org.jbo.res.rea.ser.cor.ExceptionMapping] (vert.x-eventloop-thread-10) Request failed : java.io.UncheckedIOException: java.io.IOException: Attempting a blocking read on io thread
        at java.base/java.io.BufferedReader$1.hasNext(BufferedReader.java:577)
        at java.base/java.util.Iterator.forEachRemaining(Iterator.java:132)
        at java.base/java.util.Spliterators$IteratorSpliterator.forEachRemaining(Spliterators.java:1801)
        at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484)
        at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
        at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:913)
        at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
        at java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:578)
        at com.filters.LogginFilter.getFilter(LogginFilter.java:45)
        at com.filters.LogginFilter$GeneratedServerRequestFilter$getFilter.filter(LogginFilter$GeneratedServerRequestFilter$getFilter.zig:73)
        at com.filters.LogginFilter$GeneratedServerRequestFilter$getFilter_Subclass.filter$$superaccessor1(LogginFilter$GeneratedServerRequestFilter$getFilter_Subclass.zig:201)
        at com.filters.LogginFilter$GeneratedServerRequestFilter$getFilter_Subclass$$function$$3.apply(LogginFilter$GeneratedServerRequestFilter$getFilter_Subclass$$function$$3.zig:33)
        at io.quarkus.arc.impl.AroundInvokeInvocationContext.proceed(AroundInvokeInvocationContext.java:54)
        at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.proceed(InvocationInterceptor.java:63)
        at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.monitor(InvocationInterceptor.java:49)
        at io.quarkus.arc.runtime.devconsole.InvocationInterceptor_Bean.intercept(InvocationInterceptor_Bean.zig:521)
        at io.quarkus.arc.impl.InterceptorInvocation.invoke(InterceptorInvocation.java:41)
        at io.quarkus.arc.impl.AroundInvokeInvocationContext.perform(AroundInvokeInvocationContext.java:41)
        at io.quarkus.arc.impl.InvocationContexts.performAroundInvoke(InvocationContexts.java:32)

Is this the expected behavior or is this a bug? If this is the expected behavior for non blocking endpoints is there a way to log the incoming request body (without going into every individual endpoint and logging the body) ?

Output of java -version

Java version: 11.0.9, vendor: Oracle Corporation,

Quarkus version or git rev

Quarkus 1.13.3.Final

Build tool (ie. output of mvnw --version or gradlew --version)

Apache Maven 3.6.3

[quarkus-bot]

/cc @FroMage, @geoand, @stuartwdouglas

[geoand]

I am changing this to a feature request as the JAX-RS way of reading the entire body is definitely blocking, but it does make sense to provide an alternative that will allow users to read the body in request filters

[KyriacosP]

Thank you!

[geoand]

👍🏼

[geoand]

I was thinking about this and I actually would like to know more about the use case.
How come you need to log the request body in a request filter? What's the end result you are trying to achieve)

[KyriacosP]

Our main use case is business related, having logs for the request and response of API calls enables the support team to help customers and integrators identify issues more easily and also makes the integration between different services implemented by different teams easier.

[geoand]

So what kind of information do these logs contain? Just generic HTTP related info?

[KyriacosP]

They contain the endpoint, info about the caller, path and query parameters, the request body and an identifier for each request. For the response the contain the response body and the identifier.

[geoand]

I can't shake the feeling that this is something that we could do ourselves and not expose.

@FroMage @stuartwdouglas WDYT?

[FroMage]

I'm not sure, I guess it depends on how filters and reader interceptors interact. I don't think we want to invoke reader interceptors while in the middie of the first filter.

[AmsterdamFilho]

+1

[tsfullman]

+1

[PvlPech]

+1

[stuartwdouglas]

I think the current solution would be to inject the org.jboss.resteasy.reactive.server.spi.ServerHttpRequest and call setReadListener to do a non-blocking read.

[PvlPech]

Can't we just use io.vertx.core.Future processing to resolve the initial issue? So, in @KyriacosP's example it's possible to add something like

Future<Buffer> body = request.body();
        body.onComplete(ar -> {
            if (ar.failed()) {
                /*...*/
            } else {
                logger.info("Request body: " + ar.result());
            }
        });

[krakzk]

+1

[GavinRay97]

+1

[silviu-negoita]

Also needed. Trying to inject org.jboss.resteasy.reactive.server.spi.ServerHttpRequest in my filters but I got javax.enterprise.inject.spi.DeploymentException: javax.enterprise.inject.UnsatisfiedResolutionException: Unsatisfied dependency for type org.jboss.resteasy.reactive.server.spi.ServerHttpRequest and qualifiers [@Default]

[GavinRay97]

PvlPech's answer works perfectly, this is what we have in our repo.
(We use resteasy-reactive, if it makes a difference)

Might be worth making a FAQ/documentation entry on.
Or creating a StackOverflow question and self-answering so others can find:

import io.vertx.core.http.HttpServerRequest
import org.jboss.logging.Logger
import org.jboss.resteasy.reactive.server.ServerRequestFilter
import javax.inject.Inject
import javax.ws.rs.container.ContainerRequestContext
import javax.ws.rs.core.UriInfo

class Filters {
    @Inject
    private lateinit var logger: Logger

    @ServerRequestFilter(priority = 0)
    fun logBodyFilter(info: UriInfo, request: HttpServerRequest, ctx: ContainerRequestContext) {
        request.body { buffer ->
            logger.debug("Request body: ${buffer.result()}")
        }
    }
}

[silviu-negoita]

@GavinRay97 I tried this solution but my body handler is never called (your logger in the example)

My code:

import io.vertx.core.http.HttpServerRequest;
import lombok.extern.slf4j.Slf4j;
import org.jboss.resteasy.reactive.server.ServerRequestFilter;

import javax.ws.rs.container.ContainerRequestContext;

@Slf4j
public class RequestFilters {


    @ServerRequestFilter(priority = 0)
    public void userInfoFilter(HttpServerRequest request, ContainerRequestContext containerRequestContext) {
        request.body(buffer -> {
            log.info("body {}", buffer.result());
        });
    }
}

[silviu-negoita]

Update: I found the issue. Apparently if you are not consuming the body in your rest controller, handler is never invoked. Not sure if this is a bug or a feature. But nice to know.

[ngannt1710]

I use Future bufferFuture = request.body();
bufferFuture.onComplete(bufferAsyncResult -> {} to read request body. But when I modify and set again to request body with context.setEntityStream(new ByteArrayInputStream(data.getBytes())); It is not working in async function

[csotiriou]

@stuartwdouglas can you please mention how we can inject ServerHttpRequest?

I get Parameter 'request' of method 'filter of class 'com.example.ModuleFilters' is not allowed when I'm trying to inject org.jboss.resteasy.reactive.server.spi.ServerHttpRequest to my resteasy reactive filter method.

Edit: Never mind, I injected it using var request = CurrentRequestManager.get().serverRequest();. However, it's never called, since Quarkus sets it again in another place in the chain, thus overriding the behavior I want.

[RamonaBarbieru]

Confirming that this feature is needed on our project.

[miladjafary]

We also need this feature 👍

[miladjafary]

the @GavinRay97 and @PvlPech workaround has also worked form me. Thanks guys.

The only issue is that the GlobalTracer.get().activeSpan() is return null so I could not make a trace :( this is my entire logger class

class HttpRequestFilter(private val logger: Logger) {

    @ServerRequestFilter(preMatching = true)
    fun logRequest(request: HttpServerRequest, requestContext: ContainerRequestContext) {
        val span = GlobalTracer.get().activeSpan()

        incomingLog("${requestContext.method} ${requestContext.uriInfo.requestUri.path}")
        requestContext.headers?.forEach { header ->
            val headerValue = getMaskValueIfHeaderIsSensitive(header)
            span?.setTag("http.headers.${header.key}", headerValue)
            incomingLog(header.toString())
        }

        request.body { buffer ->
            val requestBodyAsString = buffer.result().toString()
            span?.setTag("http.requestBody", requestBodyAsString)
            incomingLog(requestBodyAsString)
        }
    }

    private fun getMaskValueIfHeaderIsSensitive(header: Map.Entry<String, MutableList<String>>): String {
        return if (HttpHeaders.AUTHORIZATION.equals(header.key, true)) "****" else header.value.toString()
    }

    private fun incomingLog(log: String) {
        logger.info("http-incoming << $log")
    }
}

Update:
Instead of using @ServerRequestFilter(preMatching = true) I used @ServerRequestFilter and seems in this context the GlobalTracer.get().activeSpan() has return the span and it's not null.

[DudekJakub]

Hello guys.

I have very similar case but when it comes to my code: I need to obtain request body right after any exception is thrown so in result I'll have logged out only request bodies from failures.

What I've tried so far is using ContainerRequestContext like this:

@ServerExceptionMapper
   fun mapException(exception: Throwable, requestContext: ContainerRequestContext): Uni<Response> {
       filter(requestContext)
       return Uni.createFrom().nullItem() // ignore it
   }

   private fun filter(requestContext: ContainerRequestContext) {
       Vertx.vertx().executeBlocking { promise ->
           try {
               val text = requestContext.entityStream.bufferedReader().use { it.readText() }
               logger.info { "Size of entity stream: ${text.length}" }
               promise.complete(text)
           } catch (e: Exception) {
               promise.fail(e)
           }
       }.onComplete { println(it.result().toString()) }
   } 

but the problem is that at this point request body is already consumed by JAX-RS so in result log always shows " Size of entity stream: 0"

Then I tried some workaround and even tho it works as expected, it has some vulnerabilities:

companion object {
     val map: MutableMap<String, AsyncResult<Buffer>> = mutableMapOf()
 }

@ServerRequestFilter(priority = 0)
 fun populateMap(request: HttpServerRequest) {
     val id = request.headers()[ID] // all of my requests have ID header but even with it, due to some business cases it is not best to correlate it with particular requestBody
     request.body().onComplete { map[id] = it }
 }

 @Scheduled(every = "1m") // ignore time, it is just for testing
 fun cleanMap() {
     logger.info { "Cleaning the map [$map] right now..." }
     map.clear()
     logger.info { "Map has been cleaned up [$map]" }
 }
 

then I can easily read requestBody from the map wherever I want but still, it doesn't look like the most optimal solution. Imagine somebody sends 2 requests with identical ID in header but different requestBody -> there is non-0 chance that in my map key-value will be replaced and in result logged requestBody won't match for given request. Also it requires alocating some memories for my map...

Do you have any tips/solutions I can try to log request body only when response is failure (any exception thrown)? I'd be glad for any help. Thanks in advance.

[FroMage]

@geoand didn't you work on logging http requests in RR in the past? I can't find mention of it in https://quarkus.io/guides/resteasy-reactive

[geoand]

I don't think we ever came to a good solution

[FroMage]

#22213 was the issue. Would this sort of logging solve your use-cases? I'm asking all the +1 people.

[hungchu0912]

+1

[nitinty]

Hello Guys,

I’ve implemented the following to log the request body. Could you guys please check it, if it meets the requirements or helps ?

    @ServerRequestFilter
    public Uni<Void> filter(ResteasyReactiveContainerRequestContext requestContext) {

        return readRequestBody(requestContext)
                .onItem().invoke(body -> log.debug("Request body - {} received in filter for uri - {}", body, requestContext.getUriInfo().getRequestUri()))
                .replaceWithVoid();

    }

    private Uni<String> readRequestBody(ResteasyReactiveContainerRequestContext context) {
        if (isMultiPartRequest(context)) {
            return Uni.createFrom().item(""); // Return empty string for multipart requests
        }
        return Uni.createFrom().item(() -> {
                    // Blocking code to read the request body
                    try {
                        byte[] body = context.getEntityStream().readAllBytes();
                        context.setEntityStream(new ByteArrayInputStream(body)); // Reset stream
                        return new String(body, StandardCharsets.UTF_8);
                    } catch (Exception e) {
                        throw new ServiceRuntimeException(e);
                    }
                })
                .runSubscriptionOn(Infrastructure.getDefaultWorkerPool()); // Offload to worker thread
    }

    private boolean isMultiPartRequest(ResteasyReactiveContainerRequestContext context) {
        MediaType mediaType = context.getMediaType();
        return mediaType != null && mediaType.isCompatible(MediaType.MULTIPART_FORM_DATA_TYPE);
    }
    

[FroMage]

That's not the same at all as #22213 but if it works for you that's great :)

[csotiriou]

I believe that we have found a solution for this, which works under all circumstances. The solution for us seemed to be the usage of ReaderInterceptor and WriterInterceptor for retrieving the request / response instead of relying on the filters to do this.

• They are synchronous by default, even for Resteasy Reactive
• They are the last thing to be called after the filters and everything, and you can easily pass properties from the filters to the Interceptors.

The following is a snippet / pseudocode. The concept is to catch the body before it is being read by / sent to the client, read the streams, and replace them with a new one so that the context can continue like before.

public class BodyLoggerInterceprot implements ReaderInterceptor, WriterInterceptor {
    private final MyLogger logger;
    
    @Override
    public void aroundWriteTo(WriterInterceptorContext context) throws IOException, WebApplicationException {
        //retrieve the logging body
        var originalStream = context.getOutputStream();
        try (var baos = new LoggingBufferOutputStream(originalStream)) {
            context.setOutputStream(baos);
            context.proceed();
            var output = baos.toByteArray();
            logTransaction.setPayload(new String(output, StandardCharsets.UTF_8)); // <-- this is the body!
        } catch (Exception e) {
            //not sure about that, we need to check
            context.setOutputStream(originalStream);
        }
    }

    @Override
    public Object aroundReadFrom(ReaderInterceptorContext context) throws IOException, WebApplicationException {
        InputStream originalInputStream = context.getInputStream();
        // Convert InputStream to String
        String body = Streams.toString(originalInputStream); // <-- This is the line that reads the body!
        // Reset the stream so it can be read again in the actual processing
        context.setInputStream(new ByteArrayInputStream(body.getBytes()));
        // Proceed with the rest of the interceptor chain
    }
}

Two notes:

Usage of LoggingBufferOutputStream

this is required because of this issue: #41051 . So instead of something else, we are using a filtered stream to ensure that the original is not tampered with inside Quarkus.

Here is the code for your convenience:

/**
 * A {@link FilterOutputStream} that buffers all the data written to it.
 *
 *     The data can be retrieved using {@link #toByteArray()}.
 *     The buffer is closed when this stream is closed.
 * This is used to log the payload of a request or response, while at the
 * same time allowing the request or response to be processed as usual.
 */
public class LoggingBufferOutputStream extends FilterOutputStream {
    private final ByteArrayOutputStream buffer;

    public LoggingBufferOutputStream(OutputStream out) {
        super(out);
        this.buffer = new ByteArrayOutputStream();
    }

    @Override
    public void write(byte b[]) throws IOException {
        buffer.write(b);
        super.write(b);
    }

    public byte[] toByteArray() {
        return buffer.toByteArray();
    }

    @Override
    public void close() throws IOException {
        buffer.close();
        super.close();
    }

    @Override
    public void flush() throws IOException {
        buffer.flush();
        super.flush();
    }
}

Retrieving information from the ContainerRequestContext inside the Interceptors

A very nice feature that we have is that whatever you pass to the ContainerRequestContext with setProperty() you will still have it in the ReaderInterceptorContext and WriterInterceptorContext.

Which means that you can prepare all things to be logged in the ContainerRequestFilter, pass it to the context with setProperty() and finalize the logging (obtain the body, the properties, and log the entire thing) in the ReaderInterceptors and WriterInterceptors.

The above interceptors work for both the Incoming server requests / responses, and the outbound client requests / responses (microprofile rest client).

Caveats

Of course, since there is a duplicate read / write of the request body, there is a performance overhead, which is analogous to the size of the bodies that are being received by / sent to the REST clients. However, I suppose all logging methods for bodies would have a performance impact anyway in case of long bodies.

[geoand]

Thanks for sharing @csotiriou!