improved detection and monitoring

Author: hmottestad

core/http/client/src/main/java/org/eclipse/rdf4j/http/client/QueryCircuitBreaker.java

@@ -138,7 +138,7 @@ public void beforeExecution(QueryCircuitBreakerHandle handle) throws QueryInterr
 		}
 
 		if (state == QueryPressureState.WARN && configuration.getWarnAdmissionDelayMs() > 0) {
-			delay(configuration.getWarnAdmissionDelayMs(), configuration, false);
+			delay(configuration.getWarnAdmissionDelayMs(), state, configuration, false);
 		}
 
 		if (state.rejectsNewQueries()) {
@@ -180,8 +180,8 @@ public void checkpoint(QueryCircuitBreakerHandle handle, String operator) throws
 			throw CircuitBreakerException.cancelled(handle.getCancellationState(), configuration.getRetryAfterSeconds(),
 					handle.getCancellationReason());
 		}
-		if (state == QueryPressureState.WARN && configuration.getCheckpointDelayMs() > 0) {
-			delay(configuration.getCheckpointDelayMs(), configuration, true);
+		if (state.ordinal() >= QueryPressureState.HIGH.ordinal() && configuration.getCheckpointDelayMs() > 0) {
+			delay(configuration.getCheckpointDelayMs(), state, configuration, true);
 		}
 	}
 
@@ -224,13 +224,13 @@ public static CircuitBreakerException asCircuitBreakerException(Throwable throwa
 		return null;
 	}
 
-	private void delay(long delayMs, Configuration configuration, boolean checkpointDelay)
+	private void delay(long delayMs, QueryPressureState state, Configuration configuration, boolean checkpointDelay)
 			throws QueryInterruptedException {
 		try {
 			sleeper.sleep(delayMs);
 		} catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
-			throw CircuitBreakerException.cancelled(QueryPressureState.WARN, configuration.getRetryAfterSeconds(),
+			throw CircuitBreakerException.cancelled(state, configuration.getRetryAfterSeconds(),
 					checkpointDelay ? "Query interrupted during breaker checkpoint delay"
 							: "Query interrupted during breaker admission delay",
 					e);
@@ -644,6 +644,8 @@ public boolean isCancelRequested() {
 	}
 
 	public static final class Configuration {
+		private static final long BYTES_PER_MB = 1024L * 1024L;
+
 		private final boolean enabled;
 		private final int warnGcMs;
 		private final int highGcMs;
@@ -677,15 +679,29 @@ static Configuration fromSystemProperties() {
 					getIntProperty(WARN_GC_MS_PROPERTY, 200),
 					getIntProperty(HIGH_GC_MS_PROPERTY, 400),
 					getIntProperty(CRITICAL_GC_MS_PROPERTY, 800),
-					getIntProperty(WARN_FREE_MB_PROPERTY, 256),
-					getIntProperty(HIGH_FREE_MB_PROPERTY, 128),
+					getIntProperty(WARN_FREE_MB_PROPERTY, defaultWarnFreeMb()),
+					getIntProperty(HIGH_FREE_MB_PROPERTY, defaultHighFreeMb()),
 					getIntProperty(CRITICAL_FREE_MB_PROPERTY, 96),
 					getIntProperty(WARN_ADMISSION_DELAY_MS_PROPERTY, 50),
 					getIntProperty(CHECKPOINT_DELAY_MS_PROPERTY, 1),
 					getIntProperty(CANCEL_COOLDOWN_MS_PROPERTY, 1000),
 					getIntProperty(RETRY_AFTER_SECONDS_PROPERTY, 3));
 		}
 
+		private static int defaultWarnFreeMb() {
+			return Math.max(256, percentOfMaxMemoryMb(2));
+		}
+
+		private static int defaultHighFreeMb() {
+			return Math.max(128, percentOfMaxMemoryMb(1));
+		}
+
+		private static int percentOfMaxMemoryMb(int percent) {
+			long scaledBytes = Runtime.getRuntime().maxMemory() * percent;
+			long divisor = 100L * BYTES_PER_MB;
+			return (int) ((scaledBytes + divisor - 1) / divisor);
+		}
+
 		private static int getIntProperty(String propertyName, int defaultValue) {
 			String rawValue = System.getProperty(propertyName);
 			if (rawValue == null || rawValue.isBlank()) {

core/http/client/src/test/java/org/eclipse/rdf4j/http/client/QueryCircuitBreakerTest.java

@@ -21,6 +21,8 @@
 
 import java.lang.reflect.Constructor;
 import java.lang.reflect.Field;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
@@ -66,6 +68,14 @@ void shouldStartBackgroundMonitorThreadForHighMemoryPressure() throws Exception
 		}
 	}
 
+	@Test
+	void shouldDeriveDefaultFreeMemoryThresholdsFromMaxHeap() throws Exception {
+		Thresholds thresholds = readDefaultThresholdsFromFreshJvm("12801m");
+
+		assertEquals(257, thresholds.warnFreeMb);
+		assertEquals(129, thresholds.highFreeMb);
+	}
+
 	@Test
 	void shouldThrottleExecutionContextCheckpointToEvery16384Calls() throws Exception {
 		PropertiesScope properties = new PropertiesScope().with(QueryCircuitBreaker.ENABLED_PROPERTY, "false");
@@ -221,6 +231,25 @@ void shouldHoldCriticalStateUntilRecoveryCooldownExpires() {
 		assertEquals("NORMAL", breaker.snapshotStatus().getState());
 	}
 
+	@Test
+	void shouldThrottleCheckpointsOnlyAtHighOrAbove() {
+		Fixture fixture = new Fixture();
+		QueryCircuitBreaker breaker = fixture.breaker(configuration(true, 100, 200, 300, 400, 300, 200, 25, 10, 1000,
+				7), 1000);
+		QueryCircuitBreakerHandle handle = breaker.register(QueryCircuitBreakerHandle.Source.SERVER, "repo",
+				"checkpoint-query");
+
+		fixture.freeMemoryMb.set(350);
+		breaker.checkpoint(handle, "JOIN");
+		assertEquals(List.of(), fixture.sleeps);
+
+		fixture.freeMemoryMb.set(250);
+		breaker.checkpoint(handle, "JOIN");
+		assertEquals(List.of(10L), fixture.sleeps);
+
+		breaker.complete(handle);
+	}
+
 	@Test
 	void shouldDelayWarnAdmissionAndRejectHighAdmission() {
 		Fixture fixture = new Fixture();
@@ -388,6 +417,23 @@ private static QueryPressureState waitForCurrentState(QueryCircuitBreaker breake
 		return state;
 	}
 
+	private static Thresholds readDefaultThresholdsFromFreshJvm(String maxHeap) throws Exception {
+		String classpath = System.getProperty("java.class.path");
+		String javaExecutable = Path.of(System.getProperty("java.home"), "bin", "java").toString();
+		Process process = new ProcessBuilder(javaExecutable, "-Xmx" + maxHeap,
+				"-D" + QueryCircuitBreaker.WARN_FREE_MB_PROPERTY + "=",
+				"-D" + QueryCircuitBreaker.HIGH_FREE_MB_PROPERTY + "=",
+				"-cp", classpath, DefaultThresholdProbe.class.getName())
+				.redirectErrorStream(true)
+				.start();
+		String output = new String(process.getInputStream().readAllBytes(), StandardCharsets.UTF_8).trim();
+		int exitCode = process.waitFor();
+		assertEquals(0, exitCode, output);
+
+		String[] parts = output.split(",");
+		return new Thresholds(Integer.parseInt(parts[0]), Integer.parseInt(parts[1]));
+	}
+
 	private static final class PropertiesScope {
 		private final List<String> keys = new ArrayList<>();
 		private final List<String> previousValues = new ArrayList<>();
@@ -400,9 +446,21 @@ private PropertiesScope with(String key, String value) {
 			return this;
 		}
 
+		private PropertiesScope without(String key) {
+			keys.add(key);
+			previousValues.add(System.getProperty(key));
+			nextValues.add(null);
+			return this;
+		}
+
 		private void apply() {
 			for (int i = 0; i < keys.size(); i++) {
-				System.setProperty(keys.get(i), nextValues.get(i));
+				String nextValue = nextValues.get(i);
+				if (nextValue == null) {
+					System.clearProperty(keys.get(i));
+				} else {
+					System.setProperty(keys.get(i), nextValue);
+				}
 			}
 		}
 
@@ -436,4 +494,23 @@ private QueryCircuitBreaker breaker(QueryCircuitBreaker.Configuration configurat
 					recoveryCooldownMs, 1, false);
 		}
 	}
+
+	public static final class DefaultThresholdProbe {
+		public static void main(String[] args) {
+			QueryCircuitBreaker.Configuration configuration = QueryCircuitBreaker.Configuration.fromSystemProperties();
+			System.out.print(configuration.getWarnFreeMb());
+			System.out.print(",");
+			System.out.print(configuration.getHighFreeMb());
+		}
+	}
+
+	private static final class Thresholds {
+		private final int warnFreeMb;
+		private final int highFreeMb;
+
+		private Thresholds(int warnFreeMb, int highFreeMb) {
+			this.warnFreeMb = warnFreeMb;
+			this.highFreeMb = highFreeMb;
+		}
+	}
 }