diff --git a/CHANGES.txt b/CHANGES.txt index 21c163ea29..3fc2300113 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.1 + * Expose gossip information in system_views.gossip_info virtual table (CASSANDRA-17002) * Add guardrails for collection items and size (CASSANDRA-17153) * Improve guardrails messages (CASSANDRA-17430) * Remove all usages of junit.framework and ban them via Checkstyle (CASSANDRA-17316) diff --git a/doc/modules/cassandra/pages/new/virtualtables.adoc b/doc/modules/cassandra/pages/new/virtualtables.adoc index 69035aee67..6f562d1273 100644 --- a/doc/modules/cassandra/pages/new/virtualtables.adoc +++ b/doc/modules/cassandra/pages/new/virtualtables.adoc @@ -74,10 +74,12 @@ recent_hit_rate_per_second, recent_request_rate_per_second, request_count, and s |coordinator_write_latency |Records counts, keyspace_name, table_name, max, median, and per_second for coordinator writes. -|cql_metrcs |Metrics specific to CQL prepared statement caching. +|cql_metrics |Metrics specific to CQL prepared statement caching. |disk_usage |Records disk usage including disk_space, keyspace_name, and table_name, sorted by system keyspaces. +|gossip_info |Lists the gossip information for the cluster. + |internode_inbound |Lists information about the inbound internode messaging. |internode_outbound |Information about the outbound internode messaging. @@ -377,6 +379,24 @@ SELECT total - progress AS remaining FROM system_views.sstable_tasks; .... +=== Gossip Information Virtual Table + +The `gossip_info` virtual table lists the Gossip information for the cluster. An example query is as follows: + +.... +cqlsh> select address, port, generation, heartbeat, load, dc, rack from system_views.gossip_info; + + address | port | generation | heartbeat | load | dc | rack +-----------+------+------------+-----------+---------+-------------+------- + 127.0.0.1 | 7000 | 1645575140 | 312 | 70542.0 | datacenter1 | rack1 + 127.0.0.2 | 7000 | 1645575135 | 318 | 70499.0 | datacenter1 | rack1 + 127.0.0.3 | 7000 | 1645575140 | 312 | 70504.0 | datacenter1 | rack1 + 127.0.0.4 | 7000 | 1645575141 | 311 | 70502.0 | datacenter1 | rack1 + 127.0.0.5 | 7000 | 1645575136 | 315 | 70500.0 | datacenter1 | rack1 + +(5 rows) +.... + === Other Virtual Tables Some examples of using other virtual tables are as follows. diff --git a/src/java/org/apache/cassandra/db/virtual/GossipInfoTable.java b/src/java/org/apache/cassandra/db/virtual/GossipInfoTable.java new file mode 100644 index 0000000000..a515feb7d6 --- /dev/null +++ b/src/java/org/apache/cassandra/db/virtual/GossipInfoTable.java @@ -0,0 +1,174 @@ +/* + * 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.cassandra.db.virtual; + +import java.util.EnumSet; +import java.util.Map; + +import org.apache.cassandra.db.marshal.InetAddressType; +import org.apache.cassandra.db.marshal.Int32Type; +import org.apache.cassandra.db.marshal.UTF8Type; +import org.apache.cassandra.dht.LocalPartitioner; +import org.apache.cassandra.gms.ApplicationState; +import org.apache.cassandra.gms.EndpointState; +import org.apache.cassandra.gms.Gossiper; +import org.apache.cassandra.gms.VersionedValue; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.schema.TableMetadata; + +import static org.apache.cassandra.gms.ApplicationState.TOKENS; + +/** + * A {@link VirtualTable} that return the Gossip information in tabular format. + */ +final class GossipInfoTable extends AbstractVirtualTable +{ + static final String TABLE_NAME = "gossip_info"; + static final String TABLE_COMMENT = "lists the gossip information for the cluster"; + + static final String ADDRESS = "address"; + static final String PORT = "port"; + static final String HOSTNAME = "hostname"; + static final String GENERATION = "generation"; + static final String HEARTBEAT = "heartbeat"; + + static final ApplicationState[] STATES_FOR_VERSIONS = ApplicationState.values(); + static final ApplicationState[] STATES_FOR_VALUES; + + static + { + EnumSet applicationStates = EnumSet.allOf(ApplicationState.class); + // do not add a column for the ApplicationState.TOKENS value + applicationStates.remove(TOKENS); + STATES_FOR_VALUES = applicationStates.toArray(new ApplicationState[0]); + } + + /** + * Construct a new {@link GossipInfoTable} for the given {@code keyspace}. + * + * @param keyspace the name of the keyspace + */ + GossipInfoTable(String keyspace) + { + super(buildTableMetadata(keyspace)); + } + + /** + * {@inheritDoc} + */ + @Override + public DataSet data() + { + SimpleDataSet result = new SimpleDataSet(metadata()); + for (Map.Entry entry : Gossiper.instance.endpointStateMap.entrySet()) + { + InetAddressAndPort endpoint = entry.getKey(); + EndpointState localState = entry.getValue(); + + SimpleDataSet dataSet = result.row(endpoint.getAddress(), endpoint.getPort()) + .column(HOSTNAME, endpoint.getHostName()) + .column(GENERATION, getGeneration(localState)) + .column(HEARTBEAT, getHeartBeat(localState)); + + for (ApplicationState state : STATES_FOR_VALUES) + dataSet.column(state.name().toLowerCase(), getValue(localState, state)); + + for (ApplicationState state : STATES_FOR_VERSIONS) + dataSet.column(state.name().toLowerCase() + "_version", getVersion(localState, state)); + } + return result; + } + + /** + * Return the heartbeat generation of a given {@link EndpointState} or null if {@code localState} is null. + * + * @param localState a nullable endpoint state + * @return the heartbeat generation if available, null otherwise + */ + private Integer getGeneration(EndpointState localState) + { + return localState == null ? null : localState.getHeartBeatState().getGeneration(); + } + + /** + * Return the heartbeat version of a given {@link EndpointState} or null if {@code localState} is null. + * + * @param localState a nullable endpoint state + * @return the heartbeat version if available, null otherwise + */ + private Integer getHeartBeat(EndpointState localState) + { + return localState == null ? null : localState.getHeartBeatState().getHeartBeatVersion(); + } + + /** + * Returns the value from the {@link VersionedValue} of a given {@link ApplicationState key}, or null + * if {@code localState} is null or the {@link VersionedValue} does not exist in the {@link ApplicationState}. + * + * @param localState a nullable endpoint state + * @param key the key to the application state + * @return the value, or null if not available + */ + private String getValue(EndpointState localState, ApplicationState key) + { + VersionedValue value; + return localState == null || (value = localState.getApplicationState(key)) == null ? null : value.value; + } + + /** + * Returns the version from the {@link VersionedValue} of a given {@link ApplicationState key}, or null + * if {@code localState} is null or the {@link VersionedValue} does not exist in the {@link ApplicationState}. + * + * @param localState a nullable endpoint state + * @param key the key to the application state + * @return the version, or null if not available + */ + private Integer getVersion(EndpointState localState, ApplicationState key) + { + VersionedValue value; + return localState == null || (value = localState.getApplicationState(key)) == null ? null : value.version; + } + + /** + * Builds the {@link TableMetadata} to be provided to the superclass + * + * @param keyspace the name of the keyspace + * @return the TableMetadata class + */ + private static TableMetadata buildTableMetadata(String keyspace) + { + TableMetadata.Builder builder = TableMetadata.builder(keyspace, TABLE_NAME) + .comment(TABLE_COMMENT) + .kind(TableMetadata.Kind.VIRTUAL) + .partitioner(new LocalPartitioner(InetAddressType.instance)) + .addPartitionKeyColumn(ADDRESS, InetAddressType.instance) + .addClusteringColumn(PORT, Int32Type.instance) + .addRegularColumn(HOSTNAME, UTF8Type.instance) + .addRegularColumn(GENERATION, Int32Type.instance) + .addRegularColumn(HEARTBEAT, Int32Type.instance); + + for (ApplicationState state : STATES_FOR_VALUES) + builder.addRegularColumn(state.name().toLowerCase(), UTF8Type.instance); + + for (ApplicationState state : STATES_FOR_VERSIONS) + builder.addRegularColumn(state.name().toLowerCase() + "_version", Int32Type.instance); + + return builder.build(); + } +} diff --git a/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java b/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java index fc0f40a3ad..c59d7d5572 100644 --- a/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java +++ b/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java @@ -46,6 +46,7 @@ public final class SystemViewsKeyspace extends VirtualKeyspace .add(new CQLMetricsTable(VIRTUAL_VIEWS)) .add(new BatchMetricsTable(VIRTUAL_VIEWS)) .add(new StreamingVirtualTable(VIRTUAL_VIEWS)) + .add(new GossipInfoTable(VIRTUAL_VIEWS)) .build()); } } diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index 05a4f219a7..cd6d4dceaf 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -159,8 +159,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean @VisibleForTesting final Set seeds = new ConcurrentSkipListSet<>(); - /* map where key is the endpoint and value is the state associated with the endpoint */ - final ConcurrentMap endpointStateMap = new ConcurrentHashMap<>(); + /* map where key is the endpoint and value is the state associated with the endpoint. + * This is made public to be consumed by the GossipInfoTable virtual table */ + public final ConcurrentMap endpointStateMap = new ConcurrentHashMap<>(); /* map where key is endpoint and value is timestamp when this endpoint was removed from * gossip. We will ignore any gossip regarding these endpoints for QUARANTINE_DELAY time diff --git a/src/java/org/apache/cassandra/gms/HeartBeatState.java b/src/java/org/apache/cassandra/gms/HeartBeatState.java index cad6a48c23..3f633cb0fa 100644 --- a/src/java/org/apache/cassandra/gms/HeartBeatState.java +++ b/src/java/org/apache/cassandra/gms/HeartBeatState.java @@ -70,7 +70,7 @@ public class HeartBeatState return version == EMPTY_VERSION; } - int getGeneration() + public int getGeneration() { return generation; } @@ -80,7 +80,7 @@ public class HeartBeatState version = VersionGenerator.getNextVersion(); } - int getHeartBeatVersion() + public int getHeartBeatVersion() { return version; } diff --git a/test/unit/org/apache/cassandra/db/virtual/GossipInfoTableTest.java b/test/unit/org/apache/cassandra/db/virtual/GossipInfoTableTest.java new file mode 100644 index 0000000000..a2bce7d3d6 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/virtual/GossipInfoTableTest.java @@ -0,0 +1,192 @@ +/* + * 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.cassandra.db.virtual; + +import java.util.Map; +import java.util.Optional; + +import com.google.common.collect.ImmutableList; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.cql3.CQLTester; +import org.apache.cassandra.cql3.UntypedResultSet; +import org.apache.cassandra.gms.ApplicationState; +import org.apache.cassandra.gms.EndpointState; +import org.apache.cassandra.gms.Gossiper; +import org.apache.cassandra.locator.InetAddressAndPort; + +import static org.assertj.core.api.Assertions.assertThat; + +public class GossipInfoTableTest extends CQLTester +{ + private static final String KS_NAME = "vts"; + + @SuppressWarnings("FieldCanBeLocal") + private GossipInfoTable table; + + @BeforeClass + public static void setUpClass() + { + CQLTester.setUpClass(); + } + + @Before + public void config() + { + table = new GossipInfoTable(KS_NAME); + VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(KS_NAME, ImmutableList.of(table))); + } + + @Test + public void testSelectAllWhenGossipInfoIsEmpty() throws Throwable + { + assertEmpty(execute("SELECT * FROM vts.gossip_info")); + } + + @SuppressWarnings("deprecation") + @Test + public void testSelectAllWithStateTransitions() throws Throwable + { + try + { + requireNetwork(); // triggers gossiper + + UntypedResultSet resultSet = execute("SELECT * FROM vts.gossip_info"); + + assertThat(resultSet.size()).isEqualTo(1); + assertThat(Gossiper.instance.endpointStateMap.size()).isEqualTo(1); + + Optional> entry = Gossiper.instance.endpointStateMap.entrySet() + .stream() + .findFirst(); + assertThat(entry).isNotEmpty(); + + UntypedResultSet.Row row = resultSet.one(); + assertThat(row.getColumns().size()).isEqualTo(62); + + InetAddressAndPort endpoint = entry.get().getKey(); + EndpointState localState = entry.get().getValue(); + + assertThat(endpoint).isNotNull(); + assertThat(localState).isNotNull(); + assertThat(row.getInetAddress("address")).isEqualTo(endpoint.getAddress()); + assertThat(row.getInt("port")).isEqualTo(endpoint.getPort()); + assertThat(row.getString("hostname")).isEqualTo(endpoint.getHostName()); + assertThat(row.getInt("generation")).isEqualTo(localState.getHeartBeatState().getGeneration()); + + assertValue(row, "status", localState, ApplicationState.STATUS); + assertValue(row, "load", localState, ApplicationState.LOAD); + assertValue(row, "schema", localState, ApplicationState.SCHEMA); + assertValue(row, "dc", localState, ApplicationState.DC); + assertValue(row, "rack", localState, ApplicationState.RACK); + assertValue(row, "release_version", localState, ApplicationState.RELEASE_VERSION); + assertValue(row, "removal_coordinator", localState, ApplicationState.REMOVAL_COORDINATOR); + assertValue(row, "internal_ip", localState, ApplicationState.INTERNAL_IP); + assertValue(row, "rpc_address", localState, ApplicationState.RPC_ADDRESS); + assertValue(row, "severity", localState, ApplicationState.SEVERITY); + assertValue(row, "net_version", localState, ApplicationState.NET_VERSION); + assertValue(row, "host_id", localState, ApplicationState.HOST_ID); + assertValue(row, "rpc_ready", localState, ApplicationState.RPC_READY); + assertValue(row, "internal_address_and_port", localState, ApplicationState.INTERNAL_ADDRESS_AND_PORT); + assertValue(row, "native_address_and_port", localState, ApplicationState.NATIVE_ADDRESS_AND_PORT); + assertValue(row, "status_with_port", localState, ApplicationState.STATUS_WITH_PORT); + assertValue(row, "sstable_versions", localState, ApplicationState.SSTABLE_VERSIONS); + assertValue(row, "x_11_padding", localState, ApplicationState.X_11_PADDING); + assertValue(row, "x1", localState, ApplicationState.X1); + assertValue(row, "x2", localState, ApplicationState.X2); + assertValue(row, "x3", localState, ApplicationState.X3); + assertValue(row, "x4", localState, ApplicationState.X4); + assertValue(row, "x5", localState, ApplicationState.X5); + assertValue(row, "x6", localState, ApplicationState.X6); + assertValue(row, "x7", localState, ApplicationState.X7); + assertValue(row, "x8", localState, ApplicationState.X8); + assertValue(row, "x9", localState, ApplicationState.X9); + assertValue(row, "x10", localState, ApplicationState.X10); + + assertVersion(row, "status_version", localState, ApplicationState.STATUS); + assertVersion(row, "load_version", localState, ApplicationState.LOAD); + assertVersion(row, "schema_version", localState, ApplicationState.SCHEMA); + assertVersion(row, "dc_version", localState, ApplicationState.DC); + assertVersion(row, "rack_version", localState, ApplicationState.RACK); + assertVersion(row, "release_version_version", localState, ApplicationState.RELEASE_VERSION); + assertVersion(row, "removal_coordinator_version", localState, ApplicationState.REMOVAL_COORDINATOR); + assertVersion(row, "internal_ip_version", localState, ApplicationState.INTERNAL_IP); + assertVersion(row, "rpc_address_version", localState, ApplicationState.RPC_ADDRESS); + assertVersion(row, "severity_version", localState, ApplicationState.SEVERITY); + assertVersion(row, "net_version_version", localState, ApplicationState.NET_VERSION); + assertVersion(row, "host_id_version", localState, ApplicationState.HOST_ID); + assertVersion(row, "tokens_version", localState, ApplicationState.TOKENS); + assertVersion(row, "rpc_ready_version", localState, ApplicationState.RPC_READY); + assertVersion(row, "internal_address_and_port_version", localState, ApplicationState.INTERNAL_ADDRESS_AND_PORT); + assertVersion(row, "native_address_and_port_version", localState, ApplicationState.NATIVE_ADDRESS_AND_PORT); + assertVersion(row, "status_with_port_version", localState, ApplicationState.STATUS_WITH_PORT); + assertVersion(row, "sstable_versions_version", localState, ApplicationState.SSTABLE_VERSIONS); + assertVersion(row, "x_11_padding", localState, ApplicationState.X_11_PADDING); + assertVersion(row, "x1", localState, ApplicationState.X1); + assertVersion(row, "x2", localState, ApplicationState.X2); + assertVersion(row, "x3", localState, ApplicationState.X3); + assertVersion(row, "x4", localState, ApplicationState.X4); + assertVersion(row, "x5", localState, ApplicationState.X5); + assertVersion(row, "x6", localState, ApplicationState.X6); + assertVersion(row, "x7", localState, ApplicationState.X7); + assertVersion(row, "x8", localState, ApplicationState.X8); + assertVersion(row, "x9", localState, ApplicationState.X9); + assertVersion(row, "x10", localState, ApplicationState.X10); + } + finally + { + // clean up the gossip states + Gossiper.instance.clearUnsafe(); + } + } + + private void assertValue(UntypedResultSet.Row row, String column, EndpointState localState, ApplicationState key) + { + if (row.has(column)) + { + assertThat(localState.getApplicationState(key)).as("'%s' is expected to be not-null", key) + .isNotNull(); + assertThat(row.getString(column)).as("'%s' is expected to match column '%s'", key, column) + .isEqualTo(localState.getApplicationState(key).value); + } + else + { + assertThat(localState.getApplicationState(key)).as("'%s' is expected to be null", key) + .isNull(); + } + } + + private void assertVersion(UntypedResultSet.Row row, String column, EndpointState localState, ApplicationState key) + { + if (row.has(column)) + { + assertThat(localState.getApplicationState(key)).as("'%s' is expected to be not-null", key) + .isNotNull(); + assertThat(row.getInt(column)).as("'%s' is expected to match column '%s'", key, column) + .isEqualTo(localState.getApplicationState(key).version); + } + else + { + assertThat(localState.getApplicationState(key)).as("'%s' is expected to be null", key) + .isNull(); + } + } +}