This is an automated email from the ASF dual-hosted git repository. FrankChen021 pushed a commit to branch codex/native-sys-servers in repository https://gitbox.apache.org/repos/asf/druid.git
commit 1ee0e62a3c6e23479318ce45de713e70555cae2e Author: Frank Chen <[email protected]> AuthorDate: Wed Sep 2 18:01:11 2026 +0800 feat(sql): add native sys.servers support --- docs/querying/sql-metadata-tables.md | 3 +- .../embedded/query/NativeSysServersQueryTest.java | 79 ++++++ .../server/system/module/SystemTableModule.java | 9 +- .../system/table/ServersTableDataProvider.java | 292 +++++++++++++++++++++ .../system/table/ServersTableDescriptor.java | 102 +++++++ .../system/ServersTableDataProviderTest.java | 210 +++++++++++++++ .../sql/calcite/schema/NativeServersTable.java | 70 +++++ .../druid/sql/calcite/schema/SystemSchema.java | 40 +-- .../druid/sql/calcite/schema/SystemSchemaTest.java | 10 +- .../schema/SystemTableDataProviderTest.java | 8 + 10 files changed, 789 insertions(+), 34 deletions(-) diff --git a/docs/querying/sql-metadata-tables.md b/docs/querying/sql-metadata-tables.md index dc9c4cf03bd..0ffa44fc47c 100644 --- a/docs/querying/sql-metadata-tables.md +++ b/docs/querying/sql-metadata-tables.md @@ -170,6 +170,7 @@ execution: |-----|--------------| |[`sys.tasks`](#tasks-table)|The Overlord that owns task state. Supported filters are pushed into task storage when the configured task storage implementation supports filter pushdown.| |[`sys.server_properties`](#server_properties-table)|The Druid server processes discovered in the cluster. Filters on `server` and `service_name` can avoid reading properties from nodes that don't match.| +|[`sys.servers`](#servers-table)|The current Coordinator leader's discovered-cluster view of Druid servers.| After Druid retrieves the system-table rows, the native engine applies the remaining filters, expressions, aggregations, sorting, and result processing. A system table that doesn't advertise native query support continues to @@ -286,7 +287,7 @@ Servers table lists all discovered servers in the cluster. |start_time|STRING|Timestamp in ISO8601 format when the server was announced in the cluster| |version|VARCHAR|Druid version running on the server| |build_revision|VARCHAR|The git commit of the build that produced the server binary| -|labels|VARCHAR|Labels for the server configured using the property [`druid.labels`](../configuration/index.md)| +|labels|JSON|Labels for the server configured using the property [`druid.labels`](../configuration/index.md)| |available_processors|BIGINT|Total number of CPU processors available to the server| |total_memory|BIGINT|Total memory in bytes available to the server| diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/NativeSysServersQueryTest.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/NativeSysServersQueryTest.java new file mode 100644 index 00000000000..26153238293 --- /dev/null +++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/NativeSysServersQueryTest.java @@ -0,0 +1,79 @@ +/* + * 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.query; + +import org.apache.druid.query.QueryContexts; +import org.apache.druid.sql.calcite.planner.PlannerContext; +import org.apache.druid.sql.calcite.run.NativeSqlEngine; +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.junit.jupiter.api.Assertions; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.Map; + +public class NativeSysServersQueryTest extends EmbeddedClusterTestBase +{ + private final EmbeddedBroker broker = new EmbeddedBroker() + .addProperty("druid.labels", "{\"environment\":\"test\"}"); + + @Override + protected EmbeddedDruidCluster createCluster() + { + return EmbeddedDruidCluster.withEmbeddedDerbyAndZookeeper() + .useLatchableEmitter() + .addServer(new EmbeddedCoordinator()) + .addServer(new EmbeddedOverlord()) + .addServer(broker); + } + + @ParameterizedTest(name = "plannerStrategy = {0}") + @ValueSource(strings = { + QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED, + QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED + }) + public void testLabelsAreQueryableAsJson(final String plannerStrategy) + { + final String result = cluster.runSql( + "SELECT JSON_VALUE(labels, '$.environment') " + + "FROM sys.servers " + + "WHERE server_type = 'broker'", + nativeQueryContext(plannerStrategy) + ); + + Assertions.assertEquals("test", result); + } + + private static Map<String, Object> nativeQueryContext(final String plannerStrategy) + { + return Map.of( + QueryContexts.ENGINE, + NativeSqlEngine.NAME, + PlannerContext.CTX_USE_NATIVE_QUERY_FOR_SYSTEM_TABLES, + true, + QueryContexts.CTX_NATIVE_QUERY_SQL_PLANNING_MODE, + plannerStrategy + ); + } +} diff --git a/server/src/main/java/org/apache/druid/server/system/module/SystemTableModule.java b/server/src/main/java/org/apache/druid/server/system/module/SystemTableModule.java index 2f83d2347ea..799aefef061 100644 --- a/server/src/main/java/org/apache/druid/server/system/module/SystemTableModule.java +++ b/server/src/main/java/org/apache/druid/server/system/module/SystemTableModule.java @@ -27,12 +27,14 @@ import org.apache.druid.guice.LazySingleton; import org.apache.druid.query.SystemTableDataSource; import org.apache.druid.server.system.table.ServerPropertiesTableDataProvider; import org.apache.druid.server.system.table.ServerPropertiesTableDescriptor; +import org.apache.druid.server.system.table.ServersTableDataProvider; +import org.apache.druid.server.system.table.ServersTableDescriptor; import org.apache.druid.server.system.table.SystemTableDataProvider; import org.apache.druid.server.system.table.SystemTableDescriptor; import org.apache.druid.server.system.table.TaskTableDescriptor; /** - * Registers native system-table routing and the node-local server-properties supplier. + * Registers native system-table routing and the built-in system-table suppliers. * * <p>Table-specific integrations, such as the task supplier in indexing-service, contribute their own entries to the * native system-table multibinders.</p> @@ -50,6 +52,8 @@ public class SystemTableModule implements Module final MapBinder<String, SystemTableDescriptor> descriptorBinder = MapBinder.newMapBinder(binder, String.class, SystemTableDescriptor.class); descriptorBinder.addBinding(ServerPropertiesTableDescriptor.TABLE_NAME) .toInstance(new ServerPropertiesTableDescriptor()); + descriptorBinder.addBinding(ServersTableDescriptor.TABLE_NAME) + .toInstance(new ServersTableDescriptor()); descriptorBinder.addBinding(TaskTableDescriptor.TABLE_NAME) .toInstance(new TaskTableDescriptor()); @@ -57,5 +61,8 @@ public class SystemTableModule implements Module dataProviderBinder.addBinding(ServerPropertiesTableDescriptor.TABLE_NAME) .to(ServerPropertiesTableDataProvider.class) .in(LazySingleton.class); + dataProviderBinder.addBinding(ServersTableDescriptor.TABLE_NAME) + .to(ServersTableDataProvider.class) + .in(LazySingleton.class); } } diff --git a/server/src/main/java/org/apache/druid/server/system/table/ServersTableDataProvider.java b/server/src/main/java/org/apache/druid/server/system/table/ServersTableDataProvider.java new file mode 100644 index 00000000000..696309f5ef7 --- /dev/null +++ b/server/src/main/java/org/apache/druid/server/system/table/ServersTableDataProvider.java @@ -0,0 +1,292 @@ +/* + * 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.system.table; + +import com.google.inject.Inject; +import org.apache.druid.client.DruidServer; +import org.apache.druid.client.FilteredServerInventoryView; +import org.apache.druid.client.coordinator.CoordinatorClient; +import org.apache.druid.common.guava.FutureUtils; +import org.apache.druid.discovery.DataNodeService; +import org.apache.druid.discovery.DiscoveryDruidNode; +import org.apache.druid.discovery.DruidNodeDiscoveryProvider; +import org.apache.druid.discovery.NodeRole; +import org.apache.druid.java.util.common.ISE; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.query.filter.DimFilter; +import org.apache.druid.rpc.indexing.OverlordClient; +import org.apache.druid.server.DruidNode; +import org.apache.druid.server.security.Action; +import org.apache.druid.server.security.AuthenticationResult; +import org.apache.druid.server.security.AuthorizationResult; +import org.apache.druid.server.security.AuthorizationUtils; +import org.apache.druid.server.security.AuthorizerMapper; +import org.apache.druid.server.security.ForbiddenException; +import org.apache.druid.server.security.Resource; +import org.apache.druid.server.security.ResourceAction; + +import javax.annotation.Nullable; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; + +/** Native row supplier for {@code sys.servers}. */ +public class ServersTableDataProvider implements SystemTableDataProvider +{ + // This is used for maxSize and currentSize when they are unknown. + // The unknown size doesn't have to be 0, it's better to be null. + // However, this table is returning 0 for them for some reason and we keep the behavior for backwards compatibility. + // Maybe we can remove this and return nulls instead when we remove the bindable query path which is currently + // used to query system tables. + private static final long UNKNOWN_SIZE = 0L; + + private final DruidNodeDiscoveryProvider druidNodeDiscoveryProvider; + private final FilteredServerInventoryView serverInventoryView; + private final AuthorizerMapper authorizerMapper; + private final OverlordClient overlordClient; + private final CoordinatorClient coordinatorClient; + + @Inject + public ServersTableDataProvider( + final DruidNodeDiscoveryProvider druidNodeDiscoveryProvider, + final FilteredServerInventoryView serverInventoryView, + final AuthorizerMapper authorizerMapper, + final OverlordClient overlordClient, + final CoordinatorClient coordinatorClient + ) + { + this.druidNodeDiscoveryProvider = druidNodeDiscoveryProvider; + this.serverInventoryView = serverInventoryView; + this.authorizerMapper = authorizerMapper; + this.overlordClient = overlordClient; + this.coordinatorClient = coordinatorClient; + } + + @Override + public Iterable<Object[]> getRows( + final List<DimFilter> filters, + final AuthenticationResult internalAuthenticationResult + ) + { + authorizeServerRead(internalAuthenticationResult); + + final Iterator<DiscoveryDruidNode> druidServers = getDruidServers(); + String tmpCoordinatorLeader = ""; + String tmpOverlordLeader = ""; + + try { + tmpCoordinatorLeader = FutureUtils.getUnchecked(coordinatorClient.findCurrentLeader(), true).toString(); + } + catch (Exception ignored) { + // no reason to kill the results if something is sad and there are no leaders + } + + try { + tmpOverlordLeader = FutureUtils.getUnchecked(overlordClient.findCurrentLeader(), true).toString(); + } + catch (Exception ignored) { + // no reason to kill the results if something is sad and there are no leaders + } + + final String coordinatorLeader = tmpCoordinatorLeader; + final String overlordLeader = tmpOverlordLeader; + final List<Object[]> rows = new ArrayList<>(); + + druidServers.forEachRemaining(discoveryDruidNode -> { + final DataNodeService dataNodeService = discoveryDruidNode.getService( + DataNodeService.DISCOVERY_SERVICE_KEY, + DataNodeService.class + ); + final boolean isDiscoverableDataServer = isDiscoverableDataServer(dataNodeService); + final NodeRole serverRole = discoveryDruidNode.getNodeRole(); + + if (isDiscoverableDataServer) { + final DruidServer druidServer = serverInventoryView.getInventoryValue( + discoveryDruidNode.getDruidNode().getHostAndPortToUse() + ); + if (druidServer != null || NodeRole.HISTORICAL.equals(serverRole)) { + // Build a row for the data server if that server is in the server view, or the node type is historical. + // The historicals are usually supposed to be found in the server view. If some historicals are + // missing, it could mean that there are some problems in them to announce themselves. We just fill + // their status with nulls in this case. + rows.add(buildRowForDiscoverableDataServer(discoveryDruidNode, druidServer)); + } else { + rows.add(buildRowForNonDataServer(discoveryDruidNode)); + } + } else if (NodeRole.COORDINATOR.equals(serverRole)) { + rows.add( + buildRowForNonDataServerWithLeadership( + discoveryDruidNode, + coordinatorLeader.contains(discoveryDruidNode.getDruidNode().getHostAndPortToUse()) + ) + ); + } else if (NodeRole.OVERLORD.equals(serverRole)) { + rows.add( + buildRowForNonDataServerWithLeadership( + discoveryDruidNode, + overlordLeader.contains(discoveryDruidNode.getDruidNode().getHostAndPortToUse()) + ) + ); + } else { + rows.add(buildRowForNonDataServer(discoveryDruidNode)); + } + }); + + return rows; + } + + private void authorizeServerRead(final AuthenticationResult authenticationResult) + { + final AuthorizationResult authorizationResult = AuthorizationUtils.authorizeAllResourceActions( + authenticationResult, + Collections.singletonList(new ResourceAction(Resource.STATE_RESOURCE, Action.READ)), + authorizerMapper + ); + if (!authorizationResult.allowAccessWithNoRestriction()) { + throw new ForbiddenException( + "Insufficient permission to view servers: " + authorizationResult.getErrorMessage() + ); + } + } + + private Iterator<DiscoveryDruidNode> getDruidServers() + { + return Arrays.stream(NodeRole.values()) + .flatMap(nodeRole -> druidNodeDiscoveryProvider.getForNodeRole(nodeRole).getAllNodes().stream()) + .iterator(); + } + + private Object[] buildRowForNonDataServer(final DiscoveryDruidNode discoveryDruidNode) + { + final DruidNode node = discoveryDruidNode.getDruidNode(); + return new Object[]{ + node.getHostAndPortToUse(), + node.getHost(), + (long) node.getPlaintextPort(), + (long) node.getTlsPort(), + StringUtils.toLowerCase(discoveryDruidNode.getNodeRole().toString()), + null, + UNKNOWN_SIZE, + UNKNOWN_SIZE, + UNKNOWN_SIZE, + null, + toStringOrNull(discoveryDruidNode.getStartTime()), + node.getVersion(), + node.getBuildRevision(), + node.getLabels(), + (long) discoveryDruidNode.getAvailableProcessors(), + discoveryDruidNode.getTotalMemory() + }; + } + + private Object[] buildRowForNonDataServerWithLeadership( + final DiscoveryDruidNode discoveryDruidNode, + final boolean isLeader + ) + { + final DruidNode node = discoveryDruidNode.getDruidNode(); + return new Object[]{ + node.getHostAndPortToUse(), + node.getHost(), + (long) node.getPlaintextPort(), + (long) node.getTlsPort(), + StringUtils.toLowerCase(discoveryDruidNode.getNodeRole().toString()), + null, + UNKNOWN_SIZE, + UNKNOWN_SIZE, + UNKNOWN_SIZE, + isLeader ? 1L : 0L, + toStringOrNull(discoveryDruidNode.getStartTime()), + node.getVersion(), + node.getBuildRevision(), + node.getLabels(), + (long) discoveryDruidNode.getAvailableProcessors(), + discoveryDruidNode.getTotalMemory() + }; + } + + private Object[] buildRowForDiscoverableDataServer( + final DiscoveryDruidNode discoveryDruidNode, + @Nullable final DruidServer serverFromInventoryView + ) + { + final DruidNode node = discoveryDruidNode.getDruidNode(); + final DruidServer druidServerToUse = serverFromInventoryView == null + ? toDruidServer(discoveryDruidNode) + : serverFromInventoryView; + final long currentSize = serverFromInventoryView == null + ? UNKNOWN_SIZE + : serverFromInventoryView.getCurrSize(); + return new Object[]{ + node.getHostAndPortToUse(), + node.getHost(), + (long) node.getPlaintextPort(), + (long) node.getTlsPort(), + StringUtils.toLowerCase(discoveryDruidNode.getNodeRole().toString()), + druidServerToUse.getTier(), + currentSize, + druidServerToUse.getMaxSize(), + druidServerToUse.getStorageSize(), + null, + toStringOrNull(discoveryDruidNode.getStartTime()), + node.getVersion(), + node.getBuildRevision(), + node.getLabels(), + (long) discoveryDruidNode.getAvailableProcessors(), + discoveryDruidNode.getTotalMemory() + }; + } + + private static boolean isDiscoverableDataServer(@Nullable final DataNodeService dataNodeService) + { + return dataNodeService != null && dataNodeService.isDiscoverable(); + } + + private static DruidServer toDruidServer(final DiscoveryDruidNode discoveryDruidNode) + { + final DruidNode druidNode = discoveryDruidNode.getDruidNode(); + final DataNodeService dataNodeService = discoveryDruidNode.getService( + DataNodeService.DISCOVERY_SERVICE_KEY, + DataNodeService.class + ); + if (isDiscoverableDataServer(dataNodeService)) { + return new DruidServer( + druidNode.getHostAndPortToUse(), + druidNode.getHostAndPort(), + druidNode.getHostAndTlsPort(), + dataNodeService.getMaxSize(), + dataNodeService.getStorageSize(), + dataNodeService.getServerType(), + dataNodeService.getTier(), + dataNodeService.getPriority() + ); + } else { + throw new ISE("[%s] is not a discoverable data server", discoveryDruidNode); + } + } + + @Nullable + private static String toStringOrNull(@Nullable final Object object) + { + return object == null ? null : object.toString(); + } +} diff --git a/server/src/main/java/org/apache/druid/server/system/table/ServersTableDescriptor.java b/server/src/main/java/org/apache/druid/server/system/table/ServersTableDescriptor.java new file mode 100644 index 00000000000..e198dbeed23 --- /dev/null +++ b/server/src/main/java/org/apache/druid/server/system/table/ServersTableDescriptor.java @@ -0,0 +1,102 @@ +/* + * 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.system.table; + +import org.apache.druid.discovery.NodeRole; +import org.apache.druid.segment.column.ColumnType; +import org.apache.druid.segment.column.RowSignature; +import org.apache.druid.server.security.Action; +import org.apache.druid.server.security.AuthorizationResult; +import org.apache.druid.server.security.AuthorizationUtils; +import org.apache.druid.server.security.ForbiddenException; +import org.apache.druid.server.security.Resource; +import org.apache.druid.server.security.ResourceAction; + +import java.util.Collections; +import java.util.Set; + +/** Descriptor for the native {@code sys.servers} table. */ +public class ServersTableDescriptor implements SystemTableDescriptor +{ + public static final String TABLE_NAME = "servers"; + public static final RowSignature ROW_SIGNATURE = RowSignature + .builder() + .add("server", ColumnType.STRING) + .add("host", ColumnType.STRING) + .add("plaintext_port", ColumnType.LONG) + .add("tls_port", ColumnType.LONG) + .add("server_type", ColumnType.STRING) + .add("tier", ColumnType.STRING) + .add("curr_size", ColumnType.LONG) + .add("max_size", ColumnType.LONG) + .add("storage_size", ColumnType.LONG) + .add("is_leader", ColumnType.LONG) + .add("start_time", ColumnType.STRING) + .add("version", ColumnType.STRING) + .add("build_revision", ColumnType.STRING) + .add("labels", ColumnType.NESTED_DATA) + .add("available_processors", ColumnType.LONG) + .add("total_memory", ColumnType.LONG) + .build(); + + private static final Set<NodeRole> NODE_ROLES = Set.of(NodeRole.COORDINATOR); + private static final SystemTableRowAuthorizer ROW_AUTHORIZER = (rows, authenticationResult, authorizerMapper) -> { + final AuthorizationResult authorizationResult = AuthorizationUtils.authorizeAllResourceActions( + authenticationResult, + Collections.singletonList(new ResourceAction(Resource.STATE_RESOURCE, Action.READ)), + authorizerMapper + ); + if (!authorizationResult.allowAccessWithNoRestriction()) { + throw new ForbiddenException(authorizationResult.getErrorMessage()); + } + return rows; + }; + + @Override + public String getTableName() + { + return TABLE_NAME; + } + + @Override + public Set<NodeRole> getNodeRoles() + { + return NODE_ROLES; + } + + @Override + public SystemTableRoutingMode getRoutingMode() + { + return SystemTableRoutingMode.LEADER_ONLY; + } + + @Override + public RowSignature getRowSignature() + { + return ROW_SIGNATURE; + } + + @Override + public SystemTableRowAuthorizer getRowAuthorizer() + { + return ROW_AUTHORIZER; + } + +} diff --git a/server/src/test/java/org/apache/druid/server/system/ServersTableDataProviderTest.java b/server/src/test/java/org/apache/druid/server/system/ServersTableDataProviderTest.java new file mode 100644 index 00000000000..1fa2e4df385 --- /dev/null +++ b/server/src/test/java/org/apache/druid/server/system/ServersTableDataProviderTest.java @@ -0,0 +1,210 @@ +/* + * 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.system; + +import com.google.common.collect.ImmutableMap; +import com.google.common.util.concurrent.Futures; +import org.apache.druid.client.DruidServer; +import org.apache.druid.client.FilteredServerInventoryView; +import org.apache.druid.client.coordinator.CoordinatorClient; +import org.apache.druid.discovery.DataNodeService; +import org.apache.druid.discovery.DiscoveryDruidNode; +import org.apache.druid.discovery.DruidNodeDiscovery; +import org.apache.druid.discovery.DruidNodeDiscoveryProvider; +import org.apache.druid.discovery.DruidService; +import org.apache.druid.discovery.NodeRole; +import org.apache.druid.rpc.indexing.OverlordClient; +import org.apache.druid.segment.column.ColumnType; +import org.apache.druid.server.DruidNode; +import org.apache.druid.server.coordination.ServerType; +import org.apache.druid.server.security.Access; +import org.apache.druid.server.security.AuthConfig; +import org.apache.druid.server.security.AuthenticationResult; +import org.apache.druid.server.security.Authorizer; +import org.apache.druid.server.security.AuthorizerMapper; +import org.apache.druid.server.security.ForbiddenException; +import org.apache.druid.server.system.table.ServersTableDataProvider; +import org.apache.druid.server.system.table.ServersTableDescriptor; +import org.apache.druid.server.system.table.SystemTableRoutingMode; +import org.easymock.EasyMock; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.net.URI; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Set; + +public class ServersTableDataProviderTest +{ + private static final AuthenticationResult AUTHENTICATION_RESULT = + new AuthenticationResult("test-user", AuthConfig.ALLOW_ALL_NAME, null, null); + + @Test + public void testReturnsDiscoveredServersWithJsonLabels() throws Exception + { + final DruidNodeDiscoveryProvider discoveryProvider = EasyMock.mock(DruidNodeDiscoveryProvider.class); + final FilteredServerInventoryView serverInventoryView = EasyMock.mock(FilteredServerInventoryView.class); + final CoordinatorClient coordinatorClient = EasyMock.mock(CoordinatorClient.class); + final OverlordClient overlordClient = EasyMock.mock(OverlordClient.class); + + final DiscoveryDruidNode coordinator = discoveryNode( + new DruidNode("coordinator", "localhost", false, 8081, null, true, false), + NodeRole.COORDINATOR, + Collections.emptyMap() + ); + final Map<String, String> labels = ImmutableMap.of("environment", "test"); + final DiscoveryDruidNode broker = discoveryNode( + new DruidNode("broker", "localhost", false, 8082, null, null, true, false, labels), + NodeRole.BROKER, + ImmutableMap.of( + DataNodeService.DISCOVERY_SERVICE_KEY, + new DataNodeService("tier", 1_000L, 900L, ServerType.BROKER, 0) + ) + ); + + expectDiscoveryNodes(discoveryProvider, NodeRole.COORDINATOR, coordinator); + expectDiscoveryNodes(discoveryProvider, NodeRole.BROKER, broker); + for (final NodeRole nodeRole : NodeRole.values()) { + if (!NodeRole.COORDINATOR.equals(nodeRole) && !NodeRole.BROKER.equals(nodeRole)) { + expectDiscoveryNodes(discoveryProvider, nodeRole); + } + } + + final DruidServer server = EasyMock.mock(DruidServer.class); + EasyMock.expect(serverInventoryView.getInventoryValue("localhost:8082")).andReturn(server).once(); + EasyMock.expect(server.getCurrSize()).andReturn(100L).once(); + EasyMock.expect(server.getTier()).andReturn("tier").once(); + EasyMock.expect(server.getMaxSize()).andReturn(1_000L).once(); + EasyMock.expect(server.getStorageSize()).andReturn(900L).once(); + EasyMock.expect(coordinatorClient.findCurrentLeader()) + .andReturn(Futures.immediateFuture(new URI("localhost:8081"))) + .once(); + EasyMock.expect(overlordClient.findCurrentLeader()) + .andReturn(Futures.immediateFuture(new URI("localhost:8090"))) + .once(); + + EasyMock.replay(discoveryProvider, serverInventoryView, coordinatorClient, overlordClient, server); + + final ServersTableDataProvider provider = new ServersTableDataProvider( + discoveryProvider, + serverInventoryView, + allowAllAuthorizerMapper(), + overlordClient, + coordinatorClient + ); + final List<Object[]> rows = toRows(provider.getRows(Collections.emptyList(), AUTHENTICATION_RESULT)); + + final Object[] coordinatorRow = rows.stream() + .filter(row -> "localhost:8081".equals(row[0])) + .findFirst() + .orElseThrow(); + Assertions.assertEquals(1L, coordinatorRow[9]); + Assertions.assertNull(coordinatorRow[13]); + + final Object[] brokerRow = rows.stream() + .filter(row -> "localhost:8082".equals(row[0])) + .findFirst() + .orElseThrow(); + Assertions.assertEquals(labels, brokerRow[13]); + Assertions.assertEquals(100L, brokerRow[6]); + Assertions.assertEquals(ColumnType.NESTED_DATA, ServersTableDescriptor.ROW_SIGNATURE.getColumnType(13).orElseThrow()); + + EasyMock.verify(discoveryProvider, serverInventoryView, coordinatorClient, overlordClient, server); + } + + @Test + public void testRejectsUnauthorizedRequest() + { + final Authorizer denyAll = (authenticationResult, resource, action) -> Access.DENIED; + final AuthorizerMapper authorizerMapper = new AuthorizerMapper(null) + { + @Override + public Authorizer getAuthorizer(final String name) + { + return denyAll; + } + }; + final ServersTableDataProvider provider = new ServersTableDataProvider( + EasyMock.mock(DruidNodeDiscoveryProvider.class), + EasyMock.mock(FilteredServerInventoryView.class), + authorizerMapper, + EasyMock.mock(OverlordClient.class), + EasyMock.mock(CoordinatorClient.class) + ); + + Assertions.assertThrows( + ForbiddenException.class, + () -> provider.getRows(Collections.emptyList(), AUTHENTICATION_RESULT) + ); + } + + @Test + public void testDescriptorRoutesToCoordinatorLeader() + { + final ServersTableDescriptor descriptor = new ServersTableDescriptor(); + + Assertions.assertEquals(Set.of(NodeRole.COORDINATOR), descriptor.getNodeRoles()); + Assertions.assertEquals(SystemTableRoutingMode.LEADER_ONLY, descriptor.getRoutingMode()); + Assertions.assertEquals(ColumnType.NESTED_DATA, descriptor.getRowSignature().getColumnType(13).orElseThrow()); + } + + private static DiscoveryDruidNode discoveryNode( + final DruidNode node, + final NodeRole nodeRole, + final Map<String, DruidService> services + ) + { + return new DiscoveryDruidNode(node, nodeRole, services, null); + } + + private static void expectDiscoveryNodes( + final DruidNodeDiscoveryProvider discoveryProvider, + final NodeRole nodeRole, + final DiscoveryDruidNode... nodes + ) + { + final DruidNodeDiscovery discovery = EasyMock.mock(DruidNodeDiscovery.class); + EasyMock.expect(discoveryProvider.getForNodeRole(nodeRole)).andReturn(discovery).once(); + EasyMock.expect(discovery.getAllNodes()).andReturn(List.of(nodes)).once(); + EasyMock.replay(discovery); + } + + private static AuthorizerMapper allowAllAuthorizerMapper() + { + return new AuthorizerMapper(null) + { + @Override + public Authorizer getAuthorizer(final String name) + { + return (authenticationResult, resource, action) -> Access.OK; + } + }; + } + + private static List<Object[]> toRows(final Iterable<Object[]> rows) + { + final List<Object[]> result = new ArrayList<>(); + rows.forEach(result::add); + return result; + } +} diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/schema/NativeServersTable.java b/sql/src/main/java/org/apache/druid/sql/calcite/schema/NativeServersTable.java new file mode 100644 index 00000000000..645289e09b6 --- /dev/null +++ b/sql/src/main/java/org/apache/druid/sql/calcite/schema/NativeServersTable.java @@ -0,0 +1,70 @@ +/* + * 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 org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.logical.LogicalTableScan; +import org.apache.calcite.schema.Schema; +import org.apache.druid.query.DataSource; +import org.apache.druid.query.SystemTableDataSource; +import org.apache.druid.server.system.table.ServersTableDescriptor; +import org.apache.druid.sql.calcite.table.DruidTable; + +/** Native-query representation of {@code sys.servers}. */ +class NativeServersTable extends DruidTable +{ + private static final DataSource DATA_SOURCE = new SystemTableDataSource(ServersTableDescriptor.TABLE_NAME); + + NativeServersTable() + { + super(ServersTableDescriptor.ROW_SIGNATURE); + } + + @Override + public DataSource getDataSource() + { + return DATA_SOURCE; + } + + @Override + public boolean isJoinable() + { + return false; + } + + @Override + public boolean isBroadcast() + { + return false; + } + + @Override + public Schema.TableType getJdbcTableType() + { + return Schema.TableType.SYSTEM_TABLE; + } + + @Override + public RelNode toRel(RelOptTable.ToRelContext context, RelOptTable table) + { + return LogicalTableScan.create(context.getCluster(), table, context.getTableHints()); + } +} diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java b/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java index 050c66a8817..db988d5ac36 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java @@ -56,7 +56,6 @@ import org.apache.druid.indexer.TaskStatusPlus; import org.apache.druid.indexing.overlord.supervisor.SupervisorStatus; import org.apache.druid.java.util.common.ISE; import org.apache.druid.java.util.common.StringUtils; -import org.apache.druid.java.util.common.jackson.JacksonUtils; import org.apache.druid.java.util.common.parsers.CloseableIterator; import org.apache.druid.java.util.http.client.HttpClient; import org.apache.druid.rpc.indexing.OverlordClient; @@ -74,6 +73,7 @@ import org.apache.druid.server.security.ForbiddenException; import org.apache.druid.server.security.Resource; import org.apache.druid.server.security.ResourceAction; import org.apache.druid.server.security.ResourceType; +import org.apache.druid.server.system.table.ServersTableDescriptor; import org.apache.druid.server.system.table.TaskTableDescriptor; import org.apache.druid.sql.calcite.planner.PlannerConfig; import org.apache.druid.sql.calcite.planner.PlannerContext; @@ -199,25 +199,7 @@ public class SystemSchema extends AbstractTableSchema } ); - static final RowSignature SERVERS_SIGNATURE = RowSignature - .builder() - .add("server", ColumnType.STRING) - .add("host", ColumnType.STRING) - .add("plaintext_port", ColumnType.LONG) - .add("tls_port", ColumnType.LONG) - .add("server_type", ColumnType.STRING) - .add("tier", ColumnType.STRING) - .add("curr_size", ColumnType.LONG) - .add("max_size", ColumnType.LONG) - .add("storage_size", ColumnType.LONG) - .add("is_leader", ColumnType.LONG) - .add("start_time", ColumnType.STRING) - .add("version", ColumnType.STRING) - .add("build_revision", ColumnType.STRING) - .add("labels", ColumnType.STRING) - .add("available_processors", ColumnType.LONG) - .add("total_memory", ColumnType.LONG) - .build(); + static final RowSignature SERVERS_SIGNATURE = ServersTableDescriptor.ROW_SIGNATURE; static final RowSignature SERVER_SEGMENTS_SIGNATURE = RowSignature .builder() @@ -326,7 +308,6 @@ public class SystemSchema extends AbstractTableSchema authorizerMapper, overlordClient, coordinatorClient, - jsonMapper, authenticationResult ); case SERVER_SEGMENTS_TABLE -> new ServerSegmentsTable(serverView, authorizerMapper, authenticationResult); @@ -654,7 +635,7 @@ public class SystemSchema extends AbstractTableSchema * This table contains row per server. It contains all the discovered servers in Druid cluster. * Some columns like tier and size are only applicable to historical nodes which contain segments. */ - static class ServersTable extends AbstractTable implements ScannableTable + static class ServersTable extends AbstractTable implements ScannableTable, NativeSystemTable { // This is used for maxSize and currentSize when they are unknown. // The unknown size doesn't have to be 0, it's better to be null. @@ -668,7 +649,6 @@ public class SystemSchema extends AbstractTableSchema private final FilteredServerInventoryView serverInventoryView; private final OverlordClient overlordClient; private final CoordinatorClient coordinatorClient; - private final ObjectMapper jsonMapper; private final AuthenticationResult authenticationResult; public ServersTable( @@ -677,7 +657,6 @@ public class SystemSchema extends AbstractTableSchema AuthorizerMapper authorizerMapper, OverlordClient overlordClient, CoordinatorClient coordinatorClient, - ObjectMapper jsonMapper, AuthenticationResult authenticationResult ) { @@ -686,7 +665,6 @@ public class SystemSchema extends AbstractTableSchema this.serverInventoryView = serverInventoryView; this.overlordClient = overlordClient; this.coordinatorClient = coordinatorClient; - this.jsonMapper = jsonMapper; this.authenticationResult = authenticationResult; } @@ -702,6 +680,12 @@ public class SystemSchema extends AbstractTableSchema return TableType.SYSTEM_TABLE; } + @Override + public DruidTable asNativeTable() + { + return new NativeServersTable(); + } + @Override public Enumerable<Object[]> scan(DataContext root) { @@ -788,7 +772,7 @@ public class SystemSchema extends AbstractTableSchema toStringOrNull(discoveryDruidNode.getStartTime()), node.getVersion(), node.getBuildRevision(), - node.getLabels() == null ? null : JacksonUtils.writeValueAsString(jsonMapper, node.getLabels()), + node.getLabels(), (long) discoveryDruidNode.getAvailableProcessors(), discoveryDruidNode.getTotalMemory() }; @@ -817,7 +801,7 @@ public class SystemSchema extends AbstractTableSchema toStringOrNull(discoveryDruidNode.getStartTime()), node.getVersion(), node.getBuildRevision(), - node.getLabels() == null ? null : JacksonUtils.writeValueAsString(jsonMapper, node.getLabels()), + node.getLabels(), (long) discoveryDruidNode.getAvailableProcessors(), discoveryDruidNode.getTotalMemory() }; @@ -858,7 +842,7 @@ public class SystemSchema extends AbstractTableSchema toStringOrNull(discoveryDruidNode.getStartTime()), node.getVersion(), node.getBuildRevision(), - node.getLabels() == null ? null : JacksonUtils.writeValueAsString(jsonMapper, node.getLabels()), + node.getLabels(), (long) discoveryDruidNode.getAvailableProcessors(), discoveryDruidNode.getTotalMemory() }; diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaTest.java index fbb027fdda5..e7839a8bd3b 100644 --- a/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaTest.java +++ b/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaTest.java @@ -896,7 +896,6 @@ public class SystemSchemaTest extends CalciteTestBase authMapper, overlordClient, coordinatorClient, - MAPPER, createAuthResult(Users.SUPER) ) .createMock(); @@ -1135,7 +1134,7 @@ public class SystemSchemaTest extends CalciteTestBase startTimeStr, version, buildRevision, - "{\"brokerKey\":\"brokerValue\",\"brokerKey2\":\"brokerValue2\"}", + ImmutableMap.of("brokerKey", "brokerValue", "brokerKey2", "brokerValue2"), availableProcessors, totalMemory ) @@ -1175,7 +1174,7 @@ public class SystemSchemaTest extends CalciteTestBase startTimeStr, version, buildRevision, - "{\"overlordKey\":\"overlordValue\"}", + ImmutableMap.of("overlordKey", "overlordValue"), availableProcessors, totalMemory ) @@ -1319,7 +1318,7 @@ public class SystemSchemaTest extends CalciteTestBase String startTime, String version, String buildRevision, - String labels, + @Nullable Map<String, String> labels, long availableProcessors, long totalMemory ) @@ -2523,6 +2522,9 @@ public class SystemSchemaTest extends CalciteTestBase case STRING: expectedClass = String.class; break; + case COMPLEX: + expectedClass = Object.class; + break; default: throw new IAE("Don't know what class to expect for valueType[%s]", columnType); } diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemTableDataProviderTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemTableDataProviderTest.java index c9334532940..3a660a8ae22 100644 --- a/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemTableDataProviderTest.java +++ b/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemTableDataProviderTest.java @@ -22,6 +22,7 @@ package org.apache.druid.sql.calcite.schema; import org.apache.calcite.plan.RelOptTable; import org.apache.calcite.schema.Schema; import org.apache.druid.query.SystemTableDataSource; +import org.apache.druid.server.system.table.ServersTableDescriptor; import org.apache.druid.sql.calcite.planner.PlannerContext; import org.apache.druid.sql.calcite.run.NativeSqlEngine; import org.apache.druid.sql.calcite.run.SqlEngine; @@ -50,6 +51,13 @@ public class SystemTableDataProviderTest Assertions.assertFalse(serverProperties.isJoinable()); Assertions.assertFalse(serverProperties.isBroadcast()); Assertions.assertEquals(Schema.TableType.SYSTEM_TABLE, serverProperties.getJdbcTableType()); + + final NativeServersTable servers = new NativeServersTable(); + Assertions.assertEquals("servers", ((SystemTableDataSource) servers.getDataSource()).getTable()); + Assertions.assertEquals(ServersTableDescriptor.ROW_SIGNATURE, servers.getRowSignature()); + Assertions.assertFalse(servers.isJoinable()); + Assertions.assertFalse(servers.isBroadcast()); + Assertions.assertEquals(Schema.TableType.SYSTEM_TABLE, servers.getJdbcTableType()); } @Test --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
