This is an automated email from the ASF dual-hosted git repository.
jtuglu1 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 9c3e58e0c97 feat: ensure statusCode is included in router logs (#19668)
9c3e58e0c97 is described below
commit 9c3e58e0c973194a9e1ee4de9c11ee1c806f053d
Author: jtuglu1 <[email protected]>
AuthorDate: Fri Jul 10 11:39:11 2026 -0700
feat: ensure statusCode is included in router logs (#19668)
We rely on query logs for observability, operations. Currently the only
source of query status code on the router is in the metric emitted by the
router. We should make sure it's emitted as part of the query log as well.
---
.../druid/server/AsyncQueryForwardingServlet.java | 103 ++++++++++++---------
.../server/AsyncQueryForwardingServletTest.java | 57 +++++++++++-
2 files changed, 113 insertions(+), 47 deletions(-)
diff --git
a/services/src/main/java/org/apache/druid/server/AsyncQueryForwardingServlet.java
b/services/src/main/java/org/apache/druid/server/AsyncQueryForwardingServlet.java
index 4054e6249b1..7b8cd453b40 100644
---
a/services/src/main/java/org/apache/druid/server/AsyncQueryForwardingServlet.java
+++
b/services/src/main/java/org/apache/druid/server/AsyncQueryForwardingServlet.java
@@ -24,7 +24,6 @@ import com.fasterxml.jackson.databind.ObjectMapper;
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;
@@ -387,7 +386,7 @@ public class AsyncQueryForwardingServlet extends
AsyncProxyServlet implements Qu
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(
@@ -395,7 +394,7 @@ public class AsyncQueryForwardingServlet extends
AsyncProxyServlet implements Qu
null,
DateTimes.nowUtc(),
request.getRemoteAddr(),
- new QueryStats(ImmutableMap.of("success", false, "exception",
errorMessage, "identity", authenticationResult.getIdentity()))
+ buildRequestLogQueryStats(false, httpStatusCode,
authenticationResult.getIdentity(), null, errorMessage)
)
);
} else {
@@ -405,7 +404,7 @@ public class AsyncQueryForwardingServlet extends
AsyncProxyServlet implements Qu
null,
DateTimes.nowUtc(),
request.getRemoteAddr(),
- new QueryStats(ImmutableMap.of("success", false, "exception",
errorMessage, "identity", authenticationResult.getIdentity()))
+ buildRequestLogQueryStats(false, httpStatusCode,
authenticationResult.getIdentity(), null, errorMessage)
)
);
}
@@ -419,6 +418,38 @@ public class AsyncQueryForwardingServlet extends
AsyncProxyServlet implements Qu
);
}
+ /**
+ * 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
@@ -770,7 +801,7 @@ public class AsyncQueryForwardingServlet extends
AsyncProxyServlet implements Qu
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.
@@ -788,15 +819,12 @@ public class AsyncQueryForwardingServlet extends
AsyncProxyServlet implements Qu
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
)
)
);
@@ -815,15 +843,12 @@ public class AsyncQueryForwardingServlet extends
AsyncProxyServlet implements Qu
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
)
)
);
@@ -863,7 +888,7 @@ public class AsyncQueryForwardingServlet extends
AsyncProxyServlet implements Qu
// 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
@@ -877,15 +902,12 @@ public class AsyncQueryForwardingServlet extends
AsyncProxyServlet implements Qu
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
)
)
);
@@ -909,15 +931,12 @@ public class AsyncQueryForwardingServlet extends
AsyncProxyServlet implements Qu
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
)
)
);
diff --git
a/services/src/test/java/org/apache/druid/server/AsyncQueryForwardingServletTest.java
b/services/src/test/java/org/apache/druid/server/AsyncQueryForwardingServletTest.java
index c3b50514e82..9903eb379db 100644
---
a/services/src/test/java/org/apache/druid/server/AsyncQueryForwardingServletTest.java
+++
b/services/src/test/java/org/apache/druid/server/AsyncQueryForwardingServletTest.java
@@ -73,6 +73,7 @@ import org.apache.druid.server.initialization.ServerConfig;
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;
@@ -506,7 +507,7 @@ public class AsyncQueryForwardingServletTest extends
BaseJettyTest
}
@Test
- public void testMetricsEmittedWithErrorStatusCodeButNoResultException()
+ public void testMetricsEmittedWithErrorStatusCodeButNoResultException()
throws IOException
{
final TimeseriesQuery query = Druids.newTimeseriesQueryBuilder()
.dataSource("foo")
@@ -539,6 +540,7 @@ public class AsyncQueryForwardingServletTest extends
BaseJettyTest
};
final StubServiceEmitter stubServiceEmitter =
StubServiceEmitter.createStarted();
+ final RequestLogger requestLogger = Mockito.mock(RequestLogger.class);
final AsyncQueryForwardingServlet servlet = new
AsyncQueryForwardingServlet(
new MapQueryToolChestWarehouse(ImmutableMap.of()),
TestHelper.makeJsonMapper(),
@@ -547,7 +549,7 @@ public class AsyncQueryForwardingServletTest extends
BaseJettyTest
null,
null,
stubServiceEmitter,
- NoopRequestLogger.instance(),
+ requestLogger,
makeRouterTestOverrideEmittingFactory(),
new AuthenticatorMapper(ImmutableMap.of()),
new Properties(),
@@ -566,6 +568,7 @@ public class AsyncQueryForwardingServletTest extends
BaseJettyTest
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"));
}
@@ -643,7 +646,7 @@ public class AsyncQueryForwardingServletTest extends
BaseJettyTest
}
@Test
- public void testOnFailureWithExceptionAndUnassignedStatusCode()
+ public void testOnFailureWithExceptionAndUnassignedStatusCode() throws
IOException
{
final TimeseriesQuery query = Druids.newTimeseriesQueryBuilder()
.dataSource("foo")
@@ -665,6 +668,7 @@ public class AsyncQueryForwardingServletTest extends
BaseJettyTest
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(),
@@ -673,7 +677,7 @@ public class AsyncQueryForwardingServletTest extends
BaseJettyTest
null,
null,
stubServiceEmitter,
- NoopRequestLogger.instance(),
+ requestLogger,
makeRouterTestOverrideEmittingFactory(),
new AuthenticatorMapper(ImmutableMap.of()),
new Properties(),
@@ -693,6 +697,7 @@ public class AsyncQueryForwardingServletTest extends
BaseJettyTest
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"));
}
@@ -913,6 +918,7 @@ public class AsyncQueryForwardingServletTest extends
BaseJettyTest
};
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,
@@ -921,7 +927,7 @@ public class AsyncQueryForwardingServletTest extends
BaseJettyTest
null,
null,
stubServiceEmitter,
- NoopRequestLogger.instance(),
+ requestLogger,
new DefaultGenericQueryMetricsFactory(),
new AuthenticatorMapper(ImmutableMap.of()),
properties,
@@ -967,12 +973,53 @@ public class AsyncQueryForwardingServletTest extends
BaseJettyTest
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) {
+ 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);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
