Introduce RequestLogListener

core/src/main/java/com/linecorp/armeria/client/observation/ObservationClient.java

@@ -29,7 +29,6 @@
 import com.linecorp.armeria.common.RequestHeadersBuilder;
 import com.linecorp.armeria.common.annotation.Nullable;
 import com.linecorp.armeria.common.annotation.UnstableApi;
-import com.linecorp.armeria.common.logging.RequestLogProperty;
 import com.linecorp.armeria.internal.common.RequestContextExtension;
 import com.linecorp.armeria.server.observation.ObservationService;
 
@@ -138,24 +137,26 @@ public HttpResponse execute(ClientRequestContext ctx, HttpRequest req) throws Ex
     private static void enrichObservation(ClientRequestContext ctx,
                                           ClientObservationContext clientObservationContext,
                                           Observation observation) {
-        ctx.log()
-           .whenAvailable(RequestLogProperty.REQUEST_FIRST_BYTES_TRANSFERRED_TIME)
-           .thenAccept(requestLog -> observation.event(Events.WIRE_SEND));
-
-        ctx.log()
-           .whenAvailable(RequestLogProperty.RESPONSE_FIRST_BYTES_TRANSFERRED_TIME)
-           .thenAccept(requestLog -> {
-               if (requestLog.responseFirstBytesTransferredTimeNanos() != null) {
-                   observation.event(Events.WIRE_RECEIVE);
-               }
-           });
-
-        ctx.log().whenComplete()
-           .thenAccept(requestLog -> {
-               // TODO: ClientConnectionTimings - there is no way to record events
-               // with a specific timestamp for an observation
-               clientObservationContext.setResponse(requestLog);
-               observation.stop();
-           });
+        ctx.log().addListener((property, log) -> {
+            switch (property) {
+                case REQUEST_FIRST_BYTES_TRANSFERRED_TIME:
+                    observation.event(Events.WIRE_SEND);
+                    break;
+                case RESPONSE_FIRST_BYTES_TRANSFERRED_TIME:
+                    if (log.responseFirstBytesTransferredTimeNanos() != null) {
+                        observation.event(Events.WIRE_RECEIVE);
+                    }
+                    break;
+                case ALL_COMPLETE:
+                    // TODO: ClientConnectionTimings - there is no way to record events
+                    // with a specific timestamp for an observation
+                    clientObservationContext.setResponse(log);
+                    observation.stop();
+                    break;
+                default:
+                    // Do nothing.
+                    break;
+            }
+        });
     }
 }

core/src/main/java/com/linecorp/armeria/common/logging/IdempotentRequestLogListener.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2025 LY Corporation
+ *
+ * LY Corporation licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.linecorp.armeria.common.logging;
+
+import static com.linecorp.armeria.common.logging.DefaultRequestLog.hasInterestedFlags;
+
+import com.linecorp.armeria.internal.common.util.ReentrantShortLock;
+
+/**
+ * A {@link RequestLogListener} that delivers each event only once to the delegate listener.
+ */
+final class IdempotentRequestLogListener extends ReentrantShortLock implements RequestLogListener {
+
+    private static final long serialVersionUID = -573237359665852226L;
+
+    private final RequestLogListener delegate;
+    private int notifiedFlags;
+
+    IdempotentRequestLogListener(RequestLogListener delegate) {
+        this.delegate = delegate;
+    }
+
+    @Override
+    public void onEvent(RequestLogProperty property, RequestLog log) {
+        lock();
+        try {
+            if (hasInterestedFlags(notifiedFlags, property)) {
+                // Already notified.
+                return;
+            }
+            notifiedFlags |= property.flag();
+        } finally {
+            unlock();
+        }
+        delegate.onEvent(property, log);
+    }
+}

core/src/main/java/com/linecorp/armeria/common/logging/RequestLogAccess.java

@@ -230,6 +230,13 @@ public interface RequestLogAccess {
      */
     RequestLog ensureAvailable(Iterable<RequestLogProperty> properties);
 
+    /**
+     * Adds the specified {@link RequestLogListener} which will be invoked when a {@link RequestLogProperty}
+     * becomes available.
+     */
+    @UnstableApi
+    void addListener(RequestLogListener listener);
+
     /**
      * Returns the {@link RequestLog} for the {@link Request}, where all properties may not be available yet.
      * Note that this method is potentially unsafe; an attempt to access an unavailable property will trigger

core/src/main/java/com/linecorp/armeria/common/logging/RequestLogListener.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2025 LY Corporation
+ *
+ * LY Corporation licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.linecorp.armeria.common.logging;
+
+import com.linecorp.armeria.common.annotation.UnstableApi;
+
+/**
+ * A listener that listens to all events of a {@link RequestLog}.
+ *
+ * <p>If a {@link RequestLogProperty} was completed before adding this listener to the {@link RequestLog},
+ * the {@link #onEvent(RequestLogProperty, RequestLog)} method will be invoked immediately with the already
+ * completed property upon adding the listener.
+ *
+ * <p>Note that this listener may be invoked in the I/O worker thread so make sure to offload any blocking
+ * operations to a separate thread pool.
+ */
+@UnstableApi
+@FunctionalInterface
+public interface RequestLogListener {
+
+    /**
+     * Invoked when the specified {@link RequestLogProperty} is completed.
+     */
+    void onEvent(RequestLogProperty property, RequestLog log);
+}

core/src/main/java/com/linecorp/armeria/common/logging/RequestLogProperty.java

@@ -37,16 +37,6 @@ public enum RequestLogProperty {
      */
     REQUEST_START_TIME(true),
 
-    /**
-     * {@link RequestLog#requestEndTimeNanos()}, {@link RequestLog#requestDurationNanos()}.
-     */
-    REQUEST_END_TIME(true),
-
-    /**
-     * {@link RequestLog#requestFirstBytesTransferredTimeNanos()}.
-     */
-    REQUEST_FIRST_BYTES_TRANSFERRED_TIME(true),
-
     /**
      * {@link RequestLog#channel()}, {@link RequestLog#sessionProtocol()}, {@link RequestLog#sslSession()},
      * {@link RequestLog#connectionTimings()}.
@@ -73,6 +63,11 @@ public enum RequestLogProperty {
      */
     REQUEST_HEADERS(true),
 
+    /**
+     * {@link RequestLog#requestFirstBytesTransferredTimeNanos()}.
+     */
+    REQUEST_FIRST_BYTES_TRANSFERRED_TIME(true),
+
     /**
      * {@link RequestLog#requestContent()}, {@link RequestLog#rawRequestContent()}.
      */
@@ -98,30 +93,40 @@ public enum RequestLogProperty {
      */
     REQUEST_CAUSE(true),
 
+    /**
+     * {@link RequestLog#requestEndTimeNanos()}, {@link RequestLog#requestDurationNanos()}.
+     */
+    REQUEST_END_TIME(true),
+
+    /**
+     * Indicates that the request is complete and all request properties are available.
+     */
+    REQUEST_COMPLETE(true),
+
     // Response properties
 
+    /**
+     * {@link RequestLog#responseCause()}.
+     */
+    // Notify the response cause before other response properties to propagate the cause as early as possible.
+    RESPONSE_CAUSE(false),
+
     /**
      * {@link RequestLog#responseStartTimeMicros()}, {@link RequestLog#responseStartTimeMillis()},
      * {@link RequestLog#responseStartTimeNanos()}.
      */
     RESPONSE_START_TIME(false),
 
     /**
-     * {@link RequestLog#responseEndTimeNanos()}, {@link RequestLog#responseDurationNanos()},
-     * {@link RequestLog#totalDurationNanos()}.
+     * {@link RequestLog#responseHeaders()}.
      */
-    RESPONSE_END_TIME(false),
+    RESPONSE_HEADERS(false),
 
     /**
      * {@link RequestLog#responseFirstBytesTransferredTimeNanos()}.
      */
     RESPONSE_FIRST_BYTES_TRANSFERRED_TIME(false),
 
-    /**
-     * {@link RequestLog#responseHeaders()}.
-     */
-    RESPONSE_HEADERS(false),
-
     /**
      * {@link RequestLog#responseContent()}.
      */
@@ -132,20 +137,32 @@ public enum RequestLogProperty {
      */
     RESPONSE_CONTENT_PREVIEW(false),
 
+    /**
+     * {@link RequestLog#responseLength()}.
+     */
+    // TODO(ikhoon): Check if this property is actually used anywhere.
+    RESPONSE_LENGTH(false),
+
     /**
      * {@link RequestLog#responseTrailers()}.
      */
     RESPONSE_TRAILERS(false),
 
     /**
-     * {@link RequestLog#responseLength()}.
+     * {@link RequestLog#responseEndTimeNanos()}, {@link RequestLog#responseDurationNanos()},
+     * {@link RequestLog#totalDurationNanos()}.
      */
-    RESPONSE_LENGTH(false),
+    RESPONSE_END_TIME(false),
 
     /**
-     * {@link RequestLog#responseCause()}.
+     * Indicates that the response is complete and all response properties are available.
+     */
+    RESPONSE_COMPLETE(false),
+
+    /**
+     * Indicates that both the request and response are complete and all properties are available.
      */
-    RESPONSE_CAUSE(false);
+    ALL_COMPLETE(false);
 
     private static final Set<RequestLogProperty> REQUEST_PROPERTIES =
             Arrays.stream(values())
@@ -154,7 +171,7 @@ public enum RequestLogProperty {
 
     private static final Set<RequestLogProperty> RESPONSE_PROPERTIES =
             Arrays.stream(values())
-                  .filter(p -> !p.isRequestProperty)
+                  .filter(p -> !p.isRequestProperty && p != ALL_COMPLETE)
                   .collect(Sets.toImmutableEnumSet());
 
     private static final Set<RequestLogProperty> ALL_PROPERTIES =
@@ -165,8 +182,8 @@ public enum RequestLogProperty {
     static final int FLAGS_ALL_COMPLETE;
 
     static {
-        FLAGS_REQUEST_COMPLETE = flags(REQUEST_PROPERTIES);
-        FLAGS_RESPONSE_COMPLETE = flags(RESPONSE_PROPERTIES);
+        FLAGS_REQUEST_COMPLETE = flags(REQUEST_PROPERTIES) & ~REQUEST_COMPLETE.flag();
+        FLAGS_RESPONSE_COMPLETE = flags(RESPONSE_PROPERTIES) & ~RESPONSE_COMPLETE.flag();
         FLAGS_ALL_COMPLETE = FLAGS_REQUEST_COMPLETE | FLAGS_RESPONSE_COMPLETE;
     }
 

core/src/main/java/com/linecorp/armeria/server/observation/ObservationService.java

@@ -26,7 +26,6 @@
 import com.linecorp.armeria.common.HttpResponse;
 import com.linecorp.armeria.common.annotation.Nullable;
 import com.linecorp.armeria.common.annotation.UnstableApi;
-import com.linecorp.armeria.common.logging.RequestLogProperty;
 import com.linecorp.armeria.internal.common.RequestContextExtension;
 import com.linecorp.armeria.server.HttpService;
 import com.linecorp.armeria.server.ServiceRequestContext;
@@ -140,23 +139,24 @@ public HttpResponse serve(ServiceRequestContext ctx, HttpRequest req) throws Exc
     private static void enrichObservation(ServiceRequestContext ctx,
                                           ServiceObservationContext serviceObservationContext,
                                           Observation observation) {
-        ctx.log()
-           .whenAvailable(RequestLogProperty.REQUEST_FIRST_BYTES_TRANSFERRED_TIME)
-           .thenAccept(requestLog -> observation.event(
-                   HttpServiceObservationDocumentation.Events.WIRE_RECEIVE));
-
-        ctx.log()
-           .whenAvailable(RequestLogProperty.RESPONSE_FIRST_BYTES_TRANSFERRED_TIME)
-           .thenAccept(requestLog -> {
-               if (requestLog.responseFirstBytesTransferredTimeNanos() != null) {
-                   observation.event(HttpServiceObservationDocumentation.Events.WIRE_SEND);
-               }
-           });
-
-        ctx.log().whenComplete()
-           .thenAccept(requestLog -> {
-               serviceObservationContext.setResponse(requestLog);
-               observation.stop();
-           });
+        ctx.log().addListener((property, log) -> {
+            switch (property) {
+                case REQUEST_FIRST_BYTES_TRANSFERRED_TIME:
+                    observation.event(HttpServiceObservationDocumentation.Events.WIRE_RECEIVE);
+                    break;
+                case RESPONSE_FIRST_BYTES_TRANSFERRED_TIME:
+                    if (log.responseFirstBytesTransferredTimeNanos() != null) {
+                        observation.event(HttpServiceObservationDocumentation.Events.WIRE_SEND);
+                    }
+                    break;
+                case ALL_COMPLETE:
+                    serviceObservationContext.setResponse(log);
+                    observation.stop();
+                    break;
+                default:
+                    // Do nothing.
+                    break;
+            }
+        });
     }
 }

core/src/test/java/com/linecorp/armeria/common/logging/DefaultRequestLogTest.java

@@ -31,7 +31,6 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.BiFunction;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -66,13 +65,6 @@
 
 class DefaultRequestLogTest {
 
-    private static final BiFunction<RequestContext, HttpHeaders, String> headersSanitizer =
-            (ctx, headers) -> "sanitized_headers";
-    private static final BiFunction<RequestContext, Object, String> contentSanitizer =
-            (ctx, content) -> "sanitized_content";
-    private static final BiFunction<RequestContext, HttpHeaders, String> trailersSanitizer =
-            (ctx, trailers) -> "sanitized_trailers";
-
     @Mock
     private RequestContext ctx;
 
@@ -188,7 +180,6 @@ void rpcFailure_responseContentWithCause() {
     @ValueSource(booleans = {true, false})
     void addChild(boolean isResponseEndingWithFirstChild) {
         when(ctx.method()).thenReturn(HttpMethod.GET);
-        when(ctx.eventLoop()).thenReturn(ContextAwareEventLoop.of(ctx, ImmediateEventLoop.INSTANCE));
         final DefaultRequestLog firstChild = new DefaultRequestLog(ctx);
         final DefaultRequestLog lastChild = new DefaultRequestLog(ctx);
         log.addChild(firstChild);