github-advanced-security[bot] commented on code in PR #19855: URL: https://github.com/apache/druid/pull/19855#discussion_r3698526471
########## server/src/main/java/org/apache/druid/server/StackTraceCollector.java: ########## @@ -0,0 +1,424 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF 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 + * + * http://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 org.apache.druid.server; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.google.common.collect.ImmutableList; +import org.apache.druid.error.InvalidInput; +import org.apache.druid.java.util.common.DateTimes; + +import javax.annotation.Nullable; +import java.lang.management.LockInfo; +import java.lang.management.ManagementFactory; +import java.lang.management.MonitorInfo; +import java.lang.management.ThreadInfo; +import java.lang.management.ThreadMXBean; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +/** + * Collects a live snapshot of the Java platform threads running in the current Druid process. + * + * <p>This class intentionally has no instance state. A new collector can be created for each request. + */ +public class StackTraceCollector +{ + public static final String MAX_STACK_TRACE_FRAME_DEPTH_KEY = "maxStackTraceFrameDepth"; + public static final int MIN_ALLOWED_STACK_TRACE_FRAME_DEPTH = 10; + public static final int DEFAULT_MAX_STACK_TRACE_FRAME_DEPTH = 100; + public static final int MAX_ALLOWED_STACK_TRACE_FRAME_DEPTH = 1000; + + public ThreadStackTraceResponse collect() + { + return collect(DEFAULT_MAX_STACK_TRACE_FRAME_DEPTH); + } + + public ThreadStackTraceResponse collect(final int maxStackTraceFrameDepth) + { + validateMaxStackTraceFrameDepth(maxStackTraceFrameDepth); + final String collectedAt = DateTimes.nowUtc().toString(); + final ThreadMXBean threadMxBean = ManagementFactory.getThreadMXBean(); + final boolean cpuTimeEnabled = isCpuTimeEnabled(threadMxBean); + final Set<Long> deadlockedThreadIds = findDeadlockedThreadIds(threadMxBean); + final long[] threadIds = threadMxBean.getAllThreadIds(); + final ThreadInfo[] threadInfos = threadMxBean.getThreadInfo( + threadIds, + threadMxBean.isObjectMonitorUsageSupported(), + threadMxBean.isSynchronizerUsageSupported(), + maxStackTraceFrameDepth + ); + final List<ThreadStackTrace> threads = new ArrayList<>(threadInfos.length); + + for (final ThreadInfo threadInfo : threadInfos) { + if (threadInfo == null) { + continue; + } + + final long threadId = threadInfo.getThreadId(); + final long rawLockOwnerId = threadInfo.getLockOwnerId(); + final Long lockOwnerId = rawLockOwnerId < 0 ? null : rawLockOwnerId; + threads.add( + new ThreadStackTrace( + threadId, + threadInfo.getThreadName(), + threadInfo.getThreadState().name(), + threadInfo.isDaemon(), + threadInfo.getPriority(), + getThreadCpuTime(threadMxBean, threadId, cpuTimeEnabled, false), + getThreadCpuTime(threadMxBean, threadId, cpuTimeEnabled, true), + threadInfo.getLockName(), + lockOwnerId, + threadInfo.getLockOwnerName(), + deadlockedThreadIds.contains(threadId), + formatThreadInfo(threadInfo) + ) + ); + } + + return new ThreadStackTraceResponse(collectedAt, threads); + } + + public static int parseMaxStackTraceFrameDepth(@Nullable final String value) + { + if (value == null) { + return DEFAULT_MAX_STACK_TRACE_FRAME_DEPTH; + } + + try { + return validateMaxStackTraceFrameDepth(Long.parseLong(value)); + } + catch (NumberFormatException e) { + throw InvalidInput.exception( + "Query parameter[%s] must be an integer, but got[%s]", + MAX_STACK_TRACE_FRAME_DEPTH_KEY, + value + ); + } + } + + public static int validateMaxStackTraceFrameDepth(final long maxStackTraceFrameDepth) + { + InvalidInput.conditionalException( + maxStackTraceFrameDepth >= MIN_ALLOWED_STACK_TRACE_FRAME_DEPTH, + "[%s] must be greater than or equal to %d, but got[%d]", + MAX_STACK_TRACE_FRAME_DEPTH_KEY, + MIN_ALLOWED_STACK_TRACE_FRAME_DEPTH, + maxStackTraceFrameDepth + ); + InvalidInput.conditionalException( + maxStackTraceFrameDepth <= MAX_ALLOWED_STACK_TRACE_FRAME_DEPTH, + "[%s] must be less than or equal to %d, but got[%d]", + MAX_STACK_TRACE_FRAME_DEPTH_KEY, + MAX_ALLOWED_STACK_TRACE_FRAME_DEPTH, + maxStackTraceFrameDepth + ); + return (int) maxStackTraceFrameDepth; Review Comment: ## CodeQL / User-controlled data in numeric cast This cast to a narrower type depends on a [user-provided value](1), potentially causing truncation. [Show more details](https://github.com/apache/druid/security/code-scanning/11458) ########## embedded-tests/src/test/java/org/apache/druid/testing/embedded/schema/SystemStackTraceTableTest.java: ########## @@ -0,0 +1,230 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF 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 + * + * http://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 org.apache.druid.testing.embedded.schema; + +import com.fasterxml.jackson.core.type.TypeReference; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.query.http.ClientSqlQuery; +import org.apache.druid.rpc.RequestBuilder; +import org.apache.druid.server.StackTraceCollector; +import org.apache.druid.sql.http.ResultFormat; +import org.apache.druid.testing.embedded.EmbeddedBroker; +import org.apache.druid.testing.embedded.EmbeddedCoordinator; +import org.apache.druid.testing.embedded.EmbeddedDruidCluster; +import org.apache.druid.testing.embedded.EmbeddedOverlord; +import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase; +import org.jboss.netty.handler.codec.http.HttpMethod; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.Map; + +public class SystemStackTraceTableTest extends EmbeddedClusterTestBase +{ + private static final String BROKER_PORT = "9082"; + private static final String BROKER_SERVICE = "test/broker"; + private static final String OVERLORD_PORT = "9090"; + private static final String OVERLORD_SERVICE = "test/overlord"; + private static final String COORDINATOR_PORT = "9081"; + private static final String COORDINATOR_SERVICE = "test/coordinator"; + + private final EmbeddedBroker broker = new EmbeddedBroker() + .addProperty("druid.service", BROKER_SERVICE) + .addProperty("druid.plaintextPort", BROKER_PORT); + + private final EmbeddedOverlord overlord = new EmbeddedOverlord() + .addProperty("druid.service", OVERLORD_SERVICE) + .addProperty("druid.plaintextPort", OVERLORD_PORT); + + private final EmbeddedCoordinator coordinator = new EmbeddedCoordinator() + .addProperty("druid.service", COORDINATOR_SERVICE) + .addProperty("druid.plaintextPort", COORDINATOR_PORT); + + @Override + protected EmbeddedDruidCluster createCluster() + { + return EmbeddedDruidCluster + .withZookeeper() + .addServer(coordinator) + .addServer(overlord) + .addServer(broker); + } + + @Test + public void test_stackTraceEndpoint() + { + final StackTraceCollector.ThreadStackTraceResponse response = cluster.callApi().serviceClient().onAnyBroker( + mapper -> new RequestBuilder(HttpMethod.GET, "/status/stack"), + new TypeReference<>(){} + ); + + Assertions.assertNotNull(response.getCollectedAt()); + Assertions.assertFalse(response.getThreads().isEmpty()); + Assertions.assertTrue( + response.getThreads().stream().allMatch(thread -> !thread.getThreadName().isEmpty()) + ); + Assertions.assertTrue( + response.getThreads().stream().anyMatch(thread -> !thread.getStackTrace().isEmpty()) + ); + Assertions.assertTrue( + response.getThreads().stream().allMatch(thread -> countStackFrames(thread.getStackTrace()) <= 100) + ); + } + + @Test + public void test_stackTraceEndpointWithMaxStackTraceFrameDepth() + { + final StackTraceCollector.ThreadStackTraceResponse response = cluster.callApi().serviceClient().onAnyBroker( + mapper -> new RequestBuilder(HttpMethod.GET, "/status/stack?maxStackTraceFrameDepth=10"), + new TypeReference<>(){} + ); + + Assertions.assertTrue( + response.getThreads().stream().allMatch(thread -> countStackFrames(thread.getStackTrace()) <= 10) + ); + } + + @Test + public void test_stackTraceEndpointRejectsInvalidMaxStackTraceFrameDepth() + { + for (final String invalidDepth : new String[]{"9", "1001", "10.5"}) { + final RuntimeException exception = Assertions.assertThrows( + RuntimeException.class, + () -> cluster.callApi().serviceClient().onAnyBroker( + mapper -> new RequestBuilder( + HttpMethod.GET, + StringUtils.format("/status/stack?maxStackTraceFrameDepth=%s", invalidDepth) + ), + new TypeReference<StackTraceCollector.ThreadStackTraceResponse>(){} + ) + ); + Assertions.assertTrue(exception.getMessage().contains("400 Bad Request"), exception.getMessage()); + Assertions.assertTrue( + exception.getMessage().contains(StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY), + exception.getMessage() + ); + } + } + + @Test + public void test_stackTraceTable() + { + final String brokerHost = StringUtils.format("localhost:%s", BROKER_PORT); + final String result = cluster.runSql( + "SELECT server, service_name, node_roles, collected_at, thread_id, " + + "thread_state, daemon, priority, cpu_time_ns, user_cpu_time_ns, is_deadlocked, error_message " + + "FROM sys.stack_trace WHERE server = '%s'", + brokerHost + ); + + Assertions.assertFalse(result.isEmpty(), "The stack trace table should return broker threads"); + for (final String row : result.split("\\n")) { + final String[] columns = row.split(",", -1); + Assertions.assertEquals(brokerHost, columns[0]); + Assertions.assertEquals(BROKER_SERVICE, columns[1]); + Assertions.assertEquals("broker", columns[2]); + Assertions.assertFalse(columns[3].isEmpty()); + Assertions.assertDoesNotThrow(() -> Long.parseLong(columns[4])); + Assertions.assertFalse(columns[6].isEmpty()); + Assertions.assertTrue(columns[6].equals("0") || columns[6].equals("1"), row); + Assertions.assertDoesNotThrow(() -> Long.parseLong(columns[7])); Review Comment: ## CodeQL / Missing catch of NumberFormatException Potential uncaught 'java.lang.NumberFormatException'. [Show more details](https://github.com/apache/druid/security/code-scanning/11460) ########## embedded-tests/src/test/java/org/apache/druid/testing/embedded/schema/SystemStackTraceTableTest.java: ########## @@ -0,0 +1,230 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF 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 + * + * http://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 org.apache.druid.testing.embedded.schema; + +import com.fasterxml.jackson.core.type.TypeReference; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.query.http.ClientSqlQuery; +import org.apache.druid.rpc.RequestBuilder; +import org.apache.druid.server.StackTraceCollector; +import org.apache.druid.sql.http.ResultFormat; +import org.apache.druid.testing.embedded.EmbeddedBroker; +import org.apache.druid.testing.embedded.EmbeddedCoordinator; +import org.apache.druid.testing.embedded.EmbeddedDruidCluster; +import org.apache.druid.testing.embedded.EmbeddedOverlord; +import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase; +import org.jboss.netty.handler.codec.http.HttpMethod; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.Map; + +public class SystemStackTraceTableTest extends EmbeddedClusterTestBase +{ + private static final String BROKER_PORT = "9082"; + private static final String BROKER_SERVICE = "test/broker"; + private static final String OVERLORD_PORT = "9090"; + private static final String OVERLORD_SERVICE = "test/overlord"; + private static final String COORDINATOR_PORT = "9081"; + private static final String COORDINATOR_SERVICE = "test/coordinator"; + + private final EmbeddedBroker broker = new EmbeddedBroker() + .addProperty("druid.service", BROKER_SERVICE) + .addProperty("druid.plaintextPort", BROKER_PORT); + + private final EmbeddedOverlord overlord = new EmbeddedOverlord() + .addProperty("druid.service", OVERLORD_SERVICE) + .addProperty("druid.plaintextPort", OVERLORD_PORT); + + private final EmbeddedCoordinator coordinator = new EmbeddedCoordinator() + .addProperty("druid.service", COORDINATOR_SERVICE) + .addProperty("druid.plaintextPort", COORDINATOR_PORT); + + @Override + protected EmbeddedDruidCluster createCluster() + { + return EmbeddedDruidCluster + .withZookeeper() + .addServer(coordinator) + .addServer(overlord) + .addServer(broker); + } + + @Test + public void test_stackTraceEndpoint() + { + final StackTraceCollector.ThreadStackTraceResponse response = cluster.callApi().serviceClient().onAnyBroker( + mapper -> new RequestBuilder(HttpMethod.GET, "/status/stack"), + new TypeReference<>(){} + ); + + Assertions.assertNotNull(response.getCollectedAt()); + Assertions.assertFalse(response.getThreads().isEmpty()); + Assertions.assertTrue( + response.getThreads().stream().allMatch(thread -> !thread.getThreadName().isEmpty()) + ); + Assertions.assertTrue( + response.getThreads().stream().anyMatch(thread -> !thread.getStackTrace().isEmpty()) + ); + Assertions.assertTrue( + response.getThreads().stream().allMatch(thread -> countStackFrames(thread.getStackTrace()) <= 100) + ); + } + + @Test + public void test_stackTraceEndpointWithMaxStackTraceFrameDepth() + { + final StackTraceCollector.ThreadStackTraceResponse response = cluster.callApi().serviceClient().onAnyBroker( + mapper -> new RequestBuilder(HttpMethod.GET, "/status/stack?maxStackTraceFrameDepth=10"), + new TypeReference<>(){} + ); + + Assertions.assertTrue( + response.getThreads().stream().allMatch(thread -> countStackFrames(thread.getStackTrace()) <= 10) + ); + } + + @Test + public void test_stackTraceEndpointRejectsInvalidMaxStackTraceFrameDepth() + { + for (final String invalidDepth : new String[]{"9", "1001", "10.5"}) { + final RuntimeException exception = Assertions.assertThrows( + RuntimeException.class, + () -> cluster.callApi().serviceClient().onAnyBroker( + mapper -> new RequestBuilder( + HttpMethod.GET, + StringUtils.format("/status/stack?maxStackTraceFrameDepth=%s", invalidDepth) + ), + new TypeReference<StackTraceCollector.ThreadStackTraceResponse>(){} + ) + ); + Assertions.assertTrue(exception.getMessage().contains("400 Bad Request"), exception.getMessage()); + Assertions.assertTrue( + exception.getMessage().contains(StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY), + exception.getMessage() + ); + } + } + + @Test + public void test_stackTraceTable() + { + final String brokerHost = StringUtils.format("localhost:%s", BROKER_PORT); + final String result = cluster.runSql( + "SELECT server, service_name, node_roles, collected_at, thread_id, " + + "thread_state, daemon, priority, cpu_time_ns, user_cpu_time_ns, is_deadlocked, error_message " + + "FROM sys.stack_trace WHERE server = '%s'", + brokerHost + ); + + Assertions.assertFalse(result.isEmpty(), "The stack trace table should return broker threads"); + for (final String row : result.split("\\n")) { + final String[] columns = row.split(",", -1); + Assertions.assertEquals(brokerHost, columns[0]); + Assertions.assertEquals(BROKER_SERVICE, columns[1]); + Assertions.assertEquals("broker", columns[2]); + Assertions.assertFalse(columns[3].isEmpty()); + Assertions.assertDoesNotThrow(() -> Long.parseLong(columns[4])); + Assertions.assertFalse(columns[6].isEmpty()); + Assertions.assertTrue(columns[6].equals("0") || columns[6].equals("1"), row); + Assertions.assertDoesNotThrow(() -> Long.parseLong(columns[7])); + if (!columns[8].isEmpty()) { + Assertions.assertDoesNotThrow(() -> Long.parseLong(columns[8])); + } + if (!columns[9].isEmpty()) { + Assertions.assertDoesNotThrow(() -> Long.parseLong(columns[9])); Review Comment: ## CodeQL / Missing catch of NumberFormatException Potential uncaught 'java.lang.NumberFormatException'. [Show more details](https://github.com/apache/druid/security/code-scanning/11462) ########## server/src/test/java/org/apache/druid/server/StatusResourceTest.java: ########## @@ -101,6 +109,175 @@ Assert.assertEquals(false, response.getEntity()); } + @Test + public void testStackTrace() + { + final StatusResource resource = new StatusResource(new Properties(), null, null, null, null); + final Response httpResponse = resource.getStackTrace(null); + Assert.assertEquals(Response.Status.OK.getStatusCode(), httpResponse.getStatus()); + final StackTraceCollector.ThreadStackTraceResponse response = + (StackTraceCollector.ThreadStackTraceResponse) httpResponse.getEntity(); + + Assert.assertNotNull(response.getCollectedAt()); + Assert.assertFalse(response.getThreads().isEmpty()); + + final long currentThreadId = Thread.currentThread().getId(); Review Comment: ## CodeQL / Deprecated method or constructor invocation Invoking [Thread.getId](1) should be avoided because it has been deprecated. [Show more details](https://github.com/apache/druid/security/code-scanning/11463) ########## embedded-tests/src/test/java/org/apache/druid/testing/embedded/schema/SystemStackTraceTableTest.java: ########## @@ -0,0 +1,230 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF 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 + * + * http://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 org.apache.druid.testing.embedded.schema; + +import com.fasterxml.jackson.core.type.TypeReference; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.query.http.ClientSqlQuery; +import org.apache.druid.rpc.RequestBuilder; +import org.apache.druid.server.StackTraceCollector; +import org.apache.druid.sql.http.ResultFormat; +import org.apache.druid.testing.embedded.EmbeddedBroker; +import org.apache.druid.testing.embedded.EmbeddedCoordinator; +import org.apache.druid.testing.embedded.EmbeddedDruidCluster; +import org.apache.druid.testing.embedded.EmbeddedOverlord; +import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase; +import org.jboss.netty.handler.codec.http.HttpMethod; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.Map; + +public class SystemStackTraceTableTest extends EmbeddedClusterTestBase +{ + private static final String BROKER_PORT = "9082"; + private static final String BROKER_SERVICE = "test/broker"; + private static final String OVERLORD_PORT = "9090"; + private static final String OVERLORD_SERVICE = "test/overlord"; + private static final String COORDINATOR_PORT = "9081"; + private static final String COORDINATOR_SERVICE = "test/coordinator"; + + private final EmbeddedBroker broker = new EmbeddedBroker() + .addProperty("druid.service", BROKER_SERVICE) + .addProperty("druid.plaintextPort", BROKER_PORT); + + private final EmbeddedOverlord overlord = new EmbeddedOverlord() + .addProperty("druid.service", OVERLORD_SERVICE) + .addProperty("druid.plaintextPort", OVERLORD_PORT); + + private final EmbeddedCoordinator coordinator = new EmbeddedCoordinator() + .addProperty("druid.service", COORDINATOR_SERVICE) + .addProperty("druid.plaintextPort", COORDINATOR_PORT); + + @Override + protected EmbeddedDruidCluster createCluster() + { + return EmbeddedDruidCluster + .withZookeeper() + .addServer(coordinator) + .addServer(overlord) + .addServer(broker); + } + + @Test + public void test_stackTraceEndpoint() + { + final StackTraceCollector.ThreadStackTraceResponse response = cluster.callApi().serviceClient().onAnyBroker( + mapper -> new RequestBuilder(HttpMethod.GET, "/status/stack"), + new TypeReference<>(){} + ); + + Assertions.assertNotNull(response.getCollectedAt()); + Assertions.assertFalse(response.getThreads().isEmpty()); + Assertions.assertTrue( + response.getThreads().stream().allMatch(thread -> !thread.getThreadName().isEmpty()) + ); + Assertions.assertTrue( + response.getThreads().stream().anyMatch(thread -> !thread.getStackTrace().isEmpty()) + ); + Assertions.assertTrue( + response.getThreads().stream().allMatch(thread -> countStackFrames(thread.getStackTrace()) <= 100) + ); + } + + @Test + public void test_stackTraceEndpointWithMaxStackTraceFrameDepth() + { + final StackTraceCollector.ThreadStackTraceResponse response = cluster.callApi().serviceClient().onAnyBroker( + mapper -> new RequestBuilder(HttpMethod.GET, "/status/stack?maxStackTraceFrameDepth=10"), + new TypeReference<>(){} + ); + + Assertions.assertTrue( + response.getThreads().stream().allMatch(thread -> countStackFrames(thread.getStackTrace()) <= 10) + ); + } + + @Test + public void test_stackTraceEndpointRejectsInvalidMaxStackTraceFrameDepth() + { + for (final String invalidDepth : new String[]{"9", "1001", "10.5"}) { + final RuntimeException exception = Assertions.assertThrows( + RuntimeException.class, + () -> cluster.callApi().serviceClient().onAnyBroker( + mapper -> new RequestBuilder( + HttpMethod.GET, + StringUtils.format("/status/stack?maxStackTraceFrameDepth=%s", invalidDepth) + ), + new TypeReference<StackTraceCollector.ThreadStackTraceResponse>(){} + ) + ); + Assertions.assertTrue(exception.getMessage().contains("400 Bad Request"), exception.getMessage()); + Assertions.assertTrue( + exception.getMessage().contains(StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY), + exception.getMessage() + ); + } + } + + @Test + public void test_stackTraceTable() + { + final String brokerHost = StringUtils.format("localhost:%s", BROKER_PORT); + final String result = cluster.runSql( + "SELECT server, service_name, node_roles, collected_at, thread_id, " + + "thread_state, daemon, priority, cpu_time_ns, user_cpu_time_ns, is_deadlocked, error_message " + + "FROM sys.stack_trace WHERE server = '%s'", + brokerHost + ); + + Assertions.assertFalse(result.isEmpty(), "The stack trace table should return broker threads"); + for (final String row : result.split("\\n")) { + final String[] columns = row.split(",", -1); + Assertions.assertEquals(brokerHost, columns[0]); + Assertions.assertEquals(BROKER_SERVICE, columns[1]); + Assertions.assertEquals("broker", columns[2]); + Assertions.assertFalse(columns[3].isEmpty()); + Assertions.assertDoesNotThrow(() -> Long.parseLong(columns[4])); Review Comment: ## CodeQL / Missing catch of NumberFormatException Potential uncaught 'java.lang.NumberFormatException'. [Show more details](https://github.com/apache/druid/security/code-scanning/11459) ########## server/src/test/java/org/apache/druid/server/StatusResourceTest.java: ########## @@ -101,6 +109,175 @@ Assert.assertEquals(false, response.getEntity()); } + @Test + public void testStackTrace() + { + final StatusResource resource = new StatusResource(new Properties(), null, null, null, null); + final Response httpResponse = resource.getStackTrace(null); + Assert.assertEquals(Response.Status.OK.getStatusCode(), httpResponse.getStatus()); + final StackTraceCollector.ThreadStackTraceResponse response = + (StackTraceCollector.ThreadStackTraceResponse) httpResponse.getEntity(); + + Assert.assertNotNull(response.getCollectedAt()); + Assert.assertFalse(response.getThreads().isEmpty()); + + final long currentThreadId = Thread.currentThread().getId(); + final StackTraceCollector.ThreadStackTrace currentThread = response.getThreads() + .stream() + .filter(thread -> thread.getThreadId() == currentThreadId) + .findFirst() + .orElse(null); + Assert.assertNotNull(currentThread); + Assert.assertEquals(Thread.currentThread().getName(), currentThread.getThreadName()); + Assert.assertEquals(Thread.currentThread().getState().name(), currentThread.getThreadState()); + Assert.assertFalse(currentThread.getStackTrace().isEmpty()); + Assert.assertTrue(currentThread.getStackTrace().contains("\n\tat ")); + Assert.assertTrue( + currentThread.getStackTrace().lines().filter(line -> line.startsWith("\tat ")).count() > 8 + ); + Assert.assertFalse(currentThread.getStackTrace().contains("\t...\n")); + if (JvmUtils.isThreadCpuTimeEnabled()) { + Assert.assertNotNull(currentThread.getCpuTimeNs()); + Assert.assertNotNull(currentThread.getUserCpuTimeNs()); + } + } + + @Test + public void testStackTraceWithMaxStackTraceFrameDepth() + { + final StatusResource resource = new StatusResource(new Properties(), null, null, null, null); + final Response httpResponse = resource.getStackTrace("10"); + Assert.assertEquals(Response.Status.OK.getStatusCode(), httpResponse.getStatus()); + final StackTraceCollector.ThreadStackTraceResponse response = + (StackTraceCollector.ThreadStackTraceResponse) httpResponse.getEntity(); + + Assert.assertTrue( + response.getThreads() + .stream() + .allMatch( + thread -> thread.getStackTrace() + .lines() + .filter(line -> line.startsWith("\tat ")) + .count() <= 10 + ) + ); + } + + @Test + public void testStackTraceRejectsInvalidMaxStackTraceFrameDepth() + { + final StatusResource resource = new StatusResource(new Properties(), null, null, null, null); + + for (final String invalidDepth : ImmutableList.of("-1", "0", "9", "1001", "10.5", "not-an-integer")) { + final Response response = resource.getStackTrace(invalidDepth); + Assert.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(), response.getStatus()); + Assert.assertTrue(response.getEntity() instanceof ErrorResponse); + final DruidException exception = ((ErrorResponse) response.getEntity()).getUnderlyingException(); + Assert.assertEquals(DruidException.Category.INVALID_INPUT, exception.getCategory()); + Assert.assertTrue(exception.getMessage().contains(StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY)); + } + } + + @Test + public void testStackTraceFormatsWaitingLockOnStackFrame() throws Exception + { + final Object monitor = new Object(); + final CountDownLatch enteredMonitor = new CountDownLatch(1); + final Thread waitingThread = new Thread( + () -> { + synchronized (monitor) { + enteredMonitor.countDown(); + try { + monitor.wait(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + }, + "stack-trace-waiting-thread" + ); + waitingThread.start(); + + try { + Assert.assertTrue(enteredMonitor.await(5, TimeUnit.SECONDS)); + final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (waitingThread.getState() != Thread.State.WAITING && System.nanoTime() < deadline) { + Thread.yield(); + } + + final StackTraceCollector.ThreadStackTrace thread = new StackTraceCollector().collect() + .getThreads() + .stream() + .filter(stackTrace -> stackTrace.getThreadId() == waitingThread.getId()) + .findFirst() + .orElse(null); + Assert.assertNotNull(thread); + Assert.assertEquals(Thread.State.WAITING.name(), thread.getThreadState()); + Assert.assertTrue( + thread.getStackTrace().contains(" - waiting on " + thread.getLockName() + "\n") + ); + Assert.assertFalse(thread.getStackTrace().contains("\n\t- waiting on ")); + } + finally { + waitingThread.interrupt(); + waitingThread.join(TimeUnit.SECONDS.toMillis(5)); + } + } + + @Test + public void testStackTraceFormatsHeldMonitorAndSynchronizer() throws Exception + { + final ThreadMXBean threadMxBean = ManagementFactory.getThreadMXBean(); + Assume.assumeTrue(threadMxBean.isObjectMonitorUsageSupported()); + Assume.assumeTrue(threadMxBean.isSynchronizerUsageSupported()); + + final Object monitor = new Object(); + final ReentrantLock synchronizer = new ReentrantLock(); + final CountDownLatch locksHeld = new CountDownLatch(1); + final CountDownLatch releaseLocks = new CountDownLatch(1); + final Thread lockHolder = new Thread( + () -> { + synchronizer.lock(); + try { + synchronized (monitor) { + locksHeld.countDown(); + try { + releaseLocks.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } + finally { + synchronizer.unlock(); + } + }, + "stack-trace-lock-holder" + ); + lockHolder.start(); + + try { + Assert.assertTrue(locksHeld.await(5, TimeUnit.SECONDS)); + final StackTraceCollector.ThreadStackTrace thread = new StackTraceCollector().collect() + .getThreads() + .stream() + .filter(stackTrace -> stackTrace.getThreadId() == lockHolder.getId()) Review Comment: ## CodeQL / Deprecated method or constructor invocation Invoking [Thread.getId](1) should be avoided because it has been deprecated. [Show more details](https://github.com/apache/druid/security/code-scanning/11465) ########## embedded-tests/src/test/java/org/apache/druid/testing/embedded/schema/SystemStackTraceTableTest.java: ########## @@ -0,0 +1,230 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF 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 + * + * http://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 org.apache.druid.testing.embedded.schema; + +import com.fasterxml.jackson.core.type.TypeReference; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.query.http.ClientSqlQuery; +import org.apache.druid.rpc.RequestBuilder; +import org.apache.druid.server.StackTraceCollector; +import org.apache.druid.sql.http.ResultFormat; +import org.apache.druid.testing.embedded.EmbeddedBroker; +import org.apache.druid.testing.embedded.EmbeddedCoordinator; +import org.apache.druid.testing.embedded.EmbeddedDruidCluster; +import org.apache.druid.testing.embedded.EmbeddedOverlord; +import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase; +import org.jboss.netty.handler.codec.http.HttpMethod; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.Map; + +public class SystemStackTraceTableTest extends EmbeddedClusterTestBase +{ + private static final String BROKER_PORT = "9082"; + private static final String BROKER_SERVICE = "test/broker"; + private static final String OVERLORD_PORT = "9090"; + private static final String OVERLORD_SERVICE = "test/overlord"; + private static final String COORDINATOR_PORT = "9081"; + private static final String COORDINATOR_SERVICE = "test/coordinator"; + + private final EmbeddedBroker broker = new EmbeddedBroker() + .addProperty("druid.service", BROKER_SERVICE) + .addProperty("druid.plaintextPort", BROKER_PORT); + + private final EmbeddedOverlord overlord = new EmbeddedOverlord() + .addProperty("druid.service", OVERLORD_SERVICE) + .addProperty("druid.plaintextPort", OVERLORD_PORT); + + private final EmbeddedCoordinator coordinator = new EmbeddedCoordinator() + .addProperty("druid.service", COORDINATOR_SERVICE) + .addProperty("druid.plaintextPort", COORDINATOR_PORT); + + @Override + protected EmbeddedDruidCluster createCluster() + { + return EmbeddedDruidCluster + .withZookeeper() + .addServer(coordinator) + .addServer(overlord) + .addServer(broker); + } + + @Test + public void test_stackTraceEndpoint() + { + final StackTraceCollector.ThreadStackTraceResponse response = cluster.callApi().serviceClient().onAnyBroker( + mapper -> new RequestBuilder(HttpMethod.GET, "/status/stack"), + new TypeReference<>(){} + ); + + Assertions.assertNotNull(response.getCollectedAt()); + Assertions.assertFalse(response.getThreads().isEmpty()); + Assertions.assertTrue( + response.getThreads().stream().allMatch(thread -> !thread.getThreadName().isEmpty()) + ); + Assertions.assertTrue( + response.getThreads().stream().anyMatch(thread -> !thread.getStackTrace().isEmpty()) + ); + Assertions.assertTrue( + response.getThreads().stream().allMatch(thread -> countStackFrames(thread.getStackTrace()) <= 100) + ); + } + + @Test + public void test_stackTraceEndpointWithMaxStackTraceFrameDepth() + { + final StackTraceCollector.ThreadStackTraceResponse response = cluster.callApi().serviceClient().onAnyBroker( + mapper -> new RequestBuilder(HttpMethod.GET, "/status/stack?maxStackTraceFrameDepth=10"), + new TypeReference<>(){} + ); + + Assertions.assertTrue( + response.getThreads().stream().allMatch(thread -> countStackFrames(thread.getStackTrace()) <= 10) + ); + } + + @Test + public void test_stackTraceEndpointRejectsInvalidMaxStackTraceFrameDepth() + { + for (final String invalidDepth : new String[]{"9", "1001", "10.5"}) { + final RuntimeException exception = Assertions.assertThrows( + RuntimeException.class, + () -> cluster.callApi().serviceClient().onAnyBroker( + mapper -> new RequestBuilder( + HttpMethod.GET, + StringUtils.format("/status/stack?maxStackTraceFrameDepth=%s", invalidDepth) + ), + new TypeReference<StackTraceCollector.ThreadStackTraceResponse>(){} + ) + ); + Assertions.assertTrue(exception.getMessage().contains("400 Bad Request"), exception.getMessage()); + Assertions.assertTrue( + exception.getMessage().contains(StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY), + exception.getMessage() + ); + } + } + + @Test + public void test_stackTraceTable() + { + final String brokerHost = StringUtils.format("localhost:%s", BROKER_PORT); + final String result = cluster.runSql( + "SELECT server, service_name, node_roles, collected_at, thread_id, " + + "thread_state, daemon, priority, cpu_time_ns, user_cpu_time_ns, is_deadlocked, error_message " + + "FROM sys.stack_trace WHERE server = '%s'", + brokerHost + ); + + Assertions.assertFalse(result.isEmpty(), "The stack trace table should return broker threads"); + for (final String row : result.split("\\n")) { + final String[] columns = row.split(",", -1); + Assertions.assertEquals(brokerHost, columns[0]); + Assertions.assertEquals(BROKER_SERVICE, columns[1]); + Assertions.assertEquals("broker", columns[2]); + Assertions.assertFalse(columns[3].isEmpty()); + Assertions.assertDoesNotThrow(() -> Long.parseLong(columns[4])); + Assertions.assertFalse(columns[6].isEmpty()); + Assertions.assertTrue(columns[6].equals("0") || columns[6].equals("1"), row); + Assertions.assertDoesNotThrow(() -> Long.parseLong(columns[7])); + if (!columns[8].isEmpty()) { + Assertions.assertDoesNotThrow(() -> Long.parseLong(columns[8])); Review Comment: ## CodeQL / Missing catch of NumberFormatException Potential uncaught 'java.lang.NumberFormatException'. [Show more details](https://github.com/apache/druid/security/code-scanning/11461) ########## server/src/test/java/org/apache/druid/server/StatusResourceTest.java: ########## @@ -101,6 +109,175 @@ Assert.assertEquals(false, response.getEntity()); } + @Test + public void testStackTrace() + { + final StatusResource resource = new StatusResource(new Properties(), null, null, null, null); + final Response httpResponse = resource.getStackTrace(null); + Assert.assertEquals(Response.Status.OK.getStatusCode(), httpResponse.getStatus()); + final StackTraceCollector.ThreadStackTraceResponse response = + (StackTraceCollector.ThreadStackTraceResponse) httpResponse.getEntity(); + + Assert.assertNotNull(response.getCollectedAt()); + Assert.assertFalse(response.getThreads().isEmpty()); + + final long currentThreadId = Thread.currentThread().getId(); + final StackTraceCollector.ThreadStackTrace currentThread = response.getThreads() + .stream() + .filter(thread -> thread.getThreadId() == currentThreadId) + .findFirst() + .orElse(null); + Assert.assertNotNull(currentThread); + Assert.assertEquals(Thread.currentThread().getName(), currentThread.getThreadName()); + Assert.assertEquals(Thread.currentThread().getState().name(), currentThread.getThreadState()); + Assert.assertFalse(currentThread.getStackTrace().isEmpty()); + Assert.assertTrue(currentThread.getStackTrace().contains("\n\tat ")); + Assert.assertTrue( + currentThread.getStackTrace().lines().filter(line -> line.startsWith("\tat ")).count() > 8 + ); + Assert.assertFalse(currentThread.getStackTrace().contains("\t...\n")); + if (JvmUtils.isThreadCpuTimeEnabled()) { + Assert.assertNotNull(currentThread.getCpuTimeNs()); + Assert.assertNotNull(currentThread.getUserCpuTimeNs()); + } + } + + @Test + public void testStackTraceWithMaxStackTraceFrameDepth() + { + final StatusResource resource = new StatusResource(new Properties(), null, null, null, null); + final Response httpResponse = resource.getStackTrace("10"); + Assert.assertEquals(Response.Status.OK.getStatusCode(), httpResponse.getStatus()); + final StackTraceCollector.ThreadStackTraceResponse response = + (StackTraceCollector.ThreadStackTraceResponse) httpResponse.getEntity(); + + Assert.assertTrue( + response.getThreads() + .stream() + .allMatch( + thread -> thread.getStackTrace() + .lines() + .filter(line -> line.startsWith("\tat ")) + .count() <= 10 + ) + ); + } + + @Test + public void testStackTraceRejectsInvalidMaxStackTraceFrameDepth() + { + final StatusResource resource = new StatusResource(new Properties(), null, null, null, null); + + for (final String invalidDepth : ImmutableList.of("-1", "0", "9", "1001", "10.5", "not-an-integer")) { + final Response response = resource.getStackTrace(invalidDepth); + Assert.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(), response.getStatus()); + Assert.assertTrue(response.getEntity() instanceof ErrorResponse); + final DruidException exception = ((ErrorResponse) response.getEntity()).getUnderlyingException(); + Assert.assertEquals(DruidException.Category.INVALID_INPUT, exception.getCategory()); + Assert.assertTrue(exception.getMessage().contains(StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY)); + } + } + + @Test + public void testStackTraceFormatsWaitingLockOnStackFrame() throws Exception + { + final Object monitor = new Object(); + final CountDownLatch enteredMonitor = new CountDownLatch(1); + final Thread waitingThread = new Thread( + () -> { + synchronized (monitor) { + enteredMonitor.countDown(); + try { + monitor.wait(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + }, + "stack-trace-waiting-thread" + ); + waitingThread.start(); + + try { + Assert.assertTrue(enteredMonitor.await(5, TimeUnit.SECONDS)); + final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (waitingThread.getState() != Thread.State.WAITING && System.nanoTime() < deadline) { + Thread.yield(); + } + + final StackTraceCollector.ThreadStackTrace thread = new StackTraceCollector().collect() + .getThreads() + .stream() + .filter(stackTrace -> stackTrace.getThreadId() == waitingThread.getId()) Review Comment: ## CodeQL / Deprecated method or constructor invocation Invoking [Thread.getId](1) should be avoided because it has been deprecated. [Show more details](https://github.com/apache/druid/security/code-scanning/11464) ########## sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemStackTraceTable.java: ########## @@ -0,0 +1,378 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF 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 + * + * http://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 org.apache.druid.sql.calcite.schema; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.base.Preconditions; +import org.apache.calcite.DataContext; +import org.apache.calcite.linq4j.Enumerable; +import org.apache.calcite.linq4j.Linq4j; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeFactory; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.schema.ProjectableFilterableTable; +import org.apache.calcite.schema.Schema; +import org.apache.calcite.schema.impl.AbstractTable; +import org.apache.druid.discovery.DiscoveryDruidNode; +import org.apache.druid.discovery.DruidNodeDiscoveryProvider; +import org.apache.druid.error.InvalidInput; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.java.util.http.client.HttpClient; +import org.apache.druid.java.util.http.client.Request; +import org.apache.druid.java.util.http.client.response.StringFullResponseHandler; +import org.apache.druid.java.util.http.client.response.StringFullResponseHolder; +import org.apache.druid.query.QueryContexts; +import org.apache.druid.segment.column.ColumnType; +import org.apache.druid.segment.column.RowSignature; +import org.apache.druid.server.DruidNode; +import org.apache.druid.server.StackTraceCollector; +import org.apache.druid.server.security.AuthenticationResult; +import org.apache.druid.server.security.AuthorizerMapper; +import org.apache.druid.sql.calcite.planner.PlannerContext; +import org.apache.druid.sql.calcite.table.RowSignatures; +import org.jboss.netty.handler.codec.http.HttpMethod; + +import javax.annotation.Nullable; +import javax.servlet.http.HttpServletResponse; +import java.net.URL; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * System schema table {@code sys.stack_trace} that contains a live Java thread-stack snapshot for + * explicitly selected Druid servers. + */ +public class SystemStackTraceTable extends AbstractTable implements ProjectableFilterableTable +{ + private static final Logger log = new Logger(SystemStackTraceTable.class); + + public static final String TABLE_NAME = "stack_trace"; + + static final RowSignature ROW_SIGNATURE = RowSignature + .builder() + .add("server", ColumnType.STRING) + .add("service_name", ColumnType.STRING) + .add("node_roles", ColumnType.STRING) + .add("collected_at", ColumnType.STRING) + .add("thread_id", ColumnType.LONG) + .add("thread_name", ColumnType.STRING) + .add("thread_state", ColumnType.STRING) + .add("daemon", ColumnType.LONG) + .add("priority", ColumnType.LONG) + .add("cpu_time_ns", ColumnType.LONG) + .add("user_cpu_time_ns", ColumnType.LONG) + .add("lock_name", ColumnType.STRING) + .add("lock_owner_id", ColumnType.LONG) + .add("lock_owner_name", ColumnType.STRING) + .add("is_deadlocked", ColumnType.LONG) + .add("stack", ColumnType.STRING) + .add("error_message", ColumnType.STRING) + .build(); + + private static final int SERVER_INDEX = ROW_SIGNATURE.indexOf("server"); + private static final int SERVICE_NAME_INDEX = ROW_SIGNATURE.indexOf("service_name"); + private static final int NODE_ROLES_INDEX = ROW_SIGNATURE.indexOf("node_roles"); + private static final int COLLECTED_AT_INDEX = ROW_SIGNATURE.indexOf("collected_at"); + private static final int THREAD_ID_INDEX = ROW_SIGNATURE.indexOf("thread_id"); + private static final int THREAD_NAME_INDEX = ROW_SIGNATURE.indexOf("thread_name"); + private static final int THREAD_STATE_INDEX = ROW_SIGNATURE.indexOf("thread_state"); + private static final int DAEMON_INDEX = ROW_SIGNATURE.indexOf("daemon"); + private static final int PRIORITY_INDEX = ROW_SIGNATURE.indexOf("priority"); + private static final int CPU_TIME_NS_INDEX = ROW_SIGNATURE.indexOf("cpu_time_ns"); + private static final int USER_CPU_TIME_NS_INDEX = ROW_SIGNATURE.indexOf("user_cpu_time_ns"); + private static final int LOCK_NAME_INDEX = ROW_SIGNATURE.indexOf("lock_name"); + private static final int LOCK_OWNER_ID_INDEX = ROW_SIGNATURE.indexOf("lock_owner_id"); + private static final int LOCK_OWNER_NAME_INDEX = ROW_SIGNATURE.indexOf("lock_owner_name"); + private static final int IS_DEADLOCKED_INDEX = ROW_SIGNATURE.indexOf("is_deadlocked"); + private static final int STACK_INDEX = ROW_SIGNATURE.indexOf("stack"); + private static final int ERROR_MESSAGE_INDEX = ROW_SIGNATURE.indexOf("error_message"); + + private final DruidNodeDiscoveryProvider druidNodeDiscoveryProvider; + private final AuthorizerMapper authorizerMapper; + private final HttpClient httpClient; + private final ObjectMapper jsonMapper; + + public SystemStackTraceTable( + final DruidNodeDiscoveryProvider druidNodeDiscoveryProvider, + final AuthorizerMapper authorizerMapper, + final HttpClient httpClient, + final ObjectMapper jsonMapper + ) + { + this.druidNodeDiscoveryProvider = druidNodeDiscoveryProvider; + this.authorizerMapper = authorizerMapper; + this.httpClient = httpClient; + this.jsonMapper = jsonMapper; + } + + @Override + public RelDataType getRowType(final RelDataTypeFactory typeFactory) + { + return RowSignatures.toRelDataType(ROW_SIGNATURE, typeFactory); + } + + @Override + public Schema.TableType getJdbcTableType() + { + return Schema.TableType.SYSTEM_TABLE; + } + + @Override + public Enumerable<Object[]> scan( + final DataContext root, + final List<RexNode> filters, + @Nullable final int[] projects + ) + { + final AuthenticationResult authenticationResult = (AuthenticationResult) Preconditions.checkNotNull( + root.get(PlannerContext.DATA_CTX_AUTHENTICATION_RESULT), + "authenticationResult in dataContext" + ); + SystemSchema.checkStateReadAccessForServers(authenticationResult, authorizerMapper); + + final Set<String> serverFilter = SystemSchemaFilters.extractColumnValues(filters, SERVER_INDEX); + InvalidInput.conditionalException( + serverFilter != null, + "sys.stack_trace requires a filter on the server column using '=' or 'IN'" + ); + final int maxStackTraceFrameDepth = getMaxStackTraceFrameDepth( + root.get(StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY) + ); + + final Iterator<DiscoveryDruidNode> druidServers = SystemSchema.getDruidServers(druidNodeDiscoveryProvider); + final Map<String, ServerStackTraceTarget> serverToTargetMap = new HashMap<>(); + druidServers.forEachRemaining(discoveryDruidNode -> { + final DruidNode druidNode = discoveryDruidNode.getDruidNode(); + final String server = druidNode.getHostAndPortToUse(); + if (!serverFilter.contains(server)) { + return; + } + + final String nodeRole = discoveryDruidNode.getNodeRole().getJsonName(); + final ServerStackTraceTarget target = serverToTargetMap.get(server); + if (target == null) { + serverToTargetMap.put( + server, + new ServerStackTraceTarget( + server, + druidNode.getServiceName(), + new ArrayList<>(Collections.singletonList(nodeRole)), + druidNode + ) + ); + } else { + target.addNodeRole(nodeRole); + } + }); + + final List<Object[]> rows = new ArrayList<>(); + for (final ServerStackTraceTarget target : serverToTargetMap.values()) { + rows.addAll(target.buildRows(this, projects, maxStackTraceFrameDepth)); + } + return Linq4j.asEnumerable(rows); + } + + static int getMaxStackTraceFrameDepth(@Nullable final Object value) + { + return StackTraceCollector.validateMaxStackTraceFrameDepth( + QueryContexts.getAsLong( + StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY, + value, + StackTraceCollector.DEFAULT_MAX_STACK_TRACE_FRAME_DEPTH + ) + ); + } + + private static Object[] projectRow(final Object[] row, @Nullable final int[] projects) + { + if (projects == null) { + return row; + } + final Object[] projectedRow = new Object[projects.length]; + for (int i = 0; i < projects.length; i++) { + projectedRow[i] = row[projects[i]]; + } + return projectedRow; + } + + private StackTraceResult getStackTrace( + final DruidNode druidNode, + final int maxStackTraceFrameDepth + ) + { + final String url = druidNode.getUriToUse().resolve( + StringUtils.format( + "/status/stack?%s=%d", + StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY, + maxStackTraceFrameDepth + ) + ).toString(); + try { + final Request request = new Request(HttpMethod.GET, new URL(url)); Review Comment: ## CodeQL / Deprecated method or constructor invocation Invoking [URL.URL](1) should be avoided because it has been deprecated. [Show more details](https://github.com/apache/druid/security/code-scanning/11466) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
