diff --git a/CHANGES.txt b/CHANGES.txt index 3544594afa..b4b1245881 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -3,7 +3,8 @@ * Fix cassandra-env.sh to use $CASSANDRA_CONF to find cassandra-jaas.config (CASSANDRA-14305) * Fixed nodetool cfstats printing index name twice (CASSANDRA-14903) * Add flag to disable SASI indexes, and warnings on creation (CASSANDRA-14866) -Merged from 3.0: +Merged from 3.0:a + * Add ability to cap max negotiable protocol version (CASSANDRA-15193) * Gossip tokens on startup if available (CASSANDRA-15335) * Fix resource leak in CompressedSequentialWriter (CASSANDRA-15340) * Fix bad merge that reverted CASSANDRA-14993 (CASSANDRA-15289) diff --git a/NEWS.txt b/NEWS.txt index 2feac814ea..2aa6deba8e 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -49,6 +49,11 @@ Upgrading --------- - repair_session_max_tree_depth setting has been added to cassandra.yaml to allow operators to reduce merkle tree size if repair is creating too much heap pressure. See CASSANDRA-14096 for details. + - native_transport_max_negotiable_protocol_version has been added to cassandra.yaml to allow operators to + enforce an upper limit on the version of the native protocol that servers will negotiate with clients. + This can be used during upgrades from 2.1 to 3.0 to prevent errors due to incompatible paging state formats + between the two versions. See CASSANDRA-15193 for details. + Experimental features --------------------- diff --git a/bin/cqlsh.py b/bin/cqlsh.py index 71f84914f7..d5af8dac65 100644 --- a/bin/cqlsh.py +++ b/bin/cqlsh.py @@ -622,9 +622,9 @@ class Shell(cmd.Cmd): result, = self.session.execute("select * from system.local where key = 'local'") vers = { 'build': result['release_version'], - 'protocol': result['native_protocol_version'], 'cql': result['cql_version'], } + vers['protocol'] = self.conn.protocol_version self.connection_versions = vers def get_keyspace_names(self): diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 1d79e2aa99..7f28546e0a 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -163,7 +163,7 @@ public class Config public boolean native_transport_flush_in_batches_legacy = true; public volatile long native_transport_max_concurrent_requests_in_bytes_per_ip = -1L; public volatile long native_transport_max_concurrent_requests_in_bytes = -1L; - + public Integer native_transport_max_negotiable_protocol_version = Integer.MIN_VALUE; @Deprecated public int thrift_max_message_length_in_mb = 16; diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 75296b6746..1e2e1a1f04 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -61,6 +61,8 @@ import org.apache.cassandra.scheduler.NoScheduler; import org.apache.cassandra.security.EncryptionContext; import org.apache.cassandra.service.CacheService.CacheType; import org.apache.cassandra.thrift.ThriftServer.ThriftServerType; +import org.apache.cassandra.transport.ProtocolVersion; +import org.apache.cassandra.transport.ProtocolVersionLimit; import org.apache.cassandra.utils.FBUtilities; import org.apache.commons.lang3.StringUtils; @@ -728,6 +730,21 @@ public class DatabaseDescriptor throw new ConfigurationException("Encryption must be enabled in client_encryption_options for native_transport_port_ssl", false); } + // If max protocol version has been set, just validate it's within an acceptable range + if (conf.native_transport_max_negotiable_protocol_version != Integer.MIN_VALUE) + { + try + { + ProtocolVersion.decode(conf.native_transport_max_negotiable_protocol_version, ProtocolVersionLimit.SERVER_DEFAULT); + logger.info("Native transport max negotiable version statically limited to {}", conf.native_transport_max_negotiable_protocol_version); + } + catch (Exception e) + { + throw new ConfigurationException("Invalid setting for native_transport_max_negotiable_protocol_version; " + + ProtocolVersion.invalidVersionMessage(conf.native_transport_max_negotiable_protocol_version)); + } + } + if (conf.max_value_size_in_mb <= 0) throw new ConfigurationException("max_value_size_in_mb must be positive", false); else if (conf.max_value_size_in_mb >= 2048) @@ -1849,6 +1866,11 @@ public class DatabaseDescriptor return conf.native_transport_flush_in_batches_legacy; } + public static int getNativeProtocolMaxVersionOverride() + { + return conf.native_transport_max_negotiable_protocol_version; + } + public static double getCommitLogSyncBatchWindow() { return conf.commitlog_sync_batch_window_in_ms; diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java index 812659c797..96e041609a 100644 --- a/src/java/org/apache/cassandra/db/SystemKeyspace.java +++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java @@ -757,6 +757,18 @@ public final class SystemKeyspace return executorService.submit((Runnable) () -> executeInternal(String.format(req, PEERS, columnName), ep, value)); } + public static void updatePeerReleaseVersion(final InetAddress ep, final Object value, Runnable postUpdateTask, ExecutorService executorService) + { + if (ep.equals(FBUtilities.getBroadcastAddress())) + return; + + String req = "INSERT INTO system.%s (peer, release_version) VALUES (?, ?)"; + executorService.execute(() -> { + executeInternal(String.format(req, PEERS), ep, value); + postUpdateTask.run(); + }); + } + public static synchronized void updateHintsDropped(InetAddress ep, UUID timePeriod, int value) { // with 30 day TTL @@ -852,6 +864,37 @@ public final class SystemKeyspace return hostIdMap; } + /** + * Return a map of IP address to C* version. If an invalid version string, or no version + * at all is stored for a given peer IP, then NULL_VERSION will be reported for that peer + */ + public static Map loadPeerVersions() + { + Map releaseVersionMap = new HashMap<>(); + for (UntypedResultSet.Row row : executeInternal("SELECT peer, release_version FROM system." + PEERS)) + { + InetAddress peer = row.getInetAddress("peer"); + if (row.has("release_version")) + { + try + { + releaseVersionMap.put(peer, new CassandraVersion(row.getString("release_version"))); + } + catch (IllegalArgumentException e) + { + logger.info("Invalid version string found for {}", peer); + releaseVersionMap.put(peer, NULL_VERSION); + } + } + else + { + logger.info("No version string found for {}", peer); + releaseVersionMap.put(peer, NULL_VERSION); + } + } + return releaseVersionMap; + } + /** * Get preferred IP for given endpoint if it is known. Otherwise this returns given endpoint itself. * diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index c32a75628b..a3d18a322f 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -707,6 +707,16 @@ public class CassandraDaemon return nativeTransportService != null ? nativeTransportService.isRunning() : false; } + public int getMaxNativeProtocolVersion() + { + return nativeTransportService.getMaxProtocolVersion(); + } + + public void refreshMaxNativeProtocolVersion() + { + if (nativeTransportService != null) + nativeTransportService.refreshMaxNegotiableProtocolVersion(); + } /** * A convenience method to stop and destroy the daemon in one shot. diff --git a/src/java/org/apache/cassandra/service/NativeTransportService.java b/src/java/org/apache/cassandra/service/NativeTransportService.java index 6343f0d86f..c58bb5e705 100644 --- a/src/java/org/apache/cassandra/service/NativeTransportService.java +++ b/src/java/org/apache/cassandra/service/NativeTransportService.java @@ -35,6 +35,7 @@ import io.netty.util.concurrent.EventExecutor; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.metrics.AuthMetrics; import org.apache.cassandra.metrics.ClientMetrics; +import org.apache.cassandra.transport.ConfiguredLimit; import org.apache.cassandra.transport.Message; import org.apache.cassandra.transport.Server; @@ -50,6 +51,7 @@ public class NativeTransportService private boolean initialized = false; private EventLoopGroup workerGroup; + private ConfiguredLimit protocolVersionLimit; /** * Creates netty thread pools and event loops. @@ -71,12 +73,15 @@ public class NativeTransportService logger.info("Netty using Java NIO event loop"); } + protocolVersionLimit = ConfiguredLimit.newLimit(); + int nativePort = DatabaseDescriptor.getNativeTransportPort(); int nativePortSSL = DatabaseDescriptor.getNativeTransportPortSSL(); InetAddress nativeAddr = DatabaseDescriptor.getRpcAddress(); org.apache.cassandra.transport.Server.Builder builder = new org.apache.cassandra.transport.Server.Builder() .withEventLoopGroup(workerGroup) + .withProtocolVersionLimit(protocolVersionLimit) .withHost(nativeAddr); if (!DatabaseDescriptor.getClientEncryptionOptions().enabled) @@ -141,6 +146,20 @@ public class NativeTransportService Message.Dispatcher.shutdown(); } + public int getMaxProtocolVersion() + { + return protocolVersionLimit.getMaxVersion().asInt(); + } + + public void refreshMaxNegotiableProtocolVersion() + { + // lowering the max negotiable protocol version is only safe if we haven't already + // allowed clients to connect with a higher version. This still allows the max + // version to be raised, as that is safe. + if (initialized) + protocolVersionLimit.updateMaxSupportedVersion(); + } + /** * @return intend to use epoll bassed event looping */ diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 346f9a473c..309be3d26b 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -441,6 +441,23 @@ public class StorageService extends NotificationBroadcasterSupport implements IE return daemon.isNativeTransportRunning(); } + public int getMaxNativeProtocolVersion() + { + if (daemon == null) + { + throw new IllegalStateException("No configured daemon"); + } + return daemon.getMaxNativeProtocolVersion(); + } + + private void refreshMaxNativeProtocolVersion() + { + if (daemon != null) + { + daemon.refreshMaxNativeProtocolVersion(); + } + } + public void stopTransports() { if (isGossipActive()) @@ -2050,7 +2067,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE switch (state) { case RELEASE_VERSION: - SystemKeyspace.updatePeerInfo(endpoint, "release_version", value.value, executor); + SystemKeyspace.updatePeerReleaseVersion(endpoint, value.value, this::refreshMaxNativeProtocolVersion, executor); break; case DC: updateTopology(endpoint); @@ -2127,7 +2144,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE switch (entry.getKey()) { case RELEASE_VERSION: - SystemKeyspace.updatePeerInfo(endpoint, "release_version", entry.getValue().value, executor); + SystemKeyspace.updatePeerReleaseVersion(endpoint, entry.getValue().value, this::refreshMaxNativeProtocolVersion, executor); break; case DC: SystemKeyspace.updatePeerInfo(endpoint, "data_center", entry.getValue().value, executor); diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index bcf55cf1a1..4548fa81ca 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -693,4 +693,7 @@ public interface StorageServiceMBean extends NotificationEmitter * @return true if the node successfully starts resuming. (this does not mean bootstrap streaming was success.) */ public boolean resumeBootstrap(); + + /** Returns the max version that this node will negotiate for native protocol connections */ + public int getMaxNativeProtocolVersion(); } diff --git a/src/java/org/apache/cassandra/transport/Client.java b/src/java/org/apache/cassandra/transport/Client.java index e428b06603..368b1d745b 100644 --- a/src/java/org/apache/cassandra/transport/Client.java +++ b/src/java/org/apache/cassandra/transport/Client.java @@ -251,7 +251,7 @@ public class Client extends SimpleClient // Parse options. String host = args[0]; int port = Integer.parseInt(args[1]); - ProtocolVersion version = args.length == 3 ? ProtocolVersion.decode(Integer.parseInt(args[2])) : ProtocolVersion.CURRENT; + ProtocolVersion version = args.length == 3 ? ProtocolVersion.decode(Integer.parseInt(args[2]), ProtocolVersionLimit.SERVER_DEFAULT) : ProtocolVersion.CURRENT; ClientEncryptionOptions encryptionOptions = new ClientEncryptionOptions(); System.out.println("CQL binary protocol console " + host + "@" + port + " using native protocol version " + version); diff --git a/src/java/org/apache/cassandra/transport/ConfiguredLimit.java b/src/java/org/apache/cassandra/transport/ConfiguredLimit.java new file mode 100644 index 0000000000..16d38671aa --- /dev/null +++ b/src/java/org/apache/cassandra/transport/ConfiguredLimit.java @@ -0,0 +1,112 @@ +/* + * 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.transport; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.SystemKeyspace; +import org.apache.cassandra.utils.CassandraVersion; + +public abstract class ConfiguredLimit implements ProtocolVersionLimit +{ + private static final Logger logger = LoggerFactory.getLogger(ConfiguredLimit.class); + static final String DISABLE_MAX_PROTOCOL_AUTO_OVERRIDE = "cassandra.disable_max_protocol_auto_override"; + static final CassandraVersion MIN_VERSION_FOR_V4 = new CassandraVersion("3.0.0"); + + public abstract ProtocolVersion getMaxVersion(); + public abstract void updateMaxSupportedVersion(); + + public static ConfiguredLimit newLimit() + { + if (Boolean.getBoolean(DISABLE_MAX_PROTOCOL_AUTO_OVERRIDE)) + return new StaticLimit(ProtocolVersion.MAX_SUPPORTED_VERSION); + + int fromConfig = DatabaseDescriptor.getNativeProtocolMaxVersionOverride(); + return fromConfig != Integer.MIN_VALUE + ? new StaticLimit(ProtocolVersion.decode(fromConfig, ProtocolVersionLimit.SERVER_DEFAULT)) + : new DynamicLimit(ProtocolVersion.MAX_SUPPORTED_VERSION); + } + + private static class StaticLimit extends ConfiguredLimit + { + private final ProtocolVersion maxVersion; + private StaticLimit(ProtocolVersion maxVersion) + { + this.maxVersion = maxVersion; + logger.info("Native transport max negotiable version statically limited to {}", maxVersion); + } + + public ProtocolVersion getMaxVersion() + { + return maxVersion; + } + + public void updateMaxSupportedVersion() + { + // statically configured, so this is a no-op + } + } + + private static class DynamicLimit extends ConfiguredLimit + { + private volatile ProtocolVersion maxVersion; + private DynamicLimit(ProtocolVersion initialLimit) + { + maxVersion = initialLimit; + maybeUpdateVersion(true); + } + + public ProtocolVersion getMaxVersion() + { + return maxVersion; + } + + public void updateMaxSupportedVersion() + { + maybeUpdateVersion(false); + } + + private void maybeUpdateVersion(boolean allowLowering) + { + boolean enforceV3Cap = SystemKeyspace.loadPeerVersions() + .values() + .stream() + .anyMatch(v -> v.compareTo(MIN_VERSION_FOR_V4) < 0); + + if (!enforceV3Cap) + { + maxVersion = ProtocolVersion.MAX_SUPPORTED_VERSION; + return; + } + + if (ProtocolVersion.V3.isSmallerThan(maxVersion) && !allowLowering) + { + logger.info("Detected peers which do not fully support protocol V4, but V4 was previously negotiable. " + + "Not enforcing cap as this can cause issues for older client versions. After the next " + + "restart the server will apply the cap"); + return; + } + + logger.info("Detected peers which do not fully support protocol V4. Capping max negotiable version to V3"); + maxVersion = ProtocolVersion.V3; + } + } +} diff --git a/src/java/org/apache/cassandra/transport/Frame.java b/src/java/org/apache/cassandra/transport/Frame.java index 388cbc25c9..e603e6cc96 100644 --- a/src/java/org/apache/cassandra/transport/Frame.java +++ b/src/java/org/apache/cassandra/transport/Frame.java @@ -142,10 +142,12 @@ public class Frame private int tooLongStreamId; private final Connection.Factory factory; + private final ProtocolVersionLimit versionCap; - public Decoder(Connection.Factory factory) + public Decoder(Connection.Factory factory, ProtocolVersionLimit versionCap) { this.factory = factory; + this.versionCap = versionCap; } @Override @@ -172,7 +174,7 @@ public class Frame int firstByte = buffer.getByte(idx++); Message.Direction direction = Message.Direction.extractFromVersion(firstByte); int versionNum = firstByte & PROTOCOL_VERSION_MASK; - ProtocolVersion version = ProtocolVersion.decode(versionNum); + ProtocolVersion version = ProtocolVersion.decode(versionNum, versionCap); // Wait until we have the complete header if (readableBytes < Header.LENGTH) diff --git a/src/java/org/apache/cassandra/transport/Message.java b/src/java/org/apache/cassandra/transport/Message.java index 271b6907ca..f899022686 100644 --- a/src/java/org/apache/cassandra/transport/Message.java +++ b/src/java/org/apache/cassandra/transport/Message.java @@ -322,11 +322,19 @@ public abstract class Message @ChannelHandler.Sharable public static class ProtocolEncoder extends MessageToMessageEncoder { + private final ProtocolVersionLimit versionCap; + + ProtocolEncoder(ProtocolVersionLimit versionCap) + { + this.versionCap = versionCap; + } + public void encode(ChannelHandlerContext ctx, Message message, List results) { Connection connection = ctx.channel().attr(Connection.attributeKey).get(); // The only case the connection can be null is when we send the initial STARTUP message (client side thus) - ProtocolVersion version = connection == null ? ProtocolVersion.CURRENT : connection.getVersion(); + ProtocolVersion version = connection == null ? versionCap.getMaxVersion() : connection.getVersion(); + EnumSet flags = EnumSet.noneOf(Frame.Header.Flag.class); Codec codec = (Codec)message.type.codec; diff --git a/src/java/org/apache/cassandra/transport/ProtocolVersion.java b/src/java/org/apache/cassandra/transport/ProtocolVersion.java index cd73c86255..05fafc2a31 100644 --- a/src/java/org/apache/cassandra/transport/ProtocolVersion.java +++ b/src/java/org/apache/cassandra/transport/ProtocolVersion.java @@ -84,9 +84,9 @@ public enum ProtocolVersion implements Comparable return ret; } - public static ProtocolVersion decode(int versionNum) + public static ProtocolVersion decode(int versionNum, ProtocolVersionLimit ceiling) { - ProtocolVersion ret = versionNum >= MIN_SUPPORTED_VERSION.num && versionNum <= MAX_SUPPORTED_VERSION.num + ProtocolVersion ret = versionNum >= MIN_SUPPORTED_VERSION.num && versionNum <= ceiling.getMaxVersion().num ? SUPPORTED_VERSIONS[versionNum - MIN_SUPPORTED_VERSION.num] : null; @@ -102,7 +102,7 @@ public enum ProtocolVersion implements Comparable } // If the version is invalid reply with the highest version that we support - throw new ProtocolException(invalidVersionMessage(versionNum), MAX_SUPPORTED_VERSION); + throw new ProtocolException(invalidVersionMessage(versionNum), ceiling.getMaxVersion()); } return ret; diff --git a/src/java/org/apache/cassandra/transport/ProtocolVersionLimit.java b/src/java/org/apache/cassandra/transport/ProtocolVersionLimit.java new file mode 100644 index 0000000000..9738a19ce0 --- /dev/null +++ b/src/java/org/apache/cassandra/transport/ProtocolVersionLimit.java @@ -0,0 +1,27 @@ +/* + * 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.transport; + +@FunctionalInterface +public interface ProtocolVersionLimit +{ + public ProtocolVersion getMaxVersion(); + + public static final ProtocolVersionLimit SERVER_DEFAULT = () -> ProtocolVersion.MAX_SUPPORTED_VERSION; +} diff --git a/src/java/org/apache/cassandra/transport/Server.java b/src/java/org/apache/cassandra/transport/Server.java index b96c517fe1..ced764fbdb 100644 --- a/src/java/org/apache/cassandra/transport/Server.java +++ b/src/java/org/apache/cassandra/transport/Server.java @@ -82,11 +82,14 @@ public class Server implements CassandraDaemon.Server private final AtomicBoolean isRunning = new AtomicBoolean(false); private EventLoopGroup workerGroup; + private final ProtocolVersionLimit protocolVersionLimit; private Server (Builder builder) { this.socket = builder.getSocket(); this.useSSL = builder.useSSL; + this.protocolVersionLimit = builder.getProtocolVersionLimit(); + if (builder.workerGroup != null) { workerGroup = builder.workerGroup; @@ -184,6 +187,7 @@ public class Server implements CassandraDaemon.Server private InetAddress hostAddr; private int port = -1; private InetSocketAddress socket; + private ProtocolVersionLimit versionLimit; public Builder withSSL(boolean useSSL) { @@ -211,6 +215,19 @@ public class Server implements CassandraDaemon.Server return this; } + public Builder withProtocolVersionLimit(ProtocolVersionLimit limit) + { + this.versionLimit = limit; + return this; + } + + ProtocolVersionLimit getProtocolVersionLimit() + { + if (versionLimit == null) + throw new IllegalArgumentException("Missing protocol version limiter"); + return versionLimit; + } + public Server build() { return new Server(this); @@ -323,7 +340,6 @@ public class Server implements CassandraDaemon.Server { // Stateless handlers private static final Message.ProtocolDecoder messageDecoder = new Message.ProtocolDecoder(); - private static final Message.ProtocolEncoder messageEncoder = new Message.ProtocolEncoder(); private static final Frame.Decompressor frameDecompressor = new Frame.Decompressor(); private static final Frame.Compressor frameCompressor = new Frame.Compressor(); private static final Frame.Encoder frameEncoder = new Frame.Encoder(); @@ -351,14 +367,14 @@ public class Server implements CassandraDaemon.Server //pipeline.addLast("debug", new LoggingHandler()); - pipeline.addLast("frameDecoder", new Frame.Decoder(server.connectionFactory)); + pipeline.addLast("frameDecoder", new Frame.Decoder(server.connectionFactory, server.protocolVersionLimit)); pipeline.addLast("frameEncoder", frameEncoder); pipeline.addLast("frameDecompressor", frameDecompressor); pipeline.addLast("frameCompressor", frameCompressor); pipeline.addLast("messageDecoder", messageDecoder); - pipeline.addLast("messageEncoder", messageEncoder); + pipeline.addLast("messageEncoder", new Message.ProtocolEncoder(server.protocolVersionLimit)); pipeline.addLast("executor", new Message.Dispatcher(DatabaseDescriptor.useNativeTransportLegacyFlusher(), EndpointPayloadTracker.get(((InetSocketAddress) channel.remoteAddress()).getAddress()))); diff --git a/src/java/org/apache/cassandra/transport/SimpleClient.java b/src/java/org/apache/cassandra/transport/SimpleClient.java index c03becdb8e..1e6ea64142 100644 --- a/src/java/org/apache/cassandra/transport/SimpleClient.java +++ b/src/java/org/apache/cassandra/transport/SimpleClient.java @@ -259,7 +259,7 @@ public class SimpleClient implements Closeable // Stateless handlers private static final Message.ProtocolDecoder messageDecoder = new Message.ProtocolDecoder(); - private static final Message.ProtocolEncoder messageEncoder = new Message.ProtocolEncoder(); + private static final Message.ProtocolEncoder messageEncoder = new Message.ProtocolEncoder(ProtocolVersionLimit.SERVER_DEFAULT); private static final Frame.Decompressor frameDecompressor = new Frame.Decompressor(); private static final Frame.Compressor frameCompressor = new Frame.Compressor(); private static final Frame.Encoder frameEncoder = new Frame.Encoder(); @@ -282,7 +282,7 @@ public class SimpleClient implements Closeable channel.attr(Connection.attributeKey).set(connection); ChannelPipeline pipeline = channel.pipeline(); - pipeline.addLast("frameDecoder", new Frame.Decoder(connectionFactory)); + pipeline.addLast("frameDecoder", new Frame.Decoder(connectionFactory, ProtocolVersionLimit.SERVER_DEFAULT)); pipeline.addLast("frameEncoder", frameEncoder); pipeline.addLast("frameDecompressor", frameDecompressor); diff --git a/test/unit/org/apache/cassandra/cql3/CQLTester.java b/test/unit/org/apache/cassandra/cql3/CQLTester.java index d43863035e..e545e9f3ea 100644 --- a/test/unit/org/apache/cassandra/cql3/CQLTester.java +++ b/test/unit/org/apache/cassandra/cql3/CQLTester.java @@ -62,8 +62,10 @@ import org.apache.cassandra.serializers.TypeSerializer; import org.apache.cassandra.service.ClientState; import org.apache.cassandra.service.QueryState; import org.apache.cassandra.service.StorageService; -import org.apache.cassandra.transport.*; import org.apache.cassandra.transport.ProtocolVersion; +import org.apache.cassandra.transport.ConfiguredLimit; +import org.apache.cassandra.transport.Event; +import org.apache.cassandra.transport.Server; import org.apache.cassandra.transport.messages.ResultMessage; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; @@ -91,6 +93,7 @@ public abstract class CQLTester protected static final InetAddress nativeAddr; private static final Map clusters = new HashMap<>(); private static final Map sessions = new HashMap<>(); + protected static ConfiguredLimit protocolVersionLimit; private static boolean isServerPrepared = false; @@ -353,11 +356,43 @@ public abstract class CQLTester if (server != null) return; + prepareNetwork(); + initializeNetwork(); + } + + protected static void prepareNetwork() + { SystemKeyspace.finishStartup(); StorageService.instance.initServer(); SchemaLoader.startGossiper(); + } - server = new Server.Builder().withHost(nativeAddr).withPort(nativePort).build(); + protected static void reinitializeNetwork() + { + if (server != null && server.isRunning()) + { + server.stop(); + server = null; + } + List futures = new ArrayList<>(); + for (Cluster cluster : clusters.values()) + futures.add(cluster.closeAsync()); + for (Session session : sessions.values()) + futures.add(session.closeAsync()); + FBUtilities.waitOnFutures(futures); + clusters.clear(); + sessions.clear(); + + initializeNetwork(); + } + + private static void initializeNetwork() + { + protocolVersionLimit = ConfiguredLimit.newLimit(); + server = new Server.Builder().withHost(nativeAddr) + .withPort(nativePort) + .withProtocolVersionLimit(protocolVersionLimit) + .build(); ClientMetrics.instance.init(Collections.singleton(server)); server.start(); @@ -366,9 +401,12 @@ public abstract class CQLTester if (clusters.containsKey(version)) continue; + if (version.isGreaterThan(protocolVersionLimit.getMaxVersion())) + continue; + Cluster cluster = Cluster.builder() .addContactPoints(nativeAddr) - .withClusterName("Test Cluster") + .withClusterName("Test Cluster-" + version.name()) .withPort(nativePort) .withProtocolVersion(com.datastax.driver.core.ProtocolVersion.fromInt(version.asInt())) .build(); @@ -379,6 +417,14 @@ public abstract class CQLTester } } + protected void updateMaxNegotiableProtocolVersion() + { + if (protocolVersionLimit == null) + throw new IllegalStateException("Native transport server has not been initialized"); + + protocolVersionLimit.updateMaxSupportedVersion(); + } + protected void dropPerTestKeyspace() throws Throwable { execute(String.format("DROP KEYSPACE IF EXISTS %s", KEYSPACE_PER_TEST)); diff --git a/test/unit/org/apache/cassandra/transport/DynamicLimitTest.java b/test/unit/org/apache/cassandra/transport/DynamicLimitTest.java new file mode 100644 index 0000000000..df06fbaa99 --- /dev/null +++ b/test/unit/org/apache/cassandra/transport/DynamicLimitTest.java @@ -0,0 +1,111 @@ +/* + * 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.transport; + +import java.net.InetAddress; + +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.cql3.CQLTester; + +import static org.apache.cassandra.transport.ProtocolTestHelper.cleanupPeers; +import static org.apache.cassandra.transport.ProtocolTestHelper.setStaticLimitInConfig; +import static org.apache.cassandra.transport.ProtocolTestHelper.setupPeer; +import static org.apache.cassandra.transport.ProtocolTestHelper.updatePeerInfo; +import static org.junit.Assert.assertEquals; + +public class DynamicLimitTest +{ + @BeforeClass + public static void setup() + { + CQLTester.prepareServer(); + } + + @Test + public void disableDynamicLimitWithSystemProperty() throws Throwable + { + // Dynamic limiting of the max negotiable protocol version can be + // disabled with a system property + + // ensure that no static limit is configured + setStaticLimitInConfig(null); + + // set the property which disables dynamic limiting + System.setProperty(ConfiguredLimit.DISABLE_MAX_PROTOCOL_AUTO_OVERRIDE, "true"); + // insert a legacy peer into system.peers and also + InetAddress peer = null; + try + { + peer = setupPeer("127.1.0.1", "2.2.0"); + ConfiguredLimit limit = ConfiguredLimit.newLimit(); + assertEquals(ProtocolVersion.MAX_SUPPORTED_VERSION, limit.getMaxVersion()); + + // clearing the property after the limit has been returned has no effect + System.clearProperty(ConfiguredLimit.DISABLE_MAX_PROTOCOL_AUTO_OVERRIDE); + limit.updateMaxSupportedVersion(); + assertEquals(ProtocolVersion.MAX_SUPPORTED_VERSION, limit.getMaxVersion()); + + // a new limit should now be dynamic + limit = ConfiguredLimit.newLimit(); + assertEquals(ProtocolVersion.V3, limit.getMaxVersion()); + } + finally + { + System.clearProperty(ConfiguredLimit.DISABLE_MAX_PROTOCOL_AUTO_OVERRIDE); + cleanupPeers(peer); + } + } + + @Test + public void disallowLoweringMaxVersion() throws Throwable + { + // Lowering the max version once connections have been established is a problem + // for some clients. So for a dynamic limit, if notifications of peer versions + // trigger a change to the max version, it's only allowed to increase the max + // negotiable version + + InetAddress peer = null; + try + { + // ensure that no static limit is configured + setStaticLimitInConfig(null); + ConfiguredLimit limit = ConfiguredLimit.newLimit(); + assertEquals(ProtocolVersion.MAX_SUPPORTED_VERSION, limit.getMaxVersion()); + + peer = setupPeer("127.1.0.1", "3.0.0"); + limit.updateMaxSupportedVersion(); + assertEquals(ProtocolVersion.MAX_SUPPORTED_VERSION, limit.getMaxVersion()); + + // learn that peer doesn't actually fully support V4, behaviour should remain the same + updatePeerInfo(peer, "2.2.0"); + limit.updateMaxSupportedVersion(); + assertEquals(ProtocolVersion.MAX_SUPPORTED_VERSION, limit.getMaxVersion()); + + // finally learn that peer2 has been upgraded, just for completeness + updatePeerInfo(peer, "3.3.0"); + limit.updateMaxSupportedVersion(); + assertEquals(ProtocolVersion.MAX_SUPPORTED_VERSION, limit.getMaxVersion()); + + } finally { + cleanupPeers(peer); + } + } +} diff --git a/test/unit/org/apache/cassandra/transport/ProtocolErrorTest.java b/test/unit/org/apache/cassandra/transport/ProtocolErrorTest.java index 5e9731a6fa..5041a943e2 100644 --- a/test/unit/org/apache/cassandra/transport/ProtocolErrorTest.java +++ b/test/unit/org/apache/cassandra/transport/ProtocolErrorTest.java @@ -52,7 +52,7 @@ public class ProtocolErrorTest { public void testInvalidProtocolVersion(int version) throws Exception { - Frame.Decoder dec = new Frame.Decoder(null); + Frame.Decoder dec = new Frame.Decoder(null, ProtocolVersionLimit.SERVER_DEFAULT); List results = new ArrayList<>(); byte[] frame = new byte[] { @@ -80,7 +80,7 @@ public class ProtocolErrorTest { public void testInvalidProtocolVersionShortFrame() throws Exception { // test for CASSANDRA-11464 - Frame.Decoder dec = new Frame.Decoder(null); + Frame.Decoder dec = new Frame.Decoder(null, ProtocolVersionLimit.SERVER_DEFAULT); List results = new ArrayList<>(); byte[] frame = new byte[] { @@ -102,7 +102,7 @@ public class ProtocolErrorTest { @Test public void testInvalidDirection() throws Exception { - Frame.Decoder dec = new Frame.Decoder(null); + Frame.Decoder dec = new Frame.Decoder(null, ProtocolVersionLimit.SERVER_DEFAULT); List results = new ArrayList<>(); // should generate a protocol exception for using a response frame with @@ -133,7 +133,7 @@ public class ProtocolErrorTest { @Test public void testBodyLengthOverLimit() throws Exception { - Frame.Decoder dec = new Frame.Decoder(null); + Frame.Decoder dec = new Frame.Decoder(null, ProtocolVersionLimit.SERVER_DEFAULT); List results = new ArrayList<>(); byte[] frame = new byte[] { diff --git a/test/unit/org/apache/cassandra/transport/ProtocolNegotiationTest.java b/test/unit/org/apache/cassandra/transport/ProtocolNegotiationTest.java new file mode 100644 index 0000000000..91c1d6aba4 --- /dev/null +++ b/test/unit/org/apache/cassandra/transport/ProtocolNegotiationTest.java @@ -0,0 +1,166 @@ +/* + * 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.transport; + +import java.net.InetAddress; +import java.util.concurrent.TimeUnit; + +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +import com.datastax.driver.core.Cluster; +import com.datastax.driver.core.ProtocolVersion; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.cql3.CQLTester; + +import static org.apache.cassandra.transport.ProtocolTestHelper.cleanupPeers; +import static org.apache.cassandra.transport.ProtocolTestHelper.setStaticLimitInConfig; +import static org.apache.cassandra.transport.ProtocolTestHelper.setupPeer; +import static org.apache.cassandra.transport.ProtocolTestHelper.updatePeerInfo; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +public class ProtocolNegotiationTest extends CQLTester +{ + // to avoid JMX naming clashes between cluster metrics + private int clusterId = 0; + + @BeforeClass + public static void setup() + { + prepareNetwork(); + } + + @Before + public void clearConfig() + { + setStaticLimitInConfig(null); + } + + @Test + public void serverSupportsV3AndV4ByDefault() throws Throwable + { + reinitializeNetwork(); + // client can explicitly request either V3 or V4 + testConnection(ProtocolVersion.V3, ProtocolVersion.V3); + testConnection(ProtocolVersion.V4, ProtocolVersion.V4); + + // if not specified, V4 is the default + testConnection(null, ProtocolVersion.V4); + } + + @Test + public void testStaticLimit() throws Throwable + { + try + { + reinitializeNetwork(); + // No limit enforced to start + assertEquals(Integer.MIN_VALUE, DatabaseDescriptor.getNativeProtocolMaxVersionOverride()); + testConnection(null, ProtocolVersion.V4); + + // Update DatabaseDescriptor, then re-initialise the server to force it to read it + setStaticLimitInConfig(ProtocolVersion.V3.toInt()); + reinitializeNetwork(); + assertEquals(3, DatabaseDescriptor.getNativeProtocolMaxVersionOverride()); + testConnection(ProtocolVersion.V4, ProtocolVersion.V3); + testConnection(ProtocolVersion.V3, ProtocolVersion.V3); + testConnection(null, ProtocolVersion.V3); + } finally { + setStaticLimitInConfig(null); + } + } + + @Test + public void testDynamicLimit() throws Throwable + { + InetAddress peer1 = setupPeer("127.1.0.1", "2.2.0"); + InetAddress peer2 = setupPeer("127.1.0.2", "2.2.0"); + InetAddress peer3 = setupPeer("127.1.0.3", "2.2.0"); + reinitializeNetwork(); + try + { + // legacy peers means max negotiable version is V3 + testConnection(ProtocolVersion.V4, ProtocolVersion.V3); + testConnection(ProtocolVersion.V3, ProtocolVersion.V3); + testConnection(null, ProtocolVersion.V3); + + // receive notification that 2 peers have upgraded to a version that fully supports V4 + updatePeerInfo(peer1, "3.0.0"); + updatePeerInfo(peer2, "3.0.0"); + updateMaxNegotiableProtocolVersion(); + // version should still be capped + testConnection(ProtocolVersion.V4, ProtocolVersion.V3); + testConnection(ProtocolVersion.V3, ProtocolVersion.V3); + testConnection(null, ProtocolVersion.V3); + + // no legacy peers so V4 is negotiable + // after the last peer upgrades, cap should be lifted + updatePeerInfo(peer3, "3.0.0"); + updateMaxNegotiableProtocolVersion(); + testConnection(ProtocolVersion.V4, ProtocolVersion.V4); + testConnection(ProtocolVersion.V3, ProtocolVersion.V3); + testConnection(null, ProtocolVersion.V4); + } finally { + cleanupPeers(peer1, peer2, peer3); + } + } + + private void testConnection(com.datastax.driver.core.ProtocolVersion requestedVersion, + com.datastax.driver.core.ProtocolVersion expectedVersion) + { + long start = System.nanoTime(); + boolean expectError = requestedVersion != null && requestedVersion != expectedVersion; + Cluster.Builder builder = Cluster.builder() + .addContactPoints(nativeAddr) + .withClusterName("Test Cluster" + clusterId++) + .withPort(nativePort); + + if (requestedVersion != null) + builder = builder.withProtocolVersion(requestedVersion) ; + + Cluster cluster = builder.build(); + logger.info("Setting up cluster took {}ms", TimeUnit.MILLISECONDS.convert(System.nanoTime() - start, TimeUnit.NANOSECONDS)); + start = System.nanoTime(); + try { + cluster.connect(); + if (expectError) + fail("Expected a protocol exception"); + } + catch (Exception e) + { + if (!expectError) + { + e.printStackTrace(); + fail("Did not expect any exception"); + } + + assertTrue(e.getMessage().contains(String.format("Host does not support protocol version %s but %s", requestedVersion, expectedVersion))); + } finally { + logger.info("Testing connection took {}ms", TimeUnit.MILLISECONDS.convert(System.nanoTime() - start, TimeUnit.NANOSECONDS)); + start = System.nanoTime(); + cluster.closeAsync(); + logger.info("Tearing down cluster connection took {}ms", TimeUnit.MILLISECONDS.convert(System.nanoTime() - start, TimeUnit.NANOSECONDS)); + + } + } + +} diff --git a/test/unit/org/apache/cassandra/transport/ProtocolTestHelper.java b/test/unit/org/apache/cassandra/transport/ProtocolTestHelper.java new file mode 100644 index 0000000000..90a2801732 --- /dev/null +++ b/test/unit/org/apache/cassandra/transport/ProtocolTestHelper.java @@ -0,0 +1,95 @@ +/* + * 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.transport; + +import java.lang.reflect.Constructor; +import java.lang.reflect.Field; +import java.net.InetAddress; +import java.net.UnknownHostException; +import java.util.concurrent.ExecutorService; + +import com.google.common.util.concurrent.MoreExecutors; + +import org.apache.cassandra.config.Config; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.SystemKeyspace; +import org.apache.cassandra.gms.VersionedValue; +import org.apache.cassandra.utils.FBUtilities; + +public class ProtocolTestHelper +{ + static ExecutorService executor = MoreExecutors.newDirectExecutorService(); + static InetAddress setupPeer(String address, String version) throws Throwable + { + InetAddress peer = peer(address); + updatePeerInfo(peer, version); + return peer; + } + + static void updatePeerInfo(InetAddress peer, String version) throws Throwable + { + SystemKeyspace.updatePeerInfo(peer, "release_version", version, executor); + } + + static InetAddress peer(String address) + { + try + { + return InetAddress.getByName(address); + } + catch (UnknownHostException e) + { + throw new RuntimeException("Error creating peer", e); + } + } + + static void cleanupPeers(InetAddress...peers) throws Throwable + { + for (InetAddress peer : peers) + if (peer != null) + SystemKeyspace.removeEndpoint(peer); + } + + static void setStaticLimitInConfig(Integer version) + { + try + { + Field field = FBUtilities.getProtectedField(DatabaseDescriptor.class, "conf"); + ((Config)field.get(null)).native_transport_max_negotiable_protocol_version = version == null ? Integer.MIN_VALUE : version; + } + catch (IllegalAccessException e) + { + throw new RuntimeException("Error setting native_transport_max_protocol_version on Config", e); + } + } + + static VersionedValue releaseVersion(String versionString) + { + try + { + Constructor ctor = VersionedValue.class.getDeclaredConstructor(String.class); + ctor.setAccessible(true); + return ctor.newInstance(versionString); + } + catch (Exception e) + { + throw new RuntimeException("Error constructing VersionedValue for release version", e); + } + } +} diff --git a/test/unit/org/apache/cassandra/transport/ProtocolVersionTest.java b/test/unit/org/apache/cassandra/transport/ProtocolVersionTest.java index e287c08de6..7b56a49c2d 100644 --- a/test/unit/org/apache/cassandra/transport/ProtocolVersionTest.java +++ b/test/unit/org/apache/cassandra/transport/ProtocolVersionTest.java @@ -27,13 +27,13 @@ public class ProtocolVersionTest public void testDecode() { for (ProtocolVersion version : ProtocolVersion.SUPPORTED) - Assert.assertEquals(version, ProtocolVersion.decode(version.asInt())); + Assert.assertEquals(version, ProtocolVersion.decode(version.asInt(), ProtocolVersionLimit.SERVER_DEFAULT)); for (ProtocolVersion version : ProtocolVersion.UNSUPPORTED) { // unsupported old versions try { - Assert.assertEquals(version, ProtocolVersion.decode(version.asInt())); + Assert.assertEquals(version, ProtocolVersion.decode(version.asInt(), ProtocolVersionLimit.SERVER_DEFAULT)); Assert.fail("Expected invalid protocol exception"); } catch (ProtocolException ex) @@ -45,7 +45,7 @@ public class ProtocolVersionTest try { // unsupported newer version - Assert.assertEquals(null, ProtocolVersion.decode(63)); + Assert.assertEquals(null, ProtocolVersion.decode(63, ProtocolVersionLimit.SERVER_DEFAULT)); Assert.fail("Expected invalid protocol exception"); } catch (ProtocolException ex)