natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

remove ssl storage port from sstableloader 

patch by Stefan Miklosovic; reviewed by Brandon Williams for CASSANDRA-17602
This commit is contained in:
Stefan Miklosovic 2022-05-16 17:17:07 +02:00
parent cd0a40d09e
commit 2531dd1eba
7 changed files with 23 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.0.5
* Remove SSL storage port from sstableloader (CASSANDRA-17602)
* Allow Java 11 to satisfy RPM/Debian packaging (CASSANDRA-17669)
* Ensure FileStreamTask cannot compromise shared channel proxy for system table when interrupted (CASSANDRA-17663)
* silence benign SslClosedEngineException (CASSANDRA-17565)

18
src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java
View File

@ -28,26 +28,24 @@ import org.apache.cassandra.streaming.StreamConnectionFactory;
public class BulkLoadConnectionFactory extends DefaultConnectionFactory implements StreamConnectionFactory
{
// TODO: what is this unused variable for?
private final boolean outboundBindAny;
private final int secureStoragePort;
private final int storagePort;
private final EncryptionOptions.ServerEncryptionOptions encryptionOptions;
public BulkLoadConnectionFactory(int secureStoragePort, EncryptionOptions.ServerEncryptionOptions encryptionOptions, boolean outboundBindAny)
public BulkLoadConnectionFactory(EncryptionOptions.ServerEncryptionOptions encryptionOptions, int storagePort)
{
this.secureStoragePort = secureStoragePort;
this.storagePort = storagePort;
this.encryptionOptions = encryptionOptions;
this.outboundBindAny = outboundBindAny;
}
public Channel createConnection(OutboundConnectionSettings template, int messagingVersion) throws IOException
{
// Connect to secure port for all peers if ServerEncryptionOptions is configured other than 'none'
// When 'all', 'dc' and 'rack', server nodes always have SSL port open, and since thin client like sstableloader
// does not know which node is in which dc/rack, connecting to SSL port is always the option.
// storage port can handle both encrypted and unencrypted traffic from 4.0
// so from sstableloader's point of view we can use just storage port for both cases
template = template.withConnectTo(template.to.withPort(storagePort));
if (encryptionOptions != null && encryptionOptions.internode_encryption != EncryptionOptions.ServerEncryptionOptions.InternodeEncryption.none)
template = template.withConnectTo(template.to.withPort(secureStoragePort)).withEncryption(encryptionOptions);
template = template.withEncryption(encryptionOptions);
return super.createConnection(template, messagingVersion);
}

8
src/java/org/apache/cassandra/tools/BulkLoader.java
View File

@ -61,7 +61,6 @@ public class BulkLoader
options.hosts,
options.storagePort,
options.authProvider,
options.sslStoragePort,
options.serverEncOptions,
buildSSLOptions(options.clientEncOptions)),
handler,
@ -283,25 +282,24 @@ public class BulkLoader
static class ExternalClient extends NativeSSTableLoaderClient
{
private final int sslStoragePort;
private final int storagePort;
private final EncryptionOptions.ServerEncryptionOptions serverEncOptions;
public ExternalClient(Set<InetSocketAddress> hosts,
int storagePort,
AuthProvider authProvider,
int sslStoragePort,
EncryptionOptions.ServerEncryptionOptions serverEncryptionOptions,
SSLOptions sslOptions)
{
super(hosts, storagePort, authProvider, sslOptions);
this.sslStoragePort = sslStoragePort;
serverEncOptions = serverEncryptionOptions;
this.storagePort = storagePort;
}
@Override
public StreamConnectionFactory getConnectionFactory()
{
return new BulkLoadConnectionFactory(sslStoragePort, serverEncOptions, false);
return new BulkLoadConnectionFactory(serverEncOptions, storagePort);
}
}

15
src/java/org/apache/cassandra/tools/LoaderOptions.java
View File

@ -48,7 +48,6 @@ public class LoaderOptions
public static final String NOPROGRESS_OPTION = "no-progress";
public static final String NATIVE_PORT_OPTION = "port";
public static final String STORAGE_PORT_OPTION = "storage-port";
public static final String SSL_STORAGE_PORT_OPTION = "ssl-storage-port";
public static final String USER_OPTION = "username";
public static final String PASSWD_OPTION = "password";
public static final String AUTH_PROVIDER_OPTION = "auth-provider";
@ -82,7 +81,6 @@ public class LoaderOptions
public final int throttle;
public final int interDcThrottle;
public final int storagePort;
public final int sslStoragePort;
public final EncryptionOptions clientEncOptions;
public final int connectionsPerHost;
public final EncryptionOptions.ServerEncryptionOptions serverEncOptions;
@ -103,7 +101,6 @@ public class LoaderOptions
throttle = builder.throttle;
interDcThrottle = builder.interDcThrottle;
storagePort = builder.storagePort;
sslStoragePort = builder.sslStoragePort;
clientEncOptions = builder.clientEncOptions;
connectionsPerHost = builder.connectionsPerHost;
serverEncOptions = builder.serverEncOptions;
@ -126,7 +123,6 @@ public class LoaderOptions
int throttle = 0;
int interDcThrottle = 0;
int storagePort;
int sslStoragePort;
EncryptionOptions clientEncOptions = new EncryptionOptions();
int connectionsPerHost = 1;
EncryptionOptions.ServerEncryptionOptions serverEncOptions = new EncryptionOptions.ServerEncryptionOptions();
@ -230,12 +226,6 @@ public class LoaderOptions
return this;
}
public Builder sslStoragePort(int sslStoragePort)
{
this.sslStoragePort = sslStoragePort;
return this;
}
public Builder encOptions(EncryptionOptions encOptions)
{
this.clientEncOptions = encOptions;
@ -411,10 +401,6 @@ public class LoaderOptions
connectionsPerHost = Integer.parseInt(cmd.getOptionValue(CONNECTIONS_PER_HOST));
}
if (cmd.hasOption(SSL_STORAGE_PORT_OPTION))
sslStoragePort = Integer.parseInt(cmd.getOptionValue(SSL_STORAGE_PORT_OPTION));
else
sslStoragePort = config.ssl_storage_port;
throttle = config.stream_throughput_outbound_megabits_per_sec;
// Copy the encryption options and apply the config so that argument parsing can accesss isEnabled.
clientEncOptions = config.client_encryption_options.applyConfig();
@ -609,7 +595,6 @@ public class LoaderOptions
options.addOption("d", INITIAL_HOST_ADDRESS_OPTION, "initial hosts", "Required. try to connect to these hosts (comma separated) initially for ring information");
options.addOption("p", NATIVE_PORT_OPTION, "native transport port", "port used for native connection (default 9042)");
options.addOption("sp", STORAGE_PORT_OPTION, "storage port", "port used for internode communication (default 7000)");
options.addOption("ssp", SSL_STORAGE_PORT_OPTION, "ssl storage port", "port used for TLS internode communication (default 7001)");
options.addOption("t", THROTTLE_MBITS, "throttle", "throttle speed in Mbits (default unlimited)");
options.addOption("idct", INTER_DC_THROTTLE_MBITS, "inter-dc-throttle", "inter-datacenter throttle speed in Mbits (default unlimited)");
options.addOption("u", USER_OPTION, "username", "username for cassandra authentication");

5
src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java
View File

@ -19,7 +19,6 @@ package org.apache.cassandra.utils;
import java.nio.ByteBuffer;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.*;
import com.datastax.driver.core.*;
@ -98,9 +97,9 @@ public class NativeSSTableLoaderClient extends SSTableLoader.Client
// We only need the TableMetadata for the views, so we only load that.
tables.putAll(fetchViews(keyspace, session, partitioner, types));
}
catch (UnknownHostException e)
catch (Exception e)
{
throw new RuntimeException(e);
throw new RuntimeException("Unable to initialise " + NativeSSTableLoaderClient.class.getName(), e);
}
}

6
test/distributed/org/apache/cassandra/distributed/test/SSTableLoaderEncryptionOptionsTest.java
View File

@ -36,6 +36,7 @@ import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.tools.BulkLoader;
import org.apache.cassandra.tools.ToolRunner;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.NativeSSTableLoaderClient;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
@ -79,6 +80,7 @@ public class SSTableLoaderEncryptionOptionsTest extends AbstractEncryptionOption
if (CLUSTER != null)
CLUSTER.close();
}
@Test
public void bulkLoaderSuccessfullyStreamsOverSsl() throws Throwable
{
@ -86,7 +88,7 @@ public class SSTableLoaderEncryptionOptionsTest extends AbstractEncryptionOption
ToolRunner.ToolResult tool = ToolRunner.invokeClass(BulkLoader.class,
"--nodes", NODES,
"--port", Integer.toString(NATIVE_PORT),
"--ssl-storage-port", Integer.toString(STORAGE_PORT),
"--storage-port", Integer.toString(STORAGE_PORT),
"--keystore", validKeyStorePath,
"--keystore-password", validKeyStorePassword,
"--truststore", validTrustStorePath,
@ -112,7 +114,7 @@ public class SSTableLoaderEncryptionOptionsTest extends AbstractEncryptionOption
"--truststore-password", validTrustStorePassword,
"test/data/legacy-sstables/na/legacy_tables/legacy_na_clust");
assertNotEquals(0, tool.getExitCode());
assertTrue(tool.getStdout().contains("SSLHandshakeException"));
assertTrue(tool.getStderr().contains("Unable to initialise " + NativeSSTableLoaderClient.class.getName()));
}
private static File prepareSstablesForUpload() throws IOException

10
test/unit/org/apache/cassandra/tools/BulkLoaderTest.java
View File

@ -55,7 +55,7 @@ public class BulkLoaderTest extends OfflineToolUtils
assertEquals(-1, tool.getExitCode());
if (!(tool.getException().getCause() instanceof BulkLoadException))
throw tool.getException();
if (!(tool.getException().getCause().getCause() instanceof NoHostAvailableException))
if (!(tool.getException().getCause().getCause().getCause() instanceof NoHostAvailableException))
throw tool.getException();
assertNoUnexpectedThreadsStarted(null, new String[]{"globalEventExecutor-1-1", "globalEventExecutor-1-2"});
@ -79,7 +79,7 @@ public class BulkLoaderTest extends OfflineToolUtils
assertEquals(-1, tool.getExitCode());
if (!(tool.getException().getCause() instanceof BulkLoadException))
throw tool.getException();
if (!(tool.getException().getCause().getCause() instanceof NoHostAvailableException))
if (!(tool.getException().getCause().getCause().getCause() instanceof NoHostAvailableException))
throw tool.getException();
assertNoUnexpectedThreadsStarted(null, new String[] { "globalEventExecutor-1-1", "globalEventExecutor-1-2" });
@ -103,7 +103,7 @@ public class BulkLoaderTest extends OfflineToolUtils
assertEquals(-1, tool.getExitCode());
if (!(tool.getException().getCause() instanceof BulkLoadException))
throw tool.getException();
if (!(tool.getException().getCause().getCause() instanceof NoHostAvailableException))
if (!(tool.getException().getCause().getCause().getCause() instanceof NoHostAvailableException))
throw tool.getException();
assertNoUnexpectedThreadsStarted(null, new String[] { "globalEventExecutor-1-1", "globalEventExecutor-1-2" });
@ -124,7 +124,7 @@ public class BulkLoaderTest extends OfflineToolUtils
"9041",
OfflineToolUtils.sstableDirName("legacy_sstables", "legacy_ma_simple"));
assertEquals(-1, tool.getExitCode());
throw tool.getException().getCause().getCause();
throw tool.getException().getCause().getCause().getCause();
}
@Test(expected = NoHostAvailableException.class)
@ -135,6 +135,6 @@ public class BulkLoaderTest extends OfflineToolUtils
"127.9.9.1:9041",
OfflineToolUtils.sstableDirName("legacy_sstables", "legacy_ma_simple"));
assertEquals(-1, tool.getExitCode());
throw tool.getException().getCause().getCause();
throw tool.getException().getCause().getCause().getCause();
}
}