[DROOLS-7589] ansible-rulebook : Throw Exception when heap reaches to…

… threshold

drools-ansible-rulebook-integration-api/src/main/java/org/drools/ansible/rulebook/integration/api/RulesExecutor.java

@@ -1,5 +1,6 @@
 package org.drools.ansible.rulebook.integration.api;
 
+import org.drools.ansible.rulebook.integration.api.rulesengine.MemoryMonitorUtil;
 import org.drools.ansible.rulebook.integration.api.rulesengine.RulesEvaluator;
 import org.drools.ansible.rulebook.integration.api.rulesengine.RulesExecutorSession;
 import org.drools.ansible.rulebook.integration.api.rulesengine.SessionStats;
@@ -62,14 +63,17 @@ public long rulesCount() {
     }
 
     public CompletableFuture<Integer> executeFacts(String json) {
+        MemoryMonitorUtil.checkMemoryOccupation();
         return rulesEvaluator.executeFacts(asFactMap(json));
     }
 
     public CompletableFuture<List<Match>> processFacts(String json) {
+        MemoryMonitorUtil.checkMemoryOccupation();
         return rulesEvaluator.processFacts(asFactMap(json));
     }
 
     public CompletableFuture<List<Match>> processEvents(String json) {
+        MemoryMonitorUtil.checkMemoryOccupation();
         return rulesEvaluator.processEvents(asFactMap(json));
     }
 

drools-ansible-rulebook-integration-api/src/main/java/org/drools/ansible/rulebook/integration/api/rulesengine/MemoryMonitorUtil.java

@@ -0,0 +1,46 @@
+package org.drools.ansible.rulebook.integration.api.rulesengine;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MemoryMonitorUtil {
+
+    private static final Logger LOG = LoggerFactory.getLogger(MemoryMonitorUtil.class.getName());
+
+    public static final String MEMORY_OCCUPATION_PERCENTAGE_THRESHOLD_PROPERTY = "drools.memory.occupation.percentage.threshold";
+
+    private static final int DEFAULT_MEMORY_OCCUPATION_PERCENTAGE_THRESHOLD = 90;
+
+    private static final int MEMORY_OCCUPATION_PERCENTAGE_THRESHOLD;
+
+    static {
+        String envValue = System.getenv("DROOLS_MEMORY_THRESHOLD");
+        if (envValue != null && !envValue.isEmpty()) {
+            // Environment variable takes precedence over system property
+            System.setProperty(MEMORY_OCCUPATION_PERCENTAGE_THRESHOLD_PROPERTY, envValue);
+        }
+        MEMORY_OCCUPATION_PERCENTAGE_THRESHOLD = Integer.getInteger(MEMORY_OCCUPATION_PERCENTAGE_THRESHOLD_PROPERTY, DEFAULT_MEMORY_OCCUPATION_PERCENTAGE_THRESHOLD); // percentage
+    }
+
+    private MemoryMonitorUtil() {
+        // do not instantiate
+    }
+
+    public static void checkMemoryOccupation() {
+        int memoryOccupationPercentage = getMemoryOccupationPercentage();
+        if (memoryOccupationPercentage > MEMORY_OCCUPATION_PERCENTAGE_THRESHOLD) {
+            System.gc(); // NOSONAR
+            // double check to avoid frequent GC
+            memoryOccupationPercentage = getMemoryOccupationPercentage();
+            if (memoryOccupationPercentage > MEMORY_OCCUPATION_PERCENTAGE_THRESHOLD) {
+                LOG.error("Memory occupation is above the threshold: {}% > {}%. MaxMemory = {}, UsedMemory = {}",
+                          memoryOccupationPercentage, MEMORY_OCCUPATION_PERCENTAGE_THRESHOLD, Runtime.getRuntime().maxMemory(), Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory());
+                throw new MemoryThresholdReachedException(MEMORY_OCCUPATION_PERCENTAGE_THRESHOLD, memoryOccupationPercentage);
+            }
+        }
+    }
+
+    private static int getMemoryOccupationPercentage() {
+        return (int) ((100 * (Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory())) / Runtime.getRuntime().maxMemory());
+    }
+}

drools-ansible-rulebook-integration-api/src/main/java/org/drools/ansible/rulebook/integration/api/rulesengine/MemoryThresholdReachedException.java

@@ -0,0 +1,17 @@
+package org.drools.ansible.rulebook.integration.api.rulesengine;
+
+public class MemoryThresholdReachedException extends RuntimeException {
+
+    private final int threshold;
+    private final int actual;
+
+    public MemoryThresholdReachedException(int threshold, int actual) {
+        this.threshold = threshold;
+        this.actual = actual;
+    }
+
+    @Override
+    public String getMessage() {
+        return "Memory threshold reached: " + actual + "% > " + threshold + "%";
+    }
+}

drools-ansible-rulebook-integration-api/src/main/java/org/drools/ansible/rulebook/integration/api/rulesengine/RegisterOnlyAgendaFilter.java

@@ -43,8 +43,8 @@ public boolean accept(Match match) {
         boolean validMatch = isValidMatch(fhs);
 
         if (validMatch) {
-            if (log.isInfoEnabled()) {
-                log.info(matchToString(match));
+            if (log.isDebugEnabled()) {
+                log.debug(matchToString(match));
             }
 
             Map<String, Object> metadata = match.getRule().getMetaData();

drools-ansible-rulebook-integration-main/pom.xml

@@ -74,6 +74,8 @@
             <!-- for SlownessTest -->
             <drools.delay.warning.threshold>2</drools.delay.warning.threshold>
           </systemPropertyVariables>
+<!--          <argLine>-Xmx500m -XX:+HeapDumpOnOutOfMemoryError</argLine>-->
+          <argLine>-Xmx500m</argLine>
         </configuration>
       </plugin>
     </plugins>

drools-ansible-rulebook-integration-main/src/main/java/org/drools/ansible/rulebook/integration/main/Payload.java

@@ -1,15 +1,13 @@
 package org.drools.ansible.rulebook.integration.main;
 
-import org.drools.ansible.rulebook.integration.api.io.JsonMapper;
-import org.drools.ansible.rulebook.integration.core.jpy.AstRulesEngine;
-
-import com.fasterxml.jackson.core.JacksonException;
-
 import java.io.UncheckedIOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 
+import org.drools.ansible.rulebook.integration.api.io.JsonMapper;
+import org.drools.ansible.rulebook.integration.core.jpy.AstRulesEngine;
+
 public class Payload {
 
     private final List<String> list;
@@ -22,6 +20,9 @@ public class Payload {
 
     private int shutdown = 0;
 
+    // set true when matchedEvents occupies too much memory
+    private boolean discardMatchedEvents = false;
+
     private Payload(List<String> list) {
         this.list = list;
     }
@@ -88,6 +89,11 @@ static Payload parsePayload(Map ruleSet) {
         } catch (NullPointerException | NumberFormatException e) {
             /* ignore */
         }
+        try {
+            payload.discardMatchedEvents = Boolean.valueOf(sourcesArgs.get("discard_matched_events").toString());
+        } catch (NullPointerException | NumberFormatException e) {
+            /* ignore */
+        }
 
         return payload;
     }
@@ -128,7 +134,9 @@ public void run() {
             for (int i = 0; i < payload.loopCount; i++) {
                 for (String p : payload.list) {
                     String resultJson = engine.assertEvent(sessionId, p);
-                    returnedMatches.addAll(JsonMapper.readValueAsListOfMapOfStringAndObject(resultJson));
+                    if (!payload.discardMatchedEvents) {
+                        returnedMatches.addAll(JsonMapper.readValueAsListOfMapOfStringAndObject(resultJson));
+                    }
                     sleepSeconds(payload.eventDelay);
                 }
                 sleepSeconds(payload.loopDelay);

drools-ansible-rulebook-integration-main/src/test/java/org/drools/ansible/rulebook/integration/main/PerfTest.java

@@ -1,8 +1,11 @@
 package org.drools.ansible.rulebook.integration.main;
 
+import org.drools.ansible.rulebook.integration.api.rulesengine.MemoryThresholdReachedException;
 import org.drools.ansible.rulebook.integration.main.Main.ExecuteResult;
+import org.junit.Ignore;
 import org.junit.Test;
 
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
 public class PerfTest {
@@ -12,6 +15,19 @@ public void testManyEvents() {
         checkDuration("100k_event_rules_ast.json", 10_000);
     }
 
+    @Ignore("Disabled by default, because it takes around 40 seconds")
+    @Test
+    public void testManyLargeEvents() {
+        // match_multiple_rules: false means events are removed after match. So this test will pass without throwing MemoryThresholdReachedException
+        checkDuration("1m_event_with_20kb_payload_rules_ast.json", 120_000);
+    }
+
+    @Test
+    public void testManyLargeEventsMatchMultipleRules() {
+        // match_multiple_rules: true means events are retained until TTL expires
+        assertThrows(MemoryThresholdReachedException.class, () -> checkDuration("1m_event_with_20kb_payload_match_multiple_rules_ast.json", 120_000));
+    }
+
     @Test
     public void testOnceAfter() {
         checkDuration("56_once_after.json", 15_000);