Add data node heap check (#20168)

* add data node hep check and notification

* fix template names

* suppress forbidden invocation as it is handled by exception handling

* add recommended heap setting to notification

* changelog

* switch to oshi

* use floating point arithmetic

* change system notification text

* fixed test

* add paragraphs

---------

Co-authored-by: Tomas Dvorak <[email protected]>

Author: moesterheld

changelog/unreleased/pr-20168.toml

@@ -0,0 +1,5 @@
+type = "a"
+message = "Add system notification if heap size for data node is potentially too small."
+
+issues = ["Graylog2/graylog-plugin-enterprise#6641"]
+pulls = ["20168"]

data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchProcessImpl.java

@@ -17,11 +17,15 @@
 package org.graylog.datanode.opensearch;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.github.joschi.jadconfig.util.Size;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.eventbus.EventBus;
+import com.google.common.eventbus.Subscribe;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import jakarta.inject.Inject;
 import org.apache.commons.collections4.queue.CircularFifoQueue;
 import org.apache.commons.exec.ExecuteException;
+import org.apache.commons.io.FileUtils;
 import org.apache.http.client.utils.URIBuilder;
 import org.graylog.datanode.Configuration;
 import org.graylog.datanode.configuration.DatanodeConfiguration;
@@ -50,11 +54,17 @@
 import org.graylog.storage.opensearch2.OpenSearchClient;
 import org.graylog2.datanode.DataNodeLifecycleEvent;
 import org.graylog2.datanode.DataNodeLifecycleTrigger;
+import org.graylog2.datanode.DataNodeNotficationEvent;
+import org.graylog2.events.ClusterEventBus;
+import org.graylog2.notifications.Notification;
 import org.graylog2.plugin.system.NodeId;
 import org.graylog2.security.CustomCAX509TrustManager;
 import org.graylog2.security.TrustManagerAggregator;
+import org.graylog2.shared.SuppressForbidden;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import oshi.SystemInfo;
+import oshi.hardware.GlobalMemory;
 
 import javax.annotation.Nonnull;
 import javax.net.ssl.TrustManager;
@@ -64,17 +74,20 @@
 import java.security.KeyStore;
 import java.util.List;
 import java.util.Locale;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Queue;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
+import static org.graylog2.shared.utilities.StringUtils.f;
+
 public class OpensearchProcessImpl implements OpensearchProcess, ProcessListener {
 
     private static final Logger LOG = LoggerFactory.getLogger(OpensearchProcessImpl.class);
-
+    private static final long MEMORY_RATIO_THRESHOLD = 2;
 
     @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
     private Optional<OpensearchConfiguration> opensearchConfiguration = Optional.empty();
@@ -96,6 +109,7 @@ public class OpensearchProcessImpl implements OpensearchProcess, ProcessListener
     private final String nodeName;
     private final NodeId nodeId;
     private final EventBus eventBus;
+    private final ClusterEventBus clusterEventBus;
 
 
     static final String CLUSTER_ROUTING_ALLOCATION_EXCLUDE_SETTING = "cluster.routing.allocation.exclude._name";
@@ -105,7 +119,8 @@ public class OpensearchProcessImpl implements OpensearchProcess, ProcessListener
     @Inject
     OpensearchProcessImpl(DatanodeConfiguration datanodeConfiguration, final CustomCAX509TrustManager trustManager,
                           final Configuration configuration,
-                          ObjectMapper objectMapper, OpensearchStateMachine processState, NodeId nodeId, EventBus eventBus) {
+                          ObjectMapper objectMapper, OpensearchStateMachine processState, NodeId nodeId, EventBus eventBus,
+                          ClusterEventBus clusterEventBus) {
         this.datanodeConfiguration = datanodeConfiguration;
         this.processState = processState;
         this.stdout = new CircularFifoQueue<>(datanodeConfiguration.processLogsBufferSize());
@@ -116,6 +131,8 @@ public class OpensearchProcessImpl implements OpensearchProcess, ProcessListener
         this.nodeName = configuration.getDatanodeNodeName();
         this.nodeId = nodeId;
         this.eventBus = eventBus;
+        this.clusterEventBus = clusterEventBus;
+        eventBus.register(this);
     }
 
     private RestHighLevelClient createRestClient(OpensearchConfiguration configuration) {
@@ -214,6 +231,37 @@ private void configure() {
         );
     }
 
+    @VisibleForTesting
+    void checkConfiguredHeap() {
+        Size heap = Size.parse(configuration.getOpensearchHeap());
+        long heapBytes = heap.toBytes();
+        final GlobalMemory memory = getGlobalMemory();
+        long buffer = 2 * 1024 * 1024 * 1024L;
+        long freeMemory = memory.getAvailable() - 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);
+            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()),
+                            "recommendedMemory", FileUtils.byteCountToDisplaySize(memory.getTotal()/2),
+                            "heapSize", FileUtils.byteCountToDisplaySize(heapBytes))));
+        }
+    }
+
+    protected GlobalMemory getGlobalMemory() {
+        SystemInfo systemInfo = new SystemInfo();
+        GlobalMemory memory = systemInfo.getHardware().getMemory();
+        return memory;
+    }
+
+    @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
+    }
+
     @Override
     public synchronized void start() {
         opensearchConfiguration.ifPresentOrElse(
@@ -228,9 +276,11 @@ public synchronized void start() {
 
                     restClient = Optional.of(createRestClient(config));
                     openSearchClient = restClient.map(c -> new OpenSearchClient(c, objectMapper));
-                }),
-                () -> {throw new IllegalArgumentException("Opensearch configuration required but not supplied!");}
-        );
+                    checkConfiguredHeap();
+
+                    }),
+                    () -> {throw new IllegalArgumentException("Opensearch configuration required but not supplied!");}
+            );
     }
 
     /**

data-node/src/test/java/org/graylog/datanode/opensearch/OpensearchProcessImplTest.java

@@ -29,6 +29,8 @@
 import org.graylog.shaded.opensearch2.org.opensearch.client.RequestOptions;
 import org.graylog.shaded.opensearch2.org.opensearch.client.RestHighLevelClient;
 import org.graylog.shaded.opensearch2.org.opensearch.common.settings.Settings;
+import org.graylog2.datanode.DataNodeNotficationEvent;
+import org.graylog2.events.ClusterEventBus;
 import org.graylog2.plugin.system.NodeId;
 import org.graylog2.plugin.system.SimpleNodeId;
 import org.graylog2.security.CustomCAX509TrustManager;
@@ -38,8 +40,12 @@
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
+import oshi.hardware.GlobalMemory;
+import oshi.hardware.PhysicalMemory;
+import oshi.hardware.VirtualMemory;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.ScheduledExecutorService;
 
@@ -49,7 +55,9 @@
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
 
@@ -76,13 +84,15 @@ public class OpensearchProcessImplTest {
     RestHighLevelClient restClient;
     @Mock
     ClusterClient clusterClient;
+    @Mock
+    ClusterEventBus clusterEventBus;
 
     @Before
     public void setup() throws IOException {
         when(datanodeConfiguration.processLogsBufferSize()).thenReturn(100);
         when(configuration.getDatanodeNodeName()).thenReturn(nodeName);
         this.opensearchProcess = spy(new OpensearchProcessImpl(datanodeConfiguration, trustmManager, configuration,
-                 objectMapper, processState, nodeId, eventBus));
+                 objectMapper, processState, nodeId, eventBus, clusterEventBus));
         when(opensearchProcess.restClient()).thenReturn(Optional.of(restClient));
         when(restClient.cluster()).thenReturn(clusterClient);
     }
@@ -130,4 +140,55 @@ public void testShutdownWhenRemovedSuccessfully() throws IOException {
         verify(executor).shutdown();
     }
 
+    @Test
+    public void testHeapThresholdWarning() {
+        when(configuration.getHostname()).thenReturn("datanode");
+        when(configuration.getOpensearchHeap()).thenReturn("1g");
+        when(opensearchProcess.getGlobalMemory()).thenReturn(mockMemory(gigabytes(8), gigabytes(16)));
+        opensearchProcess.checkConfiguredHeap();
+        verify(clusterEventBus, times(1)).post(any(DataNodeNotficationEvent.class));
+    }
+
+    @Test
+    public void testNoHeapThresholdWarning() {
+        when(configuration.getOpensearchHeap()).thenReturn("1g");
+        when(opensearchProcess.getGlobalMemory()).thenReturn(mockMemory(gigabytes(2), gigabytes(3)));
+        opensearchProcess.checkConfiguredHeap();
+        verifyNoInteractions(clusterEventBus);
+    }
+
+    private GlobalMemory mockMemory(long availableMemory, long totalMemory) {
+        return new GlobalMemory() {
+
+            @Override
+            public long getTotal() {
+                return totalMemory;
+            }
+
+            @Override
+            public long getAvailable() {
+                return availableMemory;
+            }
+
+            @Override
+            public long getPageSize() {
+                throw new UnsupportedOperationException("Not supported here");
+            }
+
+            @Override
+            public VirtualMemory getVirtualMemory() {
+                throw new UnsupportedOperationException("Not supported here");
+            }
+
+            @Override
+            public List<PhysicalMemory> getPhysicalMemory() {
+                throw new UnsupportedOperationException("Not supported here");
+            }
+        };
+    }
+
+    private static long gigabytes(int i) {
+        return i * 1024 * 1024 * 1024L;
+    }
+
 }

graylog2-server/src/main/java/org/graylog/plugins/datanode/DataNodeModule.java

@@ -18,6 +18,7 @@
 
 import org.graylog2.datanode.DataNodeCommandService;
 import org.graylog2.datanode.DataNodeCommandServiceImpl;
+import org.graylog2.datanode.DataNodeEventService;
 import org.graylog2.migrations.V20231107164300_CreateDataNodeManagerRole;
 import org.graylog2.plugin.PluginModule;
 
@@ -26,6 +27,7 @@ public class DataNodeModule extends PluginModule {
     @Override
     protected void configure() {
         bind(DataNodeCommandService.class).to(DataNodeCommandServiceImpl.class);
+        bind(DataNodeEventService.class).asEagerSingleton();
         addMigration(V20231107164300_CreateDataNodeManagerRole.class);
     }
 

graylog2-server/src/main/java/org/graylog2/commands/Server.java

@@ -35,6 +35,7 @@
 import org.graylog.metrics.prometheus.PrometheusExporterConfiguration;
 import org.graylog.metrics.prometheus.PrometheusMetricsModule;
 import org.graylog.plugins.cef.CEFInputModule;
+import org.graylog.plugins.datanode.DataNodeModule;
 import org.graylog.plugins.formatting.units.UnitsModule;
 import org.graylog.plugins.map.MapWidgetModule;
 import org.graylog.plugins.map.config.GeoIpProcessorConfig;
@@ -212,7 +213,8 @@ public boolean isLocal() {
                 new DataTieringModule(),
                 new DatanodeMigrationBindings(),
                 new CaModule(),
-                new TelemetryModule()
+                new TelemetryModule(),
+                new DataNodeModule()
         );
 
         modules.add(new FieldTypeManagementModule());

graylog2-server/src/main/java/org/graylog2/datanode/DataNodeEventService.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog2.datanode;
+
+import com.google.common.eventbus.EventBus;
+import com.google.common.eventbus.Subscribe;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import org.graylog2.notifications.Notification;
+import org.graylog2.notifications.NotificationService;
+
+@Singleton
+public class DataNodeEventService {
+
+    private final NotificationService notificationService;
+
+    @Inject
+    public DataNodeEventService(EventBus eventBus, NotificationService notificationService) {
+        this.notificationService = notificationService;
+        eventBus.register(this);
+    }
+
+    @Subscribe
+    @SuppressWarnings("unused")
+    public void onNotificationEvent(DataNodeNotficationEvent event) {
+        Notification notification = notificationService.buildNow();
+        notification.addSeverity(Notification.Severity.NORMAL);
+        notification.addType(event.notificationType());
+        event.details().forEach(notification::addDetail);
+        notificationService.publishIfFirst(notification);
+    }
+
+}

graylog2-server/src/main/java/org/graylog2/datanode/DataNodeNotficationEvent.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog2.datanode;
+
+import org.graylog2.notifications.Notification;
+
+import java.util.Map;
+
+public record DataNodeNotficationEvent(String nodeId, Notification.Type notificationType, Map<String, Object> details) {
+}

graylog2-server/src/main/java/org/graylog2/notifications/Notification.java

@@ -99,7 +99,8 @@ enum Type {
         REMOTE_REINDEX_RUNNING,
         REMOTE_REINDEX_FINISHED,
         DATA_NODE_VERSION_MISMATCH,
-        DATA_TIERING_ROLLOVER_ERROR
+        DATA_TIERING_ROLLOVER_ERROR,
+        DATA_NODE_HEAP_WARNING
     }
 
     enum Severity {

graylog2-server/src/main/resources/org/graylog2/freemarker/templates/HTML/data_node_heap_warning.ftl

@@ -0,0 +1,17 @@
+<#if _title>
+    Data Node Heap Size Warning
+</#if>
+
+<#if _description>
+    <p>
+        There are data node nodes in the cluster which could potentially run with a higher configured heap size for better performance.
+    </p>
+    <p>
+        Data node <em>${hostname}</em> only has ${heapSize} Java Heap assigned, out of a total of ${totalMemory} RAM.<br/>
+        Currently, there is ${availableMemory} free memory available on the node. We recommend to make an additional half of this available to the Java Heap.
+    </p>
+    <p>
+        <em>Note: </em>For production performance, it is recommended to configure this node to use ${recommendedMemory} Java Heap (50% of RAM).<br/>
+        The Java Heap can be configured using the <em>opensearch_heap</em> configuration parameter in the node's configuration file (<em>datanode.conf</em>).
+    </p>
+</#if>

graylog2-server/src/main/resources/org/graylog2/freemarker/templates/PLAINTEXT/data_node_heap_warning.ftl

@@ -0,0 +1,11 @@
+<#if _title>
+    Data Node Heap Size Warning
+</#if>
+
+<#if _description>
+    There are data node nodes in the cluster which could potentially run with a higher configured heap size for better performance.
+    Data node ${hostname} only has ${heapSize} Java Heap assigned, out of a total of ${totalMemory} RAM.
+    Currently, there is ${availableMemory} free memory available on the node. We recommend to make an additional half of this available to the Java Heap.
+    Note: For production performance, it is recommended to configure this node to use ${recommendedMemory} Java Heap (50% of RAM).
+    The Java Heap can be configured using the opensearch_heap configuration parameter in the node's configuration file (datanode.conf).
+</#if>