From 6d6d1890425e0372debc3acc21f86dce7c08cf37 Mon Sep 17 00:00:00 2001 From: Jeremy Hanna Date: Wed, 30 Dec 2015 13:56:32 -0800 Subject: [PATCH] Fix inter datacenter streaming default patch by jeromatron; reviewed by aweisberg for CASSANDRA-9708 --- CHANGES.txt | 1 + NEWS.txt | 7 ++++++ conf/cassandra.yaml | 3 ++- .../org/apache/cassandra/config/Config.java | 2 +- .../cassandra/service/StorageService.java | 12 +++++++++ .../service/StorageServiceMBean.java | 3 +++ .../apache/cassandra/tools/BulkLoader.java | 12 ++++++++- .../org/apache/cassandra/tools/NodeProbe.java | 10 ++++++++ .../org/apache/cassandra/tools/NodeTool.java | 25 +++++++++++++++++++ 9 files changed, 72 insertions(+), 3 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 844a28f873..14c5ee61ef 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 2.1.13 + * Sane default (200Mbps) for inter-DC streaming througput (CASSANDRA-9708) * Match cassandra-loader options in COPY FROM (CASSANDRA-9303) * Fix binding to any address in CqlBulkRecordWriter (CASSANDRA-9309) * Fix the way we replace sstables after anticompaction (CASSANDRA-10831) diff --git a/NEWS.txt b/NEWS.txt index 088efae766..845801d2a8 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -20,6 +20,13 @@ New features ------------ - New options for cqlsh COPY FROM and COPY TO, see CASSANDRA-9303 for details. +Upgrading +--------- + - The default for the inter-DC stream throughput setting + (inter_dc_stream_throughput_outbound_megabits_per_sec in cassandra.yaml) is + the same than the one for intra-DC one (200Mbps) instead of being unlimited. + Having it unlimited was never intended and was a bug. + 2.1.12 ====== diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 0d0282b947..1fa04e6e96 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -599,7 +599,8 @@ sstable_preemptive_open_interval_in_mb: 50 # this setting allows users to throttle inter dc stream throughput in addition # to throttling all network stream traffic as configured with # stream_throughput_outbound_megabits_per_sec -# inter_dc_stream_throughput_outbound_megabits_per_sec: +# When unset, the default is 200 Mbps or 25 MB/s +# inter_dc_stream_throughput_outbound_megabits_per_sec: 200 # How long the coordinator should wait for read operations to complete read_request_timeout_in_ms: 5000 diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 911dd73d3e..63bbf96670 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -151,7 +151,7 @@ public class Config public Integer max_streaming_retries = 3; public volatile Integer stream_throughput_outbound_megabits_per_sec = 200; - public volatile Integer inter_dc_stream_throughput_outbound_megabits_per_sec = 0; + public volatile Integer inter_dc_stream_throughput_outbound_megabits_per_sec = 200; public String[] data_file_directories; diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 6e38b92b66..f134e8a60a 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -1088,6 +1088,18 @@ public class StorageService extends NotificationBroadcasterSupport implements IE return DatabaseDescriptor.getStreamThroughputOutboundMegabitsPerSec(); } + public void setInterDCStreamThroughputMbPerSec(int value) + { + DatabaseDescriptor.setInterDCStreamThroughputOutboundMegabitsPerSec(value); + logger.info("setinterdcstreamthroughput: throttle set to {}", value); + } + + public int getInterDCStreamThroughputMbPerSec() + { + return DatabaseDescriptor.getInterDCStreamThroughputOutboundMegabitsPerSec(); + } + + public int getCompactionThroughputMbPerSec() { return DatabaseDescriptor.getCompactionThroughputMbPerSec(); diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index 1351fea338..8fa2433895 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -476,6 +476,9 @@ public interface StorageServiceMBean extends NotificationEmitter public void setStreamThroughputMbPerSec(int value); public int getStreamThroughputMbPerSec(); + public void setInterDCStreamThroughputMbPerSec(int value); + public int getInterDCStreamThroughputMbPerSec(); + public int getCompactionThroughputMbPerSec(); public void setCompactionThroughputMbPerSec(int value); diff --git a/src/java/org/apache/cassandra/tools/BulkLoader.java b/src/java/org/apache/cassandra/tools/BulkLoader.java index 96e826d3fd..37317d535d 100644 --- a/src/java/org/apache/cassandra/tools/BulkLoader.java +++ b/src/java/org/apache/cassandra/tools/BulkLoader.java @@ -59,6 +59,7 @@ public class BulkLoader private static final String USER_OPTION = "username"; private static final String PASSWD_OPTION = "password"; private static final String THROTTLE_MBITS = "throttle"; + private static final String INTER_DC_THROTTLE_MBITS = "inter-dc-throttle"; private static final String TRANSPORT_FACTORY = "transport-factory"; @@ -93,6 +94,7 @@ public class BulkLoader handler, options.connectionsPerHost); DatabaseDescriptor.setStreamThroughputOutboundMegabitsPerSec(options.throttle); + DatabaseDescriptor.setInterDCStreamThroughputOutboundMegabitsPerSec(options.interDcThrottle); StreamResultFuture future = null; ProgressIndicator indicator = new ProgressIndicator(); @@ -401,6 +403,7 @@ public class BulkLoader public String user; public String passwd; public int throttle = 0; + public int interDcThrottle = 0; public int storagePort; public int sslStoragePort; public ITransportFactory transportFactory = new TFramedTransportFactory(); @@ -529,6 +532,7 @@ public class BulkLoader opts.storagePort = config.storage_port; opts.sslStoragePort = config.ssl_storage_port; opts.throttle = config.stream_throughput_outbound_megabits_per_sec; + opts.interDcThrottle = config.inter_dc_stream_throughput_outbound_megabits_per_sec; opts.encOptions = config.client_encryption_options; opts.serverEncOptions = config.server_encryption_options; @@ -537,6 +541,11 @@ public class BulkLoader opts.throttle = Integer.parseInt(cmd.getOptionValue(THROTTLE_MBITS)); } + if (cmd.hasOption(INTER_DC_THROTTLE_MBITS)) + { + opts.interDcThrottle = Integer.parseInt(cmd.getOptionValue(INTER_DC_THROTTLE_MBITS)); + } + if (cmd.hasOption(SSL_TRUSTSTORE)) { opts.encOptions.truststore = cmd.getOptionValue(SSL_TRUSTSTORE); @@ -656,6 +665,7 @@ public class BulkLoader 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", RPC_PORT_OPTION, "rpc port", "port used for rpc (default 9160)"); 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"); options.addOption("pw", PASSWD_OPTION, "password", "password for cassandra authentication"); options.addOption("tf", TRANSPORT_FACTORY, "transport factory", "Fully-qualified ITransportFactory class name for creating a connection to cassandra"); @@ -683,7 +693,7 @@ public class BulkLoader "you will need to have the files Standard1-g-1-Data.db and Standard1-g-1-Index.db into a directory /path/to/Keyspace1/Standard1/."; String footer = System.lineSeparator() + "You can provide cassandra.yaml file with -f command line option to set up streaming throughput, client and server encryption options. " + - "Only stream_throughput_outbound_megabits_per_sec, server_encryption_options and client_encryption_options are read from yaml. " + + "Only stream_throughput_outbound_megabits_per_sec, inter_dc_stream_throughput_outbound_megabits_per_sec, server_encryption_options and client_encryption_options are read from yaml. " + "You can override options read from cassandra.yaml with corresponding command line options."; new HelpFormatter().printHelp(usage, header, options, footer); } diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index 5ba5a324ae..1ad1147104 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/src/java/org/apache/cassandra/tools/NodeProbe.java @@ -983,6 +983,11 @@ public class NodeProbe implements AutoCloseable return ssProxy.getStreamThroughputMbPerSec(); } + public int getInterDCStreamThroughput() + { + return ssProxy.getInterDCStreamThroughputMbPerSec(); + } + public int getExceptionCount() { return ssProxy.getExceptionCount(); @@ -1018,6 +1023,11 @@ public class NodeProbe implements AutoCloseable ssProxy.setStreamThroughputMbPerSec(value); } + public void setInterDCStreamThroughput(int value) + { + ssProxy.setInterDCStreamThroughputMbPerSec(value); + } + public void setTraceProbability(double value) { ssProxy.setTraceProbability(value); diff --git a/src/java/org/apache/cassandra/tools/NodeTool.java b/src/java/org/apache/cassandra/tools/NodeTool.java index 1d4a42024d..819049e1e3 100644 --- a/src/java/org/apache/cassandra/tools/NodeTool.java +++ b/src/java/org/apache/cassandra/tools/NodeTool.java @@ -138,6 +138,7 @@ public class NodeTool GetCompactionThreshold.class, GetCompactionThroughput.class, GetStreamThroughput.class, + GetInterDCStreamThroughput.class, GetEndpoints.class, GetSSTables.class, GossipInfo.class, @@ -160,6 +161,7 @@ public class NodeTool SetCompactionThreshold.class, SetCompactionThroughput.class, SetStreamThroughput.class, + SetInterDCStreamThroughput.class, SetTraceProbability.class, Snapshot.class, ListSnapshots.class, @@ -1608,6 +1610,16 @@ public class NodeTool } } + @Command(name = "getinterdcstreamthroughput", description = "Print the Mb/s throughput cap for inter-datacenter streaming in the system") + public static class GetInterDCStreamThroughput extends NodeToolCmd + { + @Override + public void execute(NodeProbe probe) + { + System.out.println("Current inter-datacenter stream throughput: " + probe.getInterDCStreamThroughput() + " Mb/s"); + } + } + @Command(name = "getendpoints", description = "Print the end points that owns the key") public static class GetEndpoints extends NodeToolCmd { @@ -2003,6 +2015,19 @@ public class NodeTool } } + @Command(name = "setinterdcstreamthroughput", description = "Set the Mb/s throughput cap for inter-datacenter streaming in the system, or 0 to disable throttling") + public static class SetInterDCStreamThroughput extends NodeToolCmd + { + @Arguments(title = "inter_dc_stream_throughput", usage = "", description = "Value in Mb, 0 to disable throttling", required = true) + private Integer interDCStreamThroughput = null; + + @Override + public void execute(NodeProbe probe) + { + probe.setInterDCStreamThroughput(interDCStreamThroughput); + } + } + @Command(name = "settraceprobability", description = "Sets the probability for tracing any given request to value. 0 disables, 1 enables for all requests, 0 is the default") public static class SetTraceProbability extends NodeToolCmd {