diff --git a/src/java/org/apache/cassandra/metrics/ClientMetrics.java b/src/java/org/apache/cassandra/metrics/ClientMetrics.java index 8ca3480376b0..36660adaa02d 100644 --- a/src/java/org/apache/cassandra/metrics/ClientMetrics.java +++ b/src/java/org/apache/cassandra/metrics/ClientMetrics.java @@ -18,27 +18,86 @@ */ package org.apache.cassandra.metrics; +import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; import java.util.concurrent.Callable; +import org.apache.cassandra.transport.Connection; +import org.apache.cassandra.transport.Server; + import com.codahale.metrics.Gauge; import com.codahale.metrics.Meter; - -import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics; +import com.google.common.collect.ImmutableMap; public class ClientMetrics { private static final MetricNameFactory factory = new DefaultNameFactory("Client"); - + + public static final String USER = "user"; + public static final String ADDRESS = "address"; + public static final String VERSION = "version"; + public static final String KEYSPACE = "keyspace"; + public static final String PROTOCOL = "protocol"; + public static final String CIPHER = "cipher"; + public static final String DRIVER_VERSION = "driverVersion"; + public static final String DRIVER_NAME = "driverName"; + public static final String SSL = "ssl"; + public static final String REQUESTS = "requests"; + public static final ClientMetrics instance = new ClientMetrics(); - + public boolean initialized = false; + + private Collection servers; + private ClientMetrics() { } - public void addGauge(String name, final Callable provider) + public List getConnectionStates() + { + if (servers == null) + return Collections.emptyList(); + List connections = new ArrayList<>(); + for (Server s : servers) + { + connections.addAll(s.getConnectionStates()); + } + return connections; + } + + public int getConnectedNativeClients() { - Metrics.register(factory.createMetricName(name), (Gauge) () -> { + int ret = 0; + for (Server server : servers) + ret += server.getConnectedClients(); + return ret; + } + + public Map getConnectedNativeClientsByUser() + { + Map result = new HashMap<>(); + for (Server server : servers) + { + for (Entry e : server.getConnectedClientsByUser().entrySet()) + { + String user = e.getKey(); + result.put(user, result.getOrDefault(user, 0) + e.getValue()); + } + } + return result; + } + + public Gauge addGauge(String name, final Callable provider) + { + return Metrics.register(factory.createMetricName(name), (Gauge) () -> { try { return provider.call(); @@ -53,4 +112,48 @@ public Meter addMeter(String name) { return Metrics.meter(factory.createMetricName(name)); } + + public synchronized void init(Collection servers) + { + this.servers = servers; + if (initialized) return; + initialized = true; + + // register metrics + addGauge("connectedNativeClients", () -> getConnectedNativeClients()); + addGauge("connectedNativeClientsByUser", () -> getConnectedNativeClientsByUser()); + addGauge("connections", () -> + { + List> result = new ArrayList<>(); + for (Server server : servers) + { + for (Connection.View connection : server.getConnectionStates()) + { + result.add(new ImmutableMap.Builder() + .put(USER, connection.getUser()) + .put(ADDRESS, connection.getAddress().toString()) + .put(VERSION, String.valueOf(connection.getVersion())) + .put(REQUESTS, String.valueOf(connection.getRequests())) + .put(SSL, Boolean.toString(connection.sslEnabled())) + .put(DRIVER_NAME, connection.getDriverName().orElse("undefined")) + .put(DRIVER_VERSION, connection.getDriverVersion().orElse("undefined")) + .put(CIPHER, connection.getSSLCipher().orElse("undefined")) + .put(PROTOCOL, connection.getSSLProtocol().orElse("undefined")) + .put(KEYSPACE, connection.getKeyspace().orElse("")) + .build()); + } + } + return result; + }); + addGauge("clientsByProtocolVersion", () -> + { + List> result = new ArrayList<>(); + for (Server server : servers) + { + result.addAll(server.getClientsByProtocolVersion()); + } + return result; + }); + } + } diff --git a/src/java/org/apache/cassandra/service/NativeTransportService.java b/src/java/org/apache/cassandra/service/NativeTransportService.java index 39b334e0dfb7..718baa7305ae 100644 --- a/src/java/org/apache/cassandra/service/NativeTransportService.java +++ b/src/java/org/apache/cassandra/service/NativeTransportService.java @@ -115,50 +115,7 @@ synchronized void initialize() } // register metrics - ClientMetrics.instance.addGauge("connectedNativeClients", () -> - { - int ret = 0; - for (Server server : servers) - ret += server.getConnectedClients(); - return ret; - }); - ClientMetrics.instance.addGauge("connectedNativeClientsByUser", () -> - { - Map result = new HashMap<>(); - for (Server server : servers) - { - for (Entry e : server.getConnectedClientsByUser().entrySet()) - { - String user = e.getKey(); - result.put(user, result.getOrDefault(user, 0) + e.getValue()); - } - } - return result; - }); - - ClientMetrics.instance.addGauge("connections", () -> - { - List> result = new ArrayList<>(); - for (Server server : servers) - { - for (Map e : server.getConnectionStates()) - { - result.add(e); - } - } - return result; - }); - - ClientMetrics.instance.addGauge("clientsByProtocolVersion", () -> - { - List> result = new ArrayList<>(); - for (Server server : servers) - { - result.addAll(server.getClientsByProtocolVersion()); - } - return result; - }); - + ClientMetrics.instance.init(servers); AuthMetrics.init(); initialized = true; diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index 01769b4105d4..caaa337e0d25 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/src/java/org/apache/cassandra/tools/NodeProbe.java @@ -1517,7 +1517,7 @@ public Object getCompactionMetric(String metricName) /** * Retrieve Proxy metrics - * @param connections, connectedNativeClients, connectedNativeClientsByUser + * @param connections, connectedNativeClients, connectedNativeClientsByUser, clientsByProtocolVersion */ public Object getClientMetric(String metricName) { diff --git a/src/java/org/apache/cassandra/tools/nodetool/ClientStats.java b/src/java/org/apache/cassandra/tools/nodetool/ClientStats.java index 5bd5da19d235..759b21a714f8 100644 --- a/src/java/org/apache/cassandra/tools/nodetool/ClientStats.java +++ b/src/java/org/apache/cassandra/tools/nodetool/ClientStats.java @@ -23,6 +23,7 @@ import java.util.Map; import java.util.Map.Entry; +import org.apache.cassandra.metrics.ClientMetrics; import org.apache.cassandra.tools.NodeProbe; import org.apache.cassandra.tools.NodeTool.NodeToolCmd; import org.apache.cassandra.tools.nodetool.formatter.TableBuilder; @@ -87,8 +88,16 @@ public void execute(NodeProbe probe) table.add("Address", "SSL", "Cipher", "Protocol", "Version", "User", "Keyspace", "Requests", "Driver-Name", "Driver-Version"); for (Map conn : clients) { - table.add(conn.get("address"), conn.get("ssl"), conn.get("cipher"), conn.get("protocol"), conn.get("version"), - conn.get("user"), conn.get("keyspace"), conn.get("requests"), conn.get("driverName"), conn.get("driverVersion")); + table.add(conn.get(ClientMetrics.ADDRESS), + conn.get(ClientMetrics.SSL), + conn.get(ClientMetrics.CIPHER), + conn.get(ClientMetrics.PROTOCOL), + conn.get(ClientMetrics.VERSION), + conn.get(ClientMetrics.USER), + conn.get(ClientMetrics.KEYSPACE), + conn.get(ClientMetrics.REQUESTS), + conn.get(ClientMetrics.DRIVER_NAME), + conn.get(ClientMetrics.DRIVER_VERSION)); } table.printTo(System.out); System.out.println(); diff --git a/src/java/org/apache/cassandra/transport/Connection.java b/src/java/org/apache/cassandra/transport/Connection.java index a04a0554d7fd..f95a30d9e97e 100644 --- a/src/java/org/apache/cassandra/transport/Connection.java +++ b/src/java/org/apache/cassandra/transport/Connection.java @@ -17,16 +17,19 @@ */ package org.apache.cassandra.transport; +import java.net.InetSocketAddress; +import java.util.Optional; + import io.netty.channel.Channel; import io.netty.util.AttributeKey; -public class Connection +public abstract class Connection { static final AttributeKey attributeKey = AttributeKey.valueOf("CONN"); - private final Channel channel; - private final ProtocolVersion version; - private final Tracker tracker; + protected final Channel channel; + protected final ProtocolVersion version; + protected final Tracker tracker; private volatile FrameCompressor frameCompressor; @@ -64,6 +67,8 @@ public Channel channel() return channel; } + public abstract View view(); + public interface Factory { Connection newConnection(Channel channel, ProtocolVersion version); @@ -73,4 +78,17 @@ public interface Tracker { void addConnection(Channel ch, Connection connection); } + + public interface View { + public String getUser(); + public InetSocketAddress getAddress(); + public int getVersion(); + public long getRequests(); + public boolean sslEnabled(); + public Optional getDriverName(); + public Optional getDriverVersion(); + public Optional getSSLCipher(); + public Optional getSSLProtocol(); + public Optional getKeyspace(); + } } diff --git a/src/java/org/apache/cassandra/transport/Server.java b/src/java/org/apache/cassandra/transport/Server.java index 996e5bb5b34b..070ec69fc6f0 100644 --- a/src/java/org/apache/cassandra/transport/Server.java +++ b/src/java/org/apache/cassandra/transport/Server.java @@ -24,6 +24,8 @@ import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Function; +import java.util.stream.Collectors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -179,30 +181,14 @@ public Map getConnectedClientsByUser() return connectionTracker.getConnectedClientsByUser(); } - public List> getConnectionStates() + public List getConnectionStates() { - List> result = new ArrayList<>(); - for(Channel c : connectionTracker.allChannels) + List result = new ArrayList<>(); + for (Channel c : connectionTracker.allChannels) { - Connection connection = c.attr(Connection.attributeKey).get(); - if (connection instanceof ServerConnection) - { - ServerConnection conn = (ServerConnection) connection; - SslHandler sslHandler = conn.channel().pipeline().get(SslHandler.class); - - result.add(new ImmutableMap.Builder() - .put("user", conn.getClientState().getUser().getName()) - .put("keyspace", conn.getClientState().getRawKeyspace() == null ? "" : conn.getClientState().getRawKeyspace()) - .put("address", conn.getClientState().getRemoteAddress().toString()) - .put("version", String.valueOf(conn.getVersion().asInt())) - .put("requests", String.valueOf(conn.requests.getCount())) - .put("ssl", Boolean.toString(sslHandler == null)) - .put("cipher", sslHandler != null ? sslHandler.engine().getSession().getCipherSuite() : "undefined") - .put("protocol", sslHandler != null ? sslHandler.engine().getSession().getProtocol() : "undefined") - .put("driverName", conn.getClientState().getDriverName().orElse("undefined")) - .put("driverVersion", conn.getClientState().getDriverVersion().orElse("undefined")) - .build()); - } + Connection conn = c.attr(Connection.attributeKey).get(); + if (conn != null) + result.add(conn.view()); } return result; } diff --git a/src/java/org/apache/cassandra/transport/ServerConnection.java b/src/java/org/apache/cassandra/transport/ServerConnection.java index 1ebf81c609a2..337e200fb799 100644 --- a/src/java/org/apache/cassandra/transport/ServerConnection.java +++ b/src/java/org/apache/cassandra/transport/ServerConnection.java @@ -17,10 +17,14 @@ */ package org.apache.cassandra.transport; +import java.net.InetSocketAddress; +import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import io.netty.channel.Channel; +import io.netty.handler.ssl.SslHandler; + import org.apache.cassandra.auth.IAuthenticator; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.service.ClientState; @@ -124,4 +128,62 @@ public IAuthenticator.SaslNegotiator getSaslNegotiator(QueryState queryState) saslNegotiator = DatabaseDescriptor.getAuthenticator().newSaslNegotiator(queryState.getClientAddress()); return saslNegotiator; } + + public View view() + { + return new View() + { + public String getUser() + { + return getClientState().getUser().getName(); + } + + public InetSocketAddress getAddress() + { + return getClientState().getRemoteAddress(); + } + + public int getVersion() + { + return version.asInt(); + } + + public long getRequests() + { + return requests.getCount(); + } + + public boolean sslEnabled() + { + return channel().pipeline().get(SslHandler.class) != null; + } + + public Optional getDriverName() + { + return getClientState().getDriverName(); + } + + public Optional getDriverVersion() + { + return getClientState().getDriverVersion(); + } + + public Optional getSSLCipher() + { + SslHandler sslHandler = channel().pipeline().get(SslHandler.class); + return Optional.ofNullable(sslHandler != null ? sslHandler.engine().getSession().getCipherSuite() : null); + } + + public Optional getSSLProtocol() + { + SslHandler sslHandler = channel().pipeline().get(SslHandler.class); + return Optional.ofNullable(sslHandler != null ? sslHandler.engine().getSession().getProtocol() : null); + } + + public Optional getKeyspace() + { + return Optional.ofNullable(getClientState().getRawKeyspace()); + } + }; + } } diff --git a/src/java/org/apache/cassandra/transport/SimpleClient.java b/src/java/org/apache/cassandra/transport/SimpleClient.java index db7de8dd8b92..1c87a76ab04d 100644 --- a/src/java/org/apache/cassandra/transport/SimpleClient.java +++ b/src/java/org/apache/cassandra/transport/SimpleClient.java @@ -259,7 +259,13 @@ private class Initializer extends ChannelInitializer { protected void initChannel(Channel channel) throws Exception { - connection = new Connection(channel, version, tracker); + connection = new Connection(channel, version, tracker) + { + public View view() + { + return null; + } + }; channel.attr(Connection.attributeKey).set(connection); ChannelPipeline pipeline = channel.pipeline();