Finish alarm core test.

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

@@ -24,6 +24,7 @@ import java.util.List;
 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.joda.time.LocalDateTime;
 import org.joda.time.Minutes;
 
@@ -57,10 +58,10 @@ public class AlarmCore {
         return runningContext.get(indicatorName);
     }
 
-    public void start() {
+    public void start(List<AlarmCallback> allCallbacks) {
         LocalDateTime now = LocalDateTime.now();
         lastExecuteTime = now;
-        runningContext.values().forEach(ruleList -> ruleList.forEach(runningRule -> runningRule.start(now)));
+        runningContext.values().forEach(ruleList -> ruleList.forEach(runningRule -> runningRule.start(now, allCallbacks)));
         Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() ->
             runningContext.values().forEach(ruleList -> ruleList.forEach(runningRule -> {
                 LocalDateTime checkTime = LocalDateTime.now();

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

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.oap.server.core.alarm.provider;
 
+import java.util.ArrayList;
 import java.util.List;
 import org.apache.skywalking.oap.server.core.alarm.AlarmCallback;
 import org.apache.skywalking.oap.server.core.alarm.IndicatorNotify;
@@ -27,12 +28,10 @@ import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator;
 public class NotifyHandler implements IndicatorNotify {
     private final AlarmCore core;
     private final Rules rules;
-    private List<AlarmCallback> allCallbacks;
 
     public NotifyHandler(Rules rules) {
         this.rules = rules;
         core = new AlarmCore(rules);
-        core.start();
     }
 
     @Override public void notify(MetaInAlarm meta, Indicator indicator) {
@@ -52,10 +51,12 @@ public class NotifyHandler implements IndicatorNotify {
 
     @Override
     public void init(AlarmCallback... callbacks) {
+        List<AlarmCallback> allCallbacks = new ArrayList<>();
         for (AlarmCallback callback : callbacks) {
             allCallbacks.add(callback);
         }
         allCallbacks.add(new WebhookCallback(rules.getWebhooks()));
+        core.start(allCallbacks);
     }
 
 }

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

@@ -19,7 +19,10 @@
 package org.apache.skywalking.oap.server.core.alarm.provider;
 
 import java.util.LinkedList;
+import java.util.List;
 import java.util.concurrent.locks.ReentrantLock;
+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;
 import org.apache.skywalking.oap.server.core.analysis.indicator.DoubleValueHolder;
 import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator;
@@ -36,9 +39,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
- * RunningRule represents each rule in running status.
- * Based on the {@link AlarmRule} definition,
- *
+ * RunningRule represents each rule in running status. Based on the {@link AlarmRule} definition,
  *
  * @author wusheng
  */
@@ -55,7 +56,8 @@ public class RunningRule {
     private int silenceCountdown;
     private Window window;
     private volatile boolean isStarted = false;
-    private IndicatorValueType valueType;
+    private volatile IndicatorValueType valueType;
+    private volatile List<AlarmCallback> allCallbacks;
     private Scope targetScope;
 
     public RunningRule(AlarmRule alarmRule) {
@@ -74,8 +76,8 @@ public class RunningRule {
     }
 
     /**
-     * Receive indicator result from persistence, after it is saved into storage.
-     * In alarm, only minute dimensionality indicators are expected to process.
+     * Receive indicator result from persistence, after it is saved into storage. In alarm, only minute dimensionality
+     * indicators are expected to process.
      *
      * @param indicator
      */
@@ -105,15 +107,18 @@ public class RunningRule {
 
     /**
      * Start this rule in running mode.
+     *
      * @param current
      */
-    public void start(LocalDateTime current) {
+    public void start(LocalDateTime current, List<AlarmCallback> allCallbacks) {
+        this.allCallbacks = allCallbacks;
         window.start(current);
         isStarted = true;
     }
 
     /**
      * Move the buffer window to give time.
+     *
      * @param targetTime of moving target
      */
     public void moveTo(LocalDateTime targetTime) {
@@ -136,6 +141,8 @@ public class RunningRule {
             counter++;
             if (counter >= countThreshold && silenceCountdown < 1) {
                 triggerAlarm();
+            } else {
+                silenceCountdown--;
             }
         } else {
             silenceCountdown--;
@@ -150,11 +157,13 @@ public class RunningRule {
      */
     private void triggerAlarm() {
         silenceCountdown = silencePeriod;
+        AlarmMessage message = new AlarmMessage();
+        allCallbacks.forEach(callback -> callback.doAlarm(message));
     }
 
     /**
-     * A indicator window, based on {@link AlarmRule#period}.
-     * This window slides with time, just keeps the recent N(period) buckets.
+     * A indicator window, based on {@link AlarmRule#period}. This window slides with time, just keeps the recent
+     * N(period) buckets.
      *
      * @author wusheng
      */
@@ -201,7 +210,7 @@ public class RunningRule {
 
             LocalDateTime timebucket = TIME_BUCKET_FORMATTER.parseLocalDateTime(bucket + "");
 
-            int minutes = Minutes.minutesBetween(endTime, timebucket).getMinutes();
+            int minutes = Minutes.minutesBetween(timebucket, endTime).getMinutes();
             if (minutes == -1) {
                 this.moveTo(timebucket);
 
@@ -209,7 +218,6 @@ public class RunningRule {
 
             lock.lock();
             try {
-                minutes = Minutes.minutesBetween(endTime, timebucket).getMinutes();
                 if (minutes < 0) {
                     // At any moment, should NOT be here
                     // Add this code just because of my obsession :P
@@ -222,8 +230,7 @@ public class RunningRule {
                     return;
                 }
 
-                int idx = minutes - 1;
-                values.set(idx, indicator);
+                values.set(values.size() - minutes - 1, indicator);
             } finally {
                 lock.unlock();
             }

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

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.core.alarm.provider;
+
+public class AlarmCoreTest {
+}

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

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.core.alarm.provider;
+
+import java.util.LinkedList;
+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;
+import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator;
+import org.apache.skywalking.oap.server.core.analysis.indicator.IntValueHolder;
+import org.apache.skywalking.oap.server.core.remote.grpc.proto.RemoteData;
+import org.apache.skywalking.oap.server.core.source.Scope;
+import org.joda.time.LocalDateTime;
+import org.joda.time.format.DateTimeFormat;
+import org.joda.time.format.DateTimeFormatter;
+import org.junit.Assert;
+import org.junit.Test;
+import org.powermock.reflect.Whitebox;
+
+/**
+ * Running rule is the core of how does alarm work.
+ *
+ * So in this test, we need to simulate a lot of scenario to see the reactions.
+ *
+ * @author wusheng
+ */
+public class RunningRuleTest {
+    private static DateTimeFormatter TIME_BUCKET_FORMATTER = DateTimeFormat.forPattern("yyyyMMddHHmm");
+
+    @Test
+    public void testInitAndStart() {
+        AlarmRule alarmRule = new AlarmRule();
+        alarmRule.setAlarmRuleName("endpoint_percent_rule");
+        alarmRule.setIndicatorName("endpoint_percent");
+        alarmRule.setOp("<");
+        alarmRule.setThreshold("75");
+        alarmRule.setCount(3);
+        alarmRule.setPeriod(15);
+
+        RunningRule runningRule = new RunningRule(alarmRule);
+        LocalDateTime startTime = TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301440");
+        runningRule.start(startTime, new LinkedList<>());
+
+        RunningRule.Window window = Whitebox.getInternalState(runningRule, "window");
+        LocalDateTime endTime = Whitebox.getInternalState(window, "endTime");
+        int period = Whitebox.getInternalState(window, "period");
+        LinkedList<Indicator> indicatorBuffer = Whitebox.getInternalState(window, "values");
+
+        Assert.assertTrue(startTime.equals(endTime));
+        Assert.assertEquals(15, period);
+        Assert.assertEquals(15, indicatorBuffer.size());
+    }
+
+    @Test
+    public void testAlarm() {
+        AlarmRule alarmRule = new AlarmRule();
+        alarmRule.setAlarmRuleName("endpoint_percent_rule");
+        alarmRule.setIndicatorName("endpoint_percent");
+        alarmRule.setOp("<");
+        alarmRule.setThreshold("75");
+        alarmRule.setCount(3);
+        alarmRule.setPeriod(15);
+
+        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);
+
+        long timeInPeriod1 = 201808301434L;
+        long timeInPeriod2 = 201808301436L;
+        long timeInPeriod3 = 201808301438L;
+        runningRule.in(getMetaInAlarm(), getIndicator(timeInPeriod1, 70));
+        runningRule.in(getMetaInAlarm(), getIndicator(timeInPeriod2, 71));
+        runningRule.in(getMetaInAlarm(), getIndicator(timeInPeriod3, 74));
+
+        // check at 201808301440
+        runningRule.check();
+        runningRule.moveTo(TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301441"));
+        // check at 201808301441
+        runningRule.check();
+        runningRule.moveTo(TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301442"));
+        // check at 201808301442
+        runningRule.check();
+
+        Assert.assertTrue(isAlarm[0]);
+    }
+
+    @Test
+    public void testNoAlarm() {
+        AlarmRule alarmRule = new AlarmRule();
+        alarmRule.setAlarmRuleName("endpoint_percent_rule");
+        alarmRule.setIndicatorName("endpoint_percent");
+        alarmRule.setOp(">");
+        alarmRule.setThreshold("75");
+        alarmRule.setCount(3);
+        alarmRule.setPeriod(15);
+        //alarmRule.setSilencePeriod(0);
+
+        RunningRule runningRule = new RunningRule(alarmRule);
+        LocalDateTime startTime = TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301441");
+
+        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);
+
+        long timeInPeriod1 = 201808301434L;
+        long timeInPeriod2 = 201808301436L;
+        long timeInPeriod3 = 201808301438L;
+        long timeInPeriod4 = 201808301432L;
+        long timeInPeriod5 = 201808301440L;
+        runningRule.in(getMetaInAlarm(), getIndicator(timeInPeriod1, 70));
+        runningRule.in(getMetaInAlarm(), getIndicator(timeInPeriod2, 71));
+        runningRule.in(getMetaInAlarm(), getIndicator(timeInPeriod3, 74));
+        runningRule.in(getMetaInAlarm(), getIndicator(timeInPeriod4, 90));
+        runningRule.in(getMetaInAlarm(), getIndicator(timeInPeriod5, 95));
+
+        // check at 201808301440
+        runningRule.check();
+        runningRule.moveTo(TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301442"));
+        // check at 201808301441
+        runningRule.check();
+        runningRule.moveTo(TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301443"));
+        // check at 201808301442
+        runningRule.check();
+
+        Assert.assertFalse((boolean)isAlarm[0]);
+    }
+
+    @Test
+    public void testSilence() {
+        AlarmRule alarmRule = new AlarmRule();
+        alarmRule.setAlarmRuleName("endpoint_percent_rule");
+        alarmRule.setIndicatorName("endpoint_percent");
+        alarmRule.setOp("<");
+        alarmRule.setThreshold("75");
+        alarmRule.setCount(3);
+        alarmRule.setPeriod(15);
+        alarmRule.setSilencePeriod(2);
+
+        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);
+
+        long timeInPeriod1 = 201808301434L;
+        long timeInPeriod2 = 201808301436L;
+        long timeInPeriod3 = 201808301438L;
+        runningRule.in(getMetaInAlarm(), getIndicator(timeInPeriod1, 70));
+        runningRule.in(getMetaInAlarm(), getIndicator(timeInPeriod2, 71));
+        runningRule.in(getMetaInAlarm(), getIndicator(timeInPeriod3, 74));
+
+        // check at 201808301440
+        runningRule.check(); //check matches, no alarm
+        runningRule.moveTo(TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301441"));
+        // check at 201808301441
+        runningRule.check(); //check matches, no alarm
+        runningRule.moveTo(TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301442"));
+        // check at 201808301442
+        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]);
+    }
+
+    private MetaInAlarm getMetaInAlarm() {
+        return new MetaInAlarm() {
+
+            @Override public Scope getScope() {
+                return Scope.Service;
+            }
+
+            @Override public String getName() {
+                return "Service_123";
+            }
+
+            @Override public String getIndicatorName() {
+                return "endpoint_percent";
+            }
+
+            @Override public int getId0() {
+                return 123;
+            }
+
+            @Override public int getId1() {
+                return 0;
+            }
+        };
+    }
+
+    private Indicator getIndicator(long timebucket, int value) {
+        MockIndicator indicator = new MockIndicator();
+        indicator.setValue(value);
+        indicator.setTimeBucket(timebucket);
+        return indicator;
+    }
+
+    private class MockIndicator extends Indicator implements IntValueHolder {
+        private int value;
+
+        @Override public String id() {
+            return null;
+        }
+
+        @Override public void combine(Indicator indicator) {
+
+        }
+
+        @Override public void calculate() {
+
+        }
+
+        @Override public int getValue() {
+            return value;
+        }
+
+        @Override public void deserialize(RemoteData remoteData) {
+
+        }
+
+        @Override public RemoteData.Builder serialize() {
+            return null;
+        }
+
+        public void setValue(int value) {
+            this.value = value;
+        }
+    }
+}