Skip to content

Commit

Permalink
Allow to configure indices.fielddata.breaker.limit with a ratio of th…
Browse files Browse the repository at this point in the history 
…e heap size.

Close elastic#4616
  • Loading branch information
@jpountz
jpountz committed Jan 21, 2014
1 parent 649f1b1 commit a77f1cb
Show file tree
Hide file tree
Showing 5 changed files with 41 additions and 16 deletions.
4 changes: 2 additions & 2 deletions src/main/java/org/elasticsearch/cluster/settings/ClusterDynamicSettingsModule.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,10 +27,10 @@
import org.elasticsearch.common.inject.AbstractModule;
import org.elasticsearch.discovery.zen.elect.ElectMasterService;
import org.elasticsearch.indices.cache.filter.IndicesFilterCache;
import org.elasticsearch.indices.fielddata.breaker.InternalCircuitBreakerService;
import org.elasticsearch.indices.recovery.RecoverySettings;
import org.elasticsearch.indices.store.IndicesStore;
import org.elasticsearch.indices.ttl.IndicesTTLService;
import org.elasticsearch.indices.fielddata.breaker.InternalCircuitBreakerService;
import org.elasticsearch.threadpool.ThreadPool;

/**
Expand Down Expand Up @@ -78,7 +78,7 @@ public ClusterDynamicSettingsModule() {
clusterDynamicSettings.addDynamicSetting(DiskThresholdDecider.CLUSTER_ROUTING_ALLOCATION_DISK_THRESHOLD_ENABLED);
clusterDynamicSettings.addDynamicSetting(InternalClusterInfoService.INTERNAL_CLUSTER_INFO_UPDATE_INTERVAL, Validator.TIME);
clusterDynamicSettings.addDynamicSetting(SnapshotInProgressAllocationDecider.CLUSTER_ROUTING_ALLOCATION_SNAPSHOT_RELOCATION_ENABLED);
clusterDynamicSettings.addDynamicSetting(InternalCircuitBreakerService.CIRCUIT_BREAKER_MAX_BYTES_SETTING, Validator.BYTES_SIZE);
clusterDynamicSettings.addDynamicSetting(InternalCircuitBreakerService.CIRCUIT_BREAKER_MAX_BYTES_SETTING, Validator.MEMORY_SIZE);
clusterDynamicSettings.addDynamicSetting(InternalCircuitBreakerService.CIRCUIT_BREAKER_OVERHEAD_SETTING, Validator.NON_NEGATIVE_DOUBLE);
clusterDynamicSettings.addDynamicSetting(DestructiveOperations.REQUIRES_NAME);
}
Expand Down
16 changes: 15 additions & 1 deletion src/main/java/org/elasticsearch/cluster/settings/Validator.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@
import org.elasticsearch.common.unit.TimeValue;

import static org.elasticsearch.common.unit.ByteSizeValue.parseBytesSizeValue;
import static org.elasticsearch.common.unit.MemorySizeValue.parseBytesSizeValueOrHeapRatio;


/**
* Validates a setting, returning a failure message if applicable.
Expand Down Expand Up @@ -184,7 +186,19 @@ public String validate(String setting, String value) {
return null;
}
};


public static final Validator MEMORY_SIZE = new Validator() {
@Override
public String validate(String setting, String value) {
try {
parseBytesSizeValueOrHeapRatio(value);
} catch (ElasticsearchParseException ex) {
return ex.getMessage();
}
return null;
}
};

public static final Validator BOOLEAN = new Validator() {
@Override
public String validate(String setting, String value) {
Expand Down
10 changes: 8 additions & 2 deletions src/main/java/org/elasticsearch/common/unit/MemorySizeValue.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

package org.elasticsearch.common.unit;

import org.elasticsearch.ElasticsearchParseException;
import org.elasticsearch.monitor.jvm.JvmInfo;

import static org.elasticsearch.common.unit.ByteSizeValue.parseBytesSizeValue;
Expand All @@ -32,8 +33,13 @@ public enum MemorySizeValue {
* the heap is 1G, <tt>10%</tt> will be parsed as <tt>100mb</tt>. */
public static ByteSizeValue parseBytesSizeValueOrHeapRatio(String sValue) {
if (sValue.endsWith("%")) {
double percent = Double.parseDouble(sValue.substring(0, sValue.length() - 1));
return new ByteSizeValue((long) ((percent / 100) * JvmInfo.jvmInfo().getMem().getHeapMax().bytes()), ByteSizeUnit.BYTES);
final String percentAsString = sValue.substring(0, sValue.length() - 1);
try {
final double percent = Double.parseDouble(percentAsString);
return new ByteSizeValue((long) ((percent / 100) * JvmInfo.jvmInfo().getMem().getHeapMax().bytes()), ByteSizeUnit.BYTES);
} catch (NumberFormatException e) {
throw new ElasticsearchParseException("Failed to parse [" + percentAsString + "] as a double", e);
}
} else {
return parseBytesSizeValue(sValue);
}
Expand Down
15 changes: 6 additions & 9 deletions src/main/java/org/elasticsearch/indices/fielddata/breaker/InternalCircuitBreakerService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,6 @@
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.monitor.jvm.JvmInfo;
import org.elasticsearch.node.settings.NodeSettingsService;

/**
Expand All @@ -39,9 +38,7 @@ public class InternalCircuitBreakerService extends AbstractLifecycleComponent<In
public static final String CIRCUIT_BREAKER_OVERHEAD_SETTING = "indices.fielddata.breaker.overhead";

public static final double DEFAULT_OVERHEAD_CONSTANT = 1.03;

private static final long JVM_HEAP_MAX_BYTES = JvmInfo.jvmInfo().getMem().getHeapMax().bytes();
private static final long DEFAULT_BREAKER_LIMIT = (long) (0.8 * JVM_HEAP_MAX_BYTES); // 80% of the max heap
private static final String DEFAULT_BREAKER_LIMIT = "80%";

private volatile MemoryCircuitBreaker breaker;
private volatile long maxBytes;
Expand All @@ -50,7 +47,7 @@ public class InternalCircuitBreakerService extends AbstractLifecycleComponent<In
@Inject
public InternalCircuitBreakerService(Settings settings, NodeSettingsService nodeSettingsService) {
super(settings);
this.maxBytes = settings.getAsBytesSize(CIRCUIT_BREAKER_MAX_BYTES_SETTING, new ByteSizeValue(DEFAULT_BREAKER_LIMIT)).bytes();
this.maxBytes = settings.getAsMemory(CIRCUIT_BREAKER_MAX_BYTES_SETTING, DEFAULT_BREAKER_LIMIT).bytes();
this.overhead = settings.getAsDouble(CIRCUIT_BREAKER_OVERHEAD_SETTING, DEFAULT_OVERHEAD_CONSTANT);

this.breaker = new MemoryCircuitBreaker(new ByteSizeValue(maxBytes), overhead, null, logger);
Expand All @@ -62,21 +59,21 @@ class ApplySettings implements NodeSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
// clear breaker now that settings have changed
ByteSizeValue newMaxByteSizeValue = settings.getAsBytesSize(CIRCUIT_BREAKER_MAX_BYTES_SETTING, null);
long newMaxByteSizeValue = settings.getAsMemory(CIRCUIT_BREAKER_MAX_BYTES_SETTING, DEFAULT_BREAKER_LIMIT).bytes();
boolean breakerResetNeeded = false;

if (newMaxByteSizeValue != null) {
if (newMaxByteSizeValue != maxBytes) {
logger.info("updating [{}] from [{}] to [{}]", CIRCUIT_BREAKER_MAX_BYTES_SETTING,
new ByteSizeValue(InternalCircuitBreakerService.this.maxBytes), newMaxByteSizeValue);
InternalCircuitBreakerService.this.maxBytes = newMaxByteSizeValue.bytes();
maxBytes = newMaxByteSizeValue;
breakerResetNeeded = true;
}

double newOverhead = settings.getAsDouble(CIRCUIT_BREAKER_OVERHEAD_SETTING, overhead);
if (newOverhead != overhead) {
logger.info("updating [{}] from [{}] to [{}]", CIRCUIT_BREAKER_OVERHEAD_SETTING,
overhead, newOverhead);
InternalCircuitBreakerService.this.overhead = newOverhead;
overhead = newOverhead;
breakerResetNeeded = true;
}

Expand Down
12 changes: 10 additions & 2 deletions src/test/java/org/elasticsearch/indices/fielddata/breaker/CircuitBreakerServiceTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,10 +23,13 @@
import org.elasticsearch.client.Client;
import org.elasticsearch.common.collect.MapBuilder;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.monitor.jvm.JvmInfo;
import org.elasticsearch.test.ElasticsearchIntegrationTest;
import org.elasticsearch.test.junit.annotations.TestLogging;
import org.junit.Test;

import java.util.Arrays;

import static org.elasticsearch.common.settings.ImmutableSettings.settingsBuilder;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;

Expand All @@ -36,6 +39,11 @@
@ElasticsearchIntegrationTest.ClusterScope(scope = ElasticsearchIntegrationTest.Scope.TEST)
public class CircuitBreakerServiceTests extends ElasticsearchIntegrationTest {

private String randomRidiculouslySmallLimit() {
// 3 different ways to say 100 bytes
return randomFrom(Arrays.asList("100b", "100", (10000. / JvmInfo.jvmInfo().getMem().getHeapMax().bytes()) + "%"));
}

@Test
@TestLogging("org.elasticsearch.indices.fielddata.breaker:TRACE,org.elasticsearch.index.fielddata:TRACE,org.elasticsearch.common.breaker:TRACE")
public void testMemoryBreaker() {
Expand Down Expand Up @@ -63,7 +71,7 @@ public void testMemoryBreaker() {

// Update circuit breaker settings
Settings settings = settingsBuilder()
.put(InternalCircuitBreakerService.CIRCUIT_BREAKER_MAX_BYTES_SETTING, "100b")
.put(InternalCircuitBreakerService.CIRCUIT_BREAKER_MAX_BYTES_SETTING, randomRidiculouslySmallLimit())
.put(InternalCircuitBreakerService.CIRCUIT_BREAKER_OVERHEAD_SETTING, 1.05)
.build();
client.admin().cluster().prepareUpdateSettings().setTransientSettings(settings).execute().actionGet();
Expand Down Expand Up @@ -120,7 +128,7 @@ public void testRamAccountingTermsEnum() {

// Update circuit breaker settings
Settings settings = settingsBuilder()
.put(InternalCircuitBreakerService.CIRCUIT_BREAKER_MAX_BYTES_SETTING, "100b")
.put(InternalCircuitBreakerService.CIRCUIT_BREAKER_MAX_BYTES_SETTING, randomRidiculouslySmallLimit())
.put(InternalCircuitBreakerService.CIRCUIT_BREAKER_OVERHEAD_SETTING, 1.05)
.build();
client.admin().cluster().prepareUpdateSettings().setTransientSettings(settings).execute().actionGet();
Expand Down

0 comments on commit a77f1cb

Please sign in to comment.