You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by dc...@apache.org on 2020/11/05 22:47:48 UTC
[cassandra] branch trunk updated (d68c45e -> 001767d)
This is an automated email from the ASF dual-hosted git repository.
dcapwell pushed a change to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git.
from d68c45e Merge branch 'cassandra-3.11' into trunk
new f293376 TLS connections to the storage port on a node without server encryption configured causes java.io.IOException accessing missing keystore
new e74bd9f Merge branch 'cassandra-2.2' into cassandra-3.0
new 3200bcf Merge branch 'cassandra-3.0' into cassandra-3.11
new 001767d Merge branch 'cassandra-3.11' into trunk
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
CHANGES.txt | 1 +
.../cassandra/config/DatabaseDescriptor.java | 23 +-
.../apache/cassandra/config/EncryptionOptions.java | 234 ++++++++++++----
.../cassandra/config/YamlConfigurationLoader.java | 4 +-
.../apache/cassandra/db/virtual/SettingsTable.java | 2 +-
.../cassandra/net/InboundConnectionInitiator.java | 83 +++---
.../cassandra/net/InboundConnectionSettings.java | 6 +-
.../org/apache/cassandra/net/InboundSockets.java | 4 +-
.../apache/cassandra/net/OutboundConnection.java | 3 +-
.../cassandra/net/OutboundConnectionSettings.java | 3 +-
.../org/apache/cassandra/net/SocketFactory.java | 52 ++--
.../org/apache/cassandra/security/SSLFactory.java | 14 +-
.../cassandra/service/NativeTransportService.java | 50 ++--
.../org/apache/cassandra/tools/LoaderOptions.java | 12 +-
.../org/apache/cassandra/transport/Client.java | 4 +-
.../org/apache/cassandra/transport/Server.java | 43 ++-
.../apache/cassandra/transport/SimpleClient.java | 2 +-
.../cassandra/distributed/impl/InstanceConfig.java | 21 +-
.../test/AbstractEncryptionOptionsImpl.java | 295 +++++++++++++++++++++
.../distributed/test/IncRepairTruncationTest.java | 3 +-
.../test/InternodeEncryptionOptionsTest.java | 218 +++++++++++++++
.../test/NativeTransportEncryptionOptionsTest.java | 137 ++++++++++
.../distributed/test/PreviewRepairTest.java | 3 +-
.../cassandra/config/EncryptionOptionsTest.java | 178 +++++++++++++
.../config/YamlConfigurationLoaderTest.java | 11 +-
test/unit/org/apache/cassandra/cql3/CQLTester.java | 2 +-
.../cassandra/db/virtual/SettingsTableTest.java | 6 +-
.../apache/cassandra/net/MessagingServiceTest.java | 6 +-
.../service/NativeTransportServiceTest.java | 86 +++++-
.../stress/settings/SettingsTransport.java | 2 +-
.../cassandra/stress/util/JavaDriverClient.java | 2 +-
31 files changed, 1298 insertions(+), 212 deletions(-)
create mode 100644 test/distributed/org/apache/cassandra/distributed/test/AbstractEncryptionOptionsImpl.java
create mode 100644 test/distributed/org/apache/cassandra/distributed/test/InternodeEncryptionOptionsTest.java
create mode 100644 test/distributed/org/apache/cassandra/distributed/test/NativeTransportEncryptionOptionsTest.java
create mode 100644 test/unit/org/apache/cassandra/config/EncryptionOptionsTest.java
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
[cassandra] 01/01: Merge branch 'cassandra-3.11' into trunk
Posted by dc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dcapwell pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 001767de2db1803c77867d1078149763a461ebd8
Merge: d68c45e 3200bcf
Author: David Capwell <dc...@apache.org>
AuthorDate: Thu Nov 5 14:46:12 2020 -0800
Merge branch 'cassandra-3.11' into trunk
CHANGES.txt | 1 +
.../cassandra/config/DatabaseDescriptor.java | 23 +-
.../apache/cassandra/config/EncryptionOptions.java | 234 ++++++++++++----
.../cassandra/config/YamlConfigurationLoader.java | 4 +-
.../apache/cassandra/db/virtual/SettingsTable.java | 2 +-
.../cassandra/net/InboundConnectionInitiator.java | 83 +++---
.../cassandra/net/InboundConnectionSettings.java | 6 +-
.../org/apache/cassandra/net/InboundSockets.java | 4 +-
.../apache/cassandra/net/OutboundConnection.java | 3 +-
.../cassandra/net/OutboundConnectionSettings.java | 3 +-
.../org/apache/cassandra/net/SocketFactory.java | 52 ++--
.../org/apache/cassandra/security/SSLFactory.java | 14 +-
.../cassandra/service/NativeTransportService.java | 50 ++--
.../org/apache/cassandra/tools/LoaderOptions.java | 12 +-
.../org/apache/cassandra/transport/Client.java | 4 +-
.../org/apache/cassandra/transport/Server.java | 43 ++-
.../apache/cassandra/transport/SimpleClient.java | 2 +-
.../cassandra/distributed/impl/InstanceConfig.java | 21 +-
.../test/AbstractEncryptionOptionsImpl.java | 295 +++++++++++++++++++++
.../distributed/test/IncRepairTruncationTest.java | 3 +-
.../test/InternodeEncryptionOptionsTest.java | 218 +++++++++++++++
.../test/NativeTransportEncryptionOptionsTest.java | 137 ++++++++++
.../distributed/test/PreviewRepairTest.java | 3 +-
.../cassandra/config/EncryptionOptionsTest.java | 178 +++++++++++++
.../config/YamlConfigurationLoaderTest.java | 11 +-
test/unit/org/apache/cassandra/cql3/CQLTester.java | 2 +-
.../cassandra/db/virtual/SettingsTableTest.java | 6 +-
.../apache/cassandra/net/MessagingServiceTest.java | 6 +-
.../service/NativeTransportServiceTest.java | 86 +++++-
.../stress/settings/SettingsTransport.java | 2 +-
.../cassandra/stress/util/JavaDriverClient.java | 2 +-
31 files changed, 1298 insertions(+), 212 deletions(-)
diff --cc CHANGES.txt
index bf626ab,e46731b..635fa71
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,39 -1,10 +1,40 @@@
-3.11.10
+4.0-beta4
+ * Produce consistent tombstone for reads to avoid digest mistmatch (CASSANDRA-15369)
+ * Fix SSTableloader issue when restoring a table named backups (CASSANDRA-16235)
+ * Invalid serialized size for responses caused by increasing message time by 1ms which caused extra bytes in size calculation (CASSANDRA-16103)
+ * Throw BufferOverflowException from DataOutputBuffer for better visibility (CASSANDRA-16214)
++ * TLS connections to the storage port on a node without server encryption configured causes java.io.IOException accessing missing keystore (CASSANDRA-16144)
+Merged from 3.11:
Merged from 3.0:
- * Fix invalid cell value skipping when reading from disk (CASSANDRA-16223)
* Prevent invoking enable/disable gossip when not in NORMAL (CASSANDRA-16146)
-3.11.9
- * Synchronize Keyspace instance store/clear (CASSANDRA-16210)
+4.0-beta3
+ * Segregate Network and Chunk Cache BufferPools and Recirculate Partially Freed Chunks (CASSANDRA-15229)
+ * Fail truncation requests when they fail on a replica (CASSANDRA-16208)
+ * Move compact storage validation earlier in startup process (CASSANDRA-16063)
+ * Fix ByteBufferAccessor cast exceptions are thrown when trying to query a virtual table (CASSANDRA-16155)
+ * Consolidate node liveness check for forced repair (CASSANDRA-16113)
+ * Use unsigned short in ValueAccessor.sliceWithShortLength (CASSANDRA-16147)
+ * Abort repairs when getting a truncation request (CASSANDRA-15854)
+ * Remove bad assert when getting active compactions for an sstable (CASSANDRA-15457)
+ * Avoid failing compactions with very large partitions (CASSANDRA-15164)
+ * Prevent NPE in StreamMessage in type lookup (CASSANDRA-16131)
+ * Avoid invalid state transition exception during incremental repair (CASSANDRA-16067)
+ * Allow zero padding in timestamp serialization (CASSANDRA-16105)
+ * Add byte array backed cells (CASSANDRA-15393)
+ * Correctly handle pending ranges with adjacent range movements (CASSANDRA-14801)
+ * Avoid adding locahost when streaming trivial ranges (CASSANDRA-16099)
+ * Add nodetool getfullquerylog (CASSANDRA-15988)
+ * Fix yaml format and alignment in tpstats (CASSANDRA-11402)
+ * Avoid trying to keep track of RTs for endpoints we won't write to during read repair (CASSANDRA-16084)
+ * When compaction gets interrupted, the exception should include the compactionId (CASSANDRA-15954)
+ * Make Table/Keyspace Metric Names Consistent With Each Other (CASSANDRA-15909)
+ * Mutating sstable component may race with entire-sstable-streaming(ZCS) causing checksum validation failure (CASSANDRA-15861)
+ * NPE thrown while updating speculative execution time if keyspace is removed during task execution (CASSANDRA-15949)
+ * Show the progress of data streaming and index build (CASSANDRA-15406)
+ * Add flag to disable chunk cache and disable by default (CASSANDRA-16036)
+ * Upgrade to snakeyaml >= 1.26 version for CVE-2017-18640 fix (CASSANDRA-16150)
+Merged from 3.11:
* Fix ColumnFilter to avoid querying cells of unselected complex columns (CASSANDRA-15977)
* Fix memory leak in CompressedChunkReader (CASSANDRA-15880)
* Don't attempt value skipping with mixed version cluster (CASSANDRA-15833)
diff --cc src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 0387105,fe7291b..5fbb220
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@@ -760,11 -723,26 +760,16 @@@ public class DatabaseDescripto
throw new ConfigurationException("commitlog_segment_size_in_mb must be at least twice the size of max_mutation_size_in_kb / 1024", false);
// native transport encryption options
-- if (conf.native_transport_port_ssl != null
-- && conf.native_transport_port_ssl != conf.native_transport_port
- && !conf.client_encryption_options.isEnabled())
- && !conf.client_encryption_options.enabled)
++ if (conf.client_encryption_options != null)
{
-- throw new ConfigurationException("Encryption must be enabled in client_encryption_options for native_transport_port_ssl", false);
- }
++ conf.client_encryption_options.applyConfig();
+
- // 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
++ if (conf.native_transport_port_ssl != null
++ && conf.native_transport_port_ssl != conf.native_transport_port
++ && conf.client_encryption_options.tlsEncryptionPolicy() == EncryptionOptions.TlsEncryptionPolicy.UNENCRYPTED)
+ {
- 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));
++ throw new ConfigurationException("Encryption must be enabled in client_encryption_options for native_transport_port_ssl", false);
+ }
}
if (conf.max_value_size_in_mb <= 0)
@@@ -783,56 -761,32 +788,66 @@@
break;
}
- try
+ if (conf.otc_coalescing_enough_coalesced_messages > 128)
+ throw new ConfigurationException("otc_coalescing_enough_coalesced_messages must be smaller than 128", false);
+
+ if (conf.otc_coalescing_enough_coalesced_messages <= 0)
+ throw new ConfigurationException("otc_coalescing_enough_coalesced_messages must be positive", false);
+
++ if (conf.server_encryption_options != null)
+ {
- ParameterizedClass strategy = conf.back_pressure_strategy != null ? conf.back_pressure_strategy : RateBasedBackPressure.withDefaultParams();
- Class<?> clazz = Class.forName(strategy.class_name);
- if (!BackPressureStrategy.class.isAssignableFrom(clazz))
- throw new ConfigurationException(strategy + " is not an instance of " + BackPressureStrategy.class.getCanonicalName(), false);
++ conf.server_encryption_options.applyConfig();
+
- Constructor<?> ctor = clazz.getConstructor(Map.class);
- BackPressureStrategy instance = (BackPressureStrategy) ctor.newInstance(strategy.parameters);
- logger.info("Back-pressure is {} with strategy {}.", backPressureEnabled() ? "enabled" : "disabled", conf.back_pressure_strategy);
- backPressureStrategy = instance;
++ if (conf.server_encryption_options.enable_legacy_ssl_storage_port &&
++ conf.server_encryption_options.tlsEncryptionPolicy() == EncryptionOptions.TlsEncryptionPolicy.UNENCRYPTED)
++ {
++ throw new ConfigurationException("enable_legacy_ssl_storage_port is true (enabled) with internode encryption disabled (none). Enable encryption or disable the legacy ssl storage port.");
++ }
+ }
- catch (ConfigurationException ex)
+ Integer maxMessageSize = conf.internode_max_message_size_in_bytes;
+ if (maxMessageSize != null)
{
- throw ex;
+ if (maxMessageSize > conf.internode_application_receive_queue_reserve_endpoint_capacity_in_bytes)
+ throw new ConfigurationException("internode_max_message_size_in_mb must no exceed internode_application_receive_queue_reserve_endpoint_capacity_in_bytes", false);
+
+ if (maxMessageSize > conf.internode_application_receive_queue_reserve_global_capacity_in_bytes)
+ throw new ConfigurationException("internode_max_message_size_in_mb must no exceed internode_application_receive_queue_reserve_global_capacity_in_bytes", false);
+
+ if (maxMessageSize > conf.internode_application_send_queue_reserve_endpoint_capacity_in_bytes)
+ throw new ConfigurationException("internode_max_message_size_in_mb must no exceed internode_application_send_queue_reserve_endpoint_capacity_in_bytes", false);
+
+ if (maxMessageSize > conf.internode_application_send_queue_reserve_global_capacity_in_bytes)
+ throw new ConfigurationException("internode_max_message_size_in_mb must no exceed internode_application_send_queue_reserve_global_capacity_in_bytes", false);
}
- catch (Exception ex)
+ else
{
- throw new ConfigurationException("Error configuring back-pressure strategy: " + conf.back_pressure_strategy, ex);
+ conf.internode_max_message_size_in_bytes =
+ Math.min(conf.internode_application_receive_queue_reserve_endpoint_capacity_in_bytes,
+ conf.internode_application_send_queue_reserve_endpoint_capacity_in_bytes);
}
- if (conf.otc_coalescing_enough_coalesced_messages > 128)
- throw new ConfigurationException("otc_coalescing_enough_coalesced_messages must be smaller than 128", false);
+ validateMaxConcurrentAutoUpgradeTasksConf(conf.max_concurrent_automatic_sstable_upgrades);
+ }
- if (conf.otc_coalescing_enough_coalesced_messages <= 0)
- throw new ConfigurationException("otc_coalescing_enough_coalesced_messages must be positive", false);
+ @VisibleForTesting
+ static void applyConcurrentValidations(Config config)
+ {
+ if (config.concurrent_validations < 1)
+ {
+ config.concurrent_validations = config.concurrent_compactors;
+ }
+ else if (config.concurrent_validations > config.concurrent_compactors && !allowUnlimitedConcurrentValidations)
+ {
+ throw new ConfigurationException("To set concurrent_validations > concurrent_compactors, " +
+ "set the system property cassandra.allow_unlimited_concurrent_validations=true");
+ }
+ }
+
+ @VisibleForTesting
+ static void applyRepairCommandPoolSize(Config config)
+ {
+ if (config.repair_command_pool_size < 1)
+ config.repair_command_pool_size = config.concurrent_validations;
}
private static String storagedirFor(String type)
diff --cc src/java/org/apache/cassandra/config/EncryptionOptions.java
index 8ccf6d2,d662871..26bf458
--- a/src/java/org/apache/cassandra/config/EncryptionOptions.java
+++ b/src/java/org/apache/cassandra/config/EncryptionOptions.java
@@@ -17,255 -17,25 +17,353 @@@
*/
package org.apache.cassandra.config;
-import javax.net.ssl.SSLSocketFactory;
++import java.io.File;
+import java.util.List;
+import java.util.Objects;
-public abstract class EncryptionOptions
+import com.google.common.collect.ImmutableList;
+
++import org.slf4j.Logger;
++import org.slf4j.LoggerFactory;
++
+import org.apache.cassandra.locator.IEndpointSnitch;
+import org.apache.cassandra.locator.InetAddressAndPort;
+
+public class EncryptionOptions
{
- public String keystore = "conf/.keystore";
- public String keystore_password = "cassandra";
- public String truststore = "conf/.truststore";
- public String truststore_password = "cassandra";
- public String[] cipher_suites = ((SSLSocketFactory)SSLSocketFactory.getDefault()).getDefaultCipherSuites();
- public String protocol = "TLS";
- public String algorithm = "SunX509";
- public String store_type = "JKS";
- public boolean require_client_auth = false;
- public boolean require_endpoint_verification = false;
++ Logger logger = LoggerFactory.getLogger(EncryptionOptions.class);
++
++ public enum TlsEncryptionPolicy
++ {
++ UNENCRYPTED("unencrypted"), OPTIONAL("optionally encrypted"), ENCRYPTED("encrypted");
++
++ private final String description;
++
++ TlsEncryptionPolicy(String description)
++ {
++ this.description = description;
++ }
++
++ public String description()
++ {
++ return description;
++ }
++ }
++
+ public final String keystore;
+ public final String keystore_password;
+ public final String truststore;
+ public final String truststore_password;
+ public final List<String> cipher_suites;
+ public final String protocol;
+ public final String algorithm;
+ public final String store_type;
+ public final boolean require_client_auth;
+ public final boolean require_endpoint_verification;
+ // ServerEncryptionOptions does not use the enabled flag at all instead using the existing
+ // internode_encryption option. So we force this private and expose through isEnabled
+ // so users of ServerEncryptionOptions can't accidentally use this when they should use isEnabled
+ // Long term we need to refactor ClientEncryptionOptions and ServerEncyrptionOptions to be separate
+ // classes so we can choose appropriate configuration for each.
+ // See CASSANDRA-15262 and CASSANDRA-15146
- private boolean enabled;
- public final Boolean optional;
++ protected Boolean enabled;
++ protected Boolean optional;
++
++ // Calculated by calling applyConfig() after populating/parsing
++ protected Boolean isEnabled = null;
++ protected Boolean isOptional = null;
+
+ public EncryptionOptions()
+ {
+ keystore = "conf/.keystore";
+ keystore_password = "cassandra";
+ truststore = "conf/.truststore";
+ truststore_password = "cassandra";
+ cipher_suites = ImmutableList.of();
+ protocol = "TLS";
+ algorithm = null;
+ store_type = "JKS";
+ require_client_auth = false;
+ require_endpoint_verification = false;
- enabled = false;
- optional = true;
++ enabled = null;
++ optional = null;
+ }
+
- public EncryptionOptions(String keystore, String keystore_password, String truststore, String truststore_password, List<String> cipher_suites, String protocol, String algorithm, String store_type, boolean require_client_auth, boolean require_endpoint_verification, boolean enabled, Boolean optional)
++ public EncryptionOptions(String keystore, String keystore_password, String truststore, String truststore_password, List<String> cipher_suites, String protocol, String algorithm, String store_type, boolean require_client_auth, boolean require_endpoint_verification, Boolean enabled, Boolean optional)
+ {
+ this.keystore = keystore;
+ this.keystore_password = keystore_password;
+ this.truststore = truststore;
+ this.truststore_password = truststore_password;
+ this.cipher_suites = cipher_suites;
+ this.protocol = protocol;
+ this.algorithm = algorithm;
+ this.store_type = store_type;
+ this.require_client_auth = require_client_auth;
+ this.require_endpoint_verification = require_endpoint_verification;
+ this.enabled = enabled;
- if (optional != null) {
- this.optional = optional;
- } else {
- // If someone is asking for an _insecure_ connection and not explicitly telling us to refuse
- // encrypted connections we assume they would like to be able to transition to encrypted connections
- // in the future.
- this.optional = !enabled;
- }
++ this.optional = optional;
+ }
+
+ public EncryptionOptions(EncryptionOptions options)
+ {
+ keystore = options.keystore;
+ keystore_password = options.keystore_password;
+ truststore = options.truststore;
+ truststore_password = options.truststore_password;
+ cipher_suites = options.cipher_suites;
+ protocol = options.protocol;
+ algorithm = options.algorithm;
+ store_type = options.store_type;
+ require_client_auth = options.require_client_auth;
+ require_endpoint_verification = options.require_endpoint_verification;
+ enabled = options.enabled;
- if (options.optional != null) {
- optional = options.optional;
- } else {
- // If someone is asking for an _insecure_ connection and not explicitly telling us to refuse
- // encrypted connections we assume they would like to be able to transition to encrypted connections
- // in the future.
- optional = !enabled;
++ this.optional = options.optional;
++ }
++
++ /* Computes enabled and optional before use. Because the configuration can be loaded
++ * through pluggable mechanisms this is the only safe way to make sure that
++ * enabled and optional are set correctly.
++ */
++ public EncryptionOptions applyConfig()
++ {
++ ensureConfigNotApplied();
++
++ isEnabled = this.enabled != null && enabled;
++
++ if (optional != null)
++ {
++ isOptional = optional;
++ }
++ // If someone is asking for an _insecure_ connection and not explicitly telling us to refuse
++ // encrypted connections AND they have a keystore file, we assume they would like to be able
++ // to transition to encrypted connections in the future.
++ else if (new File(keystore).exists())
++ {
++ isOptional = !isEnabled;
++ }
++ else
++ {
++ // Otherwise if there's no keystore, not possible to establish an optional secure connection
++ isOptional = false;
+ }
++ return this;
++ }
++
++ private void ensureConfigApplied()
++ {
++ if (isEnabled == null || isOptional == null)
++ throw new IllegalStateException("EncryptionOptions.applyConfig must be called first");
++ }
++
++ private void ensureConfigNotApplied()
++ {
++ if (isEnabled != null || isOptional != null)
++ throw new IllegalStateException("EncryptionOptions cannot be changed after configuration applied");
+ }
+
+ /**
+ * Indicates if the channel should be encrypted. Client and Server uses different logic to determine this
+ *
+ * @return if the channel should be encrypted
+ */
- public boolean isEnabled() {
- return this.enabled;
++ public Boolean isEnabled() {
++ ensureConfigApplied();
++ return isEnabled;
+ }
+
+ /**
+ * Sets if encryption should be enabled for this channel. Note that this should only be called by
+ * the configuration parser or tests. It is public only for that purpose, mutating enabled state
+ * is probably a bad idea.
- * @param enabled
++ * @param enabled value to set
+ */
- public void setEnabled(boolean enabled) {
++ public void setEnabled(Boolean enabled) {
++ ensureConfigNotApplied();
+ this.enabled = enabled;
+ }
+
++ /**
++ * Indicates if the channel may be encrypted (but is not required to be).
++ * Explicitly providing a value in the configuration take precedent.
++ * If no optional value is set and !isEnabled(), then optional connections are allowed
++ * if a keystore exists. Without it, it would be impossible to establish the connections.
++ *
++ * Return type is Boolean even though it can never be null so that snakeyaml can find it
++ * @return if the channel may be encrypted
++ */
++ public Boolean isOptional()
++ {
++ ensureConfigApplied();
++ return isOptional;
++ }
++
++ /**
++ * Sets if encryption should be optional for this channel. Note that this should only be called by
++ * the configuration parser or tests. It is public only for that purpose, mutating enabled state
++ * is probably a bad idea.
++ * @param optional value to set
++ */
++ public void setOptional(boolean optional) {
++ ensureConfigNotApplied();
++ this.optional = optional;
++ }
++
++ public TlsEncryptionPolicy tlsEncryptionPolicy()
++ {
++ if (isOptional())
++ {
++ return TlsEncryptionPolicy.OPTIONAL;
++ }
++ else if (isEnabled())
++ {
++ return TlsEncryptionPolicy.ENCRYPTED;
++ }
++ else
++ {
++ return TlsEncryptionPolicy.UNENCRYPTED;
++ }
++ }
++
+ public EncryptionOptions withKeyStore(String keystore)
+ {
+ return new EncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- enabled, optional);
++ enabled, optional).applyConfig();
+ }
+
+ public EncryptionOptions withKeyStorePassword(String keystore_password)
+ {
+ return new EncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- enabled, optional);
++ enabled, optional).applyConfig();
+ }
+
+ public EncryptionOptions withTrustStore(String truststore)
+ {
+ return new EncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- enabled, optional);
++ enabled, optional).applyConfig();
+ }
+
+ public EncryptionOptions withTrustStorePassword(String truststore_password)
+ {
+ return new EncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- enabled, optional);
++ enabled, optional).applyConfig();
+ }
+
+ public EncryptionOptions withCipherSuites(List<String> cipher_suites)
+ {
+ return new EncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- enabled, optional);
++ enabled, optional).applyConfig();
+ }
+
+ public EncryptionOptions withCipherSuites(String ... cipher_suites)
+ {
+ return new EncryptionOptions(keystore, keystore_password, truststore, truststore_password, ImmutableList.copyOf(cipher_suites),
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- enabled, optional);
++ enabled, optional).applyConfig();
+ }
+
+ public EncryptionOptions withProtocol(String protocol)
+ {
+ return new EncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- enabled, optional);
++ enabled, optional).applyConfig();
+ }
+
+ public EncryptionOptions withAlgorithm(String algorithm)
+ {
+ return new EncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- enabled, optional);
++ enabled, optional).applyConfig();
+ }
+
+ public EncryptionOptions withStoreType(String store_type)
+ {
+ return new EncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- enabled, optional);
++ enabled, optional).applyConfig();
+ }
- public static class ClientEncryptionOptions extends EncryptionOptions
+ public EncryptionOptions withRequireClientAuth(boolean require_client_auth)
{
- public boolean enabled = false;
- public boolean optional = false;
+ return new EncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- enabled, optional);
++ enabled, optional).applyConfig();
+ }
+
+ public EncryptionOptions withRequireEndpointVerification(boolean require_endpoint_verification)
+ {
+ return new EncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- enabled, optional);
++ enabled, optional).applyConfig();
+ }
+
+ public EncryptionOptions withEnabled(boolean enabled)
+ {
+ return new EncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- enabled, optional);
++ enabled, optional).applyConfig();
+ }
+
- public EncryptionOptions withOptional(boolean optional)
++ public EncryptionOptions withOptional(Boolean optional)
+ {
+ return new EncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- enabled, optional);
++ enabled, optional).applyConfig();
+ }
+
+ /**
+ * The method is being mainly used to cache SslContexts therefore, we only consider
+ * fields that would make a difference when the TrustStore or KeyStore files are updated
+ */
+ @Override
+ public boolean equals(Object o)
+ {
+ if (o == this)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ EncryptionOptions opt = (EncryptionOptions)o;
+ return enabled == opt.enabled &&
+ optional == opt.optional &&
+ require_client_auth == opt.require_client_auth &&
+ require_endpoint_verification == opt.require_endpoint_verification &&
+ Objects.equals(keystore, opt.keystore) &&
+ Objects.equals(keystore_password, opt.keystore_password) &&
+ Objects.equals(truststore, opt.truststore) &&
+ Objects.equals(truststore_password, opt.truststore_password) &&
+ Objects.equals(protocol, opt.protocol) &&
+ Objects.equals(algorithm, opt.algorithm) &&
+ Objects.equals(store_type, opt.store_type) &&
+ Objects.equals(cipher_suites, opt.cipher_suites);
+ }
+
+ /**
+ * The method is being mainly used to cache SslContexts therefore, we only consider
+ * fields that would make a difference when the TrustStore or KeyStore files are updated
+ */
+ @Override
+ public int hashCode()
+ {
+ int result = 0;
+ result += 31 * (keystore == null ? 0 : keystore.hashCode());
+ result += 31 * (keystore_password == null ? 0 : keystore_password.hashCode());
+ result += 31 * (truststore == null ? 0 : truststore.hashCode());
+ result += 31 * (truststore_password == null ? 0 : truststore_password.hashCode());
+ result += 31 * (protocol == null ? 0 : protocol.hashCode());
+ result += 31 * (algorithm == null ? 0 : algorithm.hashCode());
+ result += 31 * (store_type == null ? 0 : store_type.hashCode());
- result += 31 * Boolean.hashCode(enabled);
- result += 31 * Boolean.hashCode(optional);
++ result += 31 * (enabled == null ? 0 : Boolean.hashCode(enabled));
++ result += 31 * (optional == null ? 0 : Boolean.hashCode(optional));
+ result += 31 * (cipher_suites == null ? 0 : cipher_suites.hashCode());
+ result += 31 * Boolean.hashCode(require_client_auth);
+ result += 31 * Boolean.hashCode(require_endpoint_verification);
+ return result;
}
public static class ServerEncryptionOptions extends EncryptionOptions
@@@ -274,155 -44,6 +372,177 @@@
{
all, none, dc, rack
}
- public InternodeEncryption internode_encryption = InternodeEncryption.none;
+
+ public final InternodeEncryption internode_encryption;
+ public final boolean enable_legacy_ssl_storage_port;
+
+ public ServerEncryptionOptions()
+ {
+ this.internode_encryption = InternodeEncryption.none;
+ this.enable_legacy_ssl_storage_port = false;
+ }
+
+ public ServerEncryptionOptions(String keystore, String keystore_password, String truststore, String truststore_password, List<String> cipher_suites, String protocol, String algorithm, String store_type, boolean require_client_auth, boolean require_endpoint_verification, Boolean optional, InternodeEncryption internode_encryption, boolean enable_legacy_ssl_storage_port)
+ {
- super(keystore, keystore_password, truststore, truststore_password, cipher_suites, protocol, algorithm, store_type, require_client_auth, require_endpoint_verification, internode_encryption != InternodeEncryption.none, optional);
++ super(keystore, keystore_password, truststore, truststore_password, cipher_suites, protocol, algorithm, store_type, require_client_auth, require_endpoint_verification, null, optional);
+ this.internode_encryption = internode_encryption;
+ this.enable_legacy_ssl_storage_port = enable_legacy_ssl_storage_port;
+ }
+
+ public ServerEncryptionOptions(ServerEncryptionOptions options)
+ {
+ super(options);
+ this.internode_encryption = options.internode_encryption;
+ this.enable_legacy_ssl_storage_port = options.enable_legacy_ssl_storage_port;
+ }
+
- public boolean isEnabled() {
- return this.internode_encryption != InternodeEncryption.none;
++ @Override
++ public EncryptionOptions applyConfig()
++ {
++ return applyConfigInternal();
++ }
++
++ private ServerEncryptionOptions applyConfigInternal()
++ {
++
++
++ super.applyConfig();
++
++ isEnabled = this.internode_encryption != InternodeEncryption.none;
++
++ if (this.enabled != null && this.enabled && !isEnabled)
++ {
++ logger.warn("Setting server_encryption_options.enabled has no effect, use internode_encryption");
++ }
++
++ // regardless of the optional flag, if the internode encryption is set to rack or dc
++ // it must be optional so that unencrypted connections within the rack or dc can be established.
++ isOptional = super.isOptional || internode_encryption == InternodeEncryption.rack || internode_encryption == InternodeEncryption.dc;
++
++ return this;
+ }
+
+ public boolean shouldEncrypt(InetAddressAndPort endpoint)
+ {
+ IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
+ switch (internode_encryption)
+ {
+ case none:
+ return false; // if nothing needs to be encrypted then return immediately.
+ case all:
+ break;
+ case dc:
+ if (snitch.getDatacenter(endpoint).equals(snitch.getLocalDatacenter()))
+ return false;
+ break;
+ case rack:
+ // for rack then check if the DC's are the same.
+ if (snitch.getRack(endpoint).equals(snitch.getLocalRack())
+ && snitch.getDatacenter(endpoint).equals(snitch.getLocalDatacenter()))
+ return false;
+ break;
+ }
+ return true;
+ }
+
+
+ public ServerEncryptionOptions withKeyStore(String keystore)
+ {
+ return new ServerEncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- optional, internode_encryption, enable_legacy_ssl_storage_port);
++ optional, internode_encryption, enable_legacy_ssl_storage_port).applyConfigInternal();
+ }
+
+ public ServerEncryptionOptions withKeyStorePassword(String keystore_password)
+ {
+ return new ServerEncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- optional, internode_encryption, enable_legacy_ssl_storage_port);
++ optional, internode_encryption, enable_legacy_ssl_storage_port).applyConfigInternal();
+ }
+
+ public ServerEncryptionOptions withTrustStore(String truststore)
+ {
+ return new ServerEncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- optional, internode_encryption, enable_legacy_ssl_storage_port);
++ optional, internode_encryption, enable_legacy_ssl_storage_port).applyConfigInternal();
+ }
+
+ public ServerEncryptionOptions withTrustStorePassword(String truststore_password)
+ {
+ return new ServerEncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- optional, internode_encryption, enable_legacy_ssl_storage_port);
++ optional, internode_encryption, enable_legacy_ssl_storage_port).applyConfigInternal();
+ }
+
+ public ServerEncryptionOptions withCipherSuites(List<String> cipher_suites)
+ {
+ return new ServerEncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- optional, internode_encryption, enable_legacy_ssl_storage_port);
++ optional, internode_encryption, enable_legacy_ssl_storage_port).applyConfigInternal();
+ }
+
+ public ServerEncryptionOptions withCipherSuites(String ... cipher_suites)
+ {
+ return new ServerEncryptionOptions(keystore, keystore_password, truststore, truststore_password, ImmutableList.copyOf(cipher_suites),
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- optional, internode_encryption, enable_legacy_ssl_storage_port);
++ optional, internode_encryption, enable_legacy_ssl_storage_port).applyConfigInternal();
+ }
+
+ public ServerEncryptionOptions withProtocol(String protocol)
+ {
+ return new ServerEncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- optional, internode_encryption, enable_legacy_ssl_storage_port);
++ optional, internode_encryption, enable_legacy_ssl_storage_port).applyConfigInternal();
+ }
+
+ public ServerEncryptionOptions withAlgorithm(String algorithm)
+ {
+ return new ServerEncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- optional, internode_encryption, enable_legacy_ssl_storage_port);
++ optional, internode_encryption, enable_legacy_ssl_storage_port).applyConfigInternal();
+ }
+
+ public ServerEncryptionOptions withStoreType(String store_type)
+ {
+ return new ServerEncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- optional, internode_encryption, enable_legacy_ssl_storage_port);
++ optional, internode_encryption, enable_legacy_ssl_storage_port).applyConfigInternal();
+ }
+
+ public ServerEncryptionOptions withRequireClientAuth(boolean require_client_auth)
+ {
+ return new ServerEncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- optional, internode_encryption, enable_legacy_ssl_storage_port);
++ optional, internode_encryption, enable_legacy_ssl_storage_port).applyConfigInternal();
+ }
+
+ public ServerEncryptionOptions withRequireEndpointVerification(boolean require_endpoint_verification)
+ {
+ return new ServerEncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- optional, internode_encryption, enable_legacy_ssl_storage_port);
++ optional, internode_encryption, enable_legacy_ssl_storage_port).applyConfigInternal();
+ }
+
+ public ServerEncryptionOptions withOptional(boolean optional)
+ {
+ return new ServerEncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- optional, internode_encryption, enable_legacy_ssl_storage_port);
++ optional, internode_encryption, enable_legacy_ssl_storage_port).applyConfigInternal();
+ }
+
+ public ServerEncryptionOptions withInternodeEncryption(InternodeEncryption internode_encryption)
+ {
+ return new ServerEncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- optional, internode_encryption, enable_legacy_ssl_storage_port);
++ optional, internode_encryption, enable_legacy_ssl_storage_port).applyConfigInternal();
+ }
+
+ public ServerEncryptionOptions withLegacySslStoragePort(boolean enable_legacy_ssl_storage_port)
+ {
+ return new ServerEncryptionOptions(keystore, keystore_password, truststore, truststore_password, cipher_suites,
+ protocol, algorithm, store_type, require_client_auth, require_endpoint_verification,
- optional, internode_encryption, enable_legacy_ssl_storage_port);
++ optional, internode_encryption, enable_legacy_ssl_storage_port).applyConfigInternal();
+ }
+
}
}
diff --cc src/java/org/apache/cassandra/db/virtual/SettingsTable.java
index d11f69a,0000000..e47ce8c
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/db/virtual/SettingsTable.java
+++ b/src/java/org/apache/cassandra/db/virtual/SettingsTable.java
@@@ -1,189 -1,0 +1,189 @@@
+/*
+ * 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.lang.reflect.Field;
+import java.lang.reflect.Modifier;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.function.BiConsumer;
+import java.util.stream.Collectors;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Functions;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+
+import org.apache.cassandra.audit.AuditLogOptions;
+import org.apache.cassandra.config.*;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.dht.LocalPartitioner;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.transport.ServerError;
+
+final class SettingsTable extends AbstractVirtualTable
+{
+ private static final String NAME = "name";
+ private static final String VALUE = "value";
+
+ @VisibleForTesting
+ static final Map<String, Field> FIELDS =
+ Arrays.stream(Config.class.getFields())
+ .filter(f -> !Modifier.isStatic(f.getModifiers()))
+ .collect(Collectors.toMap(Field::getName, Functions.identity()));
+
+ @VisibleForTesting
+ final Map<String, BiConsumer<SimpleDataSet, Field>> overrides =
+ ImmutableMap.<String, BiConsumer<SimpleDataSet, Field>>builder()
+ .put("audit_logging_options", this::addAuditLoggingOptions)
+ .put("client_encryption_options", this::addEncryptionOptions)
+ .put("server_encryption_options", this::addEncryptionOptions)
+ .put("transparent_data_encryption_options", this::addTransparentEncryptionOptions)
+ .build();
+
+ private final Config config;
+
+ SettingsTable(String keyspace)
+ {
+ this(keyspace, DatabaseDescriptor.getRawConfig());
+ }
+
+ SettingsTable(String keyspace, Config config)
+ {
+ super(TableMetadata.builder(keyspace, "settings")
+ .comment("current settings")
+ .kind(TableMetadata.Kind.VIRTUAL)
+ .partitioner(new LocalPartitioner(UTF8Type.instance))
+ .addPartitionKeyColumn(NAME, UTF8Type.instance)
+ .addRegularColumn(VALUE, UTF8Type.instance)
+ .build());
+ this.config = config;
+ }
+
+ @VisibleForTesting
+ Object getValue(Field f)
+ {
+ Object value;
+ try
+ {
+ value = f.get(config);
+ }
+ catch (IllegalAccessException | IllegalArgumentException e)
+ {
+ throw new ServerError(e);
+ }
+ return value;
+ }
+
+ private void addValue(SimpleDataSet result, Field f)
+ {
+ Object value = getValue(f);
+ if (value == null)
+ {
+ result.row(f.getName());
+ }
+ else if (overrides.containsKey(f.getName()))
+ {
+ overrides.get(f.getName()).accept(result, f);
+ }
+ else
+ {
+ if (value.getClass().isArray())
+ value = Arrays.toString((Object[]) value);
+ result.row(f.getName()).column(VALUE, value.toString());
+ }
+ }
+
+ @Override
+ public DataSet data(DecoratedKey partitionKey)
+ {
+ SimpleDataSet result = new SimpleDataSet(metadata());
+ String name = UTF8Type.instance.compose(partitionKey.getKey());
+ Field field = FIELDS.get(name);
+ if (field != null)
+ {
+ addValue(result, field);
+ }
+ else
+ {
+ // rows created by overrides might be directly queried so include them in result to be possibly filtered
+ for (String override : overrides.keySet())
+ if (name.startsWith(override))
+ addValue(result, FIELDS.get(override));
+ }
+ return result;
+ }
+
+ @Override
+ public DataSet data()
+ {
+ SimpleDataSet result = new SimpleDataSet(metadata());
+ for (Field setting : FIELDS.values())
+ addValue(result, setting);
+ return result;
+ }
+
+ private void addAuditLoggingOptions(SimpleDataSet result, Field f)
+ {
+ Preconditions.checkArgument(AuditLogOptions.class.isAssignableFrom(f.getType()));
+
+ AuditLogOptions value = (AuditLogOptions) getValue(f);
+ result.row(f.getName() + "_enabled").column(VALUE, Boolean.toString(value.enabled));
+ result.row(f.getName() + "_logger").column(VALUE, value.logger.class_name);
+ result.row(f.getName() + "_audit_logs_dir").column(VALUE, value.audit_logs_dir);
+ result.row(f.getName() + "_included_keyspaces").column(VALUE, value.included_keyspaces);
+ result.row(f.getName() + "_excluded_keyspaces").column(VALUE, value.excluded_keyspaces);
+ result.row(f.getName() + "_included_categories").column(VALUE, value.included_categories);
+ result.row(f.getName() + "_excluded_categories").column(VALUE, value.excluded_categories);
+ result.row(f.getName() + "_included_users").column(VALUE, value.included_users);
+ result.row(f.getName() + "_excluded_users").column(VALUE, value.excluded_users);
+ }
+
+ private void addEncryptionOptions(SimpleDataSet result, Field f)
+ {
+ Preconditions.checkArgument(EncryptionOptions.class.isAssignableFrom(f.getType()));
+
+ EncryptionOptions value = (EncryptionOptions) getValue(f);
+ result.row(f.getName() + "_enabled").column(VALUE, Boolean.toString(value.isEnabled()));
+ result.row(f.getName() + "_algorithm").column(VALUE, value.algorithm);
+ result.row(f.getName() + "_protocol").column(VALUE, value.protocol);
+ result.row(f.getName() + "_cipher_suites").column(VALUE, value.cipher_suites.toString());
+ result.row(f.getName() + "_client_auth").column(VALUE, Boolean.toString(value.require_client_auth));
+ result.row(f.getName() + "_endpoint_verification").column(VALUE, Boolean.toString(value.require_endpoint_verification));
- result.row(f.getName() + "_optional").column(VALUE, Boolean.toString(value.optional));
++ result.row(f.getName() + "_optional").column(VALUE, Boolean.toString(value.isOptional()));
+
+ if (value instanceof EncryptionOptions.ServerEncryptionOptions)
+ {
+ EncryptionOptions.ServerEncryptionOptions server = (EncryptionOptions.ServerEncryptionOptions) value;
+ result.row(f.getName() + "_internode_encryption").column(VALUE, server.internode_encryption.toString());
+ result.row(f.getName() + "_legacy_ssl_storage_port").column(VALUE, Boolean.toString(server.enable_legacy_ssl_storage_port));
+ }
+ }
+
+ private void addTransparentEncryptionOptions(SimpleDataSet result, Field f)
+ {
+ Preconditions.checkArgument(TransparentDataEncryptionOptions.class.isAssignableFrom(f.getType()));
+
+ TransparentDataEncryptionOptions value = (TransparentDataEncryptionOptions) getValue(f);
+ result.row(f.getName() + "_enabled").column(VALUE, Boolean.toString(value.enabled));
+ result.row(f.getName() + "_cipher").column(VALUE, value.cipher);
+ result.row(f.getName() + "_chunk_length_kb").column(VALUE, Integer.toString(value.chunk_length_kb));
+ result.row(f.getName() + "_iv_length").column(VALUE, Integer.toString(value.iv_length));
+ }
+}
diff --cc src/java/org/apache/cassandra/net/InboundConnectionInitiator.java
index f2339eb,0000000..d21358a
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/net/InboundConnectionInitiator.java
+++ b/src/java/org/apache/cassandra/net/InboundConnectionInitiator.java
@@@ -1,506 -1,0 +1,527 @@@
+/*
+ * 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.net;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.List;
+import java.util.concurrent.Future;
+import java.util.function.Consumer;
+
- import javax.net.ssl.SSLSession;
-
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.group.ChannelGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.handler.logging.LogLevel;
+import io.netty.handler.logging.LoggingHandler;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslHandler;
+import org.apache.cassandra.config.EncryptionOptions;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.net.OutboundConnectionSettings.Framing;
+import org.apache.cassandra.security.SSLFactory;
+import org.apache.cassandra.streaming.async.StreamingInboundHandler;
+import org.apache.cassandra.utils.memory.BufferPools;
+
+import static java.lang.Math.*;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.cassandra.net.MessagingService.*;
+import static org.apache.cassandra.net.MessagingService.VERSION_40;
+import static org.apache.cassandra.net.MessagingService.current_version;
+import static org.apache.cassandra.net.MessagingService.minimum_version;
+import static org.apache.cassandra.net.SocketFactory.WIRETRACE;
- import static org.apache.cassandra.net.SocketFactory.encryptionLogStatement;
+import static org.apache.cassandra.net.SocketFactory.newSslHandler;
+
+public class InboundConnectionInitiator
+{
+ private static final Logger logger = LoggerFactory.getLogger(InboundConnectionInitiator.class);
+
+ private static class Initializer extends ChannelInitializer<SocketChannel>
+ {
+ private final InboundConnectionSettings settings;
+ private final ChannelGroup channelGroup;
+ private final Consumer<ChannelPipeline> pipelineInjector;
+
+ Initializer(InboundConnectionSettings settings, ChannelGroup channelGroup,
+ Consumer<ChannelPipeline> pipelineInjector)
+ {
+ this.settings = settings;
+ this.channelGroup = channelGroup;
+ this.pipelineInjector = pipelineInjector;
+ }
+
+ @Override
+ public void initChannel(SocketChannel channel) throws Exception
+ {
+ channelGroup.add(channel);
+
+ channel.config().setOption(ChannelOption.ALLOCATOR, GlobalBufferPoolAllocator.instance);
+ channel.config().setOption(ChannelOption.SO_KEEPALIVE, true);
+ channel.config().setOption(ChannelOption.SO_REUSEADDR, true);
+ channel.config().setOption(ChannelOption.TCP_NODELAY, true); // we only send handshake messages; no point ever delaying
+
+ ChannelPipeline pipeline = channel.pipeline();
+
+ pipelineInjector.accept(pipeline);
+
+ // order of handlers: ssl -> logger -> handshakeHandler
+ // For either unencrypted or transitional modes, allow Ssl optionally.
- if (settings.encryption.optional)
++ switch(settings.encryption.tlsEncryptionPolicy())
+ {
- pipeline.addFirst("ssl", new OptionalSslHandler(settings.encryption));
- }
- else
- {
- SslContext sslContext = SSLFactory.getOrCreateSslContext(settings.encryption, true, SSLFactory.SocketType.SERVER);
- InetSocketAddress peer = settings.encryption.require_endpoint_verification ? channel.remoteAddress() : null;
- SslHandler sslHandler = newSslHandler(channel, sslContext, peer);
- logger.trace("creating inbound netty SslContext: context={}, engine={}", sslContext.getClass().getName(), sslHandler.engine().getClass().getName());
- pipeline.addFirst("ssl", sslHandler);
++ case UNENCRYPTED:
++ // Handler checks for SSL connection attempts and cleanly rejects them if encryption is disabled
++ pipeline.addFirst("rejectssl", new RejectSslHandler());
++ break;
++ case OPTIONAL:
++ pipeline.addFirst("ssl", new OptionalSslHandler(settings.encryption));
++ break;
++ case ENCRYPTED:
++ SslHandler sslHandler = getSslHandler("creating", channel, settings.encryption);
++ pipeline.addFirst("ssl", sslHandler);
++ break;
+ }
+
+ if (WIRETRACE)
+ pipeline.addLast("logger", new LoggingHandler(LogLevel.INFO));
+
+ channel.pipeline().addLast("handshake", new Handler(settings));
+
+ }
+ }
+
+ /**
+ * Create a {@link Channel} that listens on the {@code localAddr}. This method will block while trying to bind to the address,
+ * but it does not make a remote call.
+ */
+ private static ChannelFuture bind(Initializer initializer) throws ConfigurationException
+ {
+ logger.info("Listening on {}", initializer.settings);
+
+ ServerBootstrap bootstrap = initializer.settings.socketFactory
+ .newServerBootstrap()
+ .option(ChannelOption.SO_BACKLOG, 1 << 9)
+ .option(ChannelOption.ALLOCATOR, GlobalBufferPoolAllocator.instance)
+ .option(ChannelOption.SO_REUSEADDR, true)
+ .childHandler(initializer);
+
+ int socketReceiveBufferSizeInBytes = initializer.settings.socketReceiveBufferSizeInBytes;
+ if (socketReceiveBufferSizeInBytes > 0)
+ bootstrap.childOption(ChannelOption.SO_RCVBUF, socketReceiveBufferSizeInBytes);
+
+ InetAddressAndPort bind = initializer.settings.bindAddress;
+ ChannelFuture channelFuture = bootstrap.bind(new InetSocketAddress(bind.address, bind.port));
+
+ if (!channelFuture.awaitUninterruptibly().isSuccess())
+ {
+ if (channelFuture.channel().isOpen())
+ channelFuture.channel().close();
+
+ Throwable failedChannelCause = channelFuture.cause();
+
+ String causeString = "";
+ if (failedChannelCause != null && failedChannelCause.getMessage() != null)
+ causeString = failedChannelCause.getMessage();
+
+ if (causeString.contains("in use"))
+ {
+ throw new ConfigurationException(bind + " is in use by another process. Change listen_address:storage_port " +
+ "in cassandra.yaml to values that do not conflict with other services");
+ }
+ // looking at the jdk source, solaris/windows bind failue messages both use the phrase "cannot assign requested address".
+ // windows message uses "Cannot" (with a capital 'C'), and solaris (a/k/a *nux) doe not. hence we search for "annot" <sigh>
+ else if (causeString.contains("annot assign requested address"))
+ {
+ throw new ConfigurationException("Unable to bind to address " + bind
+ + ". Set listen_address in cassandra.yaml to an interface you can bind to, e.g., your private IP address on EC2");
+ }
+ else
+ {
+ throw new ConfigurationException("failed to bind to: " + bind, failedChannelCause);
+ }
+ }
+
+ return channelFuture;
+ }
+
+ public static ChannelFuture bind(InboundConnectionSettings settings, ChannelGroup channelGroup,
+ Consumer<ChannelPipeline> pipelineInjector)
+ {
+ return bind(new Initializer(settings, channelGroup, pipelineInjector));
+ }
+
+ /**
+ * 'Server-side' component that negotiates the internode handshake when establishing a new connection.
+ * This handler will be the first in the netty channel for each incoming connection (secure socket (TLS) notwithstanding),
+ * and once the handshake is successful, it will configure the proper handlers ({@link InboundMessageHandler}
+ * or {@link StreamingInboundHandler}) and remove itself from the working pipeline.
+ */
+ static class Handler extends ByteToMessageDecoder
+ {
+ private final InboundConnectionSettings settings;
+
+ private HandshakeProtocol.Initiate initiate;
+ private HandshakeProtocol.ConfirmOutboundPre40 confirmOutboundPre40;
+
+ /**
+ * A future the essentially places a timeout on how long we'll wait for the peer
+ * to complete the next step of the handshake.
+ */
+ private Future<?> handshakeTimeout;
+
+ Handler(InboundConnectionSettings settings)
+ {
+ this.settings = settings;
+ }
+
+ /**
+ * On registration, immediately schedule a timeout to kill this connection if it does not handshake promptly,
+ * and authenticate the remote address.
+ */
+ public void handlerAdded(ChannelHandlerContext ctx) throws Exception
+ {
+ handshakeTimeout = ctx.executor().schedule(() -> {
+ logger.error("Timeout handshaking with {} (on {})", SocketFactory.addressId(initiate.from, (InetSocketAddress) ctx.channel().remoteAddress()), settings.bindAddress);
+ failHandshake(ctx);
+ }, HandshakeProtocol.TIMEOUT_MILLIS, MILLISECONDS);
+
- logSsl(ctx);
+ authenticate(ctx.channel().remoteAddress());
+ }
+
+ private void authenticate(SocketAddress socketAddress) throws IOException
+ {
+ if (socketAddress.getClass().getSimpleName().equals("EmbeddedSocketAddress"))
+ return;
+
+ if (!(socketAddress instanceof InetSocketAddress))
+ throw new IOException(String.format("Unexpected SocketAddress type: %s, %s", socketAddress.getClass(), socketAddress));
+
+ InetSocketAddress addr = (InetSocketAddress)socketAddress;
+ if (!settings.authenticate(addr.getAddress(), addr.getPort()))
+ throw new IOException("Authentication failure for inbound connection from peer " + addr);
+ }
+
- private void logSsl(ChannelHandlerContext ctx)
- {
- SslHandler sslHandler = ctx.pipeline().get(SslHandler.class);
- if (sslHandler != null)
- {
- SSLSession session = sslHandler.engine().getSession();
- logger.info("connection from peer {} to {}, protocol = {}",
- ctx.channel().remoteAddress(), ctx.channel().localAddress(), session.getProtocol());
- }
- }
-
+ @Override
+ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception
+ {
+ if (initiate == null) initiate(ctx, in);
+ else if (initiate.acceptVersions == null && confirmOutboundPre40 == null) confirmPre40(ctx, in);
+ else throw new IllegalStateException("Should no longer be on pipeline");
+ }
+
+ void initiate(ChannelHandlerContext ctx, ByteBuf in) throws IOException
+ {
+ initiate = HandshakeProtocol.Initiate.maybeDecode(in);
+ if (initiate == null)
+ return;
+
+ logger.trace("Received handshake initiation message from peer {}, message = {}", ctx.channel().remoteAddress(), initiate);
+ if (initiate.acceptVersions != null)
+ {
+ logger.trace("Connection version {} (min {}) from {}", initiate.acceptVersions.max, initiate.acceptVersions.min, initiate.from);
+
+ final AcceptVersions accept;
+
+ if (initiate.type.isStreaming())
+ accept = settings.acceptStreaming;
+ else
+ accept = settings.acceptMessaging;
+
+ int useMessagingVersion = max(accept.min, min(accept.max, initiate.acceptVersions.max));
+ ByteBuf flush = new HandshakeProtocol.Accept(useMessagingVersion, accept.max).encode(ctx.alloc());
+
+ AsyncChannelPromise.writeAndFlush(ctx, flush, (ChannelFutureListener) future -> {
+ if (!future.isSuccess())
+ exceptionCaught(future.channel(), future.cause());
+ });
+
+ if (initiate.acceptVersions.min > accept.max)
+ {
+ logger.info("peer {} only supports messaging versions higher ({}) than this node supports ({})", ctx.channel().remoteAddress(), initiate.acceptVersions.min, current_version);
+ failHandshake(ctx);
+ }
+ else if (initiate.acceptVersions.max < accept.min)
+ {
+ logger.info("peer {} only supports messaging versions lower ({}) than this node supports ({})", ctx.channel().remoteAddress(), initiate.acceptVersions.max, minimum_version);
+ failHandshake(ctx);
+ }
+ else
+ {
+ if (initiate.type.isStreaming())
+ setupStreamingPipeline(initiate.from, ctx);
+ else
+ setupMessagingPipeline(initiate.from, useMessagingVersion, initiate.acceptVersions.max, ctx.pipeline());
+ }
+ }
+ else
+ {
+ int version = initiate.requestMessagingVersion;
+ assert version < VERSION_40 && version >= settings.acceptMessaging.min;
+ logger.trace("Connection version {} from {}", version, ctx.channel().remoteAddress());
+
+ if (initiate.type.isStreaming())
+ {
+ // streaming connections are per-session and have a fixed version. we can't do anything with a wrong-version stream connection, so drop it.
+ if (version != settings.acceptStreaming.max)
+ {
+ logger.warn("Received stream using protocol version {} (my version {}). Terminating connection", version, settings.acceptStreaming.max);
+ failHandshake(ctx);
+ }
+ setupStreamingPipeline(initiate.from, ctx);
+ }
+ else
+ {
+ // if this version is < the MS version the other node is trying
+ // to connect with, the other node will disconnect
+ ByteBuf response = HandshakeProtocol.Accept.respondPre40(settings.acceptMessaging.max, ctx.alloc());
+ AsyncChannelPromise.writeAndFlush(ctx, response,
+ (ChannelFutureListener) future -> {
+ if (!future.isSuccess())
+ exceptionCaught(future.channel(), future.cause());
+ });
+
+ if (version < VERSION_30)
+ throw new IOException(String.format("Unable to read obsolete message version %s from %s; The earliest version supported is 3.0.0", version, ctx.channel().remoteAddress()));
+
+ // we don't setup the messaging pipeline here, as the legacy messaging handshake requires one more message to finish
+ }
+ }
+ }
+
+ /**
+ * Handles the third (and last) message in the internode messaging handshake protocol for pre40 nodes.
+ * Grabs the protocol version and IP addr the peer wants to use.
+ */
+ @VisibleForTesting
+ void confirmPre40(ChannelHandlerContext ctx, ByteBuf in)
+ {
+ confirmOutboundPre40 = HandshakeProtocol.ConfirmOutboundPre40.maybeDecode(in);
+ if (confirmOutboundPre40 == null)
+ return;
+
+ logger.trace("Received third handshake message from peer {}, message = {}", ctx.channel().remoteAddress(), confirmOutboundPre40);
+ setupMessagingPipeline(confirmOutboundPre40.from, initiate.requestMessagingVersion, confirmOutboundPre40.maxMessagingVersion, ctx.pipeline());
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
+ {
+ exceptionCaught(ctx.channel(), cause);
+ }
+
+ private void exceptionCaught(Channel channel, Throwable cause)
+ {
+ logger.error("Failed to properly handshake with peer {}. Closing the channel.", channel.remoteAddress(), cause);
+ try
+ {
+ failHandshake(channel);
+ }
+ catch (Throwable t)
+ {
+ logger.error("Unexpected exception in {}.exceptionCaught", this.getClass().getSimpleName(), t);
+ }
+ }
+
+ private void failHandshake(ChannelHandlerContext ctx)
+ {
+ failHandshake(ctx.channel());
+ }
+
+ private void failHandshake(Channel channel)
+ {
+ channel.close();
+ if (handshakeTimeout != null)
+ handshakeTimeout.cancel(true);
+ }
+
+ private void setupStreamingPipeline(InetAddressAndPort from, ChannelHandlerContext ctx)
+ {
+ handshakeTimeout.cancel(true);
+ assert initiate.framing == Framing.UNPROTECTED;
+
+ ChannelPipeline pipeline = ctx.pipeline();
+ Channel channel = ctx.channel();
+
+ if (from == null)
+ {
+ InetSocketAddress address = (InetSocketAddress) channel.remoteAddress();
+ from = InetAddressAndPort.getByAddressOverrideDefaults(address.getAddress(), address.getPort());
+ }
+
+ BufferPools.forNetworking().setRecycleWhenFreeForCurrentThread(false);
+ pipeline.replace(this, "streamInbound", new StreamingInboundHandler(from, current_version, null));
+
+ logger.info("{} streaming connection established, version = {}, framing = {}, encryption = {}",
+ SocketFactory.channelId(from,
+ (InetSocketAddress) channel.remoteAddress(),
+ settings.bindAddress,
+ (InetSocketAddress) channel.localAddress(),
+ ConnectionType.STREAMING,
+ channel.id().asShortText()),
+ current_version,
+ initiate.framing,
- pipeline.get("ssl") != null ? encryptionLogStatement(pipeline.channel(), settings.encryption) : "disabled");
++ SocketFactory.encryptionConnectionSummary(pipeline.channel()));
+ }
+
+ @VisibleForTesting
+ void setupMessagingPipeline(InetAddressAndPort from, int useMessagingVersion, int maxMessagingVersion, ChannelPipeline pipeline)
+ {
+ handshakeTimeout.cancel(true);
+ // record the "true" endpoint, i.e. the one the peer is identified with, as opposed to the socket it connected over
+ instance().versions.set(from, maxMessagingVersion);
+
+ BufferPools.forNetworking().setRecycleWhenFreeForCurrentThread(false);
+ BufferPoolAllocator allocator = GlobalBufferPoolAllocator.instance;
+ if (initiate.type == ConnectionType.LARGE_MESSAGES)
+ {
+ // for large messages, swap the global pool allocator for a local one, to optimise utilisation of chunks
+ allocator = new LocalBufferPoolAllocator(pipeline.channel().eventLoop());
+ pipeline.channel().config().setAllocator(allocator);
+ }
+
+ FrameDecoder frameDecoder;
+ switch (initiate.framing)
+ {
+ case LZ4:
+ {
+ if (useMessagingVersion >= VERSION_40)
+ frameDecoder = FrameDecoderLZ4.fast(allocator);
+ else
+ frameDecoder = new FrameDecoderLegacyLZ4(allocator, useMessagingVersion);
+ break;
+ }
+ case CRC:
+ {
+ if (useMessagingVersion >= VERSION_40)
+ {
+ frameDecoder = FrameDecoderCrc.create(allocator);
+ break;
+ }
+ }
+ case UNPROTECTED:
+ {
+ if (useMessagingVersion >= VERSION_40)
+ frameDecoder = new FrameDecoderUnprotected(allocator);
+ else
+ frameDecoder = new FrameDecoderLegacy(allocator, useMessagingVersion);
+ break;
+ }
+ default:
+ throw new AssertionError();
+ }
+
+ frameDecoder.addLastTo(pipeline);
+
+ InboundMessageHandler handler =
+ settings.handlers.apply(from).createHandler(frameDecoder, initiate.type, pipeline.channel(), useMessagingVersion);
+
+ logger.info("{} messaging connection established, version = {}, framing = {}, encryption = {}",
+ handler.id(true),
+ useMessagingVersion,
+ initiate.framing,
- pipeline.get("ssl") != null ? encryptionLogStatement(pipeline.channel(), settings.encryption) : "disabled");
++ SocketFactory.encryptionConnectionSummary(pipeline.channel()));
+
+ pipeline.addLast("deserialize", handler);
+
+ pipeline.remove(this);
+ }
+ }
+
++ private static SslHandler getSslHandler(String description, Channel channel, EncryptionOptions.ServerEncryptionOptions encryptionOptions) throws IOException
++ {
++ final boolean buildTrustStore = true;
++ SslContext sslContext = SSLFactory.getOrCreateSslContext(encryptionOptions, buildTrustStore, SSLFactory.SocketType.SERVER);
++ InetSocketAddress peer = encryptionOptions.require_endpoint_verification ? (InetSocketAddress) channel.remoteAddress() : null;
++ SslHandler sslHandler = newSslHandler(channel, sslContext, peer);
++ logger.trace("{} inbound netty SslContext: context={}, engine={}", description, sslContext.getClass().getName(), sslHandler.engine().getClass().getName());
++ return sslHandler;
++ }
++
+ private static class OptionalSslHandler extends ByteToMessageDecoder
+ {
+ private final EncryptionOptions.ServerEncryptionOptions encryptionOptions;
+
+ OptionalSslHandler(EncryptionOptions.ServerEncryptionOptions encryptionOptions)
+ {
+ this.encryptionOptions = encryptionOptions;
+ }
+
+ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception
+ {
+ if (in.readableBytes() < 5)
+ {
+ // To detect if SSL must be used we need to have at least 5 bytes, so return here and try again
+ // once more bytes a ready.
+ return;
+ }
+
+ if (SslHandler.isEncrypted(in))
+ {
+ // Connection uses SSL/TLS, replace the detection handler with a SslHandler and so use encryption.
- SslContext sslContext = SSLFactory.getOrCreateSslContext(encryptionOptions, true, SSLFactory.SocketType.SERVER);
- Channel channel = ctx.channel();
- InetSocketAddress peer = encryptionOptions.require_endpoint_verification ? (InetSocketAddress) channel.remoteAddress() : null;
- SslHandler sslHandler = newSslHandler(channel, sslContext, peer);
++ SslHandler sslHandler = getSslHandler("replacing optional", ctx.channel(), encryptionOptions);
+ ctx.pipeline().replace(this, "ssl", sslHandler);
+ }
+ else
+ {
+ // Connection use no TLS/SSL encryption, just remove the detection handler and continue without
+ // SslHandler in the pipeline.
+ ctx.pipeline().remove(this);
+ }
+ }
+ }
++
++ private static class RejectSslHandler extends ByteToMessageDecoder
++ {
++ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
++ {
++ if (in.readableBytes() < 5)
++ {
++ // To detect if SSL must be used we need to have at least 5 bytes, so return here and try again
++ // once more bytes a ready.
++ return;
++ }
++
++ if (SslHandler.isEncrypted(in))
++ {
++ logger.info("Rejected incoming TLS connection before negotiating from {} to {}. TLS is explicitly disabled by configuration.",
++ ctx.channel().remoteAddress(), ctx.channel().localAddress());
++ in.readBytes(in.readableBytes()); // discard the readable bytes so not called again
++ ctx.close();
++ }
++ else
++ {
++ // Incoming connection did not attempt TLS/SSL encryption, just remove the detection handler and continue without
++ // SslHandler in the pipeline.
++ ctx.pipeline().remove(this);
++ }
++ }
++ }
+}
diff --cc src/java/org/apache/cassandra/net/InboundConnectionSettings.java
index 20f185a,0000000..00def4f
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/net/InboundConnectionSettings.java
+++ b/src/java/org/apache/cassandra/net/InboundConnectionSettings.java
@@@ -1,213 -1,0 +1,213 @@@
+/*
+ * 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.net;
+
+import java.net.InetAddress;
+import java.util.function.Function;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.cassandra.auth.IInternodeAuthenticator;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static java.lang.String.format;
+import static org.apache.cassandra.net.MessagingService.*;
+
+public class InboundConnectionSettings
+{
+ public final IInternodeAuthenticator authenticator;
+ public final InetAddressAndPort bindAddress;
+ public final ServerEncryptionOptions encryption;
+ public final Integer socketReceiveBufferSizeInBytes;
+ public final Integer applicationReceiveQueueCapacityInBytes;
+ public final AcceptVersions acceptMessaging;
+ public final AcceptVersions acceptStreaming;
+ public final SocketFactory socketFactory;
+ public final Function<InetAddressAndPort, InboundMessageHandlers> handlers;
+
+ private InboundConnectionSettings(IInternodeAuthenticator authenticator,
+ InetAddressAndPort bindAddress,
+ ServerEncryptionOptions encryption,
+ Integer socketReceiveBufferSizeInBytes,
+ Integer applicationReceiveQueueCapacityInBytes,
+ AcceptVersions acceptMessaging,
+ AcceptVersions acceptStreaming,
+ SocketFactory socketFactory,
+ Function<InetAddressAndPort, InboundMessageHandlers> handlers)
+ {
+ this.authenticator = authenticator;
+ this.bindAddress = bindAddress;
+ this.encryption = encryption;
+ this.socketReceiveBufferSizeInBytes = socketReceiveBufferSizeInBytes;
+ this.applicationReceiveQueueCapacityInBytes = applicationReceiveQueueCapacityInBytes;
+ this.acceptMessaging = acceptMessaging;
+ this.acceptStreaming = acceptStreaming;
+ this.socketFactory = socketFactory;
+ this.handlers = handlers;
+ }
+
+ public InboundConnectionSettings()
+ {
+ this(null, null, null, null, null, null, null, null, null);
+ }
+
+ public boolean authenticate(InetAddressAndPort endpoint)
+ {
+ return authenticator.authenticate(endpoint.address, endpoint.port);
+ }
+
+ public boolean authenticate(InetAddress address, int port)
+ {
+ return authenticator.authenticate(address, port);
+ }
+
+ public String toString()
+ {
+ return format("address: (%s), nic: %s, encryption: %s",
- bindAddress, FBUtilities.getNetworkInterface(bindAddress.address), SocketFactory.encryptionLogStatement(null, encryption));
++ bindAddress, FBUtilities.getNetworkInterface(bindAddress.address), SocketFactory.encryptionOptionsSummary(encryption));
+ }
+
+ public InboundConnectionSettings withAuthenticator(IInternodeAuthenticator authenticator)
+ {
+ return new InboundConnectionSettings(authenticator, bindAddress, encryption,
+ socketReceiveBufferSizeInBytes, applicationReceiveQueueCapacityInBytes,
+ acceptMessaging, acceptStreaming, socketFactory, handlers);
+ }
+
+ @SuppressWarnings("unused")
+ public InboundConnectionSettings withBindAddress(InetAddressAndPort bindAddress)
+ {
+ return new InboundConnectionSettings(authenticator, bindAddress, encryption,
+ socketReceiveBufferSizeInBytes, applicationReceiveQueueCapacityInBytes,
+ acceptMessaging, acceptStreaming, socketFactory, handlers);
+ }
+
+ public InboundConnectionSettings withEncryption(ServerEncryptionOptions encryption)
+ {
+ return new InboundConnectionSettings(authenticator, bindAddress, encryption,
+ socketReceiveBufferSizeInBytes, applicationReceiveQueueCapacityInBytes,
+ acceptMessaging, acceptStreaming, socketFactory, handlers);
+ }
+
+ public InboundConnectionSettings withSocketReceiveBufferSizeInBytes(int socketReceiveBufferSizeInBytes)
+ {
+ return new InboundConnectionSettings(authenticator, bindAddress, encryption,
+ socketReceiveBufferSizeInBytes, applicationReceiveQueueCapacityInBytes,
+ acceptMessaging, acceptStreaming, socketFactory, handlers);
+ }
+
+ @SuppressWarnings("unused")
+ public InboundConnectionSettings withApplicationReceiveQueueCapacityInBytes(int applicationReceiveQueueCapacityInBytes)
+ {
+ return new InboundConnectionSettings(authenticator, bindAddress, encryption,
+ socketReceiveBufferSizeInBytes, applicationReceiveQueueCapacityInBytes,
+ acceptMessaging, acceptStreaming, socketFactory, handlers);
+ }
+
+ public InboundConnectionSettings withAcceptMessaging(AcceptVersions acceptMessaging)
+ {
+ return new InboundConnectionSettings(authenticator, bindAddress, encryption,
+ socketReceiveBufferSizeInBytes, applicationReceiveQueueCapacityInBytes,
+ acceptMessaging, acceptStreaming, socketFactory, handlers);
+ }
+
+ public InboundConnectionSettings withAcceptStreaming(AcceptVersions acceptMessaging)
+ {
+ return new InboundConnectionSettings(authenticator, bindAddress, encryption,
+ socketReceiveBufferSizeInBytes, applicationReceiveQueueCapacityInBytes,
+ acceptMessaging, acceptStreaming, socketFactory, handlers);
+ }
+
+ public InboundConnectionSettings withSocketFactory(SocketFactory socketFactory)
+ {
+ return new InboundConnectionSettings(authenticator, bindAddress, encryption,
+ socketReceiveBufferSizeInBytes, applicationReceiveQueueCapacityInBytes,
+ acceptMessaging, acceptStreaming, socketFactory, handlers);
+ }
+
+ public InboundConnectionSettings withHandlers(Function<InetAddressAndPort, InboundMessageHandlers> handlers)
+ {
+ return new InboundConnectionSettings(authenticator, bindAddress, encryption,
+ socketReceiveBufferSizeInBytes, applicationReceiveQueueCapacityInBytes,
+ acceptMessaging, acceptStreaming, socketFactory, handlers);
+ }
+
- public InboundConnectionSettings withLegacyDefaults()
++ public InboundConnectionSettings withLegacySslStoragePortDefaults()
+ {
+ ServerEncryptionOptions encryption = this.encryption;
+ if (encryption == null)
+ encryption = DatabaseDescriptor.getInternodeMessagingEncyptionOptions();
- encryption = encryption.withOptional(false);
++ encryption = encryption.withOptional(false).withInternodeEncryption(ServerEncryptionOptions.InternodeEncryption.all);
+
+ return this.withBindAddress(bindAddress.withPort(DatabaseDescriptor.getSSLStoragePort()))
+ .withEncryption(encryption)
+ .withDefaults();
+ }
+
+ // note that connectTo is updated even if specified, in the case of pre40 messaging and using encryption (to update port)
+ public InboundConnectionSettings withDefaults()
+ {
+ // this is for the socket that can be plain, only ssl, or optional plain/ssl
+ if (bindAddress.port != DatabaseDescriptor.getStoragePort() && bindAddress.port != DatabaseDescriptor.getSSLStoragePort())
+ throw new ConfigurationException(format("Local endpoint port %d doesn't match YAML configured port %d or legacy SSL port %d",
+ bindAddress.port, DatabaseDescriptor.getStoragePort(), DatabaseDescriptor.getSSLStoragePort()));
+
+ IInternodeAuthenticator authenticator = this.authenticator;
+ ServerEncryptionOptions encryption = this.encryption;
+ Integer socketReceiveBufferSizeInBytes = this.socketReceiveBufferSizeInBytes;
+ Integer applicationReceiveQueueCapacityInBytes = this.applicationReceiveQueueCapacityInBytes;
+ AcceptVersions acceptMessaging = this.acceptMessaging;
+ AcceptVersions acceptStreaming = this.acceptStreaming;
+ SocketFactory socketFactory = this.socketFactory;
+ Function<InetAddressAndPort, InboundMessageHandlers> handlersFactory = this.handlers;
+
+ if (authenticator == null)
+ authenticator = DatabaseDescriptor.getInternodeAuthenticator();
+
+ if (encryption == null)
+ encryption = DatabaseDescriptor.getInternodeMessagingEncyptionOptions();
+
+ if (socketReceiveBufferSizeInBytes == null)
+ socketReceiveBufferSizeInBytes = DatabaseDescriptor.getInternodeSocketReceiveBufferSizeInBytes();
+
+ if (applicationReceiveQueueCapacityInBytes == null)
+ applicationReceiveQueueCapacityInBytes = DatabaseDescriptor.getInternodeApplicationReceiveQueueCapacityInBytes();
+
+ if (acceptMessaging == null)
+ acceptMessaging = accept_messaging;
+
+ if (acceptStreaming == null)
+ acceptStreaming = accept_streaming;
+
+ if (socketFactory == null)
+ socketFactory = instance().socketFactory;
+
+ if (handlersFactory == null)
+ handlersFactory = instance()::getInbound;
+
+ Preconditions.checkArgument(socketReceiveBufferSizeInBytes == 0 || socketReceiveBufferSizeInBytes >= 1 << 10, "illegal socket send buffer size: " + socketReceiveBufferSizeInBytes);
+ Preconditions.checkArgument(applicationReceiveQueueCapacityInBytes >= 1 << 10, "illegal application receive queue capacity: " + applicationReceiveQueueCapacityInBytes);
+
+ return new InboundConnectionSettings(authenticator, bindAddress, encryption, socketReceiveBufferSizeInBytes, applicationReceiveQueueCapacityInBytes, acceptMessaging, acceptStreaming, socketFactory, handlersFactory);
+ }
+}
diff --cc src/java/org/apache/cassandra/net/InboundSockets.java
index 93caf85,0000000..fc57224
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/net/InboundSockets.java
+++ b/src/java/org/apache/cassandra/net/InboundSockets.java
@@@ -1,269 -1,0 +1,269 @@@
+/*
+ * 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.net;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Consumer;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.group.ChannelGroup;
+import io.netty.channel.group.DefaultChannelGroup;
+import io.netty.util.concurrent.DefaultEventExecutor;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.GlobalEventExecutor;
+import io.netty.util.concurrent.PromiseNotifier;
+import io.netty.util.concurrent.SucceededFuture;
+import org.apache.cassandra.concurrent.NamedThreadFactory;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.utils.FBUtilities;
+
+class InboundSockets
+{
+ /**
+ * A simple struct to wrap up the components needed for each listening socket.
+ */
+ @VisibleForTesting
+ static class InboundSocket
+ {
+ public final InboundConnectionSettings settings;
+
+ /**
+ * The base {@link Channel} that is doing the socket listen/accept.
+ * Null only until open() is invoked and {@link #binding} has yet to complete.
+ */
+ private volatile Channel listen;
+ /**
+ * Once open() is invoked, this holds the future result of opening the socket,
+ * so that its completion can be waited on. Once complete, it sets itself to null.
+ */
+ private volatile ChannelFuture binding;
+
+ // purely to prevent close racing with open
+ private boolean closedWithoutOpening;
+
+ // used to prevent racing on close
+ private Future<Void> closeFuture;
+
+ /**
+ * A group of the open, inbound {@link Channel}s connected to this node. This is mostly interesting so that all of
+ * the inbound connections/channels can be closed when the listening socket itself is being closed.
+ */
+ private final ChannelGroup connections;
+ private final DefaultEventExecutor executor;
+
+ private InboundSocket(InboundConnectionSettings settings)
+ {
+ this.settings = settings;
+ this.executor = new DefaultEventExecutor(new NamedThreadFactory("Listen-" + settings.bindAddress));
+ this.connections = new DefaultChannelGroup(settings.bindAddress.toString(), executor);
+ }
+
+ private Future<Void> open()
+ {
+ return open(pipeline -> {});
+ }
+
+ private Future<Void> open(Consumer<ChannelPipeline> pipelineInjector)
+ {
+ synchronized (this)
+ {
+ if (listen != null)
+ return new SucceededFuture<>(GlobalEventExecutor.INSTANCE, null);
+ if (binding != null)
+ return binding;
+ if (closedWithoutOpening)
+ throw new IllegalStateException();
+ binding = InboundConnectionInitiator.bind(settings, connections, pipelineInjector);
+ }
+
+ return binding.addListener(ignore -> {
+ synchronized (this)
+ {
+ if (binding.isSuccess())
+ listen = binding.channel();
+ binding = null;
+ }
+ });
+ }
+
+ /**
+ * Close this socket and any connections created on it. Once closed, this socket may not be re-opened.
+ *
+ * This may not execute synchronously, so a Future is returned encapsulating its result.
+ * @param shutdownExecutors consumer invoked with the internal executor on completion
+ * Note that the consumer will only be invoked once per InboundSocket.
+ * Subsequent calls to close will not register a callback to different consumers.
+ */
+ private Future<Void> close(Consumer<? super ExecutorService> shutdownExecutors)
+ {
+ AsyncPromise<Void> done = AsyncPromise.uncancellable(GlobalEventExecutor.INSTANCE);
+
+ Runnable close = () -> {
+ List<Future<Void>> closing = new ArrayList<>();
+ if (listen != null)
+ closing.add(listen.close());
+ closing.add(connections.close());
+ new FutureCombiner(closing)
+ .addListener(future -> {
+ executor.shutdownGracefully();
+ shutdownExecutors.accept(executor);
+ })
+ .addListener(new PromiseNotifier<>(done));
+ };
+
+ synchronized (this)
+ {
+ if (listen == null && binding == null)
+ {
+ closedWithoutOpening = true;
+ return new SucceededFuture<>(GlobalEventExecutor.INSTANCE, null);
+ }
+
+ if (closeFuture != null)
+ {
+ return closeFuture;
+ }
+
+ closeFuture = done;
+
+ if (listen != null)
+ {
+ close.run();
+ }
+ else
+ {
+ binding.cancel(true);
+ binding.addListener(future -> close.run());
+ }
+
+ return done;
+ }
+ }
+
+ public boolean isOpen()
+ {
+ return listen != null && listen.isOpen();
+ }
+ }
+
+ private final List<InboundSocket> sockets;
+
+ InboundSockets(InboundConnectionSettings template)
+ {
+ this(withDefaultBindAddresses(template));
+ }
+
+ InboundSockets(List<InboundConnectionSettings> templates)
+ {
+ this.sockets = bindings(templates);
+ }
+
+ private static List<InboundConnectionSettings> withDefaultBindAddresses(InboundConnectionSettings template)
+ {
+ ImmutableList.Builder<InboundConnectionSettings> templates = ImmutableList.builder();
+ templates.add(template.withBindAddress(FBUtilities.getLocalAddressAndPort()));
+ if (shouldListenOnBroadcastAddress())
+ templates.add(template.withBindAddress(FBUtilities.getBroadcastAddressAndPort()));
+ return templates.build();
+ }
+
+ private static List<InboundSocket> bindings(List<InboundConnectionSettings> templates)
+ {
+ ImmutableList.Builder<InboundSocket> sockets = ImmutableList.builder();
+ for (InboundConnectionSettings template : templates)
+ addBindings(template, sockets);
+ return sockets.build();
+ }
+
+ private static void addBindings(InboundConnectionSettings template, ImmutableList.Builder<InboundSocket> out)
+ {
+ InboundConnectionSettings settings = template.withDefaults();
- InboundConnectionSettings legacySettings = template.withLegacyDefaults();
++ InboundConnectionSettings legacySettings = template.withLegacySslStoragePortDefaults();
+
+ if (settings.encryption.enable_legacy_ssl_storage_port)
+ {
+ out.add(new InboundSocket(legacySettings));
+
+ /*
+ * If the legacy ssl storage port and storage port match, only bind to the
+ * legacy ssl port. This makes it possible to configure a 4.0 node like a 3.0
+ * node with only the ssl_storage_port if required.
+ */
+ if (settings.bindAddress.equals(legacySettings.bindAddress))
+ return;
+ }
+
+ out.add(new InboundSocket(settings));
+ }
+
+ public Future<Void> open(Consumer<ChannelPipeline> pipelineInjector)
+ {
+ List<Future<Void>> opening = new ArrayList<>();
+ for (InboundSocket socket : sockets)
+ opening.add(socket.open(pipelineInjector));
+
+ return new FutureCombiner(opening);
+ }
+
+ public Future<Void> open()
+ {
+ List<Future<Void>> opening = new ArrayList<>();
+ for (InboundSocket socket : sockets)
+ opening.add(socket.open());
+ return new FutureCombiner(opening);
+ }
+
+ public boolean isListening()
+ {
+ for (InboundSocket socket : sockets)
+ if (socket.isOpen())
+ return true;
+ return false;
+ }
+
+ public Future<Void> close(Consumer<? super ExecutorService> shutdownExecutors)
+ {
+ List<Future<Void>> closing = new ArrayList<>();
+ for (InboundSocket address : sockets)
+ closing.add(address.close(shutdownExecutors));
+ return new FutureCombiner(closing);
+ }
+ public Future<Void> close()
+ {
+ return close(e -> {});
+ }
+
+ private static boolean shouldListenOnBroadcastAddress()
+ {
+ return DatabaseDescriptor.shouldListenOnBroadcastAddress()
+ && !FBUtilities.getLocalAddressAndPort().equals(FBUtilities.getBroadcastAddressAndPort());
+ }
+
+ @VisibleForTesting
+ public List<InboundSocket> sockets()
+ {
+ return sockets;
+ }
- }
++}
diff --cc src/java/org/apache/cassandra/net/OutboundConnection.java
index 66f14db,0000000..79c0459
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/net/OutboundConnection.java
+++ b/src/java/org/apache/cassandra/net/OutboundConnection.java
@@@ -1,1768 -1,0 +1,1767 @@@
+/*
+ * 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.net;
+
+import java.io.IOException;
+import java.net.ConnectException;
+import java.net.InetSocketAddress;
+import java.nio.channels.ClosedChannelException;
+import java.util.Objects;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
- import java.util.stream.Stream;
+
+import javax.annotation.Nullable;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.Uninterruptibles;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.EventLoop;
+import io.netty.channel.unix.Errors;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.Promise;
+import io.netty.util.concurrent.PromiseNotifier;
+import io.netty.util.concurrent.SucceededFuture;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.io.util.DataOutputBufferFixed;
+import org.apache.cassandra.net.OutboundConnectionInitiator.Result.MessagingSuccess;
+import org.apache.cassandra.tracing.Tracing;
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.JVMStabilityInspector;
+import org.apache.cassandra.utils.NoSpamLogger;
+
+import static java.lang.Math.max;
+import static java.lang.Math.min;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.cassandra.net.MessagingService.current_version;
+import static org.apache.cassandra.net.OutboundConnectionInitiator.*;
+import static org.apache.cassandra.net.OutboundConnections.LARGE_MESSAGE_THRESHOLD;
+import static org.apache.cassandra.net.ResourceLimits.*;
+import static org.apache.cassandra.net.ResourceLimits.Outcome.*;
+import static org.apache.cassandra.net.SocketFactory.*;
+import static org.apache.cassandra.utils.FBUtilities.prettyPrintMemory;
+import static org.apache.cassandra.utils.MonotonicClock.approxTime;
+import static org.apache.cassandra.utils.Throwables.isCausedBy;
+
+/**
+ * Represents a connection type to a peer, and handles the state transistions on the connection and the netty {@link Channel}.
+ * The underlying socket is not opened until explicitly requested (by sending a message).
+ *
+ * TODO: complete this description
+ *
+ * Aside from a few administrative methods, the main entry point to sending a message is {@link #enqueue(Message)}.
+ * Any thread may send a message (enqueueing it to {@link #queue}), but only one thread may consume messages from this
+ * queue. There is a single delivery thread - either the event loop, or a companion thread - that has logical ownership
+ * of the queue, but other threads may temporarily take ownership in order to perform book keeping, pruning, etc.,
+ * to ensure system stability.
+ *
+ * {@link Delivery#run()} is the main entry point for consuming messages from the queue, and executes either on the event
+ * loop or on a non-dedicated companion thread. This processing is activated via {@link Delivery#execute()}.
+ *
+ * Almost all internal state maintenance on this class occurs on the eventLoop, a single threaded executor which is
+ * assigned in the constructor. Further details are outlined below in the class. Some behaviours require coordination
+ * between the eventLoop and the companion thread (if any). Some minimal set of behaviours are permitted to occur on
+ * producers to ensure the connection remains healthy and does not overcommit resources.
+ *
+ * All methods are safe to invoke from any thread unless otherwise stated.
+ */
+@SuppressWarnings({ "WeakerAccess", "FieldMayBeFinal", "NonAtomicOperationOnVolatileField", "SameParameterValue" })
+public class OutboundConnection
+{
+ static final Logger logger = LoggerFactory.getLogger(OutboundConnection.class);
+ private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 30L, TimeUnit.SECONDS);
+
+ private static final AtomicLongFieldUpdater<OutboundConnection> submittedUpdater = AtomicLongFieldUpdater.newUpdater(OutboundConnection.class, "submittedCount");
+ private static final AtomicLongFieldUpdater<OutboundConnection> pendingCountAndBytesUpdater = AtomicLongFieldUpdater.newUpdater(OutboundConnection.class, "pendingCountAndBytes");
+ private static final AtomicLongFieldUpdater<OutboundConnection> overloadedCountUpdater = AtomicLongFieldUpdater.newUpdater(OutboundConnection.class, "overloadedCount");
+ private static final AtomicLongFieldUpdater<OutboundConnection> overloadedBytesUpdater = AtomicLongFieldUpdater.newUpdater(OutboundConnection.class, "overloadedBytes");
+ private static final AtomicReferenceFieldUpdater<OutboundConnection, Future> closingUpdater = AtomicReferenceFieldUpdater.newUpdater(OutboundConnection.class, Future.class, "closing");
+ private static final AtomicReferenceFieldUpdater<OutboundConnection, Future> scheduledCloseUpdater = AtomicReferenceFieldUpdater.newUpdater(OutboundConnection.class, Future.class, "scheduledClose");
+
+ private final EventLoop eventLoop;
+ private final Delivery delivery;
+
+ private final OutboundMessageCallbacks callbacks;
+ private final OutboundDebugCallbacks debug;
+ @VisibleForTesting
+ final OutboundMessageQueue queue;
+ /** the number of bytes we permit to queue to the network without acquiring any shared resource permits */
+ private final long pendingCapacityInBytes;
+ /** the number of messages and bytes queued for flush to the network,
+ * including those that are being flushed but have not been completed,
+ * packed into a long (top 20 bits for count, bottom 42 for bytes)*/
+ private volatile long pendingCountAndBytes = 0;
+ /** global shared limits that we use only if our local limits are exhausted;
+ * we allocate from here whenever queueSize > queueCapacity */
+ private final EndpointAndGlobal reserveCapacityInBytes;
+
+ /** Used in logging statements to lazily build a human-readable number of pending bytes. */
+ private final Object readablePendingBytes =
+ new Object() { @Override public String toString() { return prettyPrintMemory(pendingBytes()); } };
+
+ /** Used in logging statements to lazily build a human-readable number of reserve endpoint bytes in use. */
+ private final Object readableReserveEndpointUsing =
+ new Object() { @Override public String toString() { return prettyPrintMemory(reserveCapacityInBytes.endpoint.using()); } };
+
+ /** Used in logging statements to lazily build a human-readable number of reserve global bytes in use. */
+ private final Object readableReserveGlobalUsing =
+ new Object() { @Override public String toString() { return prettyPrintMemory(reserveCapacityInBytes.global.using()); } };
+
+ private volatile long submittedCount = 0; // updated with cas
+ private volatile long overloadedCount = 0; // updated with cas
+ private volatile long overloadedBytes = 0; // updated with cas
+ private long expiredCount = 0; // updated with queue lock held
+ private long expiredBytes = 0; // updated with queue lock held
+ private long errorCount = 0; // updated only by delivery thread
+ private long errorBytes = 0; // updated by delivery thread only
+ private long sentCount; // updated by delivery thread only
+ private long sentBytes; // updated by delivery thread only
+ private long successfulConnections; // updated by event loop only
+ private long connectionAttempts; // updated by event loop only
+
+ private static final int pendingByteBits = 42;
+ private static boolean isMaxPendingCount(long pendingCountAndBytes)
+ {
+ return (pendingCountAndBytes & (-1L << pendingByteBits)) == (-1L << pendingByteBits);
+ }
+
+ private static int pendingCount(long pendingCountAndBytes)
+ {
+ return (int) (pendingCountAndBytes >>> pendingByteBits);
+ }
+
+ private static long pendingBytes(long pendingCountAndBytes)
+ {
+ return pendingCountAndBytes & (-1L >>> (64 - pendingByteBits));
+ }
+
+ private static long pendingCountAndBytes(long pendingCount, long pendingBytes)
+ {
+ return (pendingCount << pendingByteBits) | pendingBytes;
+ }
+
+ private final ConnectionType type;
+
+ /**
+ * Contains the base settings for this connection, _including_ any defaults filled in.
+ *
+ */
+ private OutboundConnectionSettings template;
+
+ private static class State
+ {
+ static final State CLOSED = new State(Kind.CLOSED);
+
+ enum Kind { ESTABLISHED, CONNECTING, DORMANT, CLOSED }
+
+ final Kind kind;
+
+ State(Kind kind)
+ {
+ this.kind = kind;
+ }
+
+ boolean isEstablished() { return kind == Kind.ESTABLISHED; }
+ boolean isConnecting() { return kind == Kind.CONNECTING; }
+ boolean isDisconnected() { return kind == Kind.CONNECTING || kind == Kind.DORMANT; }
+ boolean isClosed() { return kind == Kind.CLOSED; }
+
+ Established established() { return (Established) this; }
+ Connecting connecting() { return (Connecting) this; }
+ Disconnected disconnected() { return (Disconnected) this; }
+ }
+
+ /**
+ * We have successfully negotiated a channel, and believe it to still be valid.
+ *
+ * Before using this, we should check isConnected() to check the Channel hasn't
+ * become invalid.
+ */
+ private static class Established extends State
+ {
+ final int messagingVersion;
+ final Channel channel;
+ final FrameEncoder.PayloadAllocator payloadAllocator;
+ final OutboundConnectionSettings settings;
+
+ Established(int messagingVersion, Channel channel, FrameEncoder.PayloadAllocator payloadAllocator, OutboundConnectionSettings settings)
+ {
+ super(Kind.ESTABLISHED);
+ this.messagingVersion = messagingVersion;
+ this.channel = channel;
+ this.payloadAllocator = payloadAllocator;
+ this.settings = settings;
+ }
+
+ boolean isConnected() { return channel.isOpen(); }
+ }
+
+ private static class Disconnected extends State
+ {
+ /** Periodic message expiry scheduled while we are disconnected; this will be cancelled and cleared each time we connect */
+ final Future<?> maintenance;
+ Disconnected(Kind kind, Future<?> maintenance)
+ {
+ super(kind);
+ this.maintenance = maintenance;
+ }
+
+ public static Disconnected dormant(Future<?> maintenance)
+ {
+ return new Disconnected(Kind.DORMANT, maintenance);
+ }
+ }
+
+ private static class Connecting extends Disconnected
+ {
+ /**
+ * Currently (or scheduled to) (re)connect; this may be cancelled (if closing) or waited on (for delivery)
+ *
+ * - The work managed by this future is partially performed asynchronously, not necessarily on the eventLoop.
+ * - It is only completed on the eventLoop
+ * - It may not be executing, but might be scheduled to be submitted if {@link #scheduled} is not null
+ */
+ final Future<Result<MessagingSuccess>> attempt;
+
+ /**
+ * If we are retrying to connect with some delay, this represents the scheduled inititation of another attempt
+ */
+ @Nullable
+ final Future<?> scheduled;
+
+ /**
+ * true iff we are retrying to connect after some failure (immediately or following a delay)
+ */
+ final boolean isFailingToConnect;
+
+ Connecting(Disconnected previous, Future<Result<MessagingSuccess>> attempt)
+ {
+ this(previous, attempt, null);
+ }
+
+ Connecting(Disconnected previous, Future<Result<MessagingSuccess>> attempt, Future<?> scheduled)
+ {
+ super(Kind.CONNECTING, previous.maintenance);
+ this.attempt = attempt;
+ this.scheduled = scheduled;
+ this.isFailingToConnect = scheduled != null || (previous.isConnecting() && previous.connecting().isFailingToConnect);
+ }
+
+ /**
+ * Cancel the connection attempt
+ *
+ * No cleanup is needed here, as {@link #attempt} is only completed on the eventLoop,
+ * so we have either already invoked the callbacks and are no longer in {@link #state},
+ * or the {@link OutboundConnectionInitiator} will handle our successful cancellation
+ * when it comes to complete, by closing the channel (if we could not cancel it before then)
+ */
+ void cancel()
+ {
+ if (scheduled != null)
+ scheduled.cancel(true);
+
+ // we guarantee that attempt is only ever completed by the eventLoop
+ boolean cancelled = attempt.cancel(true);
+ assert cancelled;
+ }
+ }
+
+ private volatile State state;
+
+ /** The connection is being permanently closed */
+ private volatile Future<Void> closing;
+ /** The connection is being permanently closed in the near future */
+ private volatile Future<Void> scheduledClose;
+
+ OutboundConnection(ConnectionType type, OutboundConnectionSettings settings, EndpointAndGlobal reserveCapacityInBytes)
+ {
+ this.template = settings.withDefaults(ConnectionCategory.MESSAGING);
+ this.type = type;
+ this.eventLoop = template.socketFactory.defaultGroup().next();
+ this.pendingCapacityInBytes = template.applicationSendQueueCapacityInBytes;
+ this.reserveCapacityInBytes = reserveCapacityInBytes;
+ this.callbacks = template.callbacks;
+ this.debug = template.debug;
+ this.queue = new OutboundMessageQueue(approxTime, this::onExpired);
+ this.delivery = type == ConnectionType.LARGE_MESSAGES
+ ? new LargeMessageDelivery(template.socketFactory.synchronousWorkExecutor)
+ : new EventLoopDelivery();
+ setDisconnected();
+ }
+
+ /**
+ * This is the main entry point for enqueuing a message to be sent to the remote peer.
+ */
+ public void enqueue(Message message) throws ClosedChannelException
+ {
+ if (isClosing())
+ throw new ClosedChannelException();
+
+ final int canonicalSize = canonicalSize(message);
+ if (canonicalSize > DatabaseDescriptor.getInternodeMaxMessageSizeInBytes())
+ throw new Message.OversizedMessageException(canonicalSize);
+
+ submittedUpdater.incrementAndGet(this);
+ switch (acquireCapacity(canonicalSize))
+ {
+ case INSUFFICIENT_ENDPOINT:
+ // if we're overloaded to one endpoint, we may be accumulating expirable messages, so
+ // attempt an expiry to see if this makes room for our newer message.
+ // this is an optimisation only; messages will be expired on ~100ms cycle, and by Delivery when it runs
+ if (queue.maybePruneExpired() && SUCCESS == acquireCapacity(canonicalSize))
+ break;
+ case INSUFFICIENT_GLOBAL:
+ onOverloaded(message);
+ return;
+ }
+
+ queue.add(message);
+ delivery.execute();
+
+ // we might race with the channel closing; if this happens, to ensure this message eventually arrives
+ // we need to remove ourselves from the queue and throw a ClosedChannelException, so that another channel
+ // can be opened in our place to try and send on.
+ if (isClosing() && queue.remove(message))
+ {
+ releaseCapacity(1, canonicalSize);
+ throw new ClosedChannelException();
+ }
+ }
+
+ /**
+ * Try to acquire the necessary resource permits for a number of pending bytes for this connection.
+ *
+ * Since the owner limit is shared amongst multiple connections, our semantics cannot be super trivial.
+ * Were they per-connection, we could simply perform an atomic increment of the queue size, then
+ * allocate any excess we need in the reserve, and on release free everything we see from both.
+ * Since we are coordinating two independent atomic variables we have to track every byte we allocate in reserve
+ * and ensure it is matched by a corresponding released byte. We also need to be sure we do not permit another
+ * releasing thread to release reserve bytes we have not yet - and may never - actually reserve.
+ *
+ * As such, we have to first check if we would need reserve bytes, then allocate them *before* we increment our
+ * queue size. We only increment the queue size if the reserve bytes are definitely not needed, or we could first
+ * obtain them. If in the process of obtaining any reserve bytes the queue size changes, we have some bytes that are
+ * reserved for us, but may be a different number to that we need. So we must continue to track these.
+ *
+ * In the happy path, this is still efficient as we simply CAS
+ */
+ private Outcome acquireCapacity(long bytes)
+ {
+ return acquireCapacity(1, bytes);
+ }
+
+ private Outcome acquireCapacity(long count, long bytes)
+ {
+ long increment = pendingCountAndBytes(count, bytes);
+ long unusedClaimedReserve = 0;
+ Outcome outcome = null;
+ loop: while (true)
+ {
+ long current = pendingCountAndBytes;
+ if (isMaxPendingCount(current))
+ {
+ outcome = INSUFFICIENT_ENDPOINT;
+ break;
+ }
+
+ long next = current + increment;
+ if (pendingBytes(next) <= pendingCapacityInBytes)
+ {
+ if (pendingCountAndBytesUpdater.compareAndSet(this, current, next))
+ {
+ outcome = SUCCESS;
+ break;
+ }
+ continue;
+ }
+
+ State state = this.state;
+ if (state.isConnecting() && state.connecting().isFailingToConnect)
+ {
+ outcome = INSUFFICIENT_ENDPOINT;
+ break;
+ }
+
+ long requiredReserve = min(bytes, pendingBytes(next) - pendingCapacityInBytes);
+ if (unusedClaimedReserve < requiredReserve)
+ {
+ long extraGlobalReserve = requiredReserve - unusedClaimedReserve;
+ switch (outcome = reserveCapacityInBytes.tryAllocate(extraGlobalReserve))
+ {
+ case INSUFFICIENT_ENDPOINT:
+ case INSUFFICIENT_GLOBAL:
+ break loop;
+ case SUCCESS:
+ unusedClaimedReserve += extraGlobalReserve;
+ }
+ }
+
+ if (pendingCountAndBytesUpdater.compareAndSet(this, current, next))
+ {
+ unusedClaimedReserve -= requiredReserve;
+ break;
+ }
+ }
+
+ if (unusedClaimedReserve > 0)
+ reserveCapacityInBytes.release(unusedClaimedReserve);
+
+ return outcome;
+ }
+
+ /**
+ * Mark a number of pending bytes as flushed to the network, releasing their capacity for new outbound messages.
+ */
+ private void releaseCapacity(long count, long bytes)
+ {
+ long decrement = pendingCountAndBytes(count, bytes);
+ long prev = pendingCountAndBytesUpdater.getAndAdd(this, -decrement);
+ if (pendingBytes(prev) > pendingCapacityInBytes)
+ {
+ long excess = min(pendingBytes(prev) - pendingCapacityInBytes, bytes);
+ reserveCapacityInBytes.release(excess);
+ }
+ }
+
+ private void onOverloaded(Message<?> message)
+ {
+ overloadedCountUpdater.incrementAndGet(this);
+
+ int canonicalSize = canonicalSize(message);
+ overloadedBytesUpdater.addAndGet(this, canonicalSize);
+
+ noSpamLogger.warn("{} overloaded; dropping {} message (queue: {} local, {} endpoint, {} global)",
+ this, FBUtilities.prettyPrintMemory(canonicalSize),
+ readablePendingBytes, readableReserveEndpointUsing, readableReserveGlobalUsing);
+
+ callbacks.onOverloaded(message, template.to);
+ }
+
+ /**
+ * Take any necessary cleanup action after a message has been selected to be discarded from the queue.
+ *
+ * Only to be invoked while holding OutboundMessageQueue.WithLock
+ */
+ private boolean onExpired(Message<?> message)
+ {
+ releaseCapacity(1, canonicalSize(message));
+ expiredCount += 1;
+ expiredBytes += canonicalSize(message);
+ noSpamLogger.warn("{} dropping message of type {} whose timeout expired before reaching the network", id(), message.verb());
+ callbacks.onExpired(message, template.to);
+ return true;
+ }
+
+ /**
+ * Take any necessary cleanup action after a message has been selected to be discarded from the queue.
+ *
+ * Only to be invoked by the delivery thread
+ */
+ private void onFailedSerialize(Message<?> message, int messagingVersion, int bytesWrittenToNetwork, Throwable t)
+ {
+ JVMStabilityInspector.inspectThrowable(t, false);
+ releaseCapacity(1, canonicalSize(message));
+ errorCount += 1;
+ errorBytes += message.serializedSize(messagingVersion);
+ logger.warn("{} dropping message of type {} due to error", id(), message.verb(), t);
+ callbacks.onFailedSerialize(message, template.to, messagingVersion, bytesWrittenToNetwork, t);
+ }
+
+ /**
+ * Take any necessary cleanup action after a message has been selected to be discarded from the queue on close.
+ * Note that this is only for messages that were queued prior to closing without graceful flush, OR
+ * for those that are unceremoniously dropped when we decide close has been trying to complete for too long.
+ */
+ private void onClosed(Message<?> message)
+ {
+ releaseCapacity(1, canonicalSize(message));
+ callbacks.onDiscardOnClose(message, template.to);
+ }
+
+ /**
+ * Delivery bundles the following:
+ *
+ * - the work that is necessary to actually deliver messages safely, and handle any exceptional states
+ * - the ability to schedule delivery for some time in the future
+ * - the ability to schedule some non-delivery work to happen some time in the future, that is guaranteed
+ * NOT to coincide with delivery for its duration, including any data that is being flushed (e.g. for closing channels)
+ * - this feature is *not* efficient, and should only be used for infrequent operations
+ */
+ private abstract class Delivery extends AtomicInteger implements Runnable
+ {
+ final ExecutorService executor;
+
+ // the AtomicInteger we extend always contains some combination of these bit flags, representing our current run state
+
+ /** Not running, and will not be scheduled again until transitioned to a new state */
+ private static final int STOPPED = 0;
+ /** Currently executing (may only be scheduled to execute, or may be about to terminate);
+ * will stop at end of this run, without rescheduling */
+ private static final int EXECUTING = 1;
+ /** Another execution has been requested; a new execution will begin some time after this state is taken */
+ private static final int EXECUTE_AGAIN = 2;
+ /** We are currently executing and will submit another execution before we terminate */
+ private static final int EXECUTING_AGAIN = EXECUTING | EXECUTE_AGAIN;
+ /** Will begin a new execution some time after this state is taken, but only once some condition is met.
+ * This state will initially be taken in tandem with EXECUTING, but if delivery completes without clearing
+ * the state, the condition will be held on its own until {@link #executeAgain} is invoked */
+ private static final int WAITING_TO_EXECUTE = 4;
+
+ /**
+ * Force all task execution to stop, once any currently in progress work is completed
+ */
+ private volatile boolean terminated;
+
+ /**
+ * Is there asynchronous delivery work in progress.
+ *
+ * This temporarily prevents any {@link #stopAndRun} work from being performed.
+ * Once both inProgress and stopAndRun are set we perform no more delivery work until one is unset,
+ * to ensure we eventually run stopAndRun.
+ *
+ * This should be updated and read only on the Delivery thread.
+ */
+ private boolean inProgress = false;
+
+ /**
+ * Request a task's execution while there is no delivery work in progress.
+ *
+ * This is to permit cleanly tearing down a connection without interrupting any messages that might be in flight.
+ * If stopAndRun is set, we should not enter doRun() until a corresponding setInProgress(false) occurs.
+ */
+ final AtomicReference<Runnable> stopAndRun = new AtomicReference<>();
+
+ Delivery(ExecutorService executor)
+ {
+ this.executor = executor;
+ }
+
+ /**
+ * Ensure that any messages or stopAndRun that were queued prior to this invocation will be seen by at least
+ * one future invocation of the delivery task, unless delivery has already been terminated.
+ */
+ public void execute()
+ {
+ if (get() < EXECUTE_AGAIN && STOPPED == getAndUpdate(i -> i == STOPPED ? EXECUTING: i | EXECUTE_AGAIN))
+ executor.execute(this);
+ }
+
+ private boolean isExecuting(int state)
+ {
+ return 0 != (state & EXECUTING);
+ }
+
+ /**
+ * This method is typically invoked after WAITING_TO_EXECUTE is set.
+ *
+ * However WAITING_TO_EXECUTE does not need to be set; all this method needs to ensure is that
+ * delivery unconditionally performs one new execution promptly.
+ */
+ void executeAgain()
+ {
+ // if we are already executing, set EXECUTING_AGAIN and leave scheduling to the currently running one.
+ // otherwise, set ourselves unconditionally to EXECUTING and schedule ourselves immediately
+ if (!isExecuting(getAndUpdate(i -> !isExecuting(i) ? EXECUTING : EXECUTING_AGAIN)))
+ executor.execute(this);
+ }
+
+ /**
+ * Invoke this when we cannot make further progress now, but we guarantee that we will execute later when we can.
+ * This simply communicates to {@link #run} that we should not schedule ourselves again, just unset the EXECUTING bit.
+ */
+ void promiseToExecuteLater()
+ {
+ set(EXECUTING | WAITING_TO_EXECUTE);
+ }
+
+ /**
+ * Called when exiting {@link #run} to schedule another run if necessary.
+ *
+ * If we are currently executing, we only reschedule if the present state is EXECUTING_AGAIN.
+ * If this is the case, we clear the EXECUTE_AGAIN bit (setting ourselves to EXECUTING), and reschedule.
+ * Otherwise, we clear the EXECUTING bit and terminate, which will set us to either STOPPED or WAITING_TO_EXECUTE
+ * (or possibly WAITING_TO_EXECUTE | EXECUTE_AGAIN, which is logically the same as WAITING_TO_EXECUTE)
+ */
+ private void maybeExecuteAgain()
+ {
+ if (EXECUTING_AGAIN == getAndUpdate(i -> i == EXECUTING_AGAIN ? EXECUTING : (i & ~EXECUTING)))
+ executor.execute(this);
+ }
+
+ /**
+ * No more tasks or delivery will be executed, once any in progress complete.
+ */
+ public void terminate()
+ {
+ terminated = true;
+ }
+
+ /**
+ * Only to be invoked by the Delivery task.
+ *
+ * If true, indicates that we have begun asynchronous delivery work, so that
+ * we cannot safely stopAndRun until it completes.
+ *
+ * Once it completes, we ensure any stopAndRun task has a chance to execute
+ * by ensuring delivery is scheduled.
+ *
+ * If stopAndRun is also set, we should not enter doRun() until a corresponding
+ * setInProgress(false) occurs.
+ */
+ void setInProgress(boolean inProgress)
+ {
+ boolean wasInProgress = this.inProgress;
+ this.inProgress = inProgress;
+ if (!inProgress && wasInProgress)
+ executeAgain();
+ }
+
+ /**
+ * Perform some delivery work.
+ *
+ * Must never be invoked directly, only via {@link #execute()}
+ */
+ public void run()
+ {
+ /* do/while handling setup for {@link #doRun()}, and repeat invocations thereof */
+ while (true)
+ {
+ if (terminated)
+ return;
+
+ if (null != stopAndRun.get())
+ {
+ // if we have an external request to perform, attempt it - if no async delivery is in progress
+
+ if (inProgress)
+ {
+ // if we are in progress, we cannot do anything;
+ // so, exit and rely on setInProgress(false) executing us
+ // (which must happen later, since it must happen on this thread)
+ promiseToExecuteLater();
+ break;
+ }
+
+ stopAndRun.getAndSet(null).run();
+ }
+
+ State state = OutboundConnection.this.state;
+ if (!state.isEstablished() || !state.established().isConnected())
+ {
+ // if we have messages yet to deliver, or a task to run, we need to reconnect and try again
+ // we try to reconnect before running another stopAndRun so that we do not infinite loop in close
+ if (hasPending() || null != stopAndRun.get())
+ {
+ promiseToExecuteLater();
+ requestConnect().addListener(f -> executeAgain());
+ }
+ break;
+ }
+
+ if (!doRun(state.established()))
+ break;
+ }
+
+ maybeExecuteAgain();
+ }
+
+ /**
+ * @return true if we should run again immediately;
+ * always false for eventLoop executor, as want to service other channels
+ */
+ abstract boolean doRun(Established established);
+
+ /**
+ * Schedule a task to run later on the delivery thread while delivery is not in progress,
+ * i.e. there are no bytes in flight to the network buffer.
+ *
+ * Does not guarantee to run promptly if there is no current connection to the remote host.
+ * May wait until a new connection is established, or a connection timeout elapses, before executing.
+ *
+ * Update the shared atomic property containing work we want to interrupt message processing to perform,
+ * the invoke schedule() to be certain it gets run.
+ */
+ void stopAndRun(Runnable run)
+ {
+ stopAndRun.accumulateAndGet(run, OutboundConnection::andThen);
+ execute();
+ }
+
+ /**
+ * Schedule a task to run on the eventLoop, guaranteeing that delivery will not occur while the task is performed.
+ */
+ abstract void stopAndRunOnEventLoop(Runnable run);
+
+ }
+
+ /**
+ * Delivery that runs entirely on the eventLoop
+ *
+ * Since this has single threaded access to most of its environment, it can be simple and efficient, however
+ * it must also have bounded run time, and limit its resource consumption to ensure other channels serviced by the
+ * eventLoop can also make progress.
+ *
+ * This operates on modest buffers, no larger than the {@link OutboundConnections#LARGE_MESSAGE_THRESHOLD} and
+ * filling at most one at a time before writing (potentially asynchronously) to the socket.
+ *
+ * We track the number of bytes we have in flight, ensuring no more than a user-defined maximum at any one time.
+ */
+ class EventLoopDelivery extends Delivery
+ {
+ private int flushingBytes;
+ private boolean isWritable = true;
+
+ EventLoopDelivery()
+ {
+ super(eventLoop);
+ }
+
+ /**
+ * {@link Delivery#doRun}
+ *
+ * Since we are on the eventLoop, in order to ensure other channels are serviced
+ * we never return true to request another run immediately.
+ *
+ * If there is more work to be done, we submit ourselves for execution once the eventLoop has time.
+ */
+ @SuppressWarnings("resource")
+ boolean doRun(Established established)
+ {
+ if (!isWritable)
+ return false;
+
+ // pendingBytes is updated before queue.size() (which triggers notEmpty, and begins delivery),
+ // so it is safe to use it here to exit delivery
+ // this number is inaccurate for old versions, but we don't mind terribly - we'll send at least one message,
+ // and get round to it eventually (though we could add a fudge factor for some room for older versions)
+ int maxSendBytes = (int) min(pendingBytes() - flushingBytes, LARGE_MESSAGE_THRESHOLD);
+ if (maxSendBytes == 0)
+ return false;
+
+ OutboundConnectionSettings settings = established.settings;
+ int messagingVersion = established.messagingVersion;
+
+ FrameEncoder.Payload sending = null;
+ int canonicalSize = 0; // number of bytes we must use for our resource accounting
+ int sendingBytes = 0;
+ int sendingCount = 0;
+ try (OutboundMessageQueue.WithLock withLock = queue.lockOrCallback(approxTime.now(), this::execute))
+ {
+ if (withLock == null)
+ return false; // we failed to acquire the queue lock, so return; we will be scheduled again when the lock is available
+
+ sending = established.payloadAllocator.allocate(true, maxSendBytes);
+ DataOutputBufferFixed out = new DataOutputBufferFixed(sending.buffer);
+
+ Message<?> next;
+ while ( null != (next = withLock.peek()) )
+ {
+ try
+ {
+ int messageSize = next.serializedSize(messagingVersion);
+
+ // actual message size for this version is larger than permitted maximum
+ if (messageSize > DatabaseDescriptor.getInternodeMaxMessageSizeInBytes())
+ throw new Message.OversizedMessageException(messageSize);
+
+ if (messageSize > sending.remaining())
+ {
+ // if we don't have enough room to serialize the next message, we have either
+ // 1) run out of room after writing some messages successfully; this might mean that we are
+ // overflowing our highWaterMark, or that we have just filled our buffer
+ // 2) we have a message that is too large for this connection; this can happen if a message's
+ // size was calculated for the wrong messaging version when enqueued.
+ // In this case we want to write it anyway, so simply allocate a large enough buffer.
+
+ if (sendingBytes > 0)
+ break;
+
+ sending.release();
+ sending = null; // set to null to prevent double-release if we fail to allocate our new buffer
+ sending = established.payloadAllocator.allocate(true, messageSize);
+ //noinspection IOResourceOpenedButNotSafelyClosed
+ out = new DataOutputBufferFixed(sending.buffer);
+ }
+
+ Tracing.instance.traceOutgoingMessage(next, messageSize, settings.connectTo);
+ Message.serializer.serialize(next, out, messagingVersion);
+
+ if (sending.length() != sendingBytes + messageSize)
+ throw new InvalidSerializedSizeException(next.verb(), messageSize, sending.length() - sendingBytes);
+
+ canonicalSize += canonicalSize(next);
+ sendingCount += 1;
+ sendingBytes += messageSize;
+ }
+ catch (Throwable t)
+ {
+ onFailedSerialize(next, messagingVersion, 0, t);
+
+ assert sending != null;
+ // reset the buffer to ignore the message we failed to serialize
+ sending.trim(sendingBytes);
+ }
+ withLock.removeHead(next);
+ }
+ if (0 == sendingBytes)
+ return false;
+
+ sending.finish();
+ debug.onSendSmallFrame(sendingCount, sendingBytes);
+ ChannelFuture flushResult = AsyncChannelPromise.writeAndFlush(established.channel, sending);
+ sending = null;
+
+ if (flushResult.isSuccess())
+ {
+ sentCount += sendingCount;
+ sentBytes += sendingBytes;
+ debug.onSentSmallFrame(sendingCount, sendingBytes);
+ }
+ else
+ {
+ flushingBytes += canonicalSize;
+ setInProgress(true);
+
+ boolean hasOverflowed = flushingBytes >= settings.flushHighWaterMark;
+ if (hasOverflowed)
+ {
+ isWritable = false;
+ promiseToExecuteLater();
+ }
+
+ int releaseBytesFinal = canonicalSize;
+ int sendingBytesFinal = sendingBytes;
+ int sendingCountFinal = sendingCount;
+ flushResult.addListener(future -> {
+
+ releaseCapacity(sendingCountFinal, releaseBytesFinal);
+ flushingBytes -= releaseBytesFinal;
+ if (flushingBytes == 0)
+ setInProgress(false);
+
+ if (!isWritable && flushingBytes <= settings.flushLowWaterMark)
+ {
+ isWritable = true;
+ executeAgain();
+ }
+
+ if (future.isSuccess())
+ {
+ sentCount += sendingCountFinal;
+ sentBytes += sendingBytesFinal;
+ debug.onSentSmallFrame(sendingCountFinal, sendingBytesFinal);
+ }
+ else
+ {
+ errorCount += sendingCountFinal;
+ errorBytes += sendingBytesFinal;
+ invalidateChannel(established, future.cause());
+ debug.onFailedSmallFrame(sendingCountFinal, sendingBytesFinal);
+ }
+ });
+ canonicalSize = 0;
+ }
+ }
+ catch (Throwable t)
+ {
+ errorCount += sendingCount;
+ errorBytes += sendingBytes;
+ invalidateChannel(established, t);
+ }
+ finally
+ {
+ if (canonicalSize > 0)
+ releaseCapacity(sendingCount, canonicalSize);
+
+ if (sending != null)
+ sending.release();
+
+ if (pendingBytes() > flushingBytes && isWritable)
+ execute();
+ }
+
+ return false;
+ }
+
+ void stopAndRunOnEventLoop(Runnable run)
+ {
+ stopAndRun(run);
+ }
+ }
+
+ /**
+ * Delivery that coordinates between the eventLoop and another (non-dedicated) thread
+ *
+ * This is to service messages that are too large to fully serialize on the eventLoop, as they could block
+ * prompt service of other requests. Since our serializers assume blocking IO, the easiest approach is to
+ * ensure a companion thread performs blocking IO that, under the hood, is serviced by async IO on the eventLoop.
+ *
+ * Most of the work here is handed off to {@link AsyncChannelOutputPlus}, with our main job being coordinating
+ * when and what we should run.
+ *
+ * To avoid allocating a huge number of threads across a cluster, we utilise the shared methods of {@link Delivery}
+ * to ensure that only one run() is actually scheduled to run at a time - this permits us to use any {@link ExecutorService}
+ * as a backing, with the number of threads defined only by the maximum concurrency needed to deliver all large messages.
+ * We use a shared caching {@link java.util.concurrent.ThreadPoolExecutor}, and rename the Threads that service
+ * our connection on entry and exit.
+ */
+ class LargeMessageDelivery extends Delivery
+ {
+ static final int DEFAULT_BUFFER_SIZE = 32 * 1024;
+
+ LargeMessageDelivery(ExecutorService executor)
+ {
+ super(executor);
+ }
+
+ /**
+ * A simple wrapper of {@link Delivery#run} to set the current Thread name for the duration of its execution.
+ */
+ public void run()
+ {
+ String threadName, priorThreadName = null;
+ try
+ {
+ priorThreadName = Thread.currentThread().getName();
+ threadName = "Messaging-OUT-" + template.from() + "->" + template.to + '-' + type;
+ Thread.currentThread().setName(threadName);
+
+ super.run();
+ }
+ finally
+ {
+ if (priorThreadName != null)
+ Thread.currentThread().setName(priorThreadName);
+ }
+ }
+
+ @SuppressWarnings({ "resource", "RedundantSuppression" }) // make eclipse warnings go away
+ boolean doRun(Established established)
+ {
+ Message<?> send = queue.tryPoll(approxTime.now(), this::execute);
+ if (send == null)
+ return false;
+
+ AsyncMessageOutputPlus out = null;
+ try
+ {
+ int messageSize = send.serializedSize(established.messagingVersion);
+ out = new AsyncMessageOutputPlus(established.channel, DEFAULT_BUFFER_SIZE, messageSize, established.payloadAllocator);
+ // actual message size for this version is larger than permitted maximum
+ if (messageSize > DatabaseDescriptor.getInternodeMaxMessageSizeInBytes())
+ throw new Message.OversizedMessageException(messageSize);
+
+ Tracing.instance.traceOutgoingMessage(send, messageSize, established.settings.connectTo);
+ Message.serializer.serialize(send, out, established.messagingVersion);
+
+ if (out.position() != messageSize)
+ throw new InvalidSerializedSizeException(send.verb(), messageSize, out.position());
+
+ out.close();
+ sentCount += 1;
+ sentBytes += messageSize;
+ releaseCapacity(1, canonicalSize(send));
+ return hasPending();
+ }
+ catch (Throwable t)
+ {
+ boolean tryAgain = true;
+
+ if (out != null)
+ {
+ out.discard();
+ if (out.flushed() > 0 ||
+ isCausedBy(t, cause -> isConnectionReset(cause)
+ || cause instanceof Errors.NativeIoException
+ || cause instanceof AsyncChannelOutputPlus.FlushException))
+ {
+ // close the channel, and wait for eventLoop to execute
+ disconnectNow(established).awaitUninterruptibly();
+ tryAgain = false;
+ try
+ {
+ // after closing, wait until we are signalled about the in flight writes;
+ // this ensures flushedToNetwork() is correct below
+ out.waitUntilFlushed(0, 0);
+ }
+ catch (Throwable ignore)
+ {
+ // irrelevant
+ }
+ }
+ }
+
+ onFailedSerialize(send, established.messagingVersion, out == null ? 0 : (int) out.flushedToNetwork(), t);
+ return tryAgain;
+ }
+ }
+
+ void stopAndRunOnEventLoop(Runnable run)
+ {
+ stopAndRun(() -> {
+ try
+ {
+ runOnEventLoop(run).await();
+ }
+ catch (InterruptedException e)
+ {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+ }
+
+ /*
+ * Size used for capacity enforcement purposes. Using current messaging version no matter what the peer's version is.
+ */
+ private int canonicalSize(Message<?> message)
+ {
+ return message.serializedSize(current_version);
+ }
+
+ private void invalidateChannel(Established established, Throwable cause)
+ {
+ JVMStabilityInspector.inspectThrowable(cause, false);
+
+ if (state != established)
+ return; // do nothing; channel already invalidated
+
+ if (isCausedByConnectionReset(cause))
+ logger.info("{} channel closed by provider", id(), cause);
+ else
+ logger.error("{} channel in potentially inconsistent state after error; closing", id(), cause);
+
+ disconnectNow(established);
+ }
+
+ /**
+ * Attempt to open a new channel to the remote endpoint.
+ *
+ * Most of the actual work is performed by OutboundConnectionInitiator, this method just manages
+ * our book keeping on either success or failure.
+ *
+ * This method is only to be invoked by the eventLoop, and the inner class' methods should only be evaluated by the eventtLoop
+ */
+ Future<?> initiate()
+ {
+ class Initiate
+ {
+ /**
+ * If we fail to connect, we want to try and connect again before any messages timeout.
+ * However, we update this each time to ensure we do not retry unreasonably often, and settle on a periodicity
+ * that might lead to timeouts in some aggressive systems.
+ */
+ long retryRateMillis = DatabaseDescriptor.getMinRpcTimeout(MILLISECONDS) / 2;
+
+ // our connection settings, possibly updated on retry
+ int messagingVersion = template.endpointToVersion().get(template.to);
+ OutboundConnectionSettings settings;
+
+ /**
+ * If we failed for any reason, try again
+ */
+ void onFailure(Throwable cause)
+ {
+ if (cause instanceof ConnectException)
+ noSpamLogger.info("{} failed to connect", id(), cause);
+ else
+ noSpamLogger.error("{} failed to connect", id(), cause);
+
+ JVMStabilityInspector.inspectThrowable(cause, false);
+
+ if (hasPending())
+ {
+ Promise<Result<MessagingSuccess>> result = new AsyncPromise<>(eventLoop);
+ state = new Connecting(state.disconnected(), result, eventLoop.schedule(() -> attempt(result), max(100, retryRateMillis), MILLISECONDS));
+ retryRateMillis = min(1000, retryRateMillis * 2);
+ }
+ else
+ {
+ // this Initiate will be discarded
+ state = Disconnected.dormant(state.disconnected().maintenance);
+ }
+ }
+
+ void onCompletedHandshake(Result<MessagingSuccess> result)
+ {
+ switch (result.outcome)
+ {
+ case SUCCESS:
+ // it is expected that close, if successful, has already cancelled us; so we do not need to worry about leaking connections
+ assert !state.isClosed();
+
+ MessagingSuccess success = result.success();
+ debug.onConnect(success.messagingVersion, settings);
+ state.disconnected().maintenance.cancel(false);
+
+ FrameEncoder.PayloadAllocator payloadAllocator = success.allocator;
+ Channel channel = success.channel;
+ Established established = new Established(messagingVersion, channel, payloadAllocator, settings);
+ state = established;
+ channel.pipeline().addLast("handleExceptionalStates", new ChannelInboundHandlerAdapter() {
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx)
+ {
+ disconnectNow(established);
+ ctx.fireChannelInactive();
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
+ {
+ try
+ {
+ invalidateChannel(established, cause);
+ }
+ catch (Throwable t)
+ {
+ logger.error("Unexpected exception in {}.exceptionCaught", this.getClass().getSimpleName(), t);
+ }
+ }
+ });
+ ++successfulConnections;
+
+ logger.info("{} successfully connected, version = {}, framing = {}, encryption = {}",
+ id(true),
+ success.messagingVersion,
+ settings.framing,
- encryptionLogStatement(channel, settings.encryption));
++ encryptionConnectionSummary(channel));
+ break;
+
+ case RETRY:
+ if (logger.isTraceEnabled())
+ logger.trace("{} incorrect legacy peer version predicted; reconnecting", id());
+
+ // the messaging version we connected with was incorrect; try again with the one supplied by the remote host
+ messagingVersion = result.retry().withMessagingVersion;
+ settings.endpointToVersion.set(settings.to, messagingVersion);
+
+ initiate();
+ break;
+
+ case INCOMPATIBLE:
+ // we cannot communicate with this peer given its messaging version; mark this as any other failure, and continue trying
+ Throwable t = new IOException(String.format("Incompatible peer: %s, messaging version: %s",
+ settings.to, result.incompatible().maxMessagingVersion));
+ t.fillInStackTrace();
+ onFailure(t);
+ break;
+
+ default:
+ throw new AssertionError();
+ }
+ }
+
+ /**
+ * Initiate all the actions required to establish a working, valid connection. This includes
+ * opening the socket, negotiating the internode messaging handshake, and setting up the working
+ * Netty {@link Channel}. However, this method will not block for all those actions: it will only
+ * kick off the connection attempt, setting the @{link #connecting} future to track its completion.
+ *
+ * Note: this should only be invoked on the event loop.
+ */
+ private void attempt(Promise<Result<MessagingSuccess>> result)
+ {
+ ++connectionAttempts;
+
+ /*
+ * Re-evaluate messagingVersion before re-attempting the connection in case
+ * endpointToVersion were updated. This happens if the outbound connection
+ * is made before the endpointToVersion table is initially constructed or out
+ * of date (e.g. if outbound connections are established for gossip
+ * as a result of an inbound connection) and can result in the wrong outbound
+ * port being selected if configured with enable_legacy_ssl_storage_port=true.
+ */
+ int knownMessagingVersion = messagingVersion();
+ if (knownMessagingVersion != messagingVersion)
+ {
+ logger.trace("Endpoint version changed from {} to {} since connection initialized, updating.",
+ messagingVersion, knownMessagingVersion);
+ messagingVersion = knownMessagingVersion;
+ }
+
+ settings = template;
+ if (messagingVersion > settings.acceptVersions.max)
+ messagingVersion = settings.acceptVersions.max;
+
+ // ensure we connect to the correct SSL port
+ settings = settings.withLegacyPortIfNecessary(messagingVersion);
+
+ initiateMessaging(eventLoop, type, settings, messagingVersion, result)
+ .addListener(future -> {
+ if (future.isCancelled())
+ return;
+ if (future.isSuccess()) //noinspection unchecked
+ onCompletedHandshake((Result<MessagingSuccess>) future.getNow());
+ else
+ onFailure(future.cause());
+ });
+ }
+
+ Future<Result<MessagingSuccess>> initiate()
+ {
+ Promise<Result<MessagingSuccess>> result = new AsyncPromise<>(eventLoop);
+ state = new Connecting(state.disconnected(), result);
+ attempt(result);
+ return result;
+ }
+ }
+
+ return new Initiate().initiate();
+ }
+
+ /**
+ * Returns a future that completes when we are _maybe_ reconnected.
+ *
+ * The connection attempt is guaranteed to have completed (successfully or not) by the time any listeners are invoked,
+ * so if a reconnection attempt is needed, it is already scheduled.
+ */
+ private Future<?> requestConnect()
+ {
+ // we may race with updates to this variable, but this is fine, since we only guarantee that we see a value
+ // that did at some point represent an active connection attempt - if it is stale, it will have been completed
+ // and the caller can retry (or utilise the successfully established connection)
+ {
+ State state = this.state;
+ if (state.isConnecting())
+ return state.connecting().attempt;
+ }
+
+ Promise<Object> promise = AsyncPromise.uncancellable(eventLoop);
+ runOnEventLoop(() -> {
+ if (isClosed()) // never going to connect
+ {
+ promise.tryFailure(new ClosedChannelException());
+ }
+ else if (state.isEstablished() && state.established().isConnected()) // already connected
+ {
+ promise.trySuccess(null);
+ }
+ else
+ {
+ if (state.isEstablished())
+ setDisconnected();
+
+ if (!state.isConnecting())
+ {
+ assert eventLoop.inEventLoop();
+ assert !isConnected();
+ initiate().addListener(new PromiseNotifier<>(promise));
+ }
+ else
+ {
+ state.connecting().attempt.addListener(new PromiseNotifier<>(promise));
+ }
+ }
+ });
+ return promise;
+ }
+
+ /**
+ * Change the IP address on which we connect to the peer. We will attempt to connect to the new address if there
+ * was a previous connection, and new incoming messages as well as existing {@link #queue} messages will be sent there.
+ * Any outstanding messages in the existing channel will still be sent to the previous address (we won't/can't move them from
+ * one channel to another).
+ *
+ * Returns null if the connection is closed.
+ */
+ Future<Void> reconnectWith(OutboundConnectionSettings reconnectWith)
+ {
+ OutboundConnectionSettings newTemplate = reconnectWith.withDefaults(ConnectionCategory.MESSAGING);
+ if (newTemplate.socketFactory != template.socketFactory) throw new IllegalArgumentException();
+ if (newTemplate.callbacks != template.callbacks) throw new IllegalArgumentException();
+ if (!Objects.equals(newTemplate.applicationSendQueueCapacityInBytes, template.applicationSendQueueCapacityInBytes)) throw new IllegalArgumentException();
+ if (!Objects.equals(newTemplate.applicationSendQueueReserveEndpointCapacityInBytes, template.applicationSendQueueReserveEndpointCapacityInBytes)) throw new IllegalArgumentException();
+ if (newTemplate.applicationSendQueueReserveGlobalCapacityInBytes != template.applicationSendQueueReserveGlobalCapacityInBytes) throw new IllegalArgumentException();
+
+ logger.info("{} updating connection settings", id());
+
+ Promise<Void> done = AsyncPromise.uncancellable(eventLoop);
+ delivery.stopAndRunOnEventLoop(() -> {
+ template = newTemplate;
+ // delivery will immediately continue after this, triggering a reconnect if necessary;
+ // this might mean a slight delay for large message delivery, as the connect will be scheduled
+ // asynchronously, so we must wait for a second turn on the eventLoop
+ if (state.isEstablished())
+ {
+ disconnectNow(state.established());
+ }
+ else if (state.isConnecting())
+ {
+ // cancel any in-flight connection attempt and restart with new template
+ state.connecting().cancel();
+ initiate();
+ }
+ done.setSuccess(null);
+ });
+ return done;
+ }
+
+ /**
+ * Close any currently open connection, forcing a reconnect if there are messages outstanding
+ * (or leaving it closed for now otherwise)
+ */
+ public boolean interrupt()
+ {
+ State state = this.state;
+ if (!state.isEstablished())
+ return false;
+
+ disconnectGracefully(state.established());
+ return true;
+ }
+
+ /**
+ * Schedule a safe close of the provided channel, if it has not already been closed.
+ *
+ * This means ensuring that delivery has stopped so that we do not corrupt or interrupt any
+ * in progress transmissions.
+ *
+ * The actual closing of the channel is performed asynchronously, to simplify our internal state management
+ * and promptly get the connection going again; the close is considered to have succeeded as soon as we
+ * have set our internal state.
+ */
+ private void disconnectGracefully(Established closeIfIs)
+ {
+ // delivery will immediately continue after this, triggering a reconnect if necessary;
+ // this might mean a slight delay for large message delivery, as the connect will be scheduled
+ // asynchronously, so we must wait for a second turn on the eventLoop
+ delivery.stopAndRunOnEventLoop(() -> disconnectNow(closeIfIs));
+ }
+
+ /**
+ * The channel is already known to be invalid, so there's no point waiting for a clean break in delivery.
+ *
+ * Delivery will be executed again as soon as we have logically closed the channel; we do not wait
+ * for the channel to actually be closed.
+ *
+ * The Future returned _does_ wait for the channel to be completely closed, so that callers can wait to be sure
+ * all writes have been completed either successfully or not.
+ */
+ private Future<?> disconnectNow(Established closeIfIs)
+ {
+ return runOnEventLoop(() -> {
+ if (state == closeIfIs)
+ {
+ // no need to wait until the channel is closed to set ourselves as disconnected (and potentially open a new channel)
+ setDisconnected();
+ if (hasPending())
+ delivery.execute();
+ closeIfIs.channel.close()
+ .addListener(future -> {
+ if (!future.isSuccess())
+ logger.info("Problem closing channel {}", closeIfIs, future.cause());
+ });
+ }
+ });
+ }
+
+ /**
+ * Schedules regular cleaning of the connection's state while it is disconnected from its remote endpoint.
+ *
+ * To be run only by the eventLoop or in the constructor
+ */
+ private void setDisconnected()
+ {
+ assert state == null || state.isEstablished();
+ state = Disconnected.dormant(eventLoop.scheduleAtFixedRate(queue::maybePruneExpired, 100L, 100L, TimeUnit.MILLISECONDS));
+ }
+
+ /**
+ * Schedule this connection to be permanently closed; only one close may be scheduled,
+ * any future scheduled closes are referred to the original triggering one (which may have a different schedule)
+ */
+ Future<Void> scheduleClose(long time, TimeUnit unit, boolean flushQueue)
+ {
+ Promise<Void> scheduledClose = AsyncPromise.uncancellable(eventLoop);
+ if (!scheduledCloseUpdater.compareAndSet(this, null, scheduledClose))
+ return this.scheduledClose;
+
+ eventLoop.schedule(() -> close(flushQueue).addListener(new PromiseNotifier<>(scheduledClose)), time, unit);
+ return scheduledClose;
+ }
+
+ /**
+ * Permanently close this connection.
+ *
+ * Immediately prevent any new messages from being enqueued - these will throw ClosedChannelException.
+ * The close itself happens asynchronously on the eventLoop, so a Future is returned to help callers
+ * wait for its completion.
+ *
+ * The flushQueue parameter indicates if any outstanding messages should be delivered before closing the connection.
+ *
+ * - If false, any already flushed or in-progress messages are completed, and the remaining messages are cleared
+ * before the connection is promptly torn down.
+ *
+ * - If true, we attempt delivery of all queued messages. If necessary, we will continue to open new connections
+ * to the remote host until they have been delivered. Only if we continue to fail to open a connection for
+ * an extended period of time will we drop any outstanding messages and close the connection.
+ */
+ public Future<Void> close(boolean flushQueue)
+ {
+ // ensure only one close attempt can be in flight
+ Promise<Void> closing = AsyncPromise.uncancellable(eventLoop);
+ if (!closingUpdater.compareAndSet(this, null, closing))
+ return this.closing;
+
+ /*
+ * Now define a cleanup closure, that will be deferred until it is safe to do so.
+ * Once run it:
+ * - immediately _logically_ closes the channel by updating this object's fields, but defers actually closing
+ * - cancels any in-flight connection attempts
+ * - cancels any maintenance work that might be scheduled
+ * - clears any waiting messages on the queue
+ * - terminates the delivery thread
+ * - finally, schedules any open channel's closure, and propagates its completion to the close promise
+ */
+ Runnable eventLoopCleanup = () -> {
+ Runnable onceNotConnecting = () -> {
+ // start by setting ourselves to definitionally closed
+ State state = this.state;
+ this.state = State.CLOSED;
+
+ try
+ {
+ // note that we never clear the queue, to ensure that an enqueue has the opportunity to remove itself
+ // if it raced with close, to potentially requeue the message on a replacement connection
+
+ // we terminate delivery here, to ensure that any listener to {@link connecting} do not schedule more work
+ delivery.terminate();
+
+ // stop periodic cleanup
+ if (state.isDisconnected())
+ {
+ state.disconnected().maintenance.cancel(true);
+ closing.setSuccess(null);
+ }
+ else
+ {
+ assert state.isEstablished();
+ state.established().channel.close()
+ .addListener(new PromiseNotifier<>(closing));
+ }
+ }
+ catch (Throwable t)
+ {
+ // in case of unexpected exception, signal completion and try to close the channel
+ closing.trySuccess(null);
+ try
+ {
+ if (state.isEstablished())
+ state.established().channel.close();
+ }
+ catch (Throwable t2)
+ {
+ t.addSuppressed(t2);
+ logger.error("Failed to close connection cleanly:", t);
+ }
+ throw t;
+ }
+ };
+
+ if (state.isConnecting())
+ {
+ // stop any in-flight connection attempts; these should be running on the eventLoop, so we should
+ // be able to cleanly cancel them, but executing on a listener guarantees correct semantics either way
+ Connecting connecting = state.connecting();
+ connecting.cancel();
+ connecting.attempt.addListener(future -> onceNotConnecting.run());
+ }
+ else
+ {
+ onceNotConnecting.run();
+ }
+ };
+
+ /*
+ * If we want to shutdown gracefully, flushing any outstanding messages, we have to do it very carefully.
+ * Things to note:
+ *
+ * - It is possible flushing messages will require establishing a new connection
+ * (However, if a connection cannot be established, we do not want to keep trying)
+ * - We have to negotiate with a separate thread, so we must be sure it is not in-progress before we stop (like channel close)
+ * - Cleanup must still happen on the eventLoop
+ *
+ * To achieve all of this, we schedule a recurring operation on the delivery thread, executing while delivery
+ * is between messages, that checks if the queue is empty; if it is, it schedules cleanup on the eventLoop.
+ */
+
+ Runnable clearQueue = () ->
+ {
+ CountDownLatch done = new CountDownLatch(1);
+ queue.runEventually(withLock -> {
+ withLock.consume(this::onClosed);
+ done.countDown();
+ });
+ //noinspection UnstableApiUsage
+ Uninterruptibles.awaitUninterruptibly(done);
+ };
+
+ if (flushQueue)
+ {
+ // just keep scheduling on delivery executor a check to see if we're done; there should always be one
+ // delivery attempt between each invocation, unless there is a wider problem with delivery scheduling
+ class FinishDelivery implements Runnable
+ {
+ public void run()
+ {
+ if (!hasPending())
+ delivery.stopAndRunOnEventLoop(eventLoopCleanup);
+ else
+ delivery.stopAndRun(() -> {
+ if (state.isConnecting() && state.connecting().isFailingToConnect)
+ clearQueue.run();
+ run();
+ });
+ }
+ }
+
+ delivery.stopAndRun(new FinishDelivery());
+ }
+ else
+ {
+ delivery.stopAndRunOnEventLoop(() -> {
+ clearQueue.run();
+ eventLoopCleanup.run();
+ });
+ }
+
+ return closing;
+ }
+
+ /**
+ * Run the task immediately if we are the eventLoop, otherwise queue it for execution on the eventLoop.
+ */
+ private Future<?> runOnEventLoop(Runnable runnable)
+ {
+ if (!eventLoop.inEventLoop())
+ return eventLoop.submit(runnable);
+
+ runnable.run();
+ return new SucceededFuture<>(eventLoop, null);
+ }
+
+ public boolean isConnected()
+ {
+ State state = this.state;
+ return state.isEstablished() && state.established().isConnected();
+ }
+
+ boolean isClosing()
+ {
+ return closing != null;
+ }
+
+ boolean isClosed()
+ {
+ return state.isClosed();
+ }
+
+ private String id(boolean includeReal)
+ {
+ State state = this.state;
+ if (!includeReal || !state.isEstablished())
+ return id();
+ Established established = state.established();
+ Channel channel = established.channel;
+ OutboundConnectionSettings settings = established.settings;
+ return SocketFactory.channelId(settings.from, (InetSocketAddress) channel.localAddress(),
+ settings.to, (InetSocketAddress) channel.remoteAddress(),
+ type, channel.id().asShortText());
+ }
+
+ private String id()
+ {
+ State state = this.state;
+ Channel channel = null;
+ OutboundConnectionSettings settings = template;
+ if (state.isEstablished())
+ {
+ channel = state.established().channel;
+ settings = state.established().settings;
+ }
+ String channelId = channel != null ? channel.id().asShortText() : "[no-channel]";
+ return SocketFactory.channelId(settings.from(), settings.to, type, channelId);
+ }
+
+ @Override
+ public String toString()
+ {
+ return id();
+ }
+
+ public boolean hasPending()
+ {
+ return 0 != pendingCountAndBytes;
+ }
+
+ public int pendingCount()
+ {
+ return pendingCount(pendingCountAndBytes);
+ }
+
+ public long pendingBytes()
+ {
+ return pendingBytes(pendingCountAndBytes);
+ }
+
+ public long sentCount()
+ {
+ // not volatile, but shouldn't matter
+ return sentCount;
+ }
+
+ public long sentBytes()
+ {
+ // not volatile, but shouldn't matter
+ return sentBytes;
+ }
+
+ public long submittedCount()
+ {
+ // not volatile, but shouldn't matter
+ return submittedCount;
+ }
+
+ public long dropped()
+ {
+ return overloadedCount + expiredCount;
+ }
+
+ public long overloadedBytes()
+ {
+ return overloadedBytes;
+ }
+
+ public long overloadedCount()
+ {
+ return overloadedCount;
+ }
+
+ public long expiredCount()
+ {
+ return expiredCount;
+ }
+
+ public long expiredBytes()
+ {
+ return expiredBytes;
+ }
+
+ public long errorCount()
+ {
+ return errorCount;
+ }
+
+ public long errorBytes()
+ {
+ return errorBytes;
+ }
+
+ public long successfulConnections()
+ {
+ return successfulConnections;
+ }
+
+ public long connectionAttempts()
+ {
+ return connectionAttempts;
+ }
+
+ private static Runnable andThen(Runnable a, Runnable b)
+ {
+ if (a == null || b == null)
+ return a == null ? b : a;
+ return () -> { a.run(); b.run(); };
+ }
+
+ @VisibleForTesting
+ public ConnectionType type()
+ {
+ return type;
+ }
+
+ @VisibleForTesting
+ OutboundConnectionSettings settings()
+ {
+ State state = this.state;
+ return state.isEstablished() ? state.established().settings : template;
+ }
+
+ @VisibleForTesting
+ int messagingVersion()
+ {
+ State state = this.state;
+ return state.isEstablished() ? state.established().messagingVersion
+ : template.endpointToVersion().get(template.to);
+ }
+
+ @VisibleForTesting
+ void unsafeRunOnDelivery(Runnable run)
+ {
+ delivery.stopAndRun(run);
+ }
+
+ @VisibleForTesting
+ Channel unsafeGetChannel()
+ {
+ State state = this.state;
+ return state.isEstablished() ? state.established().channel : null;
+ }
+
+ @VisibleForTesting
+ boolean unsafeAcquireCapacity(long amount)
+ {
+ return SUCCESS == acquireCapacity(amount);
+ }
+
+ @VisibleForTesting
+ boolean unsafeAcquireCapacity(long count, long amount)
+ {
+ return SUCCESS == acquireCapacity(count, amount);
+ }
+
+ @VisibleForTesting
+ void unsafeReleaseCapacity(long amount)
+ {
+ releaseCapacity(1, amount);
+ }
+
+ @VisibleForTesting
+ void unsafeReleaseCapacity(long count, long amount)
+ {
+ releaseCapacity(count, amount);
+ }
+
+ @VisibleForTesting
+ Limit unsafeGetEndpointReserveLimits()
+ {
+ return reserveCapacityInBytes.endpoint;
+ }
+}
diff --cc src/java/org/apache/cassandra/net/OutboundConnectionSettings.java
index 5f83b6a,0000000..1aab412
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/net/OutboundConnectionSettings.java
+++ b/src/java/org/apache/cassandra/net/OutboundConnectionSettings.java
@@@ -1,524 -1,0 +1,523 @@@
+/*
+ * 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.net;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+
+import io.netty.channel.WriteBufferWaterMark;
+import org.apache.cassandra.auth.IInternodeAuthenticator;
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.EncryptionOptions;
+import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions;
+import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.locator.IEndpointSnitch;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.config.DatabaseDescriptor.getEndpointSnitch;
+import static org.apache.cassandra.net.MessagingService.VERSION_40;
+import static org.apache.cassandra.net.MessagingService.instance;
- import static org.apache.cassandra.net.SocketFactory.encryptionLogStatement;
+import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
+
+/**
+ * A collection of settings to be passed around for outbound connections.
+ */
+@SuppressWarnings({ "WeakerAccess", "unused" })
+public class OutboundConnectionSettings
+{
+ private static final String INTRADC_TCP_NODELAY_PROPERTY = Config.PROPERTY_PREFIX + "otc_intradc_tcp_nodelay";
+ /**
+ * Enabled/disable TCP_NODELAY for intradc connections. Defaults to enabled.
+ */
+ private static final boolean INTRADC_TCP_NODELAY = Boolean.parseBoolean(System.getProperty(INTRADC_TCP_NODELAY_PROPERTY, "true"));
+
+ public enum Framing
+ {
+ // for < VERSION_40, implies no framing
+ // for >= VERSION_40, uses simple unprotected frames with header crc but no payload protection
+ UNPROTECTED(0),
+ // for < VERSION_40, uses the jpountz framing format
+ // for >= VERSION_40, uses our framing format with header crc24
+ LZ4(1),
+ // for < VERSION_40, implies UNPROTECTED
+ // for >= VERSION_40, uses simple frames with separate header and payload crc
+ CRC(2);
+
+ public static Framing forId(int id)
+ {
+ switch (id)
+ {
+ case 0: return UNPROTECTED;
+ case 1: return LZ4;
+ case 2: return CRC;
+ }
+ throw new IllegalStateException();
+ }
+
+ final int id;
+ Framing(int id)
+ {
+ this.id = id;
+ }
+ }
+
+ public final IInternodeAuthenticator authenticator;
+ public final InetAddressAndPort to;
+ public final InetAddressAndPort connectTo; // may be represented by a different IP address on this node's local network
+ public final EncryptionOptions encryption;
+ public final Framing framing;
+ public final Integer socketSendBufferSizeInBytes;
+ public final Integer applicationSendQueueCapacityInBytes;
+ public final Integer applicationSendQueueReserveEndpointCapacityInBytes;
+ public final ResourceLimits.Limit applicationSendQueueReserveGlobalCapacityInBytes;
+ public final Boolean tcpNoDelay;
+ public final int flushLowWaterMark, flushHighWaterMark;
+ public final Integer tcpConnectTimeoutInMS;
+ public final Integer tcpUserTimeoutInMS;
+ public final AcceptVersions acceptVersions;
+ public final InetAddressAndPort from;
+ public final SocketFactory socketFactory;
+ public final OutboundMessageCallbacks callbacks;
+ public final OutboundDebugCallbacks debug;
+ public final EndpointMessagingVersions endpointToVersion;
+
+ public OutboundConnectionSettings(InetAddressAndPort to)
+ {
+ this(to, null);
+ }
+
+ public OutboundConnectionSettings(InetAddressAndPort to, InetAddressAndPort preferred)
+ {
+ this(null, to, preferred, null, null, null, null, null, null, null, 1 << 15, 1 << 16, null, null, null, null, null, null, null, null);
+ }
+
+ private OutboundConnectionSettings(IInternodeAuthenticator authenticator,
+ InetAddressAndPort to,
+ InetAddressAndPort connectTo,
+ EncryptionOptions encryption,
+ Framing framing,
+ Integer socketSendBufferSizeInBytes,
+ Integer applicationSendQueueCapacityInBytes,
+ Integer applicationSendQueueReserveEndpointCapacityInBytes,
+ ResourceLimits.Limit applicationSendQueueReserveGlobalCapacityInBytes,
+ Boolean tcpNoDelay,
+ int flushLowWaterMark,
+ int flushHighWaterMark,
+ Integer tcpConnectTimeoutInMS,
+ Integer tcpUserTimeoutInMS,
+ AcceptVersions acceptVersions,
+ InetAddressAndPort from,
+ SocketFactory socketFactory,
+ OutboundMessageCallbacks callbacks,
+ OutboundDebugCallbacks debug,
+ EndpointMessagingVersions endpointToVersion)
+ {
+ Preconditions.checkArgument(socketSendBufferSizeInBytes == null || socketSendBufferSizeInBytes == 0 || socketSendBufferSizeInBytes >= 1 << 10, "illegal socket send buffer size: " + socketSendBufferSizeInBytes);
+ Preconditions.checkArgument(applicationSendQueueCapacityInBytes == null || applicationSendQueueCapacityInBytes >= 1 << 10, "illegal application send queue capacity: " + applicationSendQueueCapacityInBytes);
+ Preconditions.checkArgument(tcpUserTimeoutInMS == null || tcpUserTimeoutInMS >= 0, "tcp user timeout must be non negative: " + tcpUserTimeoutInMS);
+ Preconditions.checkArgument(tcpConnectTimeoutInMS == null || tcpConnectTimeoutInMS > 0, "tcp connect timeout must be positive: " + tcpConnectTimeoutInMS);
+
+ this.authenticator = authenticator;
+ this.to = to;
+ this.connectTo = connectTo;
+ this.encryption = encryption;
+ this.framing = framing;
+ this.socketSendBufferSizeInBytes = socketSendBufferSizeInBytes;
+ this.applicationSendQueueCapacityInBytes = applicationSendQueueCapacityInBytes;
+ this.applicationSendQueueReserveEndpointCapacityInBytes = applicationSendQueueReserveEndpointCapacityInBytes;
+ this.applicationSendQueueReserveGlobalCapacityInBytes = applicationSendQueueReserveGlobalCapacityInBytes;
+ this.tcpNoDelay = tcpNoDelay;
+ this.flushLowWaterMark = flushLowWaterMark;
+ this.flushHighWaterMark = flushHighWaterMark;
+ this.tcpConnectTimeoutInMS = tcpConnectTimeoutInMS;
+ this.tcpUserTimeoutInMS = tcpUserTimeoutInMS;
+ this.acceptVersions = acceptVersions;
+ this.from = from;
+ this.socketFactory = socketFactory;
+ this.callbacks = callbacks;
+ this.debug = debug;
+ this.endpointToVersion = endpointToVersion;
+ }
+
+ public boolean authenticate()
+ {
+ return authenticator.authenticate(to.address, to.port);
+ }
+
+ public boolean withEncryption()
+ {
+ return encryption != null;
+ }
+
+ public String toString()
+ {
+ return String.format("peer: (%s, %s), framing: %s, encryption: %s",
- to, connectTo, framing, encryptionLogStatement(encryption));
++ to, connectTo, framing, SocketFactory.encryptionOptionsSummary(encryption));
+ }
+
+ public OutboundConnectionSettings withAuthenticator(IInternodeAuthenticator authenticator)
+ {
+ return new OutboundConnectionSettings(authenticator, to, connectTo, encryption, framing,
+ socketSendBufferSizeInBytes, applicationSendQueueCapacityInBytes,
+ applicationSendQueueReserveEndpointCapacityInBytes, applicationSendQueueReserveGlobalCapacityInBytes,
+ tcpNoDelay, flushLowWaterMark, flushHighWaterMark, tcpConnectTimeoutInMS,
+ tcpUserTimeoutInMS, acceptVersions, from, socketFactory, callbacks, debug, endpointToVersion);
+ }
+
+ @SuppressWarnings("unused")
+ public OutboundConnectionSettings toEndpoint(InetAddressAndPort endpoint)
+ {
+ return new OutboundConnectionSettings(authenticator, endpoint, connectTo, encryption, framing,
+ socketSendBufferSizeInBytes, applicationSendQueueCapacityInBytes,
+ applicationSendQueueReserveEndpointCapacityInBytes, applicationSendQueueReserveGlobalCapacityInBytes,
+ tcpNoDelay, flushLowWaterMark, flushHighWaterMark, tcpConnectTimeoutInMS,
+ tcpUserTimeoutInMS, acceptVersions, from, socketFactory, callbacks, debug, endpointToVersion);
+ }
+
+ public OutboundConnectionSettings withConnectTo(InetAddressAndPort connectTo)
+ {
+ return new OutboundConnectionSettings(authenticator, to, connectTo, encryption, framing,
+ socketSendBufferSizeInBytes, applicationSendQueueCapacityInBytes,
+ applicationSendQueueReserveEndpointCapacityInBytes, applicationSendQueueReserveGlobalCapacityInBytes,
+ tcpNoDelay, flushLowWaterMark, flushHighWaterMark, tcpConnectTimeoutInMS,
+ tcpUserTimeoutInMS, acceptVersions, from, socketFactory, callbacks, debug, endpointToVersion);
+ }
+
+ public OutboundConnectionSettings withEncryption(ServerEncryptionOptions encryption)
+ {
+ return new OutboundConnectionSettings(authenticator, to, connectTo, encryption, framing,
+ socketSendBufferSizeInBytes, applicationSendQueueCapacityInBytes,
+ applicationSendQueueReserveEndpointCapacityInBytes, applicationSendQueueReserveGlobalCapacityInBytes,
+ tcpNoDelay, flushLowWaterMark, flushHighWaterMark, tcpConnectTimeoutInMS,
+ tcpUserTimeoutInMS, acceptVersions, from, socketFactory, callbacks, debug, endpointToVersion);
+ }
+
+ @SuppressWarnings("unused")
+ public OutboundConnectionSettings withFraming(Framing framing)
+ {
+ return new OutboundConnectionSettings(authenticator, to, connectTo, encryption, framing, socketSendBufferSizeInBytes, applicationSendQueueCapacityInBytes,
+ applicationSendQueueReserveEndpointCapacityInBytes, applicationSendQueueReserveGlobalCapacityInBytes,
+ tcpNoDelay, flushLowWaterMark, flushHighWaterMark, tcpConnectTimeoutInMS,
+ tcpUserTimeoutInMS, acceptVersions, from, socketFactory, callbacks, debug, endpointToVersion);
+ }
+
+ public OutboundConnectionSettings withSocketSendBufferSizeInBytes(int socketSendBufferSizeInBytes)
+ {
+ return new OutboundConnectionSettings(authenticator, to, connectTo, encryption, framing,
+ socketSendBufferSizeInBytes, applicationSendQueueCapacityInBytes,
+ applicationSendQueueReserveEndpointCapacityInBytes, applicationSendQueueReserveGlobalCapacityInBytes,
+ tcpNoDelay, flushLowWaterMark, flushHighWaterMark, tcpConnectTimeoutInMS,
+ tcpUserTimeoutInMS, acceptVersions, from, socketFactory, callbacks, debug, endpointToVersion);
+ }
+
+ @SuppressWarnings("unused")
+ public OutboundConnectionSettings withApplicationSendQueueCapacityInBytes(int applicationSendQueueCapacityInBytes)
+ {
+ return new OutboundConnectionSettings(authenticator, to, connectTo, encryption, framing,
+ socketSendBufferSizeInBytes, applicationSendQueueCapacityInBytes,
+ applicationSendQueueReserveEndpointCapacityInBytes, applicationSendQueueReserveGlobalCapacityInBytes,
+ tcpNoDelay, flushLowWaterMark, flushHighWaterMark, tcpConnectTimeoutInMS,
+ tcpUserTimeoutInMS, acceptVersions, from, socketFactory, callbacks, debug, endpointToVersion);
+ }
+
+ public OutboundConnectionSettings withApplicationReserveSendQueueCapacityInBytes(Integer applicationReserveSendQueueEndpointCapacityInBytes, ResourceLimits.Limit applicationReserveSendQueueGlobalCapacityInBytes)
+ {
+ return new OutboundConnectionSettings(authenticator, to, connectTo, encryption, framing,
+ socketSendBufferSizeInBytes, applicationSendQueueCapacityInBytes,
+ applicationReserveSendQueueEndpointCapacityInBytes, applicationReserveSendQueueGlobalCapacityInBytes,
+ tcpNoDelay, flushLowWaterMark, flushHighWaterMark, tcpConnectTimeoutInMS,
+ tcpUserTimeoutInMS, acceptVersions, from, socketFactory, callbacks, debug, endpointToVersion);
+ }
+
+ @SuppressWarnings("unused")
+ public OutboundConnectionSettings withTcpNoDelay(boolean tcpNoDelay)
+ {
+ return new OutboundConnectionSettings(authenticator, to, connectTo, encryption, framing,
+ socketSendBufferSizeInBytes, applicationSendQueueCapacityInBytes,
+ applicationSendQueueReserveEndpointCapacityInBytes, applicationSendQueueReserveGlobalCapacityInBytes,
+ tcpNoDelay, flushLowWaterMark, flushHighWaterMark, tcpConnectTimeoutInMS,
+ tcpUserTimeoutInMS, acceptVersions, from, socketFactory, callbacks, debug, endpointToVersion);
+ }
+
+ @SuppressWarnings("unused")
+ public OutboundConnectionSettings withNettyBufferBounds(WriteBufferWaterMark nettyBufferBounds)
+ {
+ return new OutboundConnectionSettings(authenticator, to, connectTo, encryption, framing,
+ socketSendBufferSizeInBytes, applicationSendQueueCapacityInBytes,
+ applicationSendQueueReserveEndpointCapacityInBytes, applicationSendQueueReserveGlobalCapacityInBytes,
+ tcpNoDelay, flushLowWaterMark, flushHighWaterMark, tcpConnectTimeoutInMS,
+ tcpUserTimeoutInMS, acceptVersions, from, socketFactory, callbacks, debug, endpointToVersion);
+ }
+
+ public OutboundConnectionSettings withTcpConnectTimeoutInMS(int tcpConnectTimeoutInMS)
+ {
+ return new OutboundConnectionSettings(authenticator, to, connectTo, encryption, framing,
+ socketSendBufferSizeInBytes, applicationSendQueueCapacityInBytes,
+ applicationSendQueueReserveEndpointCapacityInBytes, applicationSendQueueReserveGlobalCapacityInBytes,
+ tcpNoDelay, flushLowWaterMark, flushHighWaterMark, tcpConnectTimeoutInMS,
+ tcpUserTimeoutInMS, acceptVersions, from, socketFactory, callbacks, debug, endpointToVersion);
+ }
+
+ public OutboundConnectionSettings withTcpUserTimeoutInMS(int tcpUserTimeoutInMS)
+ {
+ return new OutboundConnectionSettings(authenticator, to, connectTo, encryption, framing,
+ socketSendBufferSizeInBytes, applicationSendQueueCapacityInBytes,
+ applicationSendQueueReserveEndpointCapacityInBytes, applicationSendQueueReserveGlobalCapacityInBytes,
+ tcpNoDelay, flushLowWaterMark, flushHighWaterMark, tcpConnectTimeoutInMS,
+ tcpUserTimeoutInMS, acceptVersions, from, socketFactory, callbacks, debug, endpointToVersion);
+ }
+
+ public OutboundConnectionSettings withAcceptVersions(AcceptVersions acceptVersions)
+ {
+ return new OutboundConnectionSettings(authenticator, to, connectTo, encryption, framing,
+ socketSendBufferSizeInBytes, applicationSendQueueCapacityInBytes,
+ applicationSendQueueReserveEndpointCapacityInBytes, applicationSendQueueReserveGlobalCapacityInBytes,
+ tcpNoDelay, flushLowWaterMark, flushHighWaterMark, tcpConnectTimeoutInMS,
+ tcpUserTimeoutInMS, acceptVersions, from, socketFactory, callbacks, debug, endpointToVersion);
+ }
+
+ public OutboundConnectionSettings withFrom(InetAddressAndPort from)
+ {
+ return new OutboundConnectionSettings(authenticator, to, connectTo, encryption, framing,
+ socketSendBufferSizeInBytes, applicationSendQueueCapacityInBytes,
+ applicationSendQueueReserveEndpointCapacityInBytes, applicationSendQueueReserveGlobalCapacityInBytes,
+ tcpNoDelay, flushLowWaterMark, flushHighWaterMark, tcpConnectTimeoutInMS,
+ tcpUserTimeoutInMS, acceptVersions, from, socketFactory, callbacks, debug, endpointToVersion);
+ }
+
+ public OutboundConnectionSettings withSocketFactory(SocketFactory socketFactory)
+ {
+ return new OutboundConnectionSettings(authenticator, to, connectTo, encryption, framing,
+ socketSendBufferSizeInBytes, applicationSendQueueCapacityInBytes,
+ applicationSendQueueReserveEndpointCapacityInBytes, applicationSendQueueReserveGlobalCapacityInBytes,
+ tcpNoDelay, flushLowWaterMark, flushHighWaterMark, tcpConnectTimeoutInMS,
+ tcpUserTimeoutInMS, acceptVersions, from, socketFactory, callbacks, debug, endpointToVersion);
+ }
+
+ public OutboundConnectionSettings withCallbacks(OutboundMessageCallbacks callbacks)
+ {
+ return new OutboundConnectionSettings(authenticator, to, connectTo, encryption, framing,
+ socketSendBufferSizeInBytes, applicationSendQueueCapacityInBytes,
+ applicationSendQueueReserveEndpointCapacityInBytes, applicationSendQueueReserveGlobalCapacityInBytes,
+ tcpNoDelay, flushLowWaterMark, flushHighWaterMark, tcpConnectTimeoutInMS,
+ tcpUserTimeoutInMS, acceptVersions, from, socketFactory, callbacks, debug, endpointToVersion);
+ }
+
+ public OutboundConnectionSettings withDebugCallbacks(OutboundDebugCallbacks debug)
+ {
+ return new OutboundConnectionSettings(authenticator, to, connectTo, encryption, framing,
+ socketSendBufferSizeInBytes, applicationSendQueueCapacityInBytes,
+ applicationSendQueueReserveEndpointCapacityInBytes, applicationSendQueueReserveGlobalCapacityInBytes,
+ tcpNoDelay, flushLowWaterMark, flushHighWaterMark, tcpConnectTimeoutInMS,
+ tcpUserTimeoutInMS, acceptVersions, from, socketFactory, callbacks, debug, endpointToVersion);
+ }
+
+ public OutboundConnectionSettings withDefaultReserveLimits()
+ {
+ Integer applicationReserveSendQueueEndpointCapacityInBytes = this.applicationSendQueueReserveEndpointCapacityInBytes;
+ ResourceLimits.Limit applicationReserveSendQueueGlobalCapacityInBytes = this.applicationSendQueueReserveGlobalCapacityInBytes;
+
+ if (applicationReserveSendQueueEndpointCapacityInBytes == null)
+ applicationReserveSendQueueEndpointCapacityInBytes = DatabaseDescriptor.getInternodeApplicationSendQueueReserveEndpointCapacityInBytes();
+ if (applicationReserveSendQueueGlobalCapacityInBytes == null)
+ applicationReserveSendQueueGlobalCapacityInBytes = MessagingService.instance().outboundGlobalReserveLimit;
+
+ return withApplicationReserveSendQueueCapacityInBytes(applicationReserveSendQueueEndpointCapacityInBytes, applicationReserveSendQueueGlobalCapacityInBytes);
+ }
+
+ public IInternodeAuthenticator authenticator()
+ {
+ return authenticator != null ? authenticator : DatabaseDescriptor.getInternodeAuthenticator();
+ }
+
+ public EndpointMessagingVersions endpointToVersion()
+ {
+ if (endpointToVersion == null)
+ return instance().versions;
+ return endpointToVersion;
+ }
+
+ public InetAddressAndPort from()
+ {
+ return from != null ? from : FBUtilities.getBroadcastAddressAndPort();
+ }
+
+ public OutboundDebugCallbacks debug()
+ {
+ return debug != null ? debug : OutboundDebugCallbacks.NONE;
+ }
+
+ public EncryptionOptions encryption()
+ {
+ return encryption != null ? encryption : defaultEncryptionOptions(to);
+ }
+
+ public SocketFactory socketFactory()
+ {
+ return socketFactory != null ? socketFactory : instance().socketFactory;
+ }
+
+ public OutboundMessageCallbacks callbacks()
+ {
+ return callbacks != null ? callbacks : instance().callbacks;
+ }
+
+ public int socketSendBufferSizeInBytes()
+ {
+ return socketSendBufferSizeInBytes != null ? socketSendBufferSizeInBytes
+ : DatabaseDescriptor.getInternodeSocketSendBufferSizeInBytes();
+ }
+
+ public int applicationSendQueueCapacityInBytes()
+ {
+ return applicationSendQueueCapacityInBytes != null ? applicationSendQueueCapacityInBytes
+ : DatabaseDescriptor.getInternodeApplicationSendQueueCapacityInBytes();
+ }
+
+ public ResourceLimits.Limit applicationSendQueueReserveGlobalCapacityInBytes()
+ {
+ return applicationSendQueueReserveGlobalCapacityInBytes != null ? applicationSendQueueReserveGlobalCapacityInBytes
+ : instance().outboundGlobalReserveLimit;
+ }
+
+ public int applicationSendQueueReserveEndpointCapacityInBytes()
+ {
+ return applicationSendQueueReserveEndpointCapacityInBytes != null ? applicationSendQueueReserveEndpointCapacityInBytes
+ : DatabaseDescriptor.getInternodeApplicationReceiveQueueReserveEndpointCapacityInBytes();
+ }
+
+ public int tcpConnectTimeoutInMS()
+ {
+ return tcpConnectTimeoutInMS != null ? tcpConnectTimeoutInMS
+ : DatabaseDescriptor.getInternodeTcpConnectTimeoutInMS();
+ }
+
+ public int tcpUserTimeoutInMS()
+ {
+ return tcpUserTimeoutInMS != null ? tcpUserTimeoutInMS
+ : DatabaseDescriptor.getInternodeTcpUserTimeoutInMS();
+ }
+
+ public boolean tcpNoDelay()
+ {
+ if (tcpNoDelay != null)
+ return tcpNoDelay;
+
+ if (isInLocalDC(getEndpointSnitch(), getBroadcastAddressAndPort(), to))
+ return INTRADC_TCP_NODELAY;
+
+ return DatabaseDescriptor.getInterDCTcpNoDelay();
+ }
+
+ public AcceptVersions acceptVersions(ConnectionCategory category)
+ {
+ return acceptVersions != null ? acceptVersions
+ : category.isStreaming()
+ ? MessagingService.accept_streaming
+ : MessagingService.accept_messaging;
+ }
+
+ public OutboundConnectionSettings withLegacyPortIfNecessary(int messagingVersion)
+ {
+ return withConnectTo(maybeWithSecurePort(connectTo(), messagingVersion, withEncryption()));
+ }
+
+ public InetAddressAndPort connectTo()
+ {
+ InetAddressAndPort connectTo = this.connectTo;
+ if (connectTo == null)
+ connectTo = SystemKeyspace.getPreferredIP(to);
+ return connectTo;
+ }
+
+ public String connectToId()
+ {
+ return !to.equals(connectTo())
+ ? to.toString()
+ : to.toString() + '(' + connectTo().toString() + ')';
+ }
+
+ public Framing framing(ConnectionCategory category)
+ {
+ if (framing != null)
+ return framing;
+
+ if (category.isStreaming())
+ return Framing.UNPROTECTED;
+
+ return shouldCompressConnection(getEndpointSnitch(), getBroadcastAddressAndPort(), to)
+ ? Framing.LZ4 : Framing.CRC;
+ }
+
+ // note that connectTo is updated even if specified, in the case of pre40 messaging and using encryption (to update port)
+ public OutboundConnectionSettings withDefaults(ConnectionCategory category)
+ {
+ if (to == null)
+ throw new IllegalArgumentException();
+
+ return new OutboundConnectionSettings(authenticator(), to, connectTo(),
+ encryption(), framing(category),
+ socketSendBufferSizeInBytes(), applicationSendQueueCapacityInBytes(),
+ applicationSendQueueReserveEndpointCapacityInBytes(),
+ applicationSendQueueReserveGlobalCapacityInBytes(),
+ tcpNoDelay(), flushLowWaterMark, flushHighWaterMark,
+ tcpConnectTimeoutInMS(), tcpUserTimeoutInMS(), acceptVersions(category),
+ from(), socketFactory(), callbacks(), debug(), endpointToVersion());
+ }
+
+ private static boolean isInLocalDC(IEndpointSnitch snitch, InetAddressAndPort localHost, InetAddressAndPort remoteHost)
+ {
+ String remoteDC = snitch.getDatacenter(remoteHost);
+ String localDC = snitch.getDatacenter(localHost);
+ return remoteDC != null && remoteDC.equals(localDC);
+ }
+
+ @VisibleForTesting
+ static EncryptionOptions defaultEncryptionOptions(InetAddressAndPort endpoint)
+ {
+ ServerEncryptionOptions options = DatabaseDescriptor.getInternodeMessagingEncyptionOptions();
+ return options.shouldEncrypt(endpoint) ? options : null;
+ }
+
+ @VisibleForTesting
+ static boolean shouldCompressConnection(IEndpointSnitch snitch, InetAddressAndPort localHost, InetAddressAndPort remoteHost)
+ {
+ return (DatabaseDescriptor.internodeCompression() == Config.InternodeCompression.all)
+ || ((DatabaseDescriptor.internodeCompression() == Config.InternodeCompression.dc) && !isInLocalDC(snitch, localHost, remoteHost));
+ }
+
+ private static InetAddressAndPort maybeWithSecurePort(InetAddressAndPort address, int messagingVersion, boolean isEncrypted)
+ {
+ if (!isEncrypted || messagingVersion >= VERSION_40)
+ return address;
+
+ // if we don't know the version of the peer, assume it is 4.0 (or higher) as the only time is would be lower
+ // (as in a 3.x version) is during a cluster upgrade (from 3.x to 4.0). In that case the outbound connection will
+ // unfortunately fail - however the peer should connect to this node (at some point), and once we learn it's version, it'll be
+ // in versions map. thus, when we attempt to reconnect to that node, we'll have the version and we can get the correct port.
+ // we will be able to remove this logic at 5.0.
+ // Also as of 4.0 we will propagate the "regular" port (which will support both SSL and non-SSL) via gossip so
+ // for SSL and version 4.0 always connect to the gossiped port because if SSL is enabled it should ALWAYS
+ // listen for SSL on the "regular" port.
+ return address.withPort(DatabaseDescriptor.getSSLStoragePort());
+ }
+
+}
diff --cc src/java/org/apache/cassandra/net/SocketFactory.java
index a8ee729,0000000..8300c2a
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/net/SocketFactory.java
+++ b/src/java/org/apache/cassandra/net/SocketFactory.java
@@@ -1,326 -1,0 +1,328 @@@
+/*
+ * 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.net;
+
+import java.io.IOException;
+import java.net.ConnectException;
+import java.net.InetSocketAddress;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.spi.SelectorProvider;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeoutException;
+import javax.annotation.Nullable;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLParameters;
+import javax.net.ssl.SSLSession;
+
+import com.google.common.collect.ImmutableList;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFactory;
+import io.netty.channel.DefaultSelectStrategyFactory;
+import io.netty.channel.EventLoop;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.ServerChannel;
+import io.netty.channel.epoll.EpollChannelOption;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollServerSocketChannel;
+import io.netty.channel.epoll.EpollSocketChannel;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.channel.unix.Errors;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.DefaultEventExecutorChooserFactory;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import io.netty.util.concurrent.RejectedExecutionHandlers;
+import io.netty.util.concurrent.ThreadPerTaskExecutor;
+import io.netty.util.internal.logging.InternalLoggerFactory;
+import io.netty.util.internal.logging.Slf4JLoggerFactory;
+import org.apache.cassandra.concurrent.NamedThreadFactory;
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.EncryptionOptions;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.security.SSLFactory;
+import org.apache.cassandra.service.NativeTransportService;
+import org.apache.cassandra.utils.ExecutorUtils;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static io.netty.channel.unix.Errors.ERRNO_ECONNRESET_NEGATIVE;
+import static io.netty.channel.unix.Errors.ERROR_ECONNREFUSED_NEGATIVE;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.cassandra.utils.Throwables.isCausedBy;
+
+/**
+ * A factory for building Netty {@link Channel}s. Channels here are setup with a pipeline to participate
+ * in the internode protocol handshake, either the inbound or outbound side as per the method invoked.
+ */
+public final class SocketFactory
+{
+ private static final Logger logger = LoggerFactory.getLogger(SocketFactory.class);
+
+ private static final int EVENT_THREADS = Integer.getInteger(Config.PROPERTY_PREFIX + "internode-event-threads", FBUtilities.getAvailableProcessors());
+
+ /**
+ * The default task queue used by {@code NioEventLoop} and {@code EpollEventLoop} is {@code MpscUnboundedArrayQueue},
+ * provided by JCTools. While efficient, it has an undesirable quality for a queue backing an event loop: it is
+ * not non-blocking, and can cause the event loop to busy-spin while waiting for a partially completed task
+ * offer, if the producer thread has been suspended mid-offer.
+ *
+ * As it happens, however, we have an MPSC queue implementation that is perfectly fit for this purpose -
+ * {@link ManyToOneConcurrentLinkedQueue}, that is non-blocking, and already used throughout the codebase,
+ * that we can and do use here as well.
+ */
+ enum Provider
+ {
+ NIO
+ {
+ @Override
+ NioEventLoopGroup makeEventLoopGroup(int threadCount, ThreadFactory threadFactory)
+ {
+ return new NioEventLoopGroup(threadCount,
+ new ThreadPerTaskExecutor(threadFactory),
+ DefaultEventExecutorChooserFactory.INSTANCE,
+ SelectorProvider.provider(),
+ DefaultSelectStrategyFactory.INSTANCE,
+ RejectedExecutionHandlers.reject(),
+ capacity -> new ManyToOneConcurrentLinkedQueue<>());
+ }
+
+ @Override
+ ChannelFactory<NioSocketChannel> clientChannelFactory()
+ {
+ return NioSocketChannel::new;
+ }
+
+ @Override
+ ChannelFactory<NioServerSocketChannel> serverChannelFactory()
+ {
+ return NioServerSocketChannel::new;
+ }
+ },
+ EPOLL
+ {
+ @Override
+ EpollEventLoopGroup makeEventLoopGroup(int threadCount, ThreadFactory threadFactory)
+ {
+ return new EpollEventLoopGroup(threadCount,
+ new ThreadPerTaskExecutor(threadFactory),
+ DefaultEventExecutorChooserFactory.INSTANCE,
+ DefaultSelectStrategyFactory.INSTANCE,
+ RejectedExecutionHandlers.reject(),
+ capacity -> new ManyToOneConcurrentLinkedQueue<>());
+ }
+
+ @Override
+ ChannelFactory<EpollSocketChannel> clientChannelFactory()
+ {
+ return EpollSocketChannel::new;
+ }
+
+ @Override
+ ChannelFactory<EpollServerSocketChannel> serverChannelFactory()
+ {
+ return EpollServerSocketChannel::new;
+ }
+ };
+
+ EventLoopGroup makeEventLoopGroup(int threadCount, String threadNamePrefix)
+ {
+ logger.debug("using netty {} event loop for pool prefix {}", name(), threadNamePrefix);
+ return makeEventLoopGroup(threadCount, new DefaultThreadFactory(threadNamePrefix, true));
+ }
+
+ abstract EventLoopGroup makeEventLoopGroup(int threadCount, ThreadFactory threadFactory);
+ abstract ChannelFactory<? extends Channel> clientChannelFactory();
+ abstract ChannelFactory<? extends ServerChannel> serverChannelFactory();
+
+ static Provider optimalProvider()
+ {
+ return NativeTransportService.useEpoll() ? EPOLL : NIO;
+ }
+ }
+
+ /** a useful addition for debugging; simply set to true to get more data in your logs */
+ static final boolean WIRETRACE = false;
+ static
+ {
+ if (WIRETRACE)
+ InternalLoggerFactory.setDefaultFactory(Slf4JLoggerFactory.INSTANCE);
+ }
+
+ private final Provider provider;
+ private final EventLoopGroup acceptGroup;
+ private final EventLoopGroup defaultGroup;
+ // we need a separate EventLoopGroup for outbound streaming because sendFile is blocking
+ private final EventLoopGroup outboundStreamingGroup;
+ final ExecutorService synchronousWorkExecutor = Executors.newCachedThreadPool(new NamedThreadFactory("Messaging-SynchronousWork"));
+
+ SocketFactory()
+ {
+ this(Provider.optimalProvider());
+ }
+
+ SocketFactory(Provider provider)
+ {
+ this.provider = provider;
+ this.acceptGroup = provider.makeEventLoopGroup(1, "Messaging-AcceptLoop");
+ this.defaultGroup = provider.makeEventLoopGroup(EVENT_THREADS, NamedThreadFactory.globalPrefix() + "Messaging-EventLoop");
+ this.outboundStreamingGroup = provider.makeEventLoopGroup(EVENT_THREADS, "Streaming-EventLoop");
+ }
+
+ Bootstrap newClientBootstrap(EventLoop eventLoop, int tcpUserTimeoutInMS)
+ {
+ if (eventLoop == null)
+ throw new IllegalArgumentException("must provide eventLoop");
+
+ Bootstrap bootstrap = new Bootstrap().group(eventLoop).channelFactory(provider.clientChannelFactory());
+
+ if (provider == Provider.EPOLL)
+ bootstrap.option(EpollChannelOption.TCP_USER_TIMEOUT, tcpUserTimeoutInMS);
+
+ return bootstrap;
+ }
+
+ ServerBootstrap newServerBootstrap()
+ {
+ return new ServerBootstrap().group(acceptGroup, defaultGroup).channelFactory(provider.serverChannelFactory());
+ }
+
+ /**
+ * Creates a new {@link SslHandler} from provided SslContext.
+ * @param peer enables endpoint verification for remote address when not null
+ */
+ static SslHandler newSslHandler(Channel channel, SslContext sslContext, @Nullable InetSocketAddress peer)
+ {
+ if (peer == null)
+ return sslContext.newHandler(channel.alloc());
+
+ logger.debug("Creating SSL handler for {}:{}", peer.getHostString(), peer.getPort());
+ SslHandler sslHandler = sslContext.newHandler(channel.alloc(), peer.getHostString(), peer.getPort());
+ SSLEngine engine = sslHandler.engine();
+ SSLParameters sslParameters = engine.getSSLParameters();
+ sslParameters.setEndpointIdentificationAlgorithm("HTTPS");
+ engine.setSSLParameters(sslParameters);
+ return sslHandler;
+ }
+
- static String encryptionLogStatement(EncryptionOptions options)
++ /**
++ * Summarizes the intended encryption options, suitable for logging. Once a connection is established, use
++ * {@link SocketFactory#encryptionConnectionSummary} below.
++ * @param options options to summarize
++ * @return description of encryption options
++ */
++ static String encryptionOptionsSummary(EncryptionOptions options)
+ {
- if (options == null)
- return "disabled";
++ if (options == null || options.tlsEncryptionPolicy() == EncryptionOptions.TlsEncryptionPolicy.UNENCRYPTED)
++ return EncryptionOptions.TlsEncryptionPolicy.UNENCRYPTED.description();
+
+ String encryptionType = SSLFactory.openSslIsAvailable() ? "openssl" : "jdk";
- return "enabled (" + encryptionType + ')';
++ return options.tlsEncryptionPolicy().description() + '(' + encryptionType + ')';
+ }
+
- static String encryptionLogStatement(Channel channel, EncryptionOptions options)
++ /**
++ * Summarizes the encryption status of a channel, suitable for logging.
++ * @return description of channel encryption
++ */
++ static String encryptionConnectionSummary(Channel channel)
+ {
- if (options == null || !options.isEnabled())
- return "disabled";
-
- StringBuilder sb = new StringBuilder(64);
- if (options.optional)
- sb.append("optional (factory=");
- else
- sb.append("enabled (factory=");
- sb.append(SSLFactory.openSslIsAvailable() ? "openssl" : "jdk");
-
- final SslHandler sslHandler = channel == null ? null : channel.pipeline().get(SslHandler.class);
- if (sslHandler != null)
++ final SslHandler sslHandler = channel.pipeline().get(SslHandler.class);
++ if (sslHandler == null)
+ {
- SSLSession session = sslHandler.engine().getSession();
- sb.append(";protocol=")
- .append(session.getProtocol())
- .append(";cipher=")
- .append(session.getCipherSuite());
++ return EncryptionOptions.TlsEncryptionPolicy.UNENCRYPTED.description();
+ }
-
- sb.append(')');
- return sb.toString();
++ SSLSession session = sslHandler.engine().getSession();
++
++ return "encrypted(factory=" +
++ (SSLFactory.openSslIsAvailable() ? "openssl" : "jdk") +
++ ";protocol=" +
++ (session != null ? session.getProtocol() : "MISSING SESSION") +
++ ";cipher=" +
++ (session != null ? session.getCipherSuite() : "MISSING SESSION") +
++ ')';
+ }
+
+ EventLoopGroup defaultGroup()
+ {
+ return defaultGroup;
+ }
+
+ public EventLoopGroup outboundStreamingGroup()
+ {
+ return outboundStreamingGroup;
+ }
+
+ public void shutdownNow()
+ {
+ acceptGroup.shutdownGracefully(0, 2, SECONDS);
+ defaultGroup.shutdownGracefully(0, 2, SECONDS);
+ outboundStreamingGroup.shutdownGracefully(0, 2, SECONDS);
+ synchronousWorkExecutor.shutdownNow();
+ }
+
+ void awaitTerminationUntil(long deadlineNanos) throws InterruptedException, TimeoutException
+ {
+ List<ExecutorService> groups = ImmutableList.of(acceptGroup, defaultGroup, outboundStreamingGroup, synchronousWorkExecutor);
+ ExecutorUtils.awaitTerminationUntil(deadlineNanos, groups);
+ }
+
+ static boolean isConnectionReset(Throwable t)
+ {
+ if (t instanceof ClosedChannelException)
+ return true;
+ if (t instanceof ConnectException)
+ return true;
+ if (t instanceof Errors.NativeIoException)
+ {
+ int errorCode = ((Errors.NativeIoException) t).expectedErr();
+ return errorCode == ERRNO_ECONNRESET_NEGATIVE || errorCode != ERROR_ECONNREFUSED_NEGATIVE;
+ }
+ return IOException.class == t.getClass() && ("Broken pipe".equals(t.getMessage()) || "Connection reset by peer".equals(t.getMessage()));
+ }
+
+ static boolean isCausedByConnectionReset(Throwable t)
+ {
+ return isCausedBy(t, SocketFactory::isConnectionReset);
+ }
+
+ static String channelId(InetAddressAndPort from, InetSocketAddress realFrom, InetAddressAndPort to, InetSocketAddress realTo, ConnectionType type, String id)
+ {
+ return addressId(from, realFrom) + "->" + addressId(to, realTo) + '-' + type + '-' + id;
+ }
+
+ static String addressId(InetAddressAndPort address, InetSocketAddress realAddress)
+ {
+ String str = address.toString();
+ if (!address.address.equals(realAddress.getAddress()) || address.port != realAddress.getPort())
+ str += '(' + InetAddressAndPort.toString(realAddress.getAddress(), realAddress.getPort()) + ')';
+ return str;
+ }
+
+ static String channelId(InetAddressAndPort from, InetAddressAndPort to, ConnectionType type, String id)
+ {
+ return from + "->" + to + '-' + type + '-' + id;
+ }
+}
diff --cc src/java/org/apache/cassandra/security/SSLFactory.java
index e51ce46,7216e2c..94db564
--- a/src/java/org/apache/cassandra/security/SSLFactory.java
+++ b/src/java/org/apache/cassandra/security/SSLFactory.java
@@@ -229,8 -198,14 +229,8 @@@ public final class SSLFactor
}
catch (Exception e)
{
- throw new IOException("failed to build trust manager store for secure connections", e);
- throw new IOException("Error creating the initializing the SSL Context", e);
- }
- finally
- {
- FileUtils.closeQuietly(tsf);
- FileUtils.closeQuietly(ksf);
++ throw new IOException("failed to build key manager store for secure connections", e);
}
- return ctx;
}
public static String[] filterCipherSuites(String[] supported, String[] desired)
@@@ -247,221 -222,4 +247,221 @@@
}
return ret;
}
+
+ /**
+ * get a netty {@link SslContext} instance
+ */
+ public static SslContext getOrCreateSslContext(EncryptionOptions options, boolean buildTruststore,
+ SocketType socketType) throws IOException
+ {
+ return getOrCreateSslContext(options, buildTruststore, socketType, openSslIsAvailable());
+ }
+
+ /**
+ * Get a netty {@link SslContext} instance.
+ */
+ @VisibleForTesting
+ static SslContext getOrCreateSslContext(EncryptionOptions options,
+ boolean buildTruststore,
+ SocketType socketType,
+ boolean useOpenSsl) throws IOException
+ {
+ CacheKey key = new CacheKey(options, socketType, useOpenSsl);
+ SslContext sslContext;
+
+ sslContext = cachedSslContexts.get(key);
+ if (sslContext != null)
+ return sslContext;
+
+ sslContext = createNettySslContext(options, buildTruststore, socketType, useOpenSsl);
+
+ SslContext previous = cachedSslContexts.putIfAbsent(key, sslContext);
+ if (previous == null)
+ return sslContext;
+
+ ReferenceCountUtil.release(sslContext);
+ return previous;
+ }
+
+ /**
+ * Create a Netty {@link SslContext}
+ */
+ static SslContext createNettySslContext(EncryptionOptions options, boolean buildTruststore,
+ SocketType socketType, boolean useOpenSsl) throws IOException
+ {
+ /*
+ There is a case where the netty/openssl combo might not support using KeyManagerFactory. specifically,
+ I've seen this with the netty-tcnative dynamic openssl implementation. using the netty-tcnative static-boringssl
+ works fine with KeyManagerFactory. If we want to support all of the netty-tcnative options, we would need
+ to fall back to passing in a file reference for both a x509 and PKCS#8 private key file in PEM format (see
+ {@link SslContextBuilder#forServer(File, File, String)}). However, we are not supporting that now to keep
+ the config/yaml API simple.
+ */
+ KeyManagerFactory kmf = buildKeyManagerFactory(options);
+ SslContextBuilder builder;
+ if (socketType == SocketType.SERVER)
+ {
+ builder = SslContextBuilder.forServer(kmf);
+ builder.clientAuth(options.require_client_auth ? ClientAuth.REQUIRE : ClientAuth.NONE);
+ }
+ else
+ {
+ builder = SslContextBuilder.forClient().keyManager(kmf);
+ }
+
+ builder.sslProvider(useOpenSsl ? SslProvider.OPENSSL : SslProvider.JDK);
+
+ // only set the cipher suites if the opertor has explicity configured values for it; else, use the default
+ // for each ssl implemention (jdk or openssl)
+ if (options.cipher_suites != null && !options.cipher_suites.isEmpty())
+ builder.ciphers(options.cipher_suites, SupportedCipherSuiteFilter.INSTANCE);
+
+ if (buildTruststore)
+ builder.trustManager(buildTrustManagerFactory(options));
+
+ return builder.build();
+ }
+
+ /**
+ * Performs a lightweight check whether the certificate files have been refreshed.
+ *
+ * @throws IllegalStateException if {@link #initHotReloading(EncryptionOptions.ServerEncryptionOptions, EncryptionOptions, boolean)}
+ * is not called first
+ */
+ public static void checkCertFilesForHotReloading(EncryptionOptions.ServerEncryptionOptions serverOpts,
+ EncryptionOptions clientOpts)
+ {
+ if (!isHotReloadingInitialized)
+ throw new IllegalStateException("Hot reloading functionality has not been initialized.");
+
+ logger.debug("Checking whether certificates have been updated {}", hotReloadableFiles);
+
+ if (hotReloadableFiles.stream().anyMatch(HotReloadableFile::shouldReload))
+ {
+ logger.info("SSL certificates have been updated. Reseting the ssl contexts for new connections.");
+ try
+ {
+ validateSslCerts(serverOpts, clientOpts);
+ cachedSslContexts.clear();
+ }
+ catch(Exception e)
+ {
+ logger.error("Failed to hot reload the SSL Certificates! Please check the certificate files.", e);
+ }
+ }
+ }
+
+ /**
+ * Determines whether to hot reload certificates and schedules a periodic task for it.
+ *
+ * @param serverOpts Server encryption options (Internode)
+ * @param clientOpts Client encryption options (Native Protocol)
+ */
+ public static synchronized void initHotReloading(EncryptionOptions.ServerEncryptionOptions serverOpts,
+ EncryptionOptions clientOpts,
+ boolean force) throws IOException
+ {
+ if (isHotReloadingInitialized && !force)
+ return;
+
+ logger.debug("Initializing hot reloading SSLContext");
+
+ validateSslCerts(serverOpts, clientOpts);
+
+ List<HotReloadableFile> fileList = new ArrayList<>();
+
- if (serverOpts != null && serverOpts.isEnabled())
++ if (serverOpts != null && serverOpts.tlsEncryptionPolicy() != EncryptionOptions.TlsEncryptionPolicy.UNENCRYPTED)
+ {
+ fileList.add(new HotReloadableFile(serverOpts.keystore));
+ fileList.add(new HotReloadableFile(serverOpts.truststore));
+ }
+
- if (clientOpts != null && clientOpts.isEnabled())
++ if (clientOpts != null && clientOpts.tlsEncryptionPolicy() != EncryptionOptions.TlsEncryptionPolicy.UNENCRYPTED)
+ {
+ fileList.add(new HotReloadableFile(clientOpts.keystore));
+ fileList.add(new HotReloadableFile(clientOpts.truststore));
+ }
+
+ hotReloadableFiles = ImmutableList.copyOf(fileList);
+
+ if (!isHotReloadingInitialized)
+ {
+ ScheduledExecutors.scheduledTasks
+ .scheduleWithFixedDelay(() -> checkCertFilesForHotReloading(
+ DatabaseDescriptor.getInternodeMessagingEncyptionOptions(),
+ DatabaseDescriptor.getNativeProtocolEncryptionOptions()),
+ DEFAULT_HOT_RELOAD_INITIAL_DELAY_SEC,
+ DEFAULT_HOT_RELOAD_PERIOD_SEC, TimeUnit.SECONDS);
+ }
+
+ isHotReloadingInitialized = true;
+ }
+
+
+ /**
+ * Sanity checks all certificates to ensure we can actually load them
+ */
+ public static void validateSslCerts(EncryptionOptions.ServerEncryptionOptions serverOpts, EncryptionOptions clientOpts) throws IOException
+ {
+ try
+ {
- // Ensure we're able to create both server & client SslContexts
- if (serverOpts != null && serverOpts.isEnabled())
++ // Ensure we're able to create both server & client SslContexts if they might ever be needed
++ if (serverOpts != null && serverOpts.tlsEncryptionPolicy() != EncryptionOptions.TlsEncryptionPolicy.UNENCRYPTED)
+ {
+ createNettySslContext(serverOpts, true, SocketType.SERVER, openSslIsAvailable());
+ createNettySslContext(serverOpts, true, SocketType.CLIENT, openSslIsAvailable());
+ }
+ }
+ catch (Exception e)
+ {
+ throw new IOException("Failed to create SSL context using server_encryption_options!", e);
+ }
+
+ try
+ {
- // Ensure we're able to create both server & client SslContexts
- if (clientOpts != null && clientOpts.isEnabled())
++ // Ensure we're able to create both server & client SslContexts if they might ever be needed
++ if (clientOpts != null && clientOpts.tlsEncryptionPolicy() != EncryptionOptions.TlsEncryptionPolicy.UNENCRYPTED)
+ {
+ createNettySslContext(clientOpts, clientOpts.require_client_auth, SocketType.SERVER, openSslIsAvailable());
+ createNettySslContext(clientOpts, clientOpts.require_client_auth, SocketType.CLIENT, openSslIsAvailable());
+ }
+ }
+ catch (Exception e)
+ {
+ throw new IOException("Failed to create SSL context using client_encryption_options!", e);
+ }
+ }
+
+ static class CacheKey
+ {
+ private final EncryptionOptions encryptionOptions;
+ private final SocketType socketType;
+ private final boolean useOpenSSL;
+
+ public CacheKey(EncryptionOptions encryptionOptions, SocketType socketType, boolean useOpenSSL)
+ {
+ this.encryptionOptions = encryptionOptions;
+ this.socketType = socketType;
+ this.useOpenSSL = useOpenSSL;
+ }
+
+ public boolean equals(Object o)
+ {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ CacheKey cacheKey = (CacheKey) o;
+ return (socketType == cacheKey.socketType &&
+ useOpenSSL == cacheKey.useOpenSSL &&
+ Objects.equals(encryptionOptions, cacheKey.encryptionOptions));
+ }
+
+ public int hashCode()
+ {
+ int result = 0;
+ result += 31 * socketType.hashCode();
+ result += 31 * encryptionOptions.hashCode();
+ result += 31 * Boolean.hashCode(useOpenSSL);
+ return result;
+ }
+ }
}
diff --cc src/java/org/apache/cassandra/service/NativeTransportService.java
index 6cdce3f,c58bb5e..f4f3585
--- a/src/java/org/apache/cassandra/service/NativeTransportService.java
+++ b/src/java/org/apache/cassandra/service/NativeTransportService.java
@@@ -32,11 -31,13 +32,13 @@@ import io.netty.channel.EventLoopGroup
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
-import io.netty.util.concurrent.EventExecutor;
++import io.netty.util.Version;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.metrics.AuthMetrics;
++import org.apache.cassandra.config.EncryptionOptions;
import org.apache.cassandra.metrics.ClientMetrics;
-import org.apache.cassandra.transport.ConfiguredLimit;
import org.apache.cassandra.transport.Message;
import org.apache.cassandra.transport.Server;
+import org.apache.cassandra.utils.NativeLibrary;
/**
* Handles native transport server lifecycle and associated resources. Lazily initialized.
@@@ -77,32 -81,36 +79,45 @@@ public class NativeTransportServic
org.apache.cassandra.transport.Server.Builder builder = new org.apache.cassandra.transport.Server.Builder()
.withEventLoopGroup(workerGroup)
- .withProtocolVersionLimit(protocolVersionLimit)
.withHost(nativeAddr);
- if (!DatabaseDescriptor.getNativeProtocolEncryptionOptions().isEnabled())
- if (!DatabaseDescriptor.getClientEncryptionOptions().enabled)
++ EncryptionOptions.TlsEncryptionPolicy encryptionPolicy = DatabaseDescriptor.getNativeProtocolEncryptionOptions().tlsEncryptionPolicy();
++ Server regularPortServer;
++ Server tlsPortServer = null;
++
++ // If an SSL port is separately supplied for the native transport, listen for unencrypted connections on the
++ // regular port, and encryption / optionally encrypted connections on the ssl port.
++ if (nativePort != nativePortSSL)
{
-- servers = Collections.singleton(builder.withSSL(false).withPort(nativePort).build());
++ regularPortServer = builder.withTlsEncryptionPolicy(EncryptionOptions.TlsEncryptionPolicy.UNENCRYPTED).withPort(nativePort).build();
++ switch(encryptionPolicy)
++ {
++ case OPTIONAL: // FALLTHRU - encryption is optional on the regular port, but encrypted on the tls port.
++ case ENCRYPTED:
++ tlsPortServer = builder.withTlsEncryptionPolicy(encryptionPolicy).withPort(nativePortSSL).build();
++ break;
++ case UNENCRYPTED: // Should have been caught by DatabaseDescriptor.applySimpleConfig
++ throw new IllegalStateException("Encryption must be enabled in client_encryption_options for native_transport_port_ssl");
++ default:
++ throw new IllegalStateException("Unrecognized TLS encryption policy: " + encryptionPolicy);
++ }
}
++ // Otherwise, if only the regular port is supplied, listen as the encryption policy specifies
else
{
-- if (nativePort != nativePortSSL)
-- {
-- // user asked for dedicated ssl port for supporting both non-ssl and ssl connections
-- servers = Collections.unmodifiableList(
-- Arrays.asList(
-- builder.withSSL(false).withPort(nativePort).build(),
-- builder.withSSL(true).withPort(nativePortSSL).build()
-- )
-- );
-- }
-- else
-- {
-- // ssl only mode using configured native port
-- servers = Collections.singleton(builder.withSSL(true).withPort(nativePort).build());
-- }
++ regularPortServer = builder.withTlsEncryptionPolicy(encryptionPolicy).withPort(nativePort).build();
+ }
+
- // register metrics
- ClientMetrics.instance.init(servers);
++ if (tlsPortServer == null)
++ {
++ servers = Collections.singleton(regularPortServer);
++ }
++ else
++ {
++ servers = Collections.unmodifiableList(Arrays.asList(regularPortServer, tlsPortServer));
+ }
- AuthMetrics.init();
+ ClientMetrics.instance.init(servers);
initialized = true;
}
@@@ -112,6 -120,6 +127,7 @@@
*/
public void start()
{
++ logger.info("Using Netty Version: {}", Version.identify().entrySet());
initialize();
servers.forEach(Server::start);
}
diff --cc src/java/org/apache/cassandra/tools/LoaderOptions.java
index 686c834,b3110f2..070ffa8
--- a/src/java/org/apache/cassandra/tools/LoaderOptions.java
+++ b/src/java/org/apache/cassandra/tools/LoaderOptions.java
@@@ -439,8 -383,8 +439,10 @@@ public class LoaderOption
else
sslStoragePort = config.ssl_storage_port;
throttle = config.stream_throughput_outbound_megabits_per_sec;
-- clientEncOptions = config.client_encryption_options;
++ // Copy the encryption options and apply the config so that argument parsing can accesss isEnabled.
++ clientEncOptions = config.client_encryption_options.applyConfig();
serverEncOptions = config.server_encryption_options;
++ serverEncOptions.applyConfig();
if (cmd.hasOption(THROTTLE_MBITS))
{
@@@ -472,54 -416,47 +474,62 @@@
if (cmd.hasOption(SSL_TRUSTSTORE))
{
- clientEncOptions.truststore = cmd.getOptionValue(SSL_TRUSTSTORE);
+ clientEncOptions = clientEncOptions.withTrustStore(cmd.getOptionValue(SSL_TRUSTSTORE));
++ clientEncOptions.applyConfig();
}
if (cmd.hasOption(SSL_TRUSTSTORE_PW))
{
- clientEncOptions.truststore_password = cmd.getOptionValue(SSL_TRUSTSTORE_PW);
+ clientEncOptions = clientEncOptions.withTrustStorePassword(cmd.getOptionValue(SSL_TRUSTSTORE_PW));
++ clientEncOptions.applyConfig();
}
if (cmd.hasOption(SSL_KEYSTORE))
{
- clientEncOptions.keystore = cmd.getOptionValue(SSL_KEYSTORE);
// if a keystore was provided, lets assume we'll need to use
- // it
- clientEncOptions.require_client_auth = true;
+ clientEncOptions = clientEncOptions.withKeyStore(cmd.getOptionValue(SSL_KEYSTORE))
+ .withRequireClientAuth(true);
++ clientEncOptions.applyConfig();
}
if (cmd.hasOption(SSL_KEYSTORE_PW))
{
- clientEncOptions.keystore_password = cmd.getOptionValue(SSL_KEYSTORE_PW);
+ clientEncOptions = clientEncOptions.withKeyStorePassword(cmd.getOptionValue(SSL_KEYSTORE_PW));
++ clientEncOptions.applyConfig();
}
if (cmd.hasOption(SSL_PROTOCOL))
{
- clientEncOptions.protocol = cmd.getOptionValue(SSL_PROTOCOL);
+ clientEncOptions = clientEncOptions.withProtocol(cmd.getOptionValue(SSL_PROTOCOL));
++ clientEncOptions.applyConfig();
}
if (cmd.hasOption(SSL_ALGORITHM))
{
- clientEncOptions.algorithm = cmd.getOptionValue(SSL_ALGORITHM);
+ clientEncOptions = clientEncOptions.withAlgorithm(cmd.getOptionValue(SSL_ALGORITHM));
++ clientEncOptions.applyConfig();
}
if (cmd.hasOption(SSL_STORE_TYPE))
{
- clientEncOptions.store_type = cmd.getOptionValue(SSL_STORE_TYPE);
+ clientEncOptions = clientEncOptions.withStoreType(cmd.getOptionValue(SSL_STORE_TYPE));
++ clientEncOptions.applyConfig();
}
if (cmd.hasOption(SSL_CIPHER_SUITES))
{
- clientEncOptions.cipher_suites = cmd.getOptionValue(SSL_CIPHER_SUITES).split(",");
+ clientEncOptions = clientEncOptions.withCipherSuites(cmd.getOptionValue(SSL_CIPHER_SUITES).split(","));
++ clientEncOptions.applyConfig();
}
+ if (cmd.hasOption(TARGET_KEYSPACE))
+ {
+ targetKeyspace = cmd.getOptionValue(TARGET_KEYSPACE);
+ if (StringUtils.isBlank(targetKeyspace))
+ {
+ errorMsg("Empty keyspace is not supported.", options);
+ }
+ }
return this;
}
catch (ParseException | ConfigurationException | MalformedURLException e)
diff --cc src/java/org/apache/cassandra/transport/Client.java
index ec86579,368b1d7..4f87cf4
--- a/src/java/org/apache/cassandra/transport/Client.java
+++ b/src/java/org/apache/cassandra/transport/Client.java
@@@ -48,9 -43,9 +48,9 @@@ public class Client extends SimpleClien
{
private final SimpleEventHandler eventHandler = new SimpleEventHandler();
- public Client(String host, int port, ProtocolVersion version, ClientEncryptionOptions encryptionOptions)
+ public Client(String host, int port, ProtocolVersion version, EncryptionOptions encryptionOptions)
{
- super(host, port, version, version.isBeta(), encryptionOptions);
- super(host, port, version, encryptionOptions);
++ super(host, port, version, version.isBeta(), new EncryptionOptions(encryptionOptions).applyConfig());
setEventHandler(eventHandler);
}
@@@ -292,9 -251,9 +292,9 @@@
// Parse options.
String host = args[0];
int port = Integer.parseInt(args[1]);
- ProtocolVersion version = args.length == 3 ? ProtocolVersion.decode(Integer.parseInt(args[2]), ProtocolVersionLimit.SERVER_DEFAULT) : ProtocolVersion.CURRENT;
+ ProtocolVersion version = args.length == 3 ? ProtocolVersion.decode(Integer.parseInt(args[2]), DatabaseDescriptor.getNativeTransportAllowOlderProtocols()) : ProtocolVersion.CURRENT;
- EncryptionOptions encryptionOptions = new EncryptionOptions();
- ClientEncryptionOptions encryptionOptions = new ClientEncryptionOptions();
++ EncryptionOptions encryptionOptions = new EncryptionOptions().applyConfig();
System.out.println("CQL binary protocol console " + host + "@" + port + " using native protocol version " + version);
new Client(host, port, version, encryptionOptions).run();
diff --cc src/java/org/apache/cassandra/transport/Server.java
index b1fad53,ced764f..52e6387
--- a/src/java/org/apache/cassandra/transport/Server.java
+++ b/src/java/org/apache/cassandra/transport/Server.java
@@@ -44,11 -42,8 +44,10 @@@ import io.netty.channel.group.DefaultCh
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslHandler;
-import io.netty.util.Version;
+import io.netty.handler.timeout.IdleStateEvent;
+import io.netty.handler.timeout.IdleStateHandler;
- import io.netty.util.Version;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.GlobalEventExecutor;
import io.netty.util.internal.logging.InternalLoggerFactory;
@@@ -88,7 -78,7 +87,7 @@@ public class Server implements Cassandr
};
public final InetSocketAddress socket;
-- public boolean useSSL = false;
++ public final EncryptionOptions.TlsEncryptionPolicy tlsEncryptionPolicy;
private final AtomicBoolean isRunning = new AtomicBoolean(false);
private EventLoopGroup workerGroup;
@@@ -96,7 -87,9 +95,7 @@@
private Server (Builder builder)
{
this.socket = builder.getSocket();
-- this.useSSL = builder.useSSL;
- this.protocolVersionLimit = builder.getProtocolVersionLimit();
-
++ this.tlsEncryptionPolicy = builder.tlsEncryptionPolicy;
if (builder.workerGroup != null)
{
workerGroup = builder.workerGroup;
@@@ -141,29 -134,29 +140,27 @@@
if (workerGroup != null)
bootstrap = bootstrap.group(workerGroup);
-- if (this.useSSL)
-- {
- final EncryptionOptions clientEnc = DatabaseDescriptor.getNativeProtocolEncryptionOptions();
- final EncryptionOptions.ClientEncryptionOptions clientEnc = DatabaseDescriptor.getClientEncryptionOptions();
++ final EncryptionOptions clientEnc = DatabaseDescriptor.getNativeProtocolEncryptionOptions();
-- if (clientEnc.optional)
-- {
-- logger.info("Enabling optionally encrypted CQL connections between client and server");
++ switch (this.tlsEncryptionPolicy)
++ {
++ case UNENCRYPTED:
++ bootstrap.childHandler(new Initializer(this));
++ break;
++ case OPTIONAL:
++ logger.debug("Enabling optionally encrypted CQL connections between client and server");
bootstrap.childHandler(new OptionalSecureInitializer(this, clientEnc));
-- }
-- else
-- {
-- logger.info("Enabling encrypted CQL connections between client and server");
++ break;
++ case ENCRYPTED:
++ logger.debug("Enabling encrypted CQL connections between client and server");
bootstrap.childHandler(new SecureInitializer(this, clientEnc));
-- }
-- }
-- else
-- {
-- bootstrap.childHandler(new Initializer(this));
++ break;
++ default:
++ throw new IllegalStateException("Unrecognized TLS encryption policy: " + this.tlsEncryptionPolicy);
}
// Bind and start to accept incoming connections.
-- logger.info("Using Netty Version: {}", Version.identify().entrySet());
-- logger.info("Starting listening for CQL clients on {} ({})...", socket, this.useSSL ? "encrypted" : "unencrypted");
++ logger.info("Starting listening for CQL clients on {} ({})...", socket, clientEnc.tlsEncryptionPolicy().description());
ChannelFuture bindFuture = bootstrap.bind(socket);
if (!bindFuture.awaitUninterruptibly().isSuccess())
@@@ -219,14 -183,15 +216,14 @@@
{
private EventLoopGroup workerGroup;
private EventExecutor eventExecutorGroup;
-- private boolean useSSL = false;
++ private EncryptionOptions.TlsEncryptionPolicy tlsEncryptionPolicy = EncryptionOptions.TlsEncryptionPolicy.UNENCRYPTED;
private InetAddress hostAddr;
private int port = -1;
private InetSocketAddress socket;
- private ProtocolVersionLimit versionLimit;
-- public Builder withSSL(boolean useSSL)
++ public Builder withTlsEncryptionPolicy(EncryptionOptions.TlsEncryptionPolicy tlsEncryptionPolicy)
{
-- this.useSSL = useSSL;
++ this.tlsEncryptionPolicy = tlsEncryptionPolicy;
return this;
}
diff --cc src/java/org/apache/cassandra/transport/SimpleClient.java
index ef3c1db,1e6ea64..b4e9373
--- a/src/java/org/apache/cassandra/transport/SimpleClient.java
+++ b/src/java/org/apache/cassandra/transport/SimpleClient.java
@@@ -114,7 -113,7 +114,7 @@@ public class SimpleClient implements Cl
throw new IllegalArgumentException(String.format("Beta version of server used (%s), but USE_BETA flag is not set", version));
this.version = version;
-- this.encryptionOptions = encryptionOptions;
++ this.encryptionOptions = new EncryptionOptions(encryptionOptions).applyConfig();
}
public SimpleClient(String host, int port)
diff --cc test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
index e243451,2ded71c..bf615cd
--- a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
@@@ -190,8 -186,8 +192,7 @@@ public class InstanceConfig implements
{
if (value == null)
value = NULL;
--
- params.put(fieldName, value);
+ getParams(fieldName).put(fieldName, value);
return this;
}
diff --cc test/distributed/org/apache/cassandra/distributed/test/AbstractEncryptionOptionsImpl.java
index 0000000,0000000..6c48299
new file mode 100644
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/AbstractEncryptionOptionsImpl.java
@@@ -1,0 -1,0 +1,295 @@@
++/*
++ * 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.distributed.test;
++
++import java.util.List;
++import java.util.Map;
++import java.util.concurrent.TimeUnit;
++import java.util.concurrent.atomic.AtomicInteger;
++import java.util.concurrent.atomic.AtomicReference;
++
++import com.google.common.collect.ImmutableMap;
++import org.junit.Assert;
++import org.slf4j.Logger;
++import org.slf4j.LoggerFactory;
++
++import io.netty.bootstrap.Bootstrap;
++import io.netty.buffer.ByteBuf;
++import io.netty.channel.Channel;
++import io.netty.channel.ChannelFuture;
++import io.netty.channel.ChannelHandler;
++import io.netty.channel.ChannelHandlerContext;
++import io.netty.channel.ChannelInitializer;
++import io.netty.channel.EventLoopGroup;
++import io.netty.channel.nio.NioEventLoopGroup;
++import io.netty.channel.socket.nio.NioSocketChannel;
++import io.netty.handler.codec.ByteToMessageDecoder;
++import io.netty.handler.ssl.SslContext;
++import io.netty.handler.ssl.SslHandler;
++import io.netty.util.concurrent.FutureListener;
++import org.apache.cassandra.config.EncryptionOptions;
++import org.apache.cassandra.distributed.Cluster;
++import org.apache.cassandra.exceptions.ConfigurationException;
++import org.apache.cassandra.security.SSLFactory;
++import org.apache.cassandra.utils.concurrent.SimpleCondition;
++
++public class AbstractEncryptionOptionsImpl extends TestBaseImpl
++{
++ Logger logger = LoggerFactory.getLogger(EncryptionOptions.class);
++ final static String validKeyStorePath = "test/conf/cassandra_ssl_test.keystore";
++ final static String validKeyStorePassword = "cassandra";
++ final static String validTrustStorePath = "test/conf/cassandra_ssl_test.truststore";
++ final static String validTrustStorePassword = "cassandra";
++ final static Map<String,Object> validKeystore = ImmutableMap.of("keystore", validKeyStorePath,
++ "keystore_password", validKeyStorePassword,
++ "truststore", validTrustStorePath,
++ "truststore_password", validTrustStorePassword);
++
++ // Result of a TlsConnection.connect call. The result is updated as the TLS connection
++ // sequence takes place. The nextOnFailure/nextOnSuccess allows the discard handler
++ // to correctly update state if an unexpected exception is thrown.
++ public enum ConnectResult {
++ UNINITIALIZED,
++ FAILED_TO_NEGOTIATE,
++ NEVER_CONNECTED,
++ NEGOTIATED,
++ CONNECTED_AND_ABOUT_TO_NEGOTIATE(FAILED_TO_NEGOTIATE, NEGOTIATED),
++ CONNECTING(NEVER_CONNECTED, CONNECTED_AND_ABOUT_TO_NEGOTIATE);
++
++ public final ConnectResult nextOnFailure;
++ public final ConnectResult nextOnSuccess;
++
++ ConnectResult()
++ {
++ nextOnFailure = null;
++ nextOnSuccess = null;
++ }
++ ConnectResult(ConnectResult nextOnFailure, ConnectResult nextOnSuccess)
++ {
++ this.nextOnFailure = nextOnFailure;
++ this.nextOnSuccess = nextOnSuccess;
++ }
++ }
++
++ public class TlsConnection
++ {
++ final String host;
++ final int port;
++ final EncryptionOptions encryptionOptions = new EncryptionOptions()
++ .withEnabled(true)
++ .withKeyStore(validKeyStorePath).withKeyStorePassword(validKeyStorePassword)
++ .withTrustStore(validTrustStorePath).withTrustStorePassword(validTrustStorePassword);
++ private Throwable lastThrowable;
++
++ public TlsConnection(String host, int port)
++ {
++ this.host = host;
++ this.port = port;
++ }
++
++ public synchronized Throwable lastThrowable()
++ {
++ return lastThrowable;
++ }
++ private synchronized void setLastThrowable(Throwable cause)
++ {
++ lastThrowable = cause;
++ }
++
++ final AtomicReference<ConnectResult> result = new AtomicReference<>(ConnectResult.UNINITIALIZED);
++
++ void setResult(String why, ConnectResult expected, ConnectResult newResult)
++ {
++ if (newResult == null)
++ return;
++ logger.debug("Setting progress from {} to {}", expected, expected.nextOnSuccess);
++ result.getAndUpdate(v -> {
++ if (v == expected)
++ return newResult;
++ else
++ throw new IllegalStateException(
++ String.format("CAS attempt on %s failed from %s to %s but %s did not match expected value",
++ why, expected, newResult, v));
++ });
++ }
++ void successProgress()
++ {
++ ConnectResult current = result.get();
++ setResult("success", current, current.nextOnSuccess);
++ }
++ void failure()
++ {
++ ConnectResult current = result.get();
++ setResult("failure", current, current.nextOnFailure);
++ }
++
++ ConnectResult connect() throws Throwable
++ {
++ AtomicInteger connectAttempts = new AtomicInteger(0);
++ result.set(ConnectResult.UNINITIALIZED);
++ setLastThrowable(null);
++
++ SslContext sslContext = SSLFactory.getOrCreateSslContext(encryptionOptions, true,
++ SSLFactory.SocketType.CLIENT);
++
++ EventLoopGroup workerGroup = new NioEventLoopGroup();
++ Bootstrap b = new Bootstrap();
++ SimpleCondition attemptCompleted = new SimpleCondition();
++
++ // Listener on the SSL handshake makes sure that the test completes immediately as
++ // the server waits to receive a message over the TLS connection, so the discardHandler.decode
++ // will likely never be called. The lambda has to handle it's own exceptions as it's a listener,
++ // not in the request pipeline to pass them on to discardHandler.
++ FutureListener<Channel> handshakeResult = channelFuture -> {
++ try
++ {
++ logger.debug("handshakeFuture() listener called");
++ channelFuture.get();
++ successProgress();
++ }
++ catch (Throwable cause)
++ {
++ logger.info("handshakeFuture() threw", cause);
++ failure();
++ setLastThrowable(cause);
++ }
++ attemptCompleted.signalAll();
++ };
++
++ ChannelHandler connectHandler = new ByteToMessageDecoder()
++ {
++ @Override
++ public void channelActive(ChannelHandlerContext ctx) throws Exception
++ {
++ logger.debug("connectHandler.channelActive");
++ int count = connectAttempts.incrementAndGet();
++ if (count > 1)
++ {
++ logger.info("connectHandler.channelActive called more than once - {}", count);
++ }
++ successProgress();
++
++ // Add the handler after the connection is established to make sure the connection
++ // progress is recorded
++ final SslHandler sslHandler = ctx.pipeline().get(SslHandler.class);
++ sslHandler.handshakeFuture().addListener(handshakeResult);
++
++ super.channelActive(ctx);
++ }
++
++ @Override
++ public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
++ {
++ logger.debug("connectHandler.decode - readable bytes {}", in.readableBytes());
++
++ ctx.pipeline().remove(this);
++ }
++
++ @Override
++ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
++ {
++ logger.debug("connectHandler.exceptionCaught", cause);
++ setLastThrowable(cause);
++ failure();
++ attemptCompleted.signalAll();
++ }
++ };
++ ChannelHandler discardHandler = new ByteToMessageDecoder()
++ {
++ @Override
++ public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
++ {
++ logger.info("discardHandler.decode - {} readable bytes made it past SSL negotiation, discarding.",
++ in.readableBytes());
++ in.readBytes(in.readableBytes());
++ attemptCompleted.signalAll();
++ }
++
++ @Override
++ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
++ {
++ logger.debug("discardHandler.exceptionCaught", cause);
++ setLastThrowable(cause);
++ failure();
++ attemptCompleted.signalAll();
++ }
++ };
++
++ b.group(workerGroup);
++ b.channel(NioSocketChannel.class);
++ b.handler(new ChannelInitializer<Channel>()
++ {
++ @Override
++ protected void initChannel(Channel channel)
++ {
++ SslHandler sslHandler = sslContext.newHandler(channel.alloc());
++ channel.pipeline().addFirst(connectHandler, sslHandler, discardHandler);
++ }
++ });
++
++ result.set(ConnectResult.CONNECTING);
++ ChannelFuture f = b.connect(host, port);
++ try
++ {
++ f.sync();
++ attemptCompleted.await(15, TimeUnit.SECONDS);
++ }
++ finally
++ {
++ f.channel().close();
++ }
++ return result.get();
++ }
++
++ void assertCannotConnect() throws Throwable
++ {
++ try
++ {
++ connect();
++ }
++ catch (java.net.ConnectException ex)
++ {
++ // verify it was not possible to connect before starting the server
++ }
++ }
++ }
++
++ /* Provde the cluster cannot start with the configured options */
++ void assertCannotStartDueToConfigurationException(Cluster cluster)
++ {
++ Throwable tr = null;
++ try
++ {
++ cluster.startup();
++ }
++ catch (Throwable maybeConfigException)
++ {
++ tr = maybeConfigException;
++ }
++
++ if (tr == null)
++ {
++ Assert.fail("Expected a ConfigurationException");
++ }
++ else
++ {
++ Assert.assertEquals(ConfigurationException.class.getName(), tr.getClass().getName());
++ }
++ }
++}
diff --cc test/distributed/org/apache/cassandra/distributed/test/IncRepairTruncationTest.java
index bd47906,0000000..4ffdaa9
mode 100644,000000..100644
--- a/test/distributed/org/apache/cassandra/distributed/test/IncRepairTruncationTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/IncRepairTruncationTest.java
@@@ -1,156 -1,0 +1,155 @@@
+/*
+ * 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.distributed.test;
+
+import java.io.IOException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.util.concurrent.Uninterruptibles;
+import org.junit.Test;
+
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.IMessage;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.net.Verb;
+import org.apache.cassandra.utils.concurrent.SimpleCondition;
+
+import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
+import static org.apache.cassandra.distributed.api.Feature.NETWORK;
+import static org.apache.cassandra.distributed.test.PreviewRepairTest.insert;
+
+public class IncRepairTruncationTest extends TestBaseImpl
+{
+ @Test
+ public void testTruncateDuringIncRepair() throws IOException, InterruptedException, ExecutionException
+ {
+ ExecutorService es = Executors.newFixedThreadPool(3);
+ try(Cluster cluster = init(Cluster.build(2)
- .withConfig(config -> config.set("disable_incremental_repair", false)
- .with(GOSSIP)
++ .withConfig(config -> config.with(GOSSIP)
+ .with(NETWORK))
+ .start()))
+ {
+ cluster.schemaChange("create table " + KEYSPACE + ".tbl (id int primary key, t int)");
+
+ insert(cluster.coordinator(1), 0, 100);
+ cluster.forEach((node) -> node.flush(KEYSPACE));
+ // mark everything repaired
+ cluster.get(1).nodetoolResult("repair", KEYSPACE, "tbl").asserts().success();
+
+ /*
+ make sure we are out-of-sync to make node2 stream data to node1:
+ */
+ cluster.get(2).executeInternal("insert into "+KEYSPACE+".tbl (id, t) values (5, 5)");
+ cluster.get(2).flush(KEYSPACE);
+ /*
+ start repair:
+ block streaming from 2 -> 1 until truncation below has executed
+ */
+ BlockMessage node2Streaming = new BlockMessage();
+ cluster.filters().inbound().verbs(Verb.VALIDATION_RSP.id).from(2).to(1).messagesMatching(node2Streaming).drop();
+
+ /*
+ block truncation on node2:
+ */
+ BlockMessage node2Truncation = new BlockMessage();
+ cluster.filters().inbound().verbs(Verb.TRUNCATE_REQ.id).from(1).to(2).messagesMatching(node2Truncation).drop();
+
+ Future<NodeToolResult> repairResult = es.submit(() -> cluster.get(1).nodetoolResult("repair", KEYSPACE, "tbl"));
+
+ Future<?> truncationFuture = es.submit(() -> {
+ try
+ {
+ /*
+ wait for streaming message to sent before truncating, to make sure we have a mismatch to make us stream later
+ */
+ node2Streaming.gotMessage.await();
+ }
+ catch (InterruptedException e)
+ {
+ throw new RuntimeException(e);
+ }
+ cluster.coordinator(1).execute("TRUNCATE "+KEYSPACE+".tbl", ConsistencyLevel.ALL);
+ });
+
+ node2Truncation.gotMessage.await();
+ // make sure node1 finishes truncation, removing its files
+ cluster.get(1).runOnInstance(() -> {
+ ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl");
+ while (!cfs.getLiveSSTables().isEmpty())
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+ });
+
+ /* let repair finish, streaming files from 2 -> 1 */
+ node2Streaming.allowMessage.signalAll();
+
+ /* and the repair should fail: */
+ repairResult.get().asserts().failure();
+
+ /*
+ and let truncation finish on node2
+ */
+ node2Truncation.allowMessage.signalAll();
+ truncationFuture.get();
+
+ /* wait for truncation to remove files on node2 */
+ cluster.get(2).runOnInstance(() -> {
+ ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl");
+ while (!cfs.getLiveSSTables().isEmpty())
+ {
+ System.out.println(cfs.getLiveSSTables());
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+ }
+ });
+
+ cluster.get(1).nodetoolResult("repair", "-vd", KEYSPACE, "tbl").asserts().success().notificationContains("Repair preview completed successfully");
+ }
+ finally
+ {
+ es.shutdown();
+ }
+ }
+
+ private static class BlockMessage implements IMessageFilters.Matcher
+ {
+ private final SimpleCondition gotMessage = new SimpleCondition();
+ private final SimpleCondition allowMessage = new SimpleCondition();
+
+ public boolean matches(int from, int to, IMessage message)
+ {
+ gotMessage.signalAll();
+ try
+ {
+ allowMessage.await();
+ }
+ catch (InterruptedException e)
+ {
+ throw new RuntimeException(e);
+ }
+ return false;
+ }
+ }
+}
diff --cc test/distributed/org/apache/cassandra/distributed/test/InternodeEncryptionOptionsTest.java
index 0000000,0000000..4b05ade
new file mode 100644
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/InternodeEncryptionOptionsTest.java
@@@ -1,0 -1,0 +1,218 @@@
++/*
++ * 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.distributed.test;
++
++import java.net.InetAddress;
++
++import com.google.common.collect.ImmutableMap;
++import org.junit.Assert;
++import org.junit.Test;
++
++import org.apache.cassandra.distributed.Cluster;
++import org.apache.cassandra.distributed.api.Feature;
++
++public class InternodeEncryptionOptionsTest extends AbstractEncryptionOptionsImpl
++{
++ @Test
++ public void nodeWillNotStartWithBadKeystoreTest() throws Throwable
++ {
++ try (Cluster cluster = builder().withNodes(1).withConfig(c -> {
++ c.with(Feature.NETWORK);
++ c.set("server_encryption_options",
++ ImmutableMap.of("optional", true,
++ "keystore", "/path/to/bad/keystore/that/should/not/exist",
++ "truststore", "/path/to/bad/truststore/that/should/not/exist"));
++ }).createWithoutStarting())
++ {
++ assertCannotStartDueToConfigurationException(cluster);
++ }
++ }
++
++ @Test
++ public void legacySslPortProvidedWithEncryptionNoneWillNotStartTest() throws Throwable
++ {
++ try (Cluster cluster = builder().withNodes(1).withConfig(c -> {
++ c.with(Feature.NETWORK);
++ c.set("ssl_storage_port", 7013);
++ c.set("server_encryption_options",
++ ImmutableMap.builder().putAll(validKeystore)
++ .put("internode_encryption", "none")
++ .put("optional", false)
++ .put("enable_legacy_ssl_storage_port", "true")
++ .build());
++ }).createWithoutStarting())
++ {
++ assertCannotStartDueToConfigurationException(cluster);
++ }
++ }
++
++ @Test
++ public void optionalTlsConnectionDisabledWithoutKeystoreTest() throws Throwable
++ {
++ try (Cluster cluster = builder().withNodes(1).withConfig(c -> c.with(Feature.NETWORK)).createWithoutStarting())
++ {
++ InetAddress address = cluster.get(1).config().broadcastAddress().getAddress();
++ int port = cluster.get(1).config().broadcastAddress().getPort();
++
++ TlsConnection tlsConnection = new TlsConnection(address.getHostAddress(), port);
++ tlsConnection.assertCannotConnect();
++
++ cluster.startup();
++
++ Assert.assertEquals("TLS connection should not be possible without keystore",
++ ConnectResult.FAILED_TO_NEGOTIATE, tlsConnection.connect());
++ }
++ }
++
++ @Test
++ public void optionalTlsConnectionAllowedWithKeystoreTest() throws Throwable
++ {
++ try (Cluster cluster = builder().withNodes(1).withConfig(c -> {
++ c.with(Feature.NETWORK);
++ c.set("server_encryption_options", validKeystore);
++ }).createWithoutStarting())
++ {
++ InetAddress address = cluster.get(1).config().broadcastAddress().getAddress();
++ int port = cluster.get(1).config().broadcastAddress().getPort();
++
++ TlsConnection tlsConnection = new TlsConnection(address.getHostAddress(), port);
++ tlsConnection.assertCannotConnect();
++
++ cluster.startup();
++
++ Assert.assertEquals("TLS connection should be possible with keystore by default",
++ ConnectResult.NEGOTIATED, tlsConnection.connect());
++ }
++ }
++
++ @Test
++ public void optionalTlsConnectionAllowedToStoragePortTest() throws Throwable
++ {
++ try (Cluster cluster = builder().withNodes(1).withConfig(c -> {
++ c.with(Feature.NETWORK);
++ c.set("storage_port", 7012);
++ c.set("ssl_storage_port", 7013);
++ c.set("server_encryption_options",
++ ImmutableMap.builder().putAll(validKeystore)
++ .put("internode_encryption", "none")
++ .put("optional", true)
++ .put("enable_legacy_ssl_storage_port", "true")
++ .build());
++ }).createWithoutStarting())
++ {
++ InetAddress address = cluster.get(1).config().broadcastAddress().getAddress();
++ int regular_port = (int) cluster.get(1).config().get("storage_port");
++ int ssl_port = (int) cluster.get(1).config().get("ssl_storage_port");
++
++ // Create the connections and prove they cannot connect before server start
++ TlsConnection connectToRegularPort = new TlsConnection(address.getHostAddress(), regular_port);
++ connectToRegularPort.assertCannotConnect();
++
++ TlsConnection connectToSslStoragePort = new TlsConnection(address.getHostAddress(), ssl_port);
++ connectToSslStoragePort.assertCannotConnect();
++
++ cluster.startup();
++
++ Assert.assertEquals("TLS native connection should be possible to ssl_storage_port",
++ ConnectResult.NEGOTIATED, connectToSslStoragePort.connect());
++ Assert.assertEquals("TLS native connection should be possible with valid keystore by default",
++ ConnectResult.NEGOTIATED, connectToRegularPort.connect());
++ }
++ }
++
++ @Test
++ public void legacySslStoragePortEnabledWithSameRegularAndSslPortTest() throws Throwable
++ {
++ try (Cluster cluster = builder().withNodes(1).withConfig(c -> {
++ c.with(Feature.NETWORK);
++ c.set("storage_port", 7012); // must match in-jvm dtest assigned ports
++ c.set("ssl_storage_port", 7012);
++ c.set("server_encryption_options",
++ ImmutableMap.builder().putAll(validKeystore)
++ .put("internode_encryption", "none")
++ .put("optional", true)
++ .put("enable_legacy_ssl_storage_port", "true")
++ .build());
++ }).createWithoutStarting())
++ {
++ InetAddress address = cluster.get(1).config().broadcastAddress().getAddress();
++ int ssl_port = (int) cluster.get(1).config().get("ssl_storage_port");
++
++ // Create the connections and prove they cannot connect before server start
++ TlsConnection connectToSslStoragePort = new TlsConnection(address.getHostAddress(), ssl_port);
++ connectToSslStoragePort.assertCannotConnect();
++
++ cluster.startup();
++
++ Assert.assertEquals("TLS native connection should be possible to ssl_storage_port",
++ ConnectResult.NEGOTIATED, connectToSslStoragePort.connect());
++ }
++ }
++
++
++ @Test
++ public void tlsConnectionRejectedWhenUnencrypted() throws Throwable
++ {
++ try (Cluster cluster = builder().withNodes(1).withConfig(c -> {
++ c.with(Feature.NETWORK);
++ c.set("server_encryption_options",
++ ImmutableMap.builder().putAll(validKeystore)
++ .put("internode_encryption", "none")
++ .put("optional", false)
++ .build());
++ }).createWithoutStarting())
++ {
++ InetAddress address = cluster.get(1).config().broadcastAddress().getAddress();
++ int regular_port = (int) cluster.get(1).config().get("storage_port");
++
++ // Create the connections and prove they cannot connect before server start
++ TlsConnection connection = new TlsConnection(address.getHostAddress(), regular_port);
++ connection.assertCannotConnect();
++
++ cluster.startup();
++
++ Assert.assertEquals("TLS native connection should be possible with valid keystore by default",
++ ConnectResult.FAILED_TO_NEGOTIATE, connection.connect());
++ }
++ }
++
++ @Test
++ public void allInternodeEncryptionEstablishedTest() throws Throwable
++ {
++ try (Cluster cluster = builder().withNodes(2).withConfig(c -> {
++ c.with(Feature.NETWORK)
++ .with(Feature.GOSSIP) // To make sure AllMembersAliveMonitor checks gossip (which uses internode conns)
++ .with(Feature.NATIVE_PROTOCOL); // For virtual tables
++ c.set("server_encryption_options",
++ ImmutableMap.builder().putAll(validKeystore)
++ .put("internode_encryption", "all")
++ .build());
++ }).start())
++ {
++ // Just check startup - cluster should not be able to establish internode connections xwithout encrypted connections
++ for (int i = 1; i <= cluster.size(); i++)
++ {
++ Object[][] result = cluster.get(i).executeInternal("SELECT successful_connection_attempts, address, port FROM system_views.internode_outbound");
++ Assert.assertEquals(1, result.length);
++ long successfulConnectionAttempts = (long) result[0][0];
++ Assert.assertTrue("At least one connection: " + successfulConnectionAttempts, successfulConnectionAttempts > 0);
++ }
++ }
++ }
++}
diff --cc test/distributed/org/apache/cassandra/distributed/test/NativeTransportEncryptionOptionsTest.java
index 0000000,0000000..d64b038
new file mode 100644
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/NativeTransportEncryptionOptionsTest.java
@@@ -1,0 -1,0 +1,137 @@@
++/*
++ * 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.distributed.test;
++
++import java.net.InetAddress;
++
++import com.google.common.collect.ImmutableMap;
++import org.junit.Assert;
++import org.junit.Test;
++
++import org.apache.cassandra.distributed.Cluster;
++import org.apache.cassandra.distributed.api.Feature;
++
++public class NativeTransportEncryptionOptionsTest extends AbstractEncryptionOptionsImpl
++{
++ @Test
++ public void nodeWillNotStartWithBadKeystore() throws Throwable
++ {
++ try (Cluster cluster = builder().withNodes(1).withConfig(c -> {
++ c.with(Feature.NATIVE_PROTOCOL);
++ c.set("client_encryption_options",
++ ImmutableMap.of("enabled", true,
++ "optional", true,
++ "keystore", "/path/to/bad/keystore/that/should/not/exist",
++ "truststore", "/path/to/bad/truststore/that/should/not/exist"));
++ }).createWithoutStarting())
++ {
++ assertCannotStartDueToConfigurationException(cluster);
++ }
++ }
++
++ @Test
++ public void optionalTlsConnectionDisabledWithoutKeystoreTest() throws Throwable
++ {
++ try (Cluster cluster = builder().withNodes(1).withConfig(c -> c.with(Feature.NATIVE_PROTOCOL)).createWithoutStarting())
++ {
++ InetAddress address = cluster.get(1).config().broadcastAddress().getAddress();
++ int port = (int) cluster.get(1).config().get("native_transport_port");
++
++ TlsConnection tlsConnection = new TlsConnection(address.getHostAddress(), port);
++ tlsConnection.assertCannotConnect();
++
++ cluster.startup();
++
++ Assert.assertEquals("TLS connection should not be possible without keystore",
++ ConnectResult.FAILED_TO_NEGOTIATE, tlsConnection.connect());
++ }
++ }
++
++
++ @Test
++ public void optionalTlsConnectionAllowedWithKeystoreTest() throws Throwable
++ {
++ try (Cluster cluster = builder().withNodes(1).withConfig(c -> {
++ c.with(Feature.NATIVE_PROTOCOL);
++ c.set("client_encryption_options", validKeystore);
++ }).createWithoutStarting())
++ {
++ InetAddress address = cluster.get(1).config().broadcastAddress().getAddress();
++ int port = (int) cluster.get(1).config().get("native_transport_port");
++
++ TlsConnection tlsConnection = new TlsConnection(address.getHostAddress(), port);
++ tlsConnection.assertCannotConnect();
++
++ cluster.startup();
++
++ Assert.assertEquals("TLS native connection should be possible with keystore by default",
++ ConnectResult.NEGOTIATED, tlsConnection.connect());
++ }
++ }
++
++ @Test
++ public void optionalTlsConnectionAllowedToRegularPortTest() throws Throwable
++ {
++ try (Cluster cluster = builder().withNodes(1).withConfig(c -> {
++ c.with(Feature.NATIVE_PROTOCOL);
++ c.set("native_transport_port_ssl", 9043);
++ c.set("client_encryption_options",
++ ImmutableMap.builder().putAll(validKeystore)
++ .put("enabled", false)
++ .put("optional", true)
++ .build());
++ }).createWithoutStarting())
++ {
++ InetAddress address = cluster.get(1).config().broadcastAddress().getAddress();
++ int unencrypted_port = (int) cluster.get(1).config().get("native_transport_port");
++ int ssl_port = (int) cluster.get(1).config().get("native_transport_port_ssl");
++
++ // Create the connections and prove they cannot connect before server start
++ TlsConnection connectionToUnencryptedPort = new TlsConnection(address.getHostAddress(), unencrypted_port);
++ connectionToUnencryptedPort.assertCannotConnect();
++
++ TlsConnection connectionToEncryptedPort = new TlsConnection(address.getHostAddress(), ssl_port);
++ connectionToEncryptedPort.assertCannotConnect();
++
++ cluster.startup();
++
++ Assert.assertEquals("TLS native connection should be possible to native_transport_port_ssl",
++ ConnectResult.NEGOTIATED, connectionToEncryptedPort.connect());
++ Assert.assertEquals("TLS native connection should not be possible on the regular port if an SSL port is specified",
++ ConnectResult.FAILED_TO_NEGOTIATE, connectionToUnencryptedPort.connect()); // but did connect
++ }
++ }
++
++ @Test
++ public void unencryptedNativeConnectionNotlisteningOnTlsPortTest() throws Throwable
++ {
++ try (Cluster cluster = builder().withNodes(1).withConfig(c -> {
++ c.with(Feature.NATIVE_PROTOCOL);
++ c.set("native_transport_port_ssl", 9043);
++ c.set("client_encryption_options",
++ ImmutableMap.builder().putAll(validKeystore)
++ .put("enabled", false)
++ .put("optional", false)
++ .build());
++ }).createWithoutStarting())
++ {
++ assertCannotStartDueToConfigurationException(cluster);
++ }
++ }
++}
diff --cc test/distributed/org/apache/cassandra/distributed/test/PreviewRepairTest.java
index 87837f2,0000000..bc9eda7
mode 100644,000000..100644
--- a/test/distributed/org/apache/cassandra/distributed/test/PreviewRepairTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/PreviewRepairTest.java
@@@ -1,500 -1,0 +1,499 @@@
+/*
+ * 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.distributed.test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.Uninterruptibles;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.compaction.CompactionManager;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.ICoordinator;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IIsolatedExecutor;
+import org.apache.cassandra.distributed.api.IMessage;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.impl.Instance;
+import org.apache.cassandra.distributed.shared.RepairResult;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.net.Verb;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.repair.RepairParallelism;
+import org.apache.cassandra.repair.messages.FinalizePropose;
+import org.apache.cassandra.repair.messages.RepairMessage;
+import org.apache.cassandra.repair.messages.RepairOption;
+import org.apache.cassandra.repair.messages.ValidationRequest;
+import org.apache.cassandra.service.ActiveRepairService;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.streaming.PreviewKind;
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.concurrent.SimpleCondition;
+import org.apache.cassandra.utils.progress.ProgressEventType;
+
+import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
+import static org.apache.cassandra.distributed.api.Feature.NETWORK;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class PreviewRepairTest extends TestBaseImpl
+{
+ @BeforeClass
+ public static void setup()
+ {
+ DatabaseDescriptor.daemonInitialization();
+ }
+
+ /**
+ * makes sure that the repaired sstables are not matching on the two
+ * nodes by disabling autocompaction on node2 and then running an
+ * incremental repair
+ */
+ @Test
+ public void testWithMismatchingPending() throws Throwable
+ {
+ try(Cluster cluster = init(Cluster.build(2).withConfig(config -> config.with(GOSSIP).with(NETWORK)).start()))
+ {
+ cluster.schemaChange("create table " + KEYSPACE + ".tbl (id int primary key, t int)");
+ insert(cluster.coordinator(1), 0, 100);
+ cluster.forEach((node) -> node.flush(KEYSPACE));
+ cluster.get(1).callOnInstance(repair(options(false, false)));
+ insert(cluster.coordinator(1), 100, 100);
+ cluster.forEach((node) -> node.flush(KEYSPACE));
+
+ // make sure that all sstables have moved to repaired by triggering a compaction
+ // also disables autocompaction on the nodes
+ cluster.forEach((node) -> node.runOnInstance(() -> {
+ ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl");
+ FBUtilities.waitOnFutures(CompactionManager.instance.submitBackground(cfs));
+ cfs.disableAutoCompaction();
+ }));
+ cluster.get(1).callOnInstance(repair(options(false, false)));
+ // now re-enable autocompaction on node1, this moves the sstables for the new repair to repaired
+ cluster.get(1).runOnInstance(() -> {
+ ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl");
+ cfs.enableAutoCompaction();
+ FBUtilities.waitOnFutures(CompactionManager.instance.submitBackground(cfs));
+ });
+
+ //IR and Preview repair can't run concurrently. In case the test is flaky, please check CASSANDRA-15685
+ Thread.sleep(1000);
+
+ RepairResult rs = cluster.get(1).callOnInstance(repair(options(true, false)));
+ assertTrue(rs.success); // preview repair should succeed
+ assertFalse(rs.wasInconsistent); // and we should see no mismatches
+ }
+ }
+
+ /**
+ * another case where the repaired datasets could mismatch is if an incremental repair finishes just as the preview
+ * repair is starting up.
+ *
+ * This tests this case:
+ * 1. we start a preview repair
+ * 2. pause the validation requests from node1 -> node2
+ * 3. node1 starts its validation
+ * 4. run an incremental repair which completes fine
+ * 5. node2 resumes its validation
+ *
+ * Now we will include sstables from the second incremental repair on node2 but not on node1
+ * This should fail since we fail any preview repair which is ongoing when an incremental repair finishes (step 4 above)
+ */
+ @Test
+ public void testFinishingIncRepairDuringPreview() throws IOException, InterruptedException, ExecutionException
+ {
+ ExecutorService es = Executors.newSingleThreadExecutor();
+ try(Cluster cluster = init(Cluster.build(2).withConfig(config -> config.with(GOSSIP).with(NETWORK)).start()))
+ {
+ cluster.schemaChange("create table " + KEYSPACE + ".tbl (id int primary key, t int)");
+
+ insert(cluster.coordinator(1), 0, 100);
+ cluster.forEach((node) -> node.flush(KEYSPACE));
+ cluster.get(1).callOnInstance(repair(options(false, false)));
+
+ insert(cluster.coordinator(1), 100, 100);
+ cluster.forEach((node) -> node.flush(KEYSPACE));
+
+ SimpleCondition previewRepairStarted = new SimpleCondition();
+ SimpleCondition continuePreviewRepair = new SimpleCondition();
+ DelayFirstRepairTypeMessageFilter filter = DelayFirstRepairTypeMessageFilter.validationRequest(previewRepairStarted, continuePreviewRepair);
+ // this pauses the validation request sent from node1 to node2 until we have run a full inc repair below
+ cluster.filters().outbound().verbs(Verb.VALIDATION_REQ.id).from(1).to(2).messagesMatching(filter).drop();
+
+ Future<RepairResult> rsFuture = es.submit(() -> cluster.get(1).callOnInstance(repair(options(true, false))));
+ previewRepairStarted.await();
+ // this needs to finish before the preview repair is unpaused on node2
+ cluster.get(1).callOnInstance(repair(options(false, false)));
+ continuePreviewRepair.signalAll();
+ RepairResult rs = rsFuture.get();
+ assertFalse(rs.success); // preview repair should have failed
+ assertFalse(rs.wasInconsistent); // and no mismatches should have been reported
+ }
+ finally
+ {
+ es.shutdown();
+ }
+ }
+
+ /**
+ * Tests that a IR is running, but not completed before validation compaction starts
+ */
+ @Test
+ public void testConcurrentIncRepairDuringPreview() throws IOException, InterruptedException, ExecutionException
+ {
+ try (Cluster cluster = init(Cluster.build(2).withConfig(config ->
- config.set("disable_incremental_repair", false)
- .with(GOSSIP)
++ config.with(GOSSIP)
+ .with(NETWORK)).start()))
+ {
+ cluster.schemaChange("create table " + KEYSPACE + ".tbl (id int primary key, t int)");
+ insert(cluster.coordinator(1), 0, 100);
+ cluster.forEach((node) -> node.flush(KEYSPACE));
+ cluster.get(1).callOnInstance(repair(options(false, false)));
+
+ insert(cluster.coordinator(1), 100, 100);
+ cluster.forEach((node) -> node.flush(KEYSPACE));
+
+ SimpleCondition previewRepairStarted = new SimpleCondition();
+ SimpleCondition continuePreviewRepair = new SimpleCondition();
+ // this pauses the validation request sent from node1 to node2 until the inc repair below has run
+ cluster.filters()
+ .outbound()
+ .verbs(Verb.VALIDATION_REQ.id)
+ .from(1).to(2)
+ .messagesMatching(DelayFirstRepairTypeMessageFilter.validationRequest(previewRepairStarted, continuePreviewRepair))
+ .drop();
+
+ SimpleCondition irRepairStarted = new SimpleCondition();
+ SimpleCondition continueIrRepair = new SimpleCondition();
+ // this blocks the IR from committing, so we can reenable the preview
+ cluster.filters()
+ .outbound()
+ .verbs(Verb.FINALIZE_PROPOSE_MSG.id)
+ .from(1).to(2)
+ .messagesMatching(DelayFirstRepairTypeMessageFilter.finalizePropose(irRepairStarted, continueIrRepair))
+ .drop();
+
+ Future<RepairResult> previewResult = cluster.get(1).asyncCallsOnInstance(repair(options(true, false))).call();
+ previewRepairStarted.await();
+
+ // trigger IR and wait till its ready to commit
+ Future<RepairResult> irResult = cluster.get(1).asyncCallsOnInstance(repair(options(false, false))).call();
+ irRepairStarted.await();
+
+ // unblock preview repair and wait for it to complete
+ continuePreviewRepair.signalAll();
+
+ RepairResult rs = previewResult.get();
+ assertFalse(rs.success); // preview repair should have failed
+ assertFalse(rs.wasInconsistent); // and no mismatches should have been reported
+
+ continueIrRepair.signalAll();
+ RepairResult ir = irResult.get();
+ assertTrue(ir.success);
+ assertFalse(ir.wasInconsistent); // not preview, so we don't care about preview notification
+ }
+ }
+
+ /**
+ * Same as testFinishingIncRepairDuringPreview but the previewed range does not intersect the incremental repair
+ * so both preview and incremental repair should finish fine (without any mismatches)
+ */
+ @Test
+ public void testFinishingNonIntersectingIncRepairDuringPreview() throws IOException, InterruptedException, ExecutionException
+ {
+ ExecutorService es = Executors.newSingleThreadExecutor();
+ try(Cluster cluster = init(Cluster.build(2).withConfig(config -> config.with(GOSSIP).with(NETWORK)).start()))
+ {
+ cluster.schemaChange("create table " + KEYSPACE + ".tbl (id int primary key, t int)");
+
+ insert(cluster.coordinator(1), 0, 100);
+ cluster.forEach((node) -> node.flush(KEYSPACE));
+ assertTrue(cluster.get(1).callOnInstance(repair(options(false, false))).success);
+
+ insert(cluster.coordinator(1), 100, 100);
+ cluster.forEach((node) -> node.flush(KEYSPACE));
+
+ // pause preview repair validation messages on node2 until node1 has finished
+ SimpleCondition previewRepairStarted = new SimpleCondition();
+ SimpleCondition continuePreviewRepair = new SimpleCondition();
+ DelayFirstRepairTypeMessageFilter filter = DelayFirstRepairTypeMessageFilter.validationRequest(previewRepairStarted, continuePreviewRepair);
+ cluster.filters().outbound().verbs(Verb.VALIDATION_REQ.id).from(1).to(2).messagesMatching(filter).drop();
+
+ // get local ranges to repair two separate ranges:
+ List<String> localRanges = cluster.get(1).callOnInstance(() -> {
+ List<String> res = new ArrayList<>();
+ for (Range<Token> r : StorageService.instance.getLocalReplicas(KEYSPACE).ranges())
+ res.add(r.left.getTokenValue()+ ":"+ r.right.getTokenValue());
+ return res;
+ });
+
+ assertEquals(2, localRanges.size());
+ Future<RepairResult> repairStatusFuture = es.submit(() -> cluster.get(1).callOnInstance(repair(options(true, false, localRanges.get(0)))));
+ previewRepairStarted.await(); // wait for node1 to start validation compaction
+ // this needs to finish before the preview repair is unpaused on node2
+ assertTrue(cluster.get(1).callOnInstance(repair(options(false, false, localRanges.get(1)))).success);
+
+ continuePreviewRepair.signalAll();
+ RepairResult rs = repairStatusFuture.get();
+ assertTrue(rs.success); // repair should succeed
+ assertFalse(rs.wasInconsistent); // and no mismatches
+ }
+ finally
+ {
+ es.shutdown();
+ }
+ }
+
+ @Test
+ public void snapshotTest() throws IOException, InterruptedException
+ {
+ try(Cluster cluster = init(Cluster.build(3).withConfig(config ->
+ config.set("snapshot_on_repaired_data_mismatch", true)
+ .with(GOSSIP)
+ .with(NETWORK))
+ .start()))
+ {
+ cluster.schemaChange("create table " + KEYSPACE + ".tbl (id int primary key, t int)");
+ cluster.schemaChange("create table " + KEYSPACE + ".tbl2 (id int primary key, t int)");
+ Thread.sleep(1000);
+
+ // populate 2 tables
+ insert(cluster.coordinator(1), 0, 100, "tbl");
+ insert(cluster.coordinator(1), 0, 100, "tbl2");
+ cluster.forEach((n) -> n.flush(KEYSPACE));
+
+ // make sure everything is marked repaired
+ cluster.get(1).callOnInstance(repair(options(false, false)));
+ waitMarkedRepaired(cluster);
+ // make node2 mismatch
+ unmarkRepaired(cluster.get(2), "tbl");
+ verifySnapshots(cluster, "tbl", true);
+ verifySnapshots(cluster, "tbl2", true);
+
+ AtomicInteger snapshotMessageCounter = new AtomicInteger();
+ cluster.filters().verbs(Verb.SNAPSHOT_REQ.id).messagesMatching((from, to, message) -> {
+ snapshotMessageCounter.incrementAndGet();
+ return false;
+ }).drop();
+ cluster.get(1).callOnInstance(repair(options(true, true)));
+ verifySnapshots(cluster, "tbl", false);
+ // tbl2 should not have a mismatch, so the snapshots should be empty here
+ verifySnapshots(cluster, "tbl2", true);
+ assertEquals(3, snapshotMessageCounter.get());
+
+ // and make sure that we don't try to snapshot again
+ snapshotMessageCounter.set(0);
+ cluster.get(3).callOnInstance(repair(options(true, true)));
+ assertEquals(0, snapshotMessageCounter.get());
+ }
+ }
+
+ private void waitMarkedRepaired(Cluster cluster)
+ {
+ cluster.forEach(node -> node.runOnInstance(() -> {
+ for (String table : Arrays.asList("tbl", "tbl2"))
+ {
+ ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(table);
+ while (true)
+ {
+ if (cfs.getLiveSSTables().stream().allMatch(SSTableReader::isRepaired))
+ return;
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+ }
+ }
+ }));
+ }
+
+ private void unmarkRepaired(IInvokableInstance instance, String table)
+ {
+ instance.runOnInstance(() -> {
+ ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(table);
+ try
+ {
+ cfs.getCompactionStrategyManager().mutateRepaired(cfs.getLiveSSTables(), ActiveRepairService.UNREPAIRED_SSTABLE, null, false);
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ private void verifySnapshots(Cluster cluster, String table, boolean shouldBeEmpty)
+ {
+ cluster.forEach(node -> node.runOnInstance(() -> {
+ ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(table);
+ if(shouldBeEmpty)
+ {
+ assertTrue(cfs.getSnapshotDetails().isEmpty());
+ }
+ else
+ {
+ while (cfs.getSnapshotDetails().isEmpty())
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+ }
+ }));
+ }
+
+ static abstract class DelayFirstRepairMessageFilter implements IMessageFilters.Matcher
+ {
+ private final SimpleCondition pause;
+ private final SimpleCondition resume;
+ private final AtomicBoolean waitForRepair = new AtomicBoolean(true);
+
+ protected DelayFirstRepairMessageFilter(SimpleCondition pause, SimpleCondition resume)
+ {
+ this.pause = pause;
+ this.resume = resume;
+ }
+
+ protected abstract boolean matchesMessage(RepairMessage message);
+
+ public final boolean matches(int from, int to, IMessage message)
+ {
+ try
+ {
+ Message<?> msg = Instance.deserializeMessage(message);
+ RepairMessage repairMessage = (RepairMessage) msg.payload;
+ // only the first message should be delayed:
+ if (matchesMessage(repairMessage) && waitForRepair.compareAndSet(true, false))
+ {
+ pause.signalAll();
+ resume.await();
+ }
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException(e);
+ }
+ return false; // don't drop the message
+ }
+ }
+
+ static class DelayFirstRepairTypeMessageFilter extends DelayFirstRepairMessageFilter
+ {
+ private final Class<? extends RepairMessage> type;
+
+ public DelayFirstRepairTypeMessageFilter(SimpleCondition pause, SimpleCondition resume, Class<? extends RepairMessage> type)
+ {
+ super(pause, resume);
+ this.type = type;
+ }
+
+ public static DelayFirstRepairTypeMessageFilter validationRequest(SimpleCondition pause, SimpleCondition resume)
+ {
+ return new DelayFirstRepairTypeMessageFilter(pause, resume, ValidationRequest.class);
+ }
+
+ public static DelayFirstRepairTypeMessageFilter finalizePropose(SimpleCondition pause, SimpleCondition resume)
+ {
+ return new DelayFirstRepairTypeMessageFilter(pause, resume, FinalizePropose.class);
+ }
+
+ protected boolean matchesMessage(RepairMessage repairMessage)
+ {
+ return repairMessage.getClass() == type;
+ }
+ }
+
+ static void insert(ICoordinator coordinator, int start, int count)
+ {
+ insert(coordinator, start, count, "tbl");
+ }
+
+ static void insert(ICoordinator coordinator, int start, int count, String table)
+ {
+ for (int i = start; i < start + count; i++)
+ coordinator.execute("insert into " + KEYSPACE + "." + table + " (id, t) values (?, ?)", ConsistencyLevel.ALL, i, i);
+ }
+
+ /**
+ * returns a pair with [repair success, was inconsistent]
+ */
+ private static IIsolatedExecutor.SerializableCallable<RepairResult> repair(Map<String, String> options)
+ {
+ return () -> {
+ SimpleCondition await = new SimpleCondition();
+ AtomicBoolean success = new AtomicBoolean(true);
+ AtomicBoolean wasInconsistent = new AtomicBoolean(false);
+ StorageService.instance.repair(KEYSPACE, options, ImmutableList.of((tag, event) -> {
+ if (event.getType() == ProgressEventType.ERROR)
+ {
+ success.set(false);
+ await.signalAll();
+ }
+ else if (event.getType() == ProgressEventType.NOTIFICATION && event.getMessage().contains("Repaired data is inconsistent"))
+ {
+ wasInconsistent.set(true);
+ }
+ else if (event.getType() == ProgressEventType.COMPLETE)
+ await.signalAll();
+ }));
+ try
+ {
+ await.await(1, TimeUnit.MINUTES);
+ }
+ catch (InterruptedException e)
+ {
+ throw new RuntimeException(e);
+ }
+ return new RepairResult(success.get(), wasInconsistent.get());
+ };
+ }
+
+ private static Map<String, String> options(boolean preview, boolean full)
+ {
+ Map<String, String> config = new HashMap<>();
+ config.put(RepairOption.INCREMENTAL_KEY, "true");
+ config.put(RepairOption.PARALLELISM_KEY, RepairParallelism.PARALLEL.toString());
+ if (preview)
+ config.put(RepairOption.PREVIEW, PreviewKind.REPAIRED.toString());
+ if (full)
+ config.put(RepairOption.INCREMENTAL_KEY, "false");
+ return config;
+ }
+
+ private static Map<String, String> options(boolean preview, boolean full, String range)
+ {
+ Map<String, String> options = options(preview, full);
+ options.put(RepairOption.RANGES_KEY, range);
+ return options;
+ }
+}
diff --cc test/unit/org/apache/cassandra/config/EncryptionOptionsTest.java
index 0000000,0000000..59638ad
new file mode 100644
--- /dev/null
+++ b/test/unit/org/apache/cassandra/config/EncryptionOptionsTest.java
@@@ -1,0 -1,0 +1,178 @@@
++/*
++ * 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.config;
++
++import java.io.File;
++import java.util.Collections;
++import java.util.Map;
++
++import com.google.common.collect.ImmutableMap;
++import org.junit.Assert;
++import org.junit.Test;
++
++import org.apache.cassandra.exceptions.ConfigurationException;
++import org.assertj.core.api.Assertions;
++
++import static org.apache.cassandra.config.EncryptionOptions.TlsEncryptionPolicy.UNENCRYPTED;
++import static org.apache.cassandra.config.EncryptionOptions.TlsEncryptionPolicy.OPTIONAL;
++import static org.apache.cassandra.config.EncryptionOptions.TlsEncryptionPolicy.ENCRYPTED;
++import static org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions.InternodeEncryption.all;
++import static org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions.InternodeEncryption.dc;
++import static org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions.InternodeEncryption.none;
++import static org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions.InternodeEncryption.rack;
++import static org.junit.Assert.*;
++
++public class EncryptionOptionsTest
++{
++ static class EncryptionOptionsTestCase
++ {
++ final EncryptionOptions encryptionOptions;
++ final EncryptionOptions.TlsEncryptionPolicy expected;
++ final String description;
++
++ public EncryptionOptionsTestCase(EncryptionOptions encryptionOptions, EncryptionOptions.TlsEncryptionPolicy expected, String description)
++ {
++ this.encryptionOptions = encryptionOptions;
++ this.expected = expected;
++ this.description = description;
++ }
++
++ public static EncryptionOptionsTestCase of(Boolean optional, String keystorePath, Boolean enabled, EncryptionOptions.TlsEncryptionPolicy expected)
++ {
++ return new EncryptionOptionsTestCase(new EncryptionOptions(keystorePath, "dummypass", "dummytruststore", "dummypass",
++ Collections.emptyList(), "TLS", null, "JKS", false, false, enabled, optional)
++ .applyConfig(),
++ expected,
++ String.format("optional=%s keystore=%s enabled=%s", optional, keystorePath, enabled));
++ }
++ }
++
++ static String absentKeystore = "test/conf/missing-keystore-is-not-here";
++ static String presentKeystore = "test/conf/keystore.jks";
++ EncryptionOptionsTestCase[] encryptionOptionTestCases = {
++ // Optional Keystore Enabled Expected
++ EncryptionOptionsTestCase.of(null, absentKeystore, false, UNENCRYPTED),
++ EncryptionOptionsTestCase.of(null, absentKeystore, true, ENCRYPTED),
++ EncryptionOptionsTestCase.of(null, presentKeystore, false, OPTIONAL),
++ EncryptionOptionsTestCase.of(null, presentKeystore, true, ENCRYPTED),
++ EncryptionOptionsTestCase.of(false, absentKeystore, false, UNENCRYPTED),
++ EncryptionOptionsTestCase.of(false, absentKeystore, true, ENCRYPTED),
++ EncryptionOptionsTestCase.of(true, presentKeystore, false, OPTIONAL),
++ EncryptionOptionsTestCase.of(true, presentKeystore, true, OPTIONAL)
++ };
++
++ @Test
++ public void testEncryptionOptionPolicy()
++ {
++ assertTrue(new File(presentKeystore).exists());
++ assertFalse(new File(absentKeystore).exists());
++ for (EncryptionOptionsTestCase testCase : encryptionOptionTestCases)
++ {
++ Assert.assertSame(testCase.description, testCase.expected, testCase.encryptionOptions.tlsEncryptionPolicy());
++ }
++ }
++
++ static class ServerEncryptionOptionsTestCase
++ {
++ final EncryptionOptions encryptionOptions;
++ final EncryptionOptions.TlsEncryptionPolicy expected;
++ final String description;
++
++ public ServerEncryptionOptionsTestCase(EncryptionOptions encryptionOptions, EncryptionOptions.TlsEncryptionPolicy expected, String description)
++ {
++ this.encryptionOptions = encryptionOptions;
++ this.expected = expected;
++ this.description = description;
++ }
++
++ public static ServerEncryptionOptionsTestCase of(Boolean optional, String keystorePath,
++ EncryptionOptions.ServerEncryptionOptions.InternodeEncryption internodeEncryption,
++ EncryptionOptions.TlsEncryptionPolicy expected)
++ {
++ return new ServerEncryptionOptionsTestCase(new EncryptionOptions.ServerEncryptionOptions(keystorePath, "dummypass", "dummytruststore", "dummypass",
++ Collections.emptyList(), "TLS", null, "JKS", false, false, optional, internodeEncryption, false)
++ .applyConfig(),
++ expected,
++ String.format("optional=%s keystore=%s internode=%s", optional, keystorePath, internodeEncryption));
++ }
++ }
++
++ @Test
++ public void isEnabledServer()
++ {
++ Map<String, Object> yaml = ImmutableMap.of(
++ "server_encryption_options", ImmutableMap.of(
++ "isEnabled", false
++ )
++ );
++
++ Assertions.assertThatThrownBy(() -> YamlConfigurationLoader.fromMap(yaml, Config.class))
++ .isInstanceOf(ConfigurationException.class)
++ .hasMessage("Invalid yaml. Please remove properties [isEnabled] from your cassandra.yaml");
++ }
++
++ @Test
++ public void isOptionalServer()
++ {
++ Map<String, Object> yaml = ImmutableMap.of(
++ "server_encryption_options", ImmutableMap.of(
++ "isOptional", false
++ )
++ );
++
++ Assertions.assertThatThrownBy(() -> YamlConfigurationLoader.fromMap(yaml, Config.class))
++ .isInstanceOf(ConfigurationException.class)
++ .hasMessage("Invalid yaml. Please remove properties [isOptional] from your cassandra.yaml");
++ }
++
++ ServerEncryptionOptionsTestCase[] serverEncryptionOptionTestCases = {
++
++ // Optional Keystore Internode Expected
++ ServerEncryptionOptionsTestCase.of(null, absentKeystore, none, UNENCRYPTED),
++ ServerEncryptionOptionsTestCase.of(null, absentKeystore, rack, OPTIONAL),
++ ServerEncryptionOptionsTestCase.of(null, absentKeystore, dc, OPTIONAL),
++ ServerEncryptionOptionsTestCase.of(null, absentKeystore, all, ENCRYPTED),
++
++ ServerEncryptionOptionsTestCase.of(null, presentKeystore, none, OPTIONAL),
++ ServerEncryptionOptionsTestCase.of(null, presentKeystore, rack, OPTIONAL),
++ ServerEncryptionOptionsTestCase.of(null, absentKeystore, dc, OPTIONAL),
++ ServerEncryptionOptionsTestCase.of(null, absentKeystore, all, ENCRYPTED),
++
++ ServerEncryptionOptionsTestCase.of(false, absentKeystore, none, UNENCRYPTED),
++ ServerEncryptionOptionsTestCase.of(false, absentKeystore, rack, OPTIONAL),
++ ServerEncryptionOptionsTestCase.of(false, absentKeystore, dc, OPTIONAL),
++ ServerEncryptionOptionsTestCase.of(false, absentKeystore, all, ENCRYPTED),
++
++ ServerEncryptionOptionsTestCase.of(true, presentKeystore, none, OPTIONAL),
++ ServerEncryptionOptionsTestCase.of(true, presentKeystore, rack, OPTIONAL),
++ ServerEncryptionOptionsTestCase.of(true, absentKeystore, dc, OPTIONAL),
++ ServerEncryptionOptionsTestCase.of(true, absentKeystore, all, OPTIONAL),
++ };
++
++ @Test
++ public void testServerEncryptionOptionPolicy()
++ {
++ assertTrue(new File(presentKeystore).exists());
++ assertFalse(new File(absentKeystore).exists());
++ for (ServerEncryptionOptionsTestCase testCase : serverEncryptionOptionTestCases)
++ {
++ Assert.assertSame(testCase.description, testCase.expected, testCase.encryptionOptions.tlsEncryptionPolicy());
++ }
++ }
++}
diff --cc test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java
index f705217,4811b4d..2aff83f
--- a/test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java
+++ b/test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java
@@@ -35,18 -36,19 +35,19 @@@ public class YamlConfigurationLoaderTes
int storagePort = 123;
Config.CommitLogSync commitLogSync = Config.CommitLogSync.batch;
ParameterizedClass seedProvider = new ParameterizedClass("org.apache.cassandra.locator.SimpleSeedProvider", Collections.emptyMap());
- EncryptionOptions encryptionOptions = new EncryptionOptions()
- .withKeyStore("myNewKeystore")
- .withCipherSuites("SomeCipher")
- .withOptional(false);
- EncryptionOptions encryptionOptions = new EncryptionOptions.ClientEncryptionOptions();
- encryptionOptions.keystore = "myNewKeystore";
- encryptionOptions.cipher_suites = new String[] {"SomeCipher"};
-
++ Map<String,Object> encryptionOptions = ImmutableMap.of("cipher_suites", Collections.singletonList("FakeCipher"),
++ "optional", false,
++ "enabled", true);
Map<String,Object> map = ImmutableMap.of("storage_port", storagePort,
"commitlog_sync", commitLogSync,
"seed_provider", seedProvider,
"client_encryption_options", encryptionOptions);
++
Config config = YamlConfigurationLoader.fromMap(map, Config.class);
assertEquals(storagePort, config.storage_port); // Check a simple integer
assertEquals(commitLogSync, config.commitlog_sync); // Check an enum
assertEquals(seedProvider, config.seed_provider); // Check a parameterized class
- assertEquals(encryptionOptions, config.client_encryption_options); // Check a nested object
- assertEquals(encryptionOptions.keystore, config.client_encryption_options.keystore); // Check a nested object
- assertArrayEquals(encryptionOptions.cipher_suites, config.client_encryption_options.cipher_suites);
++ assertEquals(false, config.client_encryption_options.optional); // Check a nested object
++ assertEquals(true, config.client_encryption_options.enabled); // Check a nested object
}
}
diff --cc test/unit/org/apache/cassandra/cql3/CQLTester.java
index d205c10,4e320ef..254c042
--- a/test/unit/org/apache/cassandra/cql3/CQLTester.java
+++ b/test/unit/org/apache/cassandra/cql3/CQLTester.java
@@@ -1043,16 -831,6 +1043,16 @@@ public abstract class CQLTeste
return sessions.get(protocolVersion);
}
+ protected SimpleClient newSimpleClient(ProtocolVersion version, boolean compression, boolean checksums, boolean isOverloadedException) throws IOException
+ {
- return new SimpleClient(nativeAddr.getHostAddress(), nativePort, version, version.isBeta(), new EncryptionOptions()).connect(compression, checksums, isOverloadedException);
++ return new SimpleClient(nativeAddr.getHostAddress(), nativePort, version, version.isBeta(), new EncryptionOptions().applyConfig()).connect(compression, checksums, isOverloadedException);
+ }
+
+ protected SimpleClient newSimpleClient(ProtocolVersion version, boolean compression, boolean checksums) throws IOException
+ {
+ return newSimpleClient(version, compression, checksums, false);
+ }
+
protected String formatQuery(String query)
{
return formatQuery(KEYSPACE, query);
diff --cc test/unit/org/apache/cassandra/db/virtual/SettingsTableTest.java
index 7a27282,0000000..65297c7
mode 100644,000000..100644
--- a/test/unit/org/apache/cassandra/db/virtual/SettingsTableTest.java
+++ b/test/unit/org/apache/cassandra/db/virtual/SettingsTableTest.java
@@@ -1,245 -1,0 +1,247 @@@
+/*
+ * 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.lang.reflect.Field;
+import java.lang.reflect.Modifier;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Row;
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions.InternodeEncryption;
+import org.apache.cassandra.config.ParameterizedClass;
+import org.apache.cassandra.cql3.CQLTester;
+
+public class SettingsTableTest extends CQLTester
+{
+ private static final String KS_NAME = "vts";
+
+ private Config config;
+ private SettingsTable table;
+
+ @BeforeClass
+ public static void setUpClass()
+ {
+ CQLTester.setUpClass();
+ }
+
+ @Before
+ public void config()
+ {
+ config = new Config();
++ config.client_encryption_options.applyConfig();
++ config.server_encryption_options.applyConfig();
+ table = new SettingsTable(KS_NAME, config);
+ VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(KS_NAME, ImmutableList.of(table)));
+ }
+
+ private String getValue(Field f)
+ {
+ Object untypedValue = table.getValue(f);
+ String value = null;
+ if (untypedValue != null)
+ {
+ if (untypedValue.getClass().isArray())
+ {
+ value = Arrays.toString((Object[]) untypedValue);
+ }
+ else
+ value = untypedValue.toString();
+ }
+ return value;
+ }
+
+ @Test
+ public void testSelectAll() throws Throwable
+ {
+ int paging = (int) (Math.random() * 100 + 1);
+ ResultSet result = executeNetWithPaging("SELECT * FROM vts.settings", paging);
+ int i = 0;
+ for (Row r : result)
+ {
+ i++;
+ String name = r.getString("name");
+ Field f = SettingsTable.FIELDS.get(name);
+ if (f != null) // skip overrides
+ Assert.assertEquals(getValue(f), r.getString("value"));
+ }
+ Assert.assertTrue(SettingsTable.FIELDS.size() <= i);
+ }
+
+ @Test
+ public void testSelectPartition() throws Throwable
+ {
+ List<Field> fields = Arrays.stream(Config.class.getFields())
+ .filter(f -> !Modifier.isStatic(f.getModifiers()))
+ .collect(Collectors.toList());
+ for (Field f : fields)
+ {
+ if (table.overrides.containsKey(f.getName()))
+ continue;
+
+ String q = "SELECT * FROM vts.settings WHERE name = '"+f.getName()+'\'';
+ assertRowsNet(executeNet(q), new Object[] {f.getName(), getValue(f)});
+ }
+ }
+
+ @Test
+ public void testSelectEmpty() throws Throwable
+ {
+ String q = "SELECT * FROM vts.settings WHERE name = 'EMPTY'";
+ assertRowsNet(executeNet(q));
+ }
+
+ @Test
+ public void testSelectOverride() throws Throwable
+ {
+ String q = "SELECT * FROM vts.settings WHERE name = 'server_encryption_options_enabled'";
+ assertRowsNet(executeNet(q), new Object[] {"server_encryption_options_enabled", "false"});
+ q = "SELECT * FROM vts.settings WHERE name = 'server_encryption_options_XYZ'";
+ assertRowsNet(executeNet(q));
+ }
+
+ private void check(String setting, String expected) throws Throwable
+ {
+ String q = "SELECT * FROM vts.settings WHERE name = '"+setting+'\'';
+ assertRowsNet(executeNet(q), new Object[] {setting, expected});
+ }
+
+ @Test
+ public void testEncryptionOverride() throws Throwable
+ {
+ String pre = "server_encryption_options_";
+ check(pre + "enabled", "false");
+ String all = "SELECT * FROM vts.settings WHERE " +
+ "name > 'server_encryption' AND name < 'server_encryptionz' ALLOW FILTERING";
+
+ Assert.assertEquals(9, executeNet(all).all().size());
+
+ check(pre + "algorithm", null);
+ config.server_encryption_options = config.server_encryption_options.withAlgorithm("SUPERSSL");
+ check(pre + "algorithm", "SUPERSSL");
+
+ check(pre + "cipher_suites", "[]");
+ config.server_encryption_options = config.server_encryption_options.withCipherSuites("c1", "c2");
+ check(pre + "cipher_suites", "[c1, c2]");
+
+ check(pre + "protocol", config.server_encryption_options.protocol);
+ config.server_encryption_options = config.server_encryption_options.withProtocol("TLSv5");
+ check(pre + "protocol", "TLSv5");
+
- check(pre + "optional", "true");
- config.server_encryption_options = config.server_encryption_options.withOptional(false);
+ check(pre + "optional", "false");
++ config.server_encryption_options = config.server_encryption_options.withOptional(true);
++ check(pre + "optional", "true");
+
+ check(pre + "client_auth", "false");
+ config.server_encryption_options = config.server_encryption_options.withRequireClientAuth(true);
+ check(pre + "client_auth", "true");
+
+ check(pre + "endpoint_verification", "false");
+ config.server_encryption_options = config.server_encryption_options.withRequireEndpointVerification(true);
+ check(pre + "endpoint_verification", "true");
+
+ check(pre + "internode_encryption", "none");
+ config.server_encryption_options = config.server_encryption_options.withInternodeEncryption(InternodeEncryption.all);
+ check(pre + "internode_encryption", "all");
+ check(pre + "enabled", "true");
+
+ check(pre + "legacy_ssl_storage_port", "false");
+ config.server_encryption_options = config.server_encryption_options.withLegacySslStoragePort(true);
+ check(pre + "legacy_ssl_storage_port", "true");
+ }
+
+ @Test
+ public void testAuditOverride() throws Throwable
+ {
+ String pre = "audit_logging_options_";
+ check(pre + "enabled", "false");
+ String all = "SELECT * FROM vts.settings WHERE " +
+ "name > 'audit_logging' AND name < 'audit_loggingz' ALLOW FILTERING";
+
+ config.audit_logging_options.enabled = true;
+ Assert.assertEquals(9, executeNet(all).all().size());
+ check(pre + "enabled", "true");
+
+ check(pre + "logger", "BinAuditLogger");
+ config.audit_logging_options.logger = new ParameterizedClass("logger", null);
+ check(pre + "logger", "logger");
+
+ config.audit_logging_options.audit_logs_dir = "dir";
+ check(pre + "audit_logs_dir", "dir");
+
+ check(pre + "included_keyspaces", "");
+ config.audit_logging_options.included_keyspaces = "included_keyspaces";
+ check(pre + "included_keyspaces", "included_keyspaces");
+
+ check(pre + "excluded_keyspaces", "system,system_schema,system_virtual_schema");
+ config.audit_logging_options.excluded_keyspaces = "excluded_keyspaces";
+ check(pre + "excluded_keyspaces", "excluded_keyspaces");
+
+ check(pre + "included_categories", "");
+ config.audit_logging_options.included_categories = "included_categories";
+ check(pre + "included_categories", "included_categories");
+
+ check(pre + "excluded_categories", "");
+ config.audit_logging_options.excluded_categories = "excluded_categories";
+ check(pre + "excluded_categories", "excluded_categories");
+
+ check(pre + "included_users", "");
+ config.audit_logging_options.included_users = "included_users";
+ check(pre + "included_users", "included_users");
+
+ check(pre + "excluded_users", "");
+ config.audit_logging_options.excluded_users = "excluded_users";
+ check(pre + "excluded_users", "excluded_users");
+ }
+
+ @Test
+ public void testTransparentEncryptionOptionsOverride() throws Throwable
+ {
+ String pre = "transparent_data_encryption_options_";
+ check(pre + "enabled", "false");
+ String all = "SELECT * FROM vts.settings WHERE " +
+ "name > 'transparent_data_encryption_options' AND " +
+ "name < 'transparent_data_encryption_optionsz' ALLOW FILTERING";
+
+ config.transparent_data_encryption_options.enabled = true;
+ Assert.assertEquals(4, executeNet(all).all().size());
+ check(pre + "enabled", "true");
+
+ check(pre + "cipher", "AES/CBC/PKCS5Padding");
+ config.transparent_data_encryption_options.cipher = "cipher";
+ check(pre + "cipher", "cipher");
+
+ check(pre + "chunk_length_kb", "64");
+ config.transparent_data_encryption_options.chunk_length_kb = 5;
+ check(pre + "chunk_length_kb", "5");
+
+ check(pre + "iv_length", "16");
+ config.transparent_data_encryption_options.iv_length = 7;
+ check(pre + "iv_length", "7");
+ }
+}
diff --cc test/unit/org/apache/cassandra/net/MessagingServiceTest.java
index 7bae3fa,82630b4..606d336
--- a/test/unit/org/apache/cassandra/net/MessagingServiceTest.java
+++ b/test/unit/org/apache/cassandra/net/MessagingServiceTest.java
@@@ -23,10 -23,10 +23,9 @@@ package org.apache.cassandra.net
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
+import java.util.ArrayList;
import java.util.Arrays;
--import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@@ -35,22 -35,16 +34,21 @@@ import java.util.concurrent.TimeUnit
import java.util.regex.*;
import java.util.regex.Matcher;
--import com.google.common.collect.Iterables;
+import com.google.common.net.InetAddresses;
+
import com.codahale.metrics.Timer;
+import org.apache.cassandra.auth.IInternodeAuthenticator;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.db.monitoring.ApproximateTime;
-import org.apache.cassandra.io.util.DataInputPlus.DataInputStreamPlus;
-import org.apache.cassandra.io.util.DataOutputStreamPlus;
-import org.apache.cassandra.io.util.WrappedDataOutputStreamPlus;
+import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions;
+import org.apache.cassandra.db.commitlog.CommitLog;
+import org.apache.cassandra.metrics.MessagingMetrics;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.utils.FBUtilities;
import org.caffinitas.ohc.histo.EstimatedHistogram;
+import org.junit.After;
+import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
@@@ -275,109 -228,107 +273,109 @@@ public class MessagingServiceTes
}
@Test
- public void testDoesntApplyBackPressureToBroadcastAddress() throws UnknownHostException
+ public void listenRequiredSecureConnection() throws InterruptedException
{
- DatabaseDescriptor.setBackPressureEnabled(true);
- messagingService.applyBackPressure(Arrays.asList(InetAddress.getByName("127.0.0.1")), ONE_SECOND);
- assertFalse(MockBackPressureStrategy.applied);
+ ServerEncryptionOptions serverEncryptionOptions = new ServerEncryptionOptions()
+ .withOptional(false)
+ .withInternodeEncryption(ServerEncryptionOptions.InternodeEncryption.all)
+ .withLegacySslStoragePort(false);
+ listen(serverEncryptionOptions, false);
}
- private static void addDCLatency(long sentAt, long nowTime) throws IOException
+ @Test
+ public void listenRequiredSecureConnectionWithBroadcastAddr() throws InterruptedException
{
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- try (DataOutputStreamPlus out = new WrappedDataOutputStreamPlus(baos))
- {
- out.writeInt((int) sentAt);
- }
- DataInputStreamPlus in = new DataInputStreamPlus(new ByteArrayInputStream(baos.toByteArray()));
- MessageIn.readConstructionTime(FBUtilities.getLocalAddress(), in, nowTime);
+ ServerEncryptionOptions serverEncryptionOptions = new ServerEncryptionOptions()
+ .withOptional(false)
+ .withInternodeEncryption(ServerEncryptionOptions.InternodeEncryption.all)
+ .withLegacySslStoragePort(false);
+ listen(serverEncryptionOptions, true);
}
- public static class MockBackPressureStrategy implements BackPressureStrategy<MockBackPressureStrategy.MockBackPressureState>
+ @Test
+ public void listenRequiredSecureConnectionWithLegacyPort() throws InterruptedException
{
- public static volatile boolean applied = false;
+ ServerEncryptionOptions serverEncryptionOptions = new ServerEncryptionOptions()
+ .withInternodeEncryption(ServerEncryptionOptions.InternodeEncryption.all)
+ .withOptional(false)
+ .withLegacySslStoragePort(true);
+ listen(serverEncryptionOptions, false);
+ }
- public MockBackPressureStrategy(Map<String, Object> args)
- {
- }
+ @Test
+ public void listenRequiredSecureConnectionWithBroadcastAddrAndLegacyPort() throws InterruptedException
+ {
+ ServerEncryptionOptions serverEncryptionOptions = new ServerEncryptionOptions()
+ .withInternodeEncryption(ServerEncryptionOptions.InternodeEncryption.all)
+ .withOptional(false)
+ .withLegacySslStoragePort(true);
+ listen(serverEncryptionOptions, true);
+ }
- @Override
- public void apply(Set<MockBackPressureState> states, long timeout, TimeUnit unit)
- {
- if (!Iterables.isEmpty(states))
- applied = true;
- }
+ @Test
+ public void listenOptionalSecureConnection() throws InterruptedException
+ {
+ ServerEncryptionOptions serverEncryptionOptions = new ServerEncryptionOptions()
+ .withOptional(true);
+ listen(serverEncryptionOptions, false);
+ }
- @Override
- public MockBackPressureState newState(InetAddress host)
+ @Test
+ public void listenOptionalSecureConnectionWithBroadcastAddr() throws InterruptedException
+ {
+ ServerEncryptionOptions serverEncryptionOptions = new ServerEncryptionOptions()
+ .withOptional(true);
+ listen(serverEncryptionOptions, true);
+ }
+
+ private void listen(ServerEncryptionOptions serverEncryptionOptions, boolean listenOnBroadcastAddr) throws InterruptedException
+ {
+ InetAddress listenAddress = FBUtilities.getJustLocalAddress();
+ if (listenOnBroadcastAddr)
{
- return new MockBackPressureState(host);
+ DatabaseDescriptor.setShouldListenOnBroadcastAddress(true);
+ listenAddress = InetAddresses.increment(FBUtilities.getBroadcastAddressAndPort().address);
+ DatabaseDescriptor.setListenAddress(listenAddress);
+ FBUtilities.reset();
}
- public static class MockBackPressureState implements BackPressureState
+ InboundConnectionSettings settings = new InboundConnectionSettings()
+ .withEncryption(serverEncryptionOptions);
+ InboundSockets connections = new InboundSockets(settings);
+ try
{
- private final InetAddress host;
- public volatile boolean onSend = false;
- public volatile boolean onReceive = false;
- public volatile boolean onTimeout = false;
-
- private MockBackPressureState(InetAddress host)
- {
- this.host = host;
- }
-
- @Override
- public void onMessageSent(MessageOut<?> message)
+ connections.open().await();
+ Assert.assertTrue(connections.isListening());
+
+ Set<InetAddressAndPort> expect = new HashSet<>();
+ expect.add(InetAddressAndPort.getByAddressOverrideDefaults(listenAddress, DatabaseDescriptor.getStoragePort()));
+ if (settings.encryption.enable_legacy_ssl_storage_port)
+ expect.add(InetAddressAndPort.getByAddressOverrideDefaults(listenAddress, DatabaseDescriptor.getSSLStoragePort()));
+ if (listenOnBroadcastAddr)
{
- onSend = true;
+ expect.add(InetAddressAndPort.getByAddressOverrideDefaults(FBUtilities.getBroadcastAddressAndPort().address, DatabaseDescriptor.getStoragePort()));
+ if (settings.encryption.enable_legacy_ssl_storage_port)
+ expect.add(InetAddressAndPort.getByAddressOverrideDefaults(FBUtilities.getBroadcastAddressAndPort().address, DatabaseDescriptor.getSSLStoragePort()));
}
- @Override
- public void onResponseReceived()
- {
- onReceive = true;
- }
-
- @Override
- public void onResponseTimeout()
- {
- onTimeout = true;
- }
-
- @Override
- public double getBackPressureRateLimit()
- {
- throw new UnsupportedOperationException("Not supported yet.");
- }
+ Assert.assertEquals(expect.size(), connections.sockets().size());
- @Override
- public InetAddress getHost()
+ final int legacySslPort = DatabaseDescriptor.getSSLStoragePort();
+ for (InboundSockets.InboundSocket socket : connections.sockets())
{
- return host;
+ Assert.assertEquals(serverEncryptionOptions.isEnabled(), socket.settings.encryption.isEnabled());
- Assert.assertEquals(serverEncryptionOptions.optional, socket.settings.encryption.optional);
++ Assert.assertEquals(serverEncryptionOptions.isOptional(), socket.settings.encryption.isOptional());
+ if (!serverEncryptionOptions.isEnabled())
+ Assert.assertFalse(legacySslPort == socket.settings.bindAddress.port);
+ if (legacySslPort == socket.settings.bindAddress.port)
- Assert.assertFalse(socket.settings.encryption.optional);
++ Assert.assertFalse(socket.settings.encryption.isOptional());
+ Assert.assertTrue(socket.settings.bindAddress.toString(), expect.remove(socket.settings.bindAddress));
}
}
- }
-
- private static class BackPressureCallback implements IAsyncCallback
- {
- @Override
- public boolean supportsBackPressure()
- {
- return true;
- }
-
- @Override
- public boolean isLatencyForSnitch()
- {
- return false;
- }
-
- @Override
- public void response(MessageIn msg)
+ finally
{
- throw new UnsupportedOperationException("Not supported.");
+ connections.close().await();
+ Assert.assertFalse(connections.isListening());
}
}
diff --cc test/unit/org/apache/cassandra/service/NativeTransportServiceTest.java
index 86b73ab,25fac21..e70ef0d
--- a/test/unit/org/apache/cassandra/service/NativeTransportServiceTest.java
+++ b/test/unit/org/apache/cassandra/service/NativeTransportServiceTest.java
@@@ -23,12 -22,13 +23,15 @@@ import java.util.function.Consumer
import java.util.stream.Collectors;
import java.util.stream.IntStream;
++import javax.xml.crypto.Data;
++
import com.google.common.collect.Sets;
import org.junit.After;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.config.DatabaseDescriptor;
++import org.apache.cassandra.config.EncryptionOptions;
import org.apache.cassandra.transport.Server;
import org.apache.cassandra.utils.Pair;
@@@ -38,16 -38,16 +41,19 @@@ import static org.junit.Assert.assertTr
public class NativeTransportServiceTest
{
++ static EncryptionOptions defaultOptions;
++
@BeforeClass
public static void setupDD()
{
DatabaseDescriptor.daemonInitialization();
++ defaultOptions = DatabaseDescriptor.getNativeProtocolEncryptionOptions();
}
@After
public void resetConfig()
{
- DatabaseDescriptor.updateNativeProtocolEncryptionOptions(options -> options.withEnabled(false));
- DatabaseDescriptor.getClientEncryptionOptions().enabled = false;
++ DatabaseDescriptor.updateNativeProtocolEncryptionOptions(update -> new EncryptionOptions(defaultOptions).applyConfig());
DatabaseDescriptor.setNativeTransportPortSSL(null);
}
@@@ -118,7 -118,7 +124,7 @@@
{
assertEquals(1, service.getServers().size());
Server server = service.getServers().iterator().next();
-- assertFalse(server.useSSL);
++ assertEquals(EncryptionOptions.TlsEncryptionPolicy.UNENCRYPTED, server.tlsEncryptionPolicy);
assertEquals(server.socket.getPort(), DatabaseDescriptor.getNativeTransportPort());
});
}
@@@ -135,7 -135,7 +141,7 @@@
service.initialize();
assertEquals(1, service.getServers().size());
Server server = service.getServers().iterator().next();
-- assertTrue(server.useSSL);
++ assertEquals(EncryptionOptions.TlsEncryptionPolicy.ENCRYPTED, server.tlsEncryptionPolicy);
assertEquals(server.socket.getPort(), DatabaseDescriptor.getNativeTransportPort());
}, false, 1);
}
@@@ -152,16 -152,16 +158,19 @@@
service.initialize();
assertEquals(1, service.getServers().size());
Server server = service.getServers().iterator().next();
-- assertTrue(server.useSSL);
++ assertEquals(EncryptionOptions.TlsEncryptionPolicy.OPTIONAL, server.tlsEncryptionPolicy);
assertEquals(server.socket.getPort(), DatabaseDescriptor.getNativeTransportPort());
}, false, 1);
}
@Test
-- public void testSSLWithNonSSL()
++ public void testSSLPortWithOptionalEncryption()
{
// ssl+non-ssl settings: client encryption enabled and additional ssl port specified
- DatabaseDescriptor.updateNativeProtocolEncryptionOptions(options -> options.withEnabled(true));
- DatabaseDescriptor.getClientEncryptionOptions().enabled = true;
++ DatabaseDescriptor.updateNativeProtocolEncryptionOptions(
++ options -> options.withEnabled(true)
++ .withOptional(true)
++ .withKeyStore("test/conf/cassandra_ssl_test.keystore"));
DatabaseDescriptor.setNativeTransportPortSSL(8432);
withService((NativeTransportService service) ->
@@@ -170,12 -170,12 +179,71 @@@
assertEquals(2, service.getServers().size());
assertEquals(
Sets.newHashSet(Arrays.asList(
-- Pair.create(true, DatabaseDescriptor.getNativeTransportPortSSL()),
-- Pair.create(false, DatabaseDescriptor.getNativeTransportPort())
++ Pair.create(EncryptionOptions.TlsEncryptionPolicy.OPTIONAL,
++ DatabaseDescriptor.getNativeTransportPortSSL()),
++ Pair.create(EncryptionOptions.TlsEncryptionPolicy.UNENCRYPTED,
++ DatabaseDescriptor.getNativeTransportPort())
)
),
service.getServers().stream().map((Server s) ->
-- Pair.create(s.useSSL, s.socket.getPort())).collect(Collectors.toSet())
++ Pair.create(s.tlsEncryptionPolicy,
++ s.socket.getPort())).collect(Collectors.toSet())
++ );
++ }, false, 1);
++ }
++
++ @Test(expected=java.lang.IllegalStateException.class)
++ public void testSSLPortWithDisabledEncryption()
++ {
++ // ssl+non-ssl settings: client encryption disabled and additional ssl port specified
++ // should get an illegal state exception
++ DatabaseDescriptor.updateNativeProtocolEncryptionOptions(
++ options -> options.withEnabled(false));
++ DatabaseDescriptor.setNativeTransportPortSSL(8432);
++
++ withService((NativeTransportService service) ->
++ {
++ service.initialize();
++ assertEquals(1, service.getServers().size());
++ assertEquals(
++ Sets.newHashSet(Arrays.asList(
++ Pair.create(EncryptionOptions.TlsEncryptionPolicy.UNENCRYPTED,
++ DatabaseDescriptor.getNativeTransportPort())
++ )
++ ),
++ service.getServers().stream().map((Server s) ->
++ Pair.create(s.tlsEncryptionPolicy,
++ s.socket.getPort())).collect(Collectors.toSet())
++ );
++ }, false, 1);
++ }
++
++ @Test
++ public void testSSLPortWithEnabledSSL()
++ {
++ // ssl+non-ssl settings: client encryption enabled and additional ssl port specified
++ // encryption is enabled and not optional, so listen on both ports requiring encryption
++ DatabaseDescriptor.updateNativeProtocolEncryptionOptions(
++ options -> options.withEnabled(true)
++ .withOptional(false)
++ .withKeyStore("test/conf/cassandra_ssl_test.keystore"));
++ DatabaseDescriptor.setNativeTransportPortSSL(8432);
++
++ withService((NativeTransportService service) ->
++ {
++ service.initialize();
++ assertEquals(2, service.getServers().size());
++ assertEquals(
++ Sets.newHashSet(Arrays.asList(
++ Pair.create(EncryptionOptions.TlsEncryptionPolicy.ENCRYPTED,
++ DatabaseDescriptor.getNativeTransportPortSSL()),
++ Pair.create(EncryptionOptions.TlsEncryptionPolicy.UNENCRYPTED,
++ DatabaseDescriptor.getNativeTransportPort())
++ )
++ ),
++ service.getServers().stream().map((Server s) ->
++ Pair.create(s.tlsEncryptionPolicy,
++ s.socket.getPort())).collect(Collectors.toSet())
);
}, false, 1);
}
diff --cc tools/stress/src/org/apache/cassandra/stress/settings/SettingsTransport.java
index 4ea4cd2,4981a87..5f22a7b
--- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsTransport.java
+++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsTransport.java
@@@ -36,25 -43,74 +36,25 @@@ public class SettingsTransport implemen
public SettingsTransport(TOptions options)
{
this.options = options;
- this.fqFactoryClass = options.factory.value();
- try
- {
- Class<?> clazz = Class.forName(fqFactoryClass);
- if (!ITransportFactory.class.isAssignableFrom(clazz))
- throw new IllegalArgumentException(clazz + " is not a valid transport factory");
- // check we can instantiate it
- clazz.newInstance();
- }
- catch (Exception e)
- {
- throw new IllegalArgumentException("Invalid transport factory class: " + options.factory.value(), e);
- }
- }
-
- private void configureTransportFactory(ITransportFactory transportFactory, TOptions options)
- {
- Map<String, String> factoryOptions = new HashMap<>();
- // If the supplied factory supports the same set of options as our SSL impl, set those
- if (transportFactory.supportedOptions().contains(SSLTransportFactory.TRUSTSTORE))
- factoryOptions.put(SSLTransportFactory.TRUSTSTORE, options.trustStore.value());
- if (transportFactory.supportedOptions().contains(SSLTransportFactory.TRUSTSTORE_PASSWORD))
- factoryOptions.put(SSLTransportFactory.TRUSTSTORE_PASSWORD, options.trustStorePw.value());
- if (transportFactory.supportedOptions().contains(SSLTransportFactory.KEYSTORE))
- factoryOptions.put(SSLTransportFactory.KEYSTORE, options.keyStore.value());
- if (transportFactory.supportedOptions().contains(SSLTransportFactory.KEYSTORE_PASSWORD))
- factoryOptions.put(SSLTransportFactory.KEYSTORE_PASSWORD, options.keyStorePw.value());
- if (transportFactory.supportedOptions().contains(SSLTransportFactory.PROTOCOL))
- factoryOptions.put(SSLTransportFactory.PROTOCOL, options.protocol.value());
- if (transportFactory.supportedOptions().contains(SSLTransportFactory.CIPHER_SUITES))
- factoryOptions.put(SSLTransportFactory.CIPHER_SUITES, options.ciphers.value());
- // Now check if any of the factory's supported options are set as system properties
- for (String optionKey : transportFactory.supportedOptions())
- if (System.getProperty(optionKey) != null)
- factoryOptions.put(optionKey, System.getProperty(optionKey));
-
- transportFactory.setOptions(factoryOptions);
- }
-
- public synchronized ITransportFactory getFactory()
- {
- if (factory == null)
- {
- try
- {
- this.factory = (ITransportFactory) Class.forName(fqFactoryClass).newInstance();
- configureTransportFactory(this.factory, this.options);
- }
- catch (Exception e)
- {
- throw new RuntimeException(e);
- }
- }
- return factory;
}
- public EncryptionOptions.ClientEncryptionOptions getEncryptionOptions()
+ public EncryptionOptions getEncryptionOptions()
{
- EncryptionOptions encOptions = new EncryptionOptions();
- EncryptionOptions.ClientEncryptionOptions encOptions = new EncryptionOptions.ClientEncryptionOptions();
++ EncryptionOptions encOptions = new EncryptionOptions().applyConfig();
if (options.trustStore.present())
{
- encOptions.enabled = true;
- encOptions.truststore = options.trustStore.value();
- encOptions.truststore_password = options.trustStorePw.value();
+ encOptions = encOptions
+ .withEnabled(true)
+ .withTrustStore(options.trustStore.value())
+ .withTrustStorePassword(options.trustStorePw.value())
+ .withAlgorithm(options.alg.value())
+ .withProtocol(options.protocol.value())
+ .withCipherSuites(options.ciphers.value().split(","));
if (options.keyStore.present())
{
- encOptions.keystore = options.keyStore.value();
- encOptions.keystore_password = options.keyStorePw.value();
+ encOptions = encOptions
+ .withKeyStore(options.keyStore.value())
+ .withKeyStorePassword(options.keyStorePw.value());
}
else
{
diff --cc tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java
index 643e58f,e0b4262..b054a6e
--- a/tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java
+++ b/tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java
@@@ -70,7 -70,7 +70,7 @@@ public class JavaDriverClien
this.username = settings.mode.username;
this.password = settings.mode.password;
this.authProvider = settings.mode.authProvider;
-- this.encryptionOptions = encryptionOptions;
++ this.encryptionOptions = new EncryptionOptions(encryptionOptions).applyConfig();
this.loadBalancingPolicy = loadBalancingPolicy(settings);
this.connectionsPerHost = settings.mode.connectionsPerHost == null ? 8 : settings.mode.connectionsPerHost;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org