diff --git a/data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchProcessImpl.java b/data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchProcessImpl.java index 5b2d7b50874b..8a949b34c958 100644 --- a/data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchProcessImpl.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchProcessImpl.java @@ -60,7 +60,10 @@ import javax.annotation.Nonnull; import javax.net.ssl.X509TrustManager; +import java.io.IOException; import java.net.URI; +import java.nio.file.Files; +import java.nio.file.Path; import java.security.KeyStore; import java.util.List; import java.util.Locale; @@ -71,6 +74,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.stream.Stream; import static org.graylog2.shared.utilities.StringUtils.f; @@ -79,6 +83,7 @@ public class OpensearchProcessImpl implements OpensearchProcess, ProcessListener private static final Logger LOG = LoggerFactory.getLogger(OpensearchProcessImpl.class); private static final long MEMORY_RATIO_THRESHOLD = 2; private static final int CLUSTER_REQUEST_TIMEOUT = 30; + private static final long CGROUP_V1_UNLIMITED_THRESHOLD = 1L << 60; @SuppressWarnings("OptionalUsedAsFieldOrParameterType") private Optional opensearchConfiguration = Optional.empty(); @@ -216,24 +221,110 @@ public void removeConfiguration() { void checkConfiguredHeap() { Size heap = Size.parse(configuration.getOpensearchHeap()); long heapBytes = heap.toBytes(); - final GlobalMemory memory = getGlobalMemory(); + final MemoryValues memoryValues = getContainerMemory().orElseGet(() -> { + final GlobalMemory memory = getGlobalMemory(); + return new MemoryValues(memory.getTotal(), memory.getAvailable()); + }); long buffer = 2 * 1024 * 1024 * 1024L; - long freeMemory = memory.getAvailable() - buffer; + long freeMemory = memoryValues.available() - buffer; float memoryRatio = (float) freeMemory / heapBytes; if (memoryRatio > MEMORY_RATIO_THRESHOLD) { LOG.warn("There appears to be about {} times more available memory than the heap size configured for this data node.", memoryRatio); - final String recommendedMemory = FileUtils.byteCountToDisplaySize(memory.getTotal() / 2); + final String recommendedMemory = FileUtils.byteCountToDisplaySize(memoryValues.total() / 2); clusterEventBus.post(new DataNodeNotficationEvent(nodeId.getNodeId(), Notification.Type.DATA_NODE_HEAP_WARNING, Map.of("hostname", configuration.getHostname(), "memoryRatio", f("%.1f", memoryRatio), - "totalMemory", FileUtils.byteCountToDisplaySize(memory.getTotal()), - "availableMemory", FileUtils.byteCountToDisplaySize(memory.getAvailable()), + "totalMemory", FileUtils.byteCountToDisplaySize(memoryValues.total()), + "availableMemory", FileUtils.byteCountToDisplaySize(memoryValues.available()), "recommendedMemory", recommendedMemory, "recommendedMemorySetting", recommendedMemorySetting(recommendedMemory), "heapSize", FileUtils.byteCountToDisplaySize(heapBytes)))); } } + @VisibleForTesting + Optional getContainerMemory() { + return readCgroupV2Memory().or(this::readCgroupV1Memory); + } + + private Optional readCgroupV2Memory() { + final Optional rawLimit = readFirstLine(cgroupV2MemoryMaxPath()); + final Optional rawUsed = readFirstLine(cgroupV2MemoryCurrentPath()); + if (rawLimit.isEmpty() || rawUsed.isEmpty()) { + return Optional.empty(); + } + + if ("max".equalsIgnoreCase(rawLimit.get().trim())) { + return Optional.empty(); + } + + final Optional limit = parseLong(rawLimit.get()); + final Optional used = parseLong(rawUsed.get()); + if (limit.isEmpty() || used.isEmpty()) { + return Optional.empty(); + } + + return memoryValues(limit.get(), used.get()); + } + + private Optional readCgroupV1Memory() { + final Optional limit = readFirstLine(cgroupV1MemoryLimitPath()).flatMap(this::parseLong); + final Optional used = readFirstLine(cgroupV1MemoryUsagePath()).flatMap(this::parseLong); + if (limit.isEmpty() || used.isEmpty()) { + return Optional.empty(); + } + + if (limit.get() >= CGROUP_V1_UNLIMITED_THRESHOLD) { + return Optional.empty(); + } + + return memoryValues(limit.get(), used.get()); + } + + private Optional memoryValues(long total, long used) { + if (total <= 0 || used < 0) { + return Optional.empty(); + } + return Optional.of(new MemoryValues(total, Math.max(0, total - used))); + } + + private Optional parseLong(String value) { + try { + return Optional.of(Long.parseLong(value.trim())); + } catch (NumberFormatException ignored) { + return Optional.empty(); + } + } + + @VisibleForTesting + Optional readFirstLine(Path path) { + try (Stream lines = Files.lines(path)) { + return lines.findFirst(); + } catch (IOException ignored) { + return Optional.empty(); + } + } + + @VisibleForTesting + Path cgroupV2MemoryMaxPath() { + return Path.of("/sys/fs/cgroup/memory.max"); + } + + @VisibleForTesting + Path cgroupV2MemoryCurrentPath() { + return Path.of("/sys/fs/cgroup/memory.current"); + } + + @VisibleForTesting + Path cgroupV1MemoryLimitPath() { + return Path.of("/sys/fs/cgroup/memory/memory.limit_in_bytes"); + } + + @VisibleForTesting + Path cgroupV1MemoryUsagePath() { + return Path.of("/sys/fs/cgroup/memory/memory.usage_in_bytes"); + } + protected static String recommendedMemorySetting(String recommendedMemory) { // opensearch values need to be in different format, same as java heap config values. Example: 7GB => 7g, 512MB => 512m return recommendedMemory.replace("B", "").replace(" ", "").toLowerCase(Locale.ROOT); @@ -245,6 +336,9 @@ protected GlobalMemory getGlobalMemory() { return memory; } + @VisibleForTesting + record MemoryValues(long total, long available) {} + @Subscribe public void onNotificationEvent(DataNodeNotficationEvent event) { // we need a subscriber in the data node, otherwise this event will be ignored due to it having no subscribers diff --git a/data-node/src/test/java/org/graylog/datanode/opensearch/OpensearchProcessImplTest.java b/data-node/src/test/java/org/graylog/datanode/opensearch/OpensearchProcessImplTest.java index c1835d25c2f4..d0536db3da79 100644 --- a/data-node/src/test/java/org/graylog/datanode/opensearch/OpensearchProcessImplTest.java +++ b/data-node/src/test/java/org/graylog/datanode/opensearch/OpensearchProcessImplTest.java @@ -33,6 +33,7 @@ import org.graylog2.security.CustomCAX509TrustManager; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; @@ -43,6 +44,8 @@ import oshi.hardware.VirtualMemory; import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.List; import java.util.Optional; import java.util.concurrent.ScheduledExecutorService; @@ -139,6 +142,7 @@ public void testShutdownWhenRemovedSuccessfully() throws IOException { public void testHeapThresholdWarning() { when(configuration.getHostname()).thenReturn("datanode"); when(configuration.getOpensearchHeap()).thenReturn("1g"); + when(opensearchProcess.getContainerMemory()).thenReturn(Optional.empty()); when(opensearchProcess.getGlobalMemory()).thenReturn(mockMemory(gigabytes(8), gigabytes(16))); opensearchProcess.checkConfiguredHeap(); verify(clusterEventBus, times(1)).post(any(DataNodeNotficationEvent.class)); @@ -147,11 +151,75 @@ public void testHeapThresholdWarning() { @Test public void testNoHeapThresholdWarning() { when(configuration.getOpensearchHeap()).thenReturn("1g"); + when(opensearchProcess.getContainerMemory()).thenReturn(Optional.empty()); when(opensearchProcess.getGlobalMemory()).thenReturn(mockMemory(gigabytes(2), gigabytes(3))); opensearchProcess.checkConfiguredHeap(); verifyNoInteractions(clusterEventBus); } + @Test + void testGetContainerMemoryCgroupV2(@TempDir Path tempDir) throws Exception { + final Path memoryMax = tempDir.resolve("memory.max"); + final Path memoryCurrent = tempDir.resolve("memory.current"); + Files.writeString(memoryMax, "4294967296\n"); + Files.writeString(memoryCurrent, "1073741824\n"); + + when(opensearchProcess.cgroupV2MemoryMaxPath()).thenReturn(memoryMax); + when(opensearchProcess.cgroupV2MemoryCurrentPath()).thenReturn(memoryCurrent); + + final Optional memory = opensearchProcess.getContainerMemory(); + + Assertions.assertThat(memory).isPresent(); + Assertions.assertThat(memory.get().total()).isEqualTo(4294967296L); + Assertions.assertThat(memory.get().available()).isEqualTo(3221225472L); + } + + @Test + void testGetContainerMemoryCgroupV1Fallback(@TempDir Path tempDir) throws Exception { + final Path memoryMax = tempDir.resolve("memory.max"); + final Path memoryCurrent = tempDir.resolve("memory.current"); + Files.writeString(memoryMax, "max\n"); + Files.writeString(memoryCurrent, "100\n"); + + final Path memoryLimit = tempDir.resolve("memory.limit_in_bytes"); + final Path memoryUsage = tempDir.resolve("memory.usage_in_bytes"); + Files.writeString(memoryLimit, "4294967296\n"); + Files.writeString(memoryUsage, "1073741824\n"); + + when(opensearchProcess.cgroupV2MemoryMaxPath()).thenReturn(memoryMax); + when(opensearchProcess.cgroupV2MemoryCurrentPath()).thenReturn(memoryCurrent); + when(opensearchProcess.cgroupV1MemoryLimitPath()).thenReturn(memoryLimit); + when(opensearchProcess.cgroupV1MemoryUsagePath()).thenReturn(memoryUsage); + + final Optional memory = opensearchProcess.getContainerMemory(); + + Assertions.assertThat(memory).isPresent(); + Assertions.assertThat(memory.get().total()).isEqualTo(4294967296L); + Assertions.assertThat(memory.get().available()).isEqualTo(3221225472L); + } + + @Test + void testGetContainerMemoryCgroupV1UnlimitedIgnored(@TempDir Path tempDir) throws Exception { + final Path memoryMax = tempDir.resolve("memory.max"); + final Path memoryCurrent = tempDir.resolve("memory.current"); + Files.writeString(memoryMax, "max\n"); + Files.writeString(memoryCurrent, "100\n"); + + final Path memoryLimit = tempDir.resolve("memory.limit_in_bytes"); + final Path memoryUsage = tempDir.resolve("memory.usage_in_bytes"); + Files.writeString(memoryLimit, Long.toString(Long.MAX_VALUE)); + Files.writeString(memoryUsage, "1073741824\n"); + + when(opensearchProcess.cgroupV2MemoryMaxPath()).thenReturn(memoryMax); + when(opensearchProcess.cgroupV2MemoryCurrentPath()).thenReturn(memoryCurrent); + when(opensearchProcess.cgroupV1MemoryLimitPath()).thenReturn(memoryLimit); + when(opensearchProcess.cgroupV1MemoryUsagePath()).thenReturn(memoryUsage); + + final Optional memory = opensearchProcess.getContainerMemory(); + + Assertions.assertThat(memory).isEmpty(); + } + private GlobalMemory mockMemory(long availableMemory, long totalMemory) { return new GlobalMemory() {