mirror of https://github.com/apache/cassandra
expose gossip information in system_views.gossip_info virtual table
patch by Francisco Guerrero; reviewed by Stefan Miklosovic and Yifan Cai for CASSANDRA-17002 This commit adds a new virtual table that exposes the gossip information in tabular format. The information is the same as the information presented through the `nodetool gossipinfo` command, but the virtual table splits the version and value from `VersionedValue` into two different columns. This is intented to help clients reading the vtable without the need of parsing the version:value information (as it currently stands in gossipinfo). The token value does not have a column. This is consistent with the gossipinfo output which always renders ":<hidden>" for the Token value. Only the token_version column is available.
This commit is contained in:
Francisco Guerrero
Stefan Miklosovic
parent
935bcf1e87
commit
61be4d8362
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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<ApplicationState> 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<InetAddressAndPort, EndpointState> 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();
|
||||
}
|
||||
}
|
||||
|
|
@ -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());
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -159,8 +159,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
|
|||
@VisibleForTesting
|
||||
final Set<InetAddressAndPort> seeds = new ConcurrentSkipListSet<>();
|
||||
|
||||
/* map where key is the endpoint and value is the state associated with the endpoint */
|
||||
final ConcurrentMap<InetAddressAndPort, EndpointState> 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<InetAddressAndPort, EndpointState> 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
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<Map.Entry<InetAddressAndPort, EndpointState>> 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();
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue