Change alarm notification to batch mode.

oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmCore.java

@@ -25,6 +25,7 @@ import java.util.Map;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import org.apache.skywalking.oap.server.core.alarm.AlarmCallback;
+import org.apache.skywalking.oap.server.core.alarm.AlarmMessage;
 import org.joda.time.LocalDateTime;
 import org.joda.time.Minutes;
 
@@ -61,8 +62,9 @@ public class AlarmCore {
     public void start(List<AlarmCallback> allCallbacks) {
         LocalDateTime now = LocalDateTime.now();
         lastExecuteTime = now;
-        runningContext.values().forEach(ruleList -> ruleList.forEach(runningRule -> runningRule.start(now, allCallbacks)));
-        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() ->
+        runningContext.values().forEach(ruleList -> ruleList.forEach(runningRule -> runningRule.start(now)));
+        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
+            List<AlarmMessage> alarmMessageList = new ArrayList<>(30);
             runningContext.values().forEach(ruleList -> ruleList.forEach(runningRule -> {
                 LocalDateTime checkTime = LocalDateTime.now();
                 int minutes = Minutes.minutesBetween(lastExecuteTime, checkTime).getMinutes();
@@ -72,11 +74,17 @@ public class AlarmCore {
                      * Don't run in the first quarter per min, avoid to trigger alarm.
                      */
                     if (checkTime.getSecondOfMinute() > 15) {
-                        runningRule.check();
+                        AlarmMessage alarmMessage = runningRule.check();
+                        if (alarmMessage != AlarmMessage.NONE) {
+                            alarmMessageList.add(alarmMessage);
+                        }
                         // Set the last execute time, and make sure the second is `00`, such as: 18:30:00
                         lastExecuteTime = checkTime.minusSeconds(checkTime.getSecondOfMinute());
                     }
                 }
-            })), 10, 10, TimeUnit.SECONDS);
+            }));
+
+            allCallbacks.forEach(callback -> callback.doAlarm(alarmMessageList));
+        }, 10, 10, TimeUnit.SECONDS);
     }
 }

oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/RunningRule.java

@@ -55,7 +55,6 @@ public class RunningRule {
     private Window window;
     private volatile boolean isStarted = false;
     private volatile IndicatorValueType valueType;
-    private volatile List<AlarmCallback> allCallbacks;
     private Scope targetScope;
 
     public RunningRule(AlarmRule alarmRule) {
@@ -108,8 +107,7 @@ public class RunningRule {
      *
      * @param current
      */
-    public void start(LocalDateTime current, List<AlarmCallback> allCallbacks) {
-        this.allCallbacks = allCallbacks;
+    public void start(LocalDateTime current) {
         window.start(current);
         isStarted = true;
     }
@@ -126,7 +124,7 @@ public class RunningRule {
     /**
      * Check the conditions, decide to whether trigger alarm.
      */
-    public void check() {
+    public AlarmMessage check() {
         boolean isMatched = window.isMatch();
 
         /**
@@ -138,7 +136,7 @@ public class RunningRule {
         if (isMatched) {
             counter++;
             if (counter >= countThreshold && silenceCountdown < 1) {
-                triggerAlarm();
+                return triggerAlarm();
             } else {
                 silenceCountdown--;
             }
@@ -148,15 +146,16 @@ public class RunningRule {
                 counter--;
             }
         }
+        return AlarmMessage.NONE;
     }
 
     /**
      * Trigger alarm callbacks.
      */
-    private void triggerAlarm() {
+    private AlarmMessage triggerAlarm() {
         silenceCountdown = silencePeriod;
         AlarmMessage message = new AlarmMessage();
-        allCallbacks.forEach(callback -> callback.doAlarm(message));
+        return message;
     }
 
     /**

oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/WebhookCallback.java

@@ -32,7 +32,7 @@ public class WebhookCallback implements AlarmCallback {
         this.remoteEndpoints = remoteEndpoints;
     }
 
-    @Override public void doAlarm(AlarmMessage alarmMessage) {
+    @Override public void doAlarm(List<AlarmMessage> alarmMessage) {
 
     }
 }

oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/RunningRuleTest.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.oap.server.core.alarm.provider;
 
 import java.util.LinkedList;
+import java.util.List;
 import org.apache.skywalking.oap.server.core.alarm.AlarmCallback;
 import org.apache.skywalking.oap.server.core.alarm.AlarmMessage;
 import org.apache.skywalking.oap.server.core.alarm.MetaInAlarm;
@@ -55,7 +56,7 @@ public class RunningRuleTest {
 
         RunningRule runningRule = new RunningRule(alarmRule);
         LocalDateTime startTime = TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301440");
-        runningRule.start(startTime, new LinkedList<>());
+        runningRule.start(startTime);
 
         RunningRule.Window window = Whitebox.getInternalState(runningRule, "window");
         LocalDateTime endTime = Whitebox.getInternalState(window, "endTime");
@@ -80,15 +81,7 @@ public class RunningRuleTest {
         RunningRule runningRule = new RunningRule(alarmRule);
         LocalDateTime startTime = TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301440");
 
-        final boolean[] isAlarm = {false};
-        AlarmCallback assertCallback = new AlarmCallback() {
-            @Override public void doAlarm(AlarmMessage alarmMessage) {
-                isAlarm[0] = true;
-            }
-        };
-        LinkedList<AlarmCallback> callbackList = new LinkedList<>();
-        callbackList.add(assertCallback);
-        runningRule.start(startTime, callbackList);
+        runningRule.start(startTime);
 
         long timeInPeriod1 = 201808301434L;
         long timeInPeriod2 = 201808301436L;
@@ -98,15 +91,14 @@ public class RunningRuleTest {
         runningRule.in(getMetaInAlarm(), getIndicator(timeInPeriod3, 74));
 
         // check at 201808301440
-        runningRule.check();
+        Assert.assertEquals(AlarmMessage.NONE, runningRule.check());
         runningRule.moveTo(TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301441"));
         // check at 201808301441
-        runningRule.check();
+        Assert.assertEquals(AlarmMessage.NONE, runningRule.check());
         runningRule.moveTo(TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301442"));
         // check at 201808301442
-        runningRule.check();
+        Assert.assertNotEquals(AlarmMessage.NONE, runningRule.check());
 
-        Assert.assertTrue(isAlarm[0]);
     }
 
     @Test
@@ -125,13 +117,13 @@ public class RunningRuleTest {
 
         final boolean[] isAlarm = {false};
         AlarmCallback assertCallback = new AlarmCallback() {
-            @Override public void doAlarm(AlarmMessage alarmMessage) {
+            @Override public void doAlarm(List<AlarmMessage> alarmMessage) {
                 isAlarm[0] = true;
             }
         };
         LinkedList<AlarmCallback> callbackList = new LinkedList<>();
         callbackList.add(assertCallback);
-        runningRule.start(startTime, callbackList);
+        runningRule.start(startTime);
 
         long timeInPeriod1 = 201808301434L;
         long timeInPeriod2 = 201808301436L;
@@ -145,15 +137,13 @@ public class RunningRuleTest {
         runningRule.in(getMetaInAlarm(), getIndicator(timeInPeriod5, 95));
 
         // check at 201808301440
-        runningRule.check();
+        Assert.assertEquals(AlarmMessage.NONE, runningRule.check());
         runningRule.moveTo(TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301442"));
         // check at 201808301441
-        runningRule.check();
+        Assert.assertEquals(AlarmMessage.NONE, runningRule.check());
         runningRule.moveTo(TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301443"));
         // check at 201808301442
-        runningRule.check();
-
-        Assert.assertFalse((boolean)isAlarm[0]);
+        Assert.assertEquals(AlarmMessage.NONE, runningRule.check());
     }
 
     @Test
@@ -170,16 +160,7 @@ public class RunningRuleTest {
         RunningRule runningRule = new RunningRule(alarmRule);
         LocalDateTime startTime = TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301440");
 
-        final Object[] isAlarm = {false, 0};
-        AlarmCallback assertCallback = new AlarmCallback() {
-            @Override public void doAlarm(AlarmMessage alarmMessage) {
-                isAlarm[0] = true;
-                isAlarm[1] = (int)isAlarm[1] + 1;
-            }
-        };
-        LinkedList<AlarmCallback> callbackList = new LinkedList<>();
-        callbackList.add(assertCallback);
-        runningRule.start(startTime, callbackList);
+        runningRule.start(startTime);
 
         long timeInPeriod1 = 201808301434L;
         long timeInPeriod2 = 201808301436L;
@@ -189,22 +170,19 @@ public class RunningRuleTest {
         runningRule.in(getMetaInAlarm(), getIndicator(timeInPeriod3, 74));
 
         // check at 201808301440
-        runningRule.check(); //check matches, no alarm
+        Assert.assertEquals(AlarmMessage.NONE, runningRule.check()); //check matches, no alarm
         runningRule.moveTo(TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301441"));
         // check at 201808301441
-        runningRule.check(); //check matches, no alarm
+        Assert.assertEquals(AlarmMessage.NONE, runningRule.check()); //check matches, no alarm
         runningRule.moveTo(TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301442"));
         // check at 201808301442
+        Assert.assertNotEquals(AlarmMessage.NONE, runningRule.check()); //alarm
+        Assert.assertEquals(AlarmMessage.NONE, runningRule.check()); //silence, no alarm
+        Assert.assertEquals(AlarmMessage.NONE, runningRule.check()); //silence, no alarm
         runningRule.check(); //alarm
-        runningRule.check(); //silence, no alarm
-        runningRule.check(); //silence, no alarm
-        runningRule.check(); //alarm
-        runningRule.check(); //silence, no alarm
-        runningRule.check(); //silence, no alarm
-        runningRule.check(); //alarm
-
-        Assert.assertTrue((boolean)isAlarm[0]);
-        Assert.assertEquals(3, (int)isAlarm[1]);
+        Assert.assertEquals(AlarmMessage.NONE, runningRule.check()); //silence, no alarm
+        Assert.assertEquals(AlarmMessage.NONE, runningRule.check()); //silence, no alarm
+        Assert.assertNotEquals(AlarmMessage.NONE, runningRule.check()); //alarm
     }
 
     private MetaInAlarm getMetaInAlarm() {

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmCallback.java

@@ -18,6 +18,8 @@
 
 package org.apache.skywalking.oap.server.core.alarm;
 
+import java.util.List;
+
 /**
  * Alarm call back will be called by alarm implementor,
  * after it decided alarm should be sent.
@@ -25,5 +27,5 @@ package org.apache.skywalking.oap.server.core.alarm;
  * @author wusheng
  */
 public interface AlarmCallback {
-    void doAlarm(AlarmMessage alarmMessage);
+    void doAlarm(List<AlarmMessage> alarmMessage);
 }

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmMessage.java

@@ -30,4 +30,9 @@ import lombok.Setter;
 @Setter(AccessLevel.PUBLIC)
 @Getter(AccessLevel.PUBLIC)
 public class AlarmMessage {
+    public static AlarmMessage NONE = new NoAlarm();
+
+    private static class NoAlarm extends AlarmMessage {
+
+    }
 }

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmStandardPersistence.java

@@ -18,11 +18,13 @@
 
 package org.apache.skywalking.oap.server.core.alarm;
 
+import java.util.List;
+
 /**
  * Save the alarm info into storage for UI query.
  */
 public class AlarmStandardPersistence implements AlarmCallback {
-    @Override public void doAlarm(AlarmMessage alarmMessage) {
+    @Override public void doAlarm(List<AlarmMessage> alarmMessage) {
         //TODO Peng-yongsheng
         /**
          * This is just a callback entrance.