diff --git a/service/definition/src/main/java/ai/timefold/solver/service/definition/impl/executionprofile/MemoryDumpExecutionProfile.java b/service/definition/src/main/java/ai/timefold/solver/service/definition/impl/executionprofile/MemoryDumpExecutionProfile.java new file mode 100644 index 00000000000..148a5b16844 --- /dev/null +++ b/service/definition/src/main/java/ai/timefold/solver/service/definition/impl/executionprofile/MemoryDumpExecutionProfile.java @@ -0,0 +1,43 @@ +package ai.timefold.solver.service.definition.impl.executionprofile; + +import java.util.Map; + +import ai.timefold.solver.service.definition.internal.executionprofile.ExecutionProfile; + +/** + * Writes a thread dump and a heap dump when the solver runs short of memory, before the container is {@code OOMKilled}. + *

+ * The profile enables the solver worker's memory watchdog, which checks the process memory against the container limit every + * second. The first time it reaches the threshold, the watchdog writes both dumps into the execution-profile artifacts + * directory and fails the run, which uploads them. A memory spike that kills the container between two checks is not + * captured. + *

+ * The profile deliberately leaves the heap size alone, since lowering it could cause out-of-memory errors that would not + * otherwise occur. Instead, the watchdog fails the run early, leaving room below the container limit to write the dumps. + */ +public final class MemoryDumpExecutionProfile implements ExecutionProfile { + + /** Environment-variable form of {@code ai.timefold.solver.monitoring.memory.enabled}. */ + static final String ENV_MEMORY_WATCHDOG_ENABLED = "AI_TIMEFOLD_SOLVER_MONITORING_MEMORY_ENABLED"; + + @Override + public String id() { + return "memory-dump"; + } + + @Override + public String name() { + return "Thread and heap dump on high memory"; + } + + @Override + public String description() { + return "Writes a thread dump and a heap dump to the run's execution artifacts and fails the run when memory usage " + + "gets close to the limit, before the container is killed."; + } + + @Override + public Map toEnvironment(Map options) { + return Map.of(ENV_MEMORY_WATCHDOG_ENABLED, "true"); + } +} diff --git a/service/definition/src/main/resources/META-INF/services/ai.timefold.solver.service.definition.internal.executionprofile.ExecutionProfile b/service/definition/src/main/resources/META-INF/services/ai.timefold.solver.service.definition.internal.executionprofile.ExecutionProfile index 51bcdb4831d..a6e23cdd8f3 100644 --- a/service/definition/src/main/resources/META-INF/services/ai.timefold.solver.service.definition.internal.executionprofile.ExecutionProfile +++ b/service/definition/src/main/resources/META-INF/services/ai.timefold.solver.service.definition.internal.executionprofile.ExecutionProfile @@ -1 +1,2 @@ ai.timefold.solver.service.definition.impl.executionprofile.SeedExecutionProfile +ai.timefold.solver.service.definition.impl.executionprofile.MemoryDumpExecutionProfile diff --git a/service/definition/src/test/java/ai/timefold/solver/service/definition/impl/executionprofile/MemoryDumpExecutionProfileTest.java b/service/definition/src/test/java/ai/timefold/solver/service/definition/impl/executionprofile/MemoryDumpExecutionProfileTest.java new file mode 100644 index 00000000000..d52272964e2 --- /dev/null +++ b/service/definition/src/test/java/ai/timefold/solver/service/definition/impl/executionprofile/MemoryDumpExecutionProfileTest.java @@ -0,0 +1,29 @@ +package ai.timefold.solver.service.definition.impl.executionprofile; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.Map; + +import org.junit.jupiter.api.Test; + +class MemoryDumpExecutionProfileTest { + + private final MemoryDumpExecutionProfile profile = new MemoryDumpExecutionProfile(); + + @Test + void enablesMemoryWatchdog() { + assertThat(profile.toEnvironment(Map.of())) + .containsExactly(Map.entry("AI_TIMEFOLD_SOLVER_MONITORING_MEMORY_ENABLED", "true")); + } + + @Test + void ignoresUnrelatedOptions() { + assertThat(profile.toEnvironment(Map.of("seed", "1"))) + .isEqualTo(profile.toEnvironment(Map.of())); + } + + @Test + void acceptsNullOptions() { + assertThat(profile.toEnvironment(null)).isEqualTo(profile.toEnvironment(Map.of())); + } +} diff --git a/service/worker/src/main/java/ai/timefold/solver/service/worker/impl/monitoring/GcOverheadMonitoring.java b/service/worker/src/main/java/ai/timefold/solver/service/worker/impl/monitoring/GcOverheadMonitoring.java index eb1051caea8..d65f9bf3751 100644 --- a/service/worker/src/main/java/ai/timefold/solver/service/worker/impl/monitoring/GcOverheadMonitoring.java +++ b/service/worker/src/main/java/ai/timefold/solver/service/worker/impl/monitoring/GcOverheadMonitoring.java @@ -2,10 +2,14 @@ import java.lang.management.GarbageCollectorMXBean; import java.lang.management.ManagementFactory; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.List; +import java.util.OptionalLong; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import jakarta.annotation.PreDestroy; import jakarta.enterprise.context.ApplicationScoped; @@ -21,9 +25,21 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.sun.management.HotSpotDiagnosticMXBean; +import com.sun.management.HotSpotDiagnosticMXBean.ThreadDumpFormat; + import io.quarkus.arc.Unremovable; import io.quarkus.runtime.StartupEvent; +/** + * Fails the run when the JVM spends too much time in garbage collection. + *

+ * When the memory watchdog is enabled ({@code ai.timefold.solver.monitoring.memory.enabled}, set by an execution profile), it + * also checks the process memory against the container memory limit at a short interval. The first time it reaches the + * threshold, it writes a thread dump and a heap dump into the execution-profile artifacts directory and fails the run, which + * uploads the artifacts. This happens before the container would be {@code OOMKilled}, so there is still room to write the + * dumps. + */ @ApplicationScoped @Unremovable public class GcOverheadMonitoring { @@ -35,6 +51,10 @@ public class GcOverheadMonitoring { private final SolverWorker solverWorker; private final List gcBeans; private final int gcThresholdPct; + private final boolean memoryWatchdogEnabled; + private final int memoryThresholdPct; + private final int memoryIntervalMillis; + private final AtomicBoolean memoryWatchdogTriggered = new AtomicBoolean(false); private long prevGcTime = 0; @@ -48,12 +68,21 @@ public class GcOverheadMonitoring { public GcOverheadMonitoring(SolverWorker solverWorker, @ConfigProperty(name = "ai.timefold.solver.monitoring.gc.thresholdPct", defaultValue = "100") Integer gcThresholdPct, - @ConfigProperty(name = "ai.timefold.solver.monitoring.gc.delaySeconds", defaultValue = "10") Integer delaySeconds) { + @ConfigProperty(name = "ai.timefold.solver.monitoring.gc.delaySeconds", defaultValue = "10") Integer delaySeconds, + @ConfigProperty(name = "ai.timefold.solver.monitoring.memory.enabled", + defaultValue = "false") Boolean memoryWatchdogEnabled, + @ConfigProperty(name = "ai.timefold.solver.monitoring.memory.thresholdPct", + defaultValue = "90") Integer memoryThresholdPct, + @ConfigProperty(name = "ai.timefold.solver.monitoring.memory.intervalMillis", + defaultValue = "1000") Integer memoryIntervalMillis) { this.solverWorker = solverWorker; this.gcBeans = ManagementFactory.getGarbageCollectorMXBeans(); this.prevGcTime = getTotalGcTime(); this.gcThresholdPct = gcThresholdPct; this.delaySeconds = delaySeconds; + this.memoryWatchdogEnabled = memoryWatchdogEnabled; + this.memoryThresholdPct = memoryThresholdPct; + this.memoryIntervalMillis = memoryIntervalMillis; } public void onStart(@Observes StartupEvent event) { @@ -83,6 +112,71 @@ public void onStart(@Observes StartupEvent event) { prevGcTime = currGcTime; }, delaySeconds, delaySeconds, TimeUnit.SECONDS); + + if (memoryWatchdogEnabled) { + LOGGER.info("Memory watchdog enabled: dumping and failing the run at {}% memory usage", memoryThresholdPct); + executor.scheduleAtFixedRate(() -> { + try { + checkMemory(id); + } catch (Throwable e) { + // An exception would cancel the scheduled task, so keep the watchdog alive. + LOGGER.warn("Memory watchdog check failed due to {}", e.getMessage()); + } + }, memoryIntervalMillis, memoryIntervalMillis, TimeUnit.MILLISECONDS); + } + } + } + + private void checkMemory(String id) { + OptionalLong rss = MemoryPressure.readRssBytes(); + OptionalLong containerLimit = MemoryPressure.containerMemoryLimitBytes(); + + String reason = MemoryPressure.exceededReason(rss, containerLimit, memoryThresholdPct); + if (reason == null || !memoryWatchdogTriggered.compareAndSet(false, true)) { + return; + } + LOGGER.error("Memory usage exceeded {}% ({}) - writing thread and heap dumps and shutting down", + memoryThresholdPct, reason); + Path dumpDir = createDumpDirectory(); + if (dumpDir != null) { + writeDumps(dumpDir); + } + solverWorker.notifyOnFailure(id, new TimefoldRuntimeException(ErrorCodes.SOLVER_UNKNOWN, + "Not enough memory available to solve dataset - configure memory via configuration profile", + new OutOfMemoryError("Memory usage exceeded " + memoryThresholdPct + "% (" + reason + ")"), + false)); + System.exit(JVM_MONITORING_EXIST_STATUS_CODE); + } + + private static Path createDumpDirectory() { + String dir = System.getenv(EnvironmentVars.ENV_TIMEFOLD_EXECUTION_PROFILE_DIR); + Path dumpDir = Path.of(dir != null && !dir.isBlank() ? dir : EnvironmentVars.DEFAULT_EXECUTION_PROFILE_DIR) + .toAbsolutePath(); + try { + return Files.createDirectories(dumpDir); + } catch (Exception e) { + LOGGER.warn("Unable to create dump directory {} due to {}", dumpDir, e.getMessage()); + return null; + } + } + + private static void writeDumps(Path dumpDir) { + long timestamp = System.currentTimeMillis(); + HotSpotDiagnosticMXBean diagnostics = ManagementFactory.getPlatformMXBean(HotSpotDiagnosticMXBean.class); + // Thread dump first: it is cheap and captures the threads before the heap dump pauses the JVM. + Path threadDump = dumpDir.resolve("thread-dump-" + timestamp + ".txt"); + try { + diagnostics.dumpThreads(threadDump.toString(), ThreadDumpFormat.TEXT_PLAIN); + LOGGER.info("Thread dump written to {}", threadDump); + } catch (Exception e) { + LOGGER.warn("Unable to write thread dump to {} due to {}", threadDump, e.getMessage()); + } + Path heapDump = dumpDir.resolve("heap-dump-" + timestamp + ".hprof"); + try { + diagnostics.dumpHeap(heapDump.toString(), true); + LOGGER.info("Heap dump written to {}", heapDump); + } catch (Exception e) { + LOGGER.warn("Unable to write heap dump to {} due to {}", heapDump, e.getMessage()); } } diff --git a/service/worker/src/main/java/ai/timefold/solver/service/worker/impl/monitoring/MemoryPressure.java b/service/worker/src/main/java/ai/timefold/solver/service/worker/impl/monitoring/MemoryPressure.java new file mode 100644 index 00000000000..ce1db265bb4 --- /dev/null +++ b/service/worker/src/main/java/ai/timefold/solver/service/worker/impl/monitoring/MemoryPressure.java @@ -0,0 +1,99 @@ +package ai.timefold.solver.service.worker.impl.monitoring; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.OptionalLong; + +import ai.timefold.solver.service.definition.internal.platform.EnvironmentVars; + +/** + * Reads the process memory the memory watchdog in {@link GcOverheadMonitoring} compares against the container memory limit. + *

+ * The process resident set size (RSS) is the memory the process actually uses, both the heap and everything outside it, so it + * is the closest single figure to what gets a container {@code OOMKilled}. It also counts heap that only holds garbage not yet + * collected, so the threshold should leave some margin for that. + */ +final class MemoryPressure { + + static final Path PROC_SELF_STATUS = Path.of("/proc/self/status"); + static final Path CGROUP_V2_MEMORY_MAX = Path.of("/sys/fs/cgroup/memory.max"); + static final Path CGROUP_V1_MEMORY_LIMIT = Path.of("/sys/fs/cgroup/memory/memory.limit_in_bytes"); + + /** cgroup v1 reports an unlimited container as a value close to {@link Long#MAX_VALUE}. */ + private static final long UNLIMITED_THRESHOLD = 1L << 60; + + private MemoryPressure() { + } + + static OptionalLong readRssBytes() { + try { + return parseRssBytes(Files.readAllLines(PROC_SELF_STATUS)); + } catch (IOException | RuntimeException e) { + // Not on Linux or not readable; the RSS signal is simply unavailable. + return OptionalLong.empty(); + } + } + + static OptionalLong parseRssBytes(List statusLines) { + for (String line : statusLines) { + if (line.startsWith("VmRSS:")) { + String[] parts = line.substring("VmRSS:".length()).trim().split("\\s+"); + try { + return OptionalLong.of(Long.parseLong(parts[0]) * 1024); + } catch (NumberFormatException e) { + return OptionalLong.empty(); + } + } + } + return OptionalLong.empty(); + } + + static OptionalLong containerMemoryLimitBytes() { + OptionalLong fromEnv = parseLimitBytes(System.getenv(EnvironmentVars.K8S_INFO_MEMORY_LIMIT)); + if (fromEnv.isPresent()) { + return fromEnv; + } + for (Path path : List.of(CGROUP_V2_MEMORY_MAX, CGROUP_V1_MEMORY_LIMIT)) { + try { + OptionalLong limit = parseLimitBytes(Files.readString(path)); + if (limit.isPresent()) { + return limit; + } + } catch (IOException | RuntimeException e) { + // Try the next location. + } + } + return OptionalLong.empty(); + } + + static OptionalLong parseLimitBytes(String value) { + if (value == null || value.isBlank()) { + return OptionalLong.empty(); + } + try { + long limit = Long.parseLong(value.trim()); + return limit > 0 && limit < UNLIMITED_THRESHOLD ? OptionalLong.of(limit) : OptionalLong.empty(); + } catch (NumberFormatException e) { + // cgroup v2 reports an unlimited container as "max". + return OptionalLong.empty(); + } + } + + /** + * @return a description of the memory usage if it reached the threshold, or null otherwise + */ + static String exceededReason(OptionalLong rss, OptionalLong containerLimit, int thresholdPct) { + if (rss.isPresent() && containerLimit.isPresent() + && rss.getAsLong() * 100 >= containerLimit.getAsLong() * thresholdPct) { + return "process memory " + toMiB(rss.getAsLong()) + " MiB of container limit " + + toMiB(containerLimit.getAsLong()) + " MiB"; + } + return null; + } + + private static long toMiB(long bytes) { + return bytes / (1024 * 1024); + } +} diff --git a/service/worker/src/test/java/ai/timefold/solver/service/worker/impl/monitoring/MemoryPressureTest.java b/service/worker/src/test/java/ai/timefold/solver/service/worker/impl/monitoring/MemoryPressureTest.java new file mode 100644 index 00000000000..e22156cfd4c --- /dev/null +++ b/service/worker/src/test/java/ai/timefold/solver/service/worker/impl/monitoring/MemoryPressureTest.java @@ -0,0 +1,58 @@ +package ai.timefold.solver.service.worker.impl.monitoring; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.List; +import java.util.OptionalLong; + +import org.junit.jupiter.api.Test; + +class MemoryPressureTest { + + private static final long MIB = 1024 * 1024; + + @Test + void parsesRssFromProcStatus() { + List status = List.of("Name:\tjava", "VmPeak:\t 9000 kB", "VmRSS:\t 2048 kB", "Threads:\t42"); + assertThat(MemoryPressure.parseRssBytes(status)).hasValue(2048 * 1024); + } + + @Test + void rssUnavailableWithoutVmRssLine() { + assertThat(MemoryPressure.parseRssBytes(List.of("Name:\tjava"))).isEmpty(); + } + + @Test + void parsesLimit() { + assertThat(MemoryPressure.parseLimitBytes("1073741824\n")).hasValue(1073741824); + } + + @Test + void treatsUnlimitedAsNoLimit() { + assertThat(MemoryPressure.parseLimitBytes("max\n")).isEmpty(); // cgroup v2 + assertThat(MemoryPressure.parseLimitBytes("9223372036854771712")).isEmpty(); // cgroup v1 + assertThat(MemoryPressure.parseLimitBytes(null)).isEmpty(); + assertThat(MemoryPressure.parseLimitBytes("")).isEmpty(); + } + + @Test + void belowThreshold() { + assertThat(MemoryPressure.exceededReason(OptionalLong.of(899 * MIB), OptionalLong.of(1000 * MIB), 90)).isNull(); + } + + @Test + void processMemoryReachesThreshold() { + assertThat(MemoryPressure.exceededReason(OptionalLong.of(900 * MIB), OptionalLong.of(1000 * MIB), 90)) + .isEqualTo("process memory 900 MiB of container limit 1000 MiB"); + } + + @Test + void ignoredWithoutContainerLimit() { + assertThat(MemoryPressure.exceededReason(OptionalLong.of(5000 * MIB), OptionalLong.empty(), 90)).isNull(); + } + + @Test + void ignoredWithoutRss() { + assertThat(MemoryPressure.exceededReason(OptionalLong.empty(), OptionalLong.of(1000 * MIB), 90)).isNull(); + } +}