Skip to content

feat: ensure statusCode is included in router logs #19668

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jtuglu1 wants to merge 2 commits into
base: master
Choose a base branch
from
+113 −47
Open

feat: ensure statusCode is included in router logs #19668

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fabe731
feat: ensure statusCode is included in router logs
jtuglu1 Jul 8, 2026
a3a9890
clean up
jtuglu1 Jul 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
103 changes: 61 additions & 42 deletions services/src/main/java/org/apache/druid/server/AsyncQueryForwardingServlet.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,6 @@
import com.fasterxml.jackson.dataformat.smile.SmileFactory;
import com.fasterxml.jackson.jaxrs.smile.SmileMediaTypes;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
import com.google.inject.Inject;
import com.google.inject.Provider;
import org.apache.calcite.avatica.remote.ProtobufTranslation;
Expand Down Expand Up @@ -387,15 +386,15 @@ private void handleQueryParseException(
final String errorMessage = exceptionToReport.getMessage() == null
? "no error message" : exceptionToReport.getMessage();

AuthenticationResult authenticationResult = AuthorizationUtils.authenticationResultFromRequest(request);
final AuthenticationResult authenticationResult = AuthorizationUtils.authenticationResultFromRequest(request);

if (isNativeQuery) {
requestLogger.logNativeQuery(
RequestLogLine.forNative(
null,
DateTimes.nowUtc(),
request.getRemoteAddr(),
new QueryStats(ImmutableMap.of("success", false, "exception", errorMessage, "identity", authenticationResult.getIdentity()))
buildRequestLogQueryStats(false, httpStatusCode, authenticationResult.getIdentity(), null, errorMessage)
)
);
} else {
Expand All @@ -405,7 +404,7 @@ private void handleQueryParseException(
null,
DateTimes.nowUtc(),
request.getRemoteAddr(),
new QueryStats(ImmutableMap.of("success", false, "exception", errorMessage, "identity", authenticationResult.getIdentity()))
buildRequestLogQueryStats(false, httpStatusCode, authenticationResult.getIdentity(), null, errorMessage)
)
);
}
Expand All @@ -419,6 +418,38 @@ private void handleQueryParseException(
);
}

/**
* Builds the {@link QueryStats} recorded in router request logs. Success paths pass {@code queryTimeMs};
* failure paths pass {@code exceptionMessage}. Exactly one of the two is non-null at every call site.
*/
private static QueryStats buildRequestLogQueryStats(
boolean success,
int statusCode,
String identity,
@Nullable Long queryTimeMs,
@Nullable String exceptionMessage
)
{
if (exceptionMessage != null) {
return new QueryStats(
Map.of(
"success", success,
DruidMetrics.STATUS_CODE, statusCode,
"exception", exceptionMessage,
"identity", identity
)
);
}
return new QueryStats(
Map.of(
"query/time", queryTimeMs,
"success", success,
DruidMetrics.STATUS_CODE, statusCode,
"identity", identity
)
);
}

protected void doService(
HttpServletRequest request,
HttpServletResponse response
Expand Down Expand Up @@ -770,7 +801,7 @@ public void onComplete(Result result)
failedQueryCount.incrementAndGet();
}

AuthenticationResult authenticationResult = AuthorizationUtils.authenticationResultFromRequest(req);
final AuthenticationResult authenticationResult = AuthorizationUtils.authenticationResultFromRequest(req);

// As router is simply a proxy, we don't make an effort to construct the error code from the exception ourselves.
// We rely on broker to set this for us if the error occurs downstream.
Expand All @@ -788,15 +819,12 @@ public void onComplete(Result result)
sqlQuery.getContext(),
DateTimes.nowUtc(),
req.getRemoteAddr(),
new QueryStats(
ImmutableMap.of(
"query/time",
TimeUnit.NANOSECONDS.toMillis(requestTimeNs),
"success",
success,
"identity",
authenticationResult.getIdentity()
)
buildRequestLogQueryStats(
success,
statusCode,
authenticationResult.getIdentity(),
TimeUnit.NANOSECONDS.toMillis(requestTimeNs),
null
)
)
);
Expand All @@ -815,15 +843,12 @@ public void onComplete(Result result)
query,
DateTimes.nowUtc(),
req.getRemoteAddr(),
new QueryStats(
ImmutableMap.of(
"query/time",
TimeUnit.NANOSECONDS.toMillis(requestTimeNs),
"success",
success,
"identity",
authenticationResult.getIdentity()
)
buildRequestLogQueryStats(
success,
statusCode,
authenticationResult.getIdentity(),
TimeUnit.NANOSECONDS.toMillis(requestTimeNs),
null
)
)
);
Expand Down Expand Up @@ -863,7 +888,7 @@ public void onFailure(Response response, Throwable failure)
// We rely on broker to set this for us if the error occurs downstream.
// Otherwise, if there's a router/client error, we log this as an unknown error.
final int statusCode = determineStatusCode(false, response.getStatus());
AuthenticationResult authenticationResult = AuthorizationUtils.authenticationResultFromRequest(req);
final AuthenticationResult authenticationResult = AuthorizationUtils.authenticationResultFromRequest(req);
emitQueryTime(requestTimeNs, false, sqlQueryId, queryId, statusCode, authenticationResult);

//noinspection VariableNotUsedInsideIf
Expand All @@ -877,15 +902,12 @@ public void onFailure(Response response, Throwable failure)
sqlQuery.getContext(),
DateTimes.nowUtc(),
req.getRemoteAddr(),
new QueryStats(
ImmutableMap.of(
"success",
false,
"exception",
errorMessage == null ? "no message" : errorMessage,
"identity",
authenticationResult.getIdentity()
)
buildRequestLogQueryStats(
false,
statusCode,
authenticationResult.getIdentity(),
null,
errorMessage == null ? "no message" : errorMessage
)
)
);
Expand All @@ -909,15 +931,12 @@ public void onFailure(Response response, Throwable failure)
query,
DateTimes.nowUtc(),
req.getRemoteAddr(),
new QueryStats(
ImmutableMap.of(
"success",
false,
"exception",
errorMessage == null ? "no message" : errorMessage,
"identity",
authenticationResult.getIdentity()
)
buildRequestLogQueryStats(
false,
statusCode,
authenticationResult.getIdentity(),
null,
errorMessage == null ? "no message" : errorMessage
)
)
);
Expand Down
57 changes: 52 additions & 5 deletions services/src/test/java/org/apache/druid/server/AsyncQueryForwardingServletTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,7 @@
import org.apache.druid.server.initialization.jetty.JettyServerInitUtils;
import org.apache.druid.server.initialization.jetty.JettyServerInitializer;
import org.apache.druid.server.log.NoopRequestLogger;
import org.apache.druid.server.log.RequestLogger;
import org.apache.druid.server.metrics.NoopServiceEmitter;
import org.apache.druid.server.router.QueryHostFinder;
import org.apache.druid.server.router.RendezvousHashAvaticaConnectionBalancer;
Expand Down Expand Up @@ -506,7 +507,7 @@ public QueryMetrics<Query<?>> makeMetrics()
}

@Test
public void testMetricsEmittedWithErrorStatusCodeButNoResultException()
public void testMetricsEmittedWithErrorStatusCodeButNoResultException() throws IOException
{
final TimeseriesQuery query = Druids.newTimeseriesQueryBuilder()
.dataSource("foo")
Expand Down Expand Up @@ -539,6 +540,7 @@ public Throwable getFailure()
};

final StubServiceEmitter stubServiceEmitter = StubServiceEmitter.createStarted();
final RequestLogger requestLogger = Mockito.mock(RequestLogger.class);
final AsyncQueryForwardingServlet servlet = new AsyncQueryForwardingServlet(
new MapQueryToolChestWarehouse(ImmutableMap.of()),
TestHelper.makeJsonMapper(),
Expand All @@ -547,7 +549,7 @@ public Throwable getFailure()
null,
null,
stubServiceEmitter,
NoopRequestLogger.instance(),
requestLogger,
makeRouterTestOverrideEmittingFactory(),
new AuthenticatorMapper(ImmutableMap.of()),
new Properties(),
Expand All @@ -566,6 +568,7 @@ public Throwable getFailure()
504,
stubServiceEmitter.getMetricEvents("query/time").get(0).toMap().get(DruidMetrics.STATUS_CODE)
);
assertNativeQueryStatusCodeMatchesMetric(requestLogger, getEmittedStatusCode(stubServiceEmitter));
Assert.assertEquals("false", stubServiceEmitter.getMetricEvents("query/time").get(0).toMap().get("success"));
Assert.assertEquals("testUser", stubServiceEmitter.getMetricEvents("query/time").get(0).toMap().get("identity"));
}
Expand Down Expand Up @@ -643,7 +646,7 @@ public Throwable getFailure()
}

@Test
public void testOnFailureWithExceptionAndUnassignedStatusCode()
public void testOnFailureWithExceptionAndUnassignedStatusCode() throws IOException
{
final TimeseriesQuery query = Druids.newTimeseriesQueryBuilder()
.dataSource("foo")
Expand All @@ -665,6 +668,7 @@ public void testOnFailureWithExceptionAndUnassignedStatusCode()
Mockito.when(responseMock.getHeaders()).thenReturn(HttpFields.build());

final StubServiceEmitter stubServiceEmitter = StubServiceEmitter.createStarted();
final RequestLogger requestLogger = Mockito.mock(RequestLogger.class);
final AsyncQueryForwardingServlet servlet = new AsyncQueryForwardingServlet(
new MapQueryToolChestWarehouse(ImmutableMap.of()),
TestHelper.makeJsonMapper(),
Expand All @@ -673,7 +677,7 @@ public void testOnFailureWithExceptionAndUnassignedStatusCode()
null,
null,
stubServiceEmitter,
NoopRequestLogger.instance(),
requestLogger,
makeRouterTestOverrideEmittingFactory(),
new AuthenticatorMapper(ImmutableMap.of()),
new Properties(),
Expand All @@ -693,6 +697,7 @@ public void testOnFailureWithExceptionAndUnassignedStatusCode()
500, // Should default to 500 when status is 0
stubServiceEmitter.getMetricEvents("query/time").get(0).toMap().get(DruidMetrics.STATUS_CODE)
);
assertNativeQueryStatusCodeMatchesMetric(requestLogger, getEmittedStatusCode(stubServiceEmitter));
Assert.assertEquals("false", stubServiceEmitter.getMetricEvents("query/time").get(0).toMap().get("success"));
Assert.assertEquals("testUser", stubServiceEmitter.getMetricEvents("query/time").get(0).toMap().get("identity"));
}
Expand Down Expand Up @@ -913,6 +918,7 @@ public CompletableFuture<Boolean> abort(Throwable throwable)
};
final Result result = new Result(proxyRequestMock, response);
final StubServiceEmitter stubServiceEmitter = StubServiceEmitter.createStarted();
final RequestLogger requestLogger = Mockito.mock(RequestLogger.class);
final AsyncQueryForwardingServlet servlet = new AsyncQueryForwardingServlet(
new MapQueryToolChestWarehouse(ImmutableMap.of()),
jsonMapper,
Expand All @@ -921,7 +927,7 @@ public CompletableFuture<Boolean> abort(Throwable throwable)
null,
null,
stubServiceEmitter,
NoopRequestLogger.instance(),
requestLogger,
new DefaultGenericQueryMetricsFactory(),
new AuthenticatorMapper(ImmutableMap.of()),
properties,
Expand Down Expand Up @@ -967,12 +973,53 @@ protected void doService(
stubServiceEmitter.getMetricEvents("query/time").get(0).toMap().get(DruidMetrics.STATUS_CODE)
);
}
if (!isJDBCSql) {
assertStatusCodeMatchesMetric(requestLogger, isNativeSql, getEmittedStatusCode(stubServiceEmitter));
}

// This test is mostly about verifying that the servlet calls the right methods the right number of times.
EasyMock.verify(hostFinder, requestMock);
Assert.assertEquals(1, didService.get());
}

private static int getEmittedStatusCode(StubServiceEmitter stubServiceEmitter)
{
return (int) stubServiceEmitter.getMetricEvents("query/time").get(0).toMap().get(DruidMetrics.STATUS_CODE);
}

private static void assertStatusCodeMatchesMetric(
RequestLogger requestLogger,
boolean isSqlQuery,
int metricStatusCode
) throws IOException
{
if (isSqlQuery) {

@kfaraz kfaraz Jul 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Seems like this method is always being called for SQL queries only.

final ArgumentCaptor<RequestLogLine> requestLogLineCaptor = ArgumentCaptor.forClass(RequestLogLine.class);
Mockito.verify(requestLogger).logSqlQuery(requestLogLineCaptor.capture());
assertStatusCode(requestLogLineCaptor.getValue(), metricStatusCode);
} else {
assertNativeQueryStatusCodeMatchesMetric(requestLogger, metricStatusCode);
}
}

private static void assertNativeQueryStatusCodeMatchesMetric(
RequestLogger requestLogger,
int metricStatusCode
) throws IOException
{
final ArgumentCaptor<RequestLogLine> requestLogLineCaptor = ArgumentCaptor.forClass(RequestLogLine.class);
Mockito.verify(requestLogger).logNativeQuery(requestLogLineCaptor.capture());
assertStatusCode(requestLogLineCaptor.getValue(), metricStatusCode);
}

private static void assertStatusCode(RequestLogLine requestLogLine, int metricStatusCode)
{
Assert.assertEquals(
metricStatusCode,
requestLogLine.getQueryStats().getStats().get(DruidMetrics.STATUS_CODE)
);
}

private static Server makeTestDeleteServer(int port, final CountDownLatch latch)
{
Server server = new Server(port);
Expand Down
Loading