Generate the corresponding event information for the alert. (#6702)

CHANGES.md

@@ -18,6 +18,7 @@ Release Notes.
 #### OAP-Backend
 * BugFix: filter invalid Envoy access logs whose socket address is empty.
 * Fix K8s monitoring the incorrect metrics calculate. 
+* Loop alarm into event system.
 
 #### UI
 * Add logo for kong plugin.

docs/en/concepts-and-designs/event.md

@@ -54,3 +54,11 @@ The end time of the event. This field may be empty if the event has not ended ye
 
 **NOTE:** When reporting an event, you typically call the report function twice, the first time for starting of the event and the second time for ending of the event, both with the same UUID.
 There are also cases where you would already have both the start time and end time. For example, when exporting events from a third-party system, the start time and end time are already known so you may simply call the report function once.
+
+## Known Events
+
+| Name | Type | When |
+| :----: | :----: | :-----|
+| Start | Normal | When your Java Application starts with SkyWalking Agent installed, the `Start` Event will be created. |
+| Shutdown | Normal | When your Java Application stops with SkyWalking Agent installed, the `Shutdown` Event will be created.  |
+| Alarm | Error | When the Alarm is triggered, the corresponding `Alarm` Event will is created. |
\ No newline at end of file

oap-server/server-alarm-plugin/pom.xml

@@ -38,6 +38,11 @@
             <artifactId>library-util</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>event-analyzer</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>io.grpc</groupId>
             <artifactId>grpc-testing</artifactId>

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

@@ -66,7 +66,7 @@ public class AlarmModuleProvider extends ModuleProvider {
 
         alarmRulesWatcher = new AlarmRulesWatcher(rules, this);
 
-        notifyHandler = new NotifyHandler(alarmRulesWatcher);
+        notifyHandler = new NotifyHandler(alarmRulesWatcher, getManager());
         notifyHandler.init(new AlarmStandardPersistence());
         this.registerServiceImplementation(MetricsNotify.class, notifyHandler);
     }
@@ -91,4 +91,5 @@ public class AlarmModuleProvider extends ModuleProvider {
             ConfigurationModule.NAME
         };
     }
+
 }

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

@@ -84,6 +84,7 @@ public class CompositeRuleEvaluator {
                     message.setRuleName(compositeAlarmRule.getAlarmRuleName());
                     String alarmMessage = formatMessage(message, compositeAlarmRule.getMessage(), compositeAlarmRule.getExpression());
                     message.setAlarmMessage(alarmMessage);
+                    message.setPeriod(headMsg.getPeriod());
                     compositeRuleMessages.add(message);
                 }
             });

oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/EventHookCallback.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 org.apache.skywalking.apm.network.event.v3.Event;
+import org.apache.skywalking.apm.network.event.v3.Source;
+import org.apache.skywalking.apm.network.event.v3.Type;
+import org.apache.skywalking.oap.server.analyzer.event.EventAnalyzerModule;
+import org.apache.skywalking.oap.server.analyzer.event.EventAnalyzerService;
+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.analysis.IDManager;
+import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * EventCallBack: When an alert is present, an event is generated for each alert message. These events are then sent to the internal event analyzer.
+ *
+ */
+public class EventHookCallback implements AlarmCallback {
+
+    private final ModuleManager manager;
+
+    public EventHookCallback(ModuleManager manager) {
+        this.manager = manager;
+    }
+
+    @Override
+    public void doAlarm(List<AlarmMessage> alarmMessage) {
+        EventAnalyzerService analyzerService = manager.find(EventAnalyzerModule.NAME).provider().getService(EventAnalyzerService.class);
+        alarmMessage.forEach(a -> {
+            for (Event event : constructCurrentEvent(a)) {
+                analyzerService.analyze(event);
+            }
+        });
+    }
+
+    private List<Event> constructCurrentEvent(AlarmMessage msg) {
+        List<Event> events = new ArrayList<>(2);
+        long now = System.currentTimeMillis();
+        Event.Builder builder = Event.newBuilder()
+                .setUuid(UUID.randomUUID().toString())
+                .setName("Alarm")
+                .setStartTime(now - (msg.getPeriod() * 60 * 1000))
+                .setMessage(msg.getAlarmMessage())
+                .setType(Type.Error)
+                .setEndTime(now);
+        switch (msg.getScopeId()) {
+            case DefaultScopeDefine.SERVICE :
+                IDManager.ServiceID.ServiceIDDefinition serviceIdDef = IDManager.ServiceID.analysisId(msg.getId0());
+                builder.setSource(
+                    Source.newBuilder()
+                            .setService(serviceIdDef.getName())
+                            .build()
+                );
+                events.add(builder.build());
+                break;
+            case DefaultScopeDefine.SERVICE_RELATION :
+                IDManager.ServiceID.ServiceIDDefinition sourceServiceIdDef = IDManager.ServiceID.analysisId(msg.getId0());
+                builder.setSource(
+                        Source.newBuilder()
+                            .setService(sourceServiceIdDef.getName())
+                            .build()
+                );
+                events.add(builder.build());
+                IDManager.ServiceID.ServiceIDDefinition destServiceIdDef = IDManager.ServiceID.analysisId(msg.getId1());
+                builder.setSource(
+                        Source.newBuilder()
+                                .setService(destServiceIdDef.getName())
+                                .build()
+                ).setUuid(UUID.randomUUID().toString());
+                events.add(builder.build());
+                break;
+            case DefaultScopeDefine.SERVICE_INSTANCE :
+                IDManager.ServiceInstanceID.InstanceIDDefinition instanceIdDef = IDManager.ServiceInstanceID.analysisId(msg.getId0());
+                builder.setSource(
+                        Source.newBuilder()
+                                .setServiceInstance(instanceIdDef.getName())
+                                .setService(IDManager.ServiceID.analysisId(instanceIdDef.getServiceId()).getName())
+                                .build()
+                );
+                events.add(builder.build());
+                break;
+            case DefaultScopeDefine.SERVICE_INSTANCE_RELATION :
+                IDManager.ServiceInstanceID.InstanceIDDefinition sourceInstanceIdDef = IDManager.ServiceInstanceID.analysisId(msg.getId0());
+                builder.setSource(
+                        Source.newBuilder()
+                                .setServiceInstance(sourceInstanceIdDef.getName())
+                                .setService(IDManager.ServiceID.analysisId(sourceInstanceIdDef.getServiceId()).getName())
+                                .build()
+                );
+                events.add(builder.build());
+                IDManager.ServiceInstanceID.InstanceIDDefinition destInstanceIdDef = IDManager.ServiceInstanceID.analysisId(msg.getId1());
+                builder.setSource(
+                        Source.newBuilder()
+                                .setServiceInstance(destInstanceIdDef.getName())
+                                .setService(IDManager.ServiceID.analysisId(destInstanceIdDef.getServiceId()).getName())
+                                .build()
+                ).setUuid(UUID.randomUUID().toString());
+                events.add(builder.build());
+                break;
+            case DefaultScopeDefine.ENDPOINT :
+                IDManager.EndpointID.EndpointIDDefinition endpointIDDef = IDManager.EndpointID.analysisId(msg.getId0());
+                builder.setSource(
+                        Source.newBuilder()
+                                .setEndpoint(endpointIDDef.getEndpointName())
+                                .setService(IDManager.ServiceID.analysisId(endpointIDDef.getServiceId()).getName())
+                                .build()
+                );
+                events.add(builder.build());
+                break;
+            case DefaultScopeDefine.ENDPOINT_RELATION :
+                IDManager.EndpointID.EndpointIDDefinition sourceEndpointIDDef = IDManager.EndpointID.analysisId(msg.getId0());
+                builder.setSource(
+                        Source.newBuilder()
+                                .setEndpoint(sourceEndpointIDDef.getEndpointName())
+                                .setService(IDManager.ServiceID.analysisId(sourceEndpointIDDef.getServiceId()).getName())
+                                .build()
+                );
+                events.add(builder.build());
+                IDManager.EndpointID.EndpointIDDefinition destEndpointIDDef = IDManager.EndpointID.analysisId(msg.getId1());
+                builder.setSource(
+                        Source.newBuilder()
+                                .setEndpoint(destEndpointIDDef.getEndpointName())
+                                .setService(IDManager.ServiceID.analysisId(destEndpointIDDef.getServiceId()).getName())
+                                .build()
+                ).setUuid(UUID.randomUUID().toString());
+                events.add(builder.build());
+                break;
+        }
+        return events;
+    }
+}

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

@@ -41,15 +41,18 @@ import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
 import org.apache.skywalking.oap.server.core.analysis.metrics.MetricsMetaInfo;
 import org.apache.skywalking.oap.server.core.analysis.metrics.WithMetadata;
 import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
 
 @Slf4j
 public class NotifyHandler implements MetricsNotify {
     private final AlarmCore core;
     private final AlarmRulesWatcher alarmRulesWatcher;
+    private final ModuleManager manager;
 
-    public NotifyHandler(AlarmRulesWatcher alarmRulesWatcher) {
+    public NotifyHandler(AlarmRulesWatcher alarmRulesWatcher, ModuleManager manager) {
         this.alarmRulesWatcher = alarmRulesWatcher;
         core = new AlarmCore(alarmRulesWatcher);
+        this.manager = manager;
     }
 
     @Override
@@ -166,6 +169,7 @@ public class NotifyHandler implements MetricsNotify {
         allCallbacks.add(new WechatHookCallback(alarmRulesWatcher));
         allCallbacks.add(new DingtalkHookCallback(alarmRulesWatcher));
         allCallbacks.add(new FeishuHookCallback(alarmRulesWatcher));
+        allCallbacks.add(new EventHookCallback(this.manager));
         core.start(allCallbacks);
     }
 }

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

@@ -236,6 +236,7 @@ public class RunningRule {
                 alarmMessage.setAlarmMessage(formatter.format(meta));
                 alarmMessage.setOnlyAsCondition(this.onlyAsCondition);
                 alarmMessage.setStartTime(System.currentTimeMillis());
+                alarmMessage.setPeriod(this.period);
                 alarmMessageList.add(alarmMessage);
             }
         });

oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/EventHookCallbackTest.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 org.apache.skywalking.apm.network.event.v3.Event;
+import org.apache.skywalking.oap.server.analyzer.event.EventAnalyzerService;
+import org.apache.skywalking.oap.server.analyzer.event.EventAnalyzerServiceImpl;
+import org.apache.skywalking.oap.server.core.alarm.AlarmMessage;
+import org.apache.skywalking.oap.server.core.analysis.IDManager;
+import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.module.ModuleProviderHolder;
+import org.apache.skywalking.oap.server.library.module.ModuleServiceHolder;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static junit.framework.TestCase.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class EventHookCallbackTest {
+
+    private ModuleManager moduleManager = mock(ModuleManager.class);
+
+    private ModuleProviderHolder moduleProviderHolder = mock(ModuleProviderHolder.class);
+
+    private ModuleServiceHolder moduleServiceHolder = mock(ModuleServiceHolder.class);
+
+    private MockEventAnalyzerService mockEventAnalyzerService = mock(MockEventAnalyzerService.class);
+
+    private EventAnalyzerService eventAnalyzerService = mock(EventAnalyzerServiceImpl.class);
+
+    @Test
+    public void testEventCallbackHasRightFlow() throws Exception {
+        List<AlarmMessage> msgs = mockAlarmMessagesHasSingleElement();
+        EventHookCallback callback = new EventHookCallback(this.moduleManager);
+        when(moduleManager.find("event-analyzer")).thenReturn(moduleProviderHolder);
+        when(moduleProviderHolder.provider()).thenReturn(moduleServiceHolder);
+        when(moduleServiceHolder.getService(EventAnalyzerService.class)).thenReturn(mockEventAnalyzerService);
+
+        //make sure current service be called.
+        callback.doAlarm(msgs);
+        verify(mockEventAnalyzerService).analyze(any(Event.class));
+
+        when(moduleServiceHolder.getService(EventAnalyzerService.class)).thenReturn(eventAnalyzerService);
+        callback.doAlarm(msgs);
+        //Ensure that the current Event is properly constructed
+        ArgumentCaptor<Event> argument = ArgumentCaptor.forClass(Event.class);
+        verify(eventAnalyzerService).analyze(argument.capture());
+        Event value = argument.getValue();
+        AlarmMessage msg = msgs.get(0);
+        assertEquals(msg.getName(), value.getSource().getService());
+        assertEquals("Alarm", value.getName());
+        assertEquals(msg.getAlarmMessage(), value.getMessage());
+        assertEquals(msg.getPeriod(), (value.getEndTime() - value.getStartTime()) / 1000 / 60);
+
+    }
+
+    @Test
+    public void testRelationEventBeProperlyConstructed() {
+        List<AlarmMessage> msgs = mockAlarmMessagesHasSourceAndDest();
+        EventHookCallback callback = new EventHookCallback(this.moduleManager);
+        when(moduleManager.find("event-analyzer")).thenReturn(moduleProviderHolder);
+        when(moduleProviderHolder.provider()).thenReturn(moduleServiceHolder);
+        when(moduleServiceHolder.getService(EventAnalyzerService.class)).thenReturn(eventAnalyzerService);
+        callback.doAlarm(msgs);
+
+        ArgumentCaptor<Event> argument = ArgumentCaptor.forClass(Event.class);
+        verify(eventAnalyzerService, times(2)).analyze(argument.capture());
+        List<Event> events = argument.getAllValues();
+        assertEquals(events.size(), 2);
+        Event sourceEvent = events.get(0);
+        Event destEvent = events.get(1);
+        AlarmMessage msg = msgs.get(0);
+        assertEquals(sourceEvent.getSource().getService(), IDManager.ServiceID.analysisId(msg.getId0()).getName());
+        assertEquals((sourceEvent.getEndTime() - sourceEvent.getStartTime()) / 1000 / 60, msg.getPeriod());
+        assertEquals(destEvent.getSource().getService(), IDManager.ServiceID.analysisId(msg.getId1()).getName());
+        assertEquals((destEvent.getEndTime() - destEvent.getStartTime()) / 1000 / 60, msg.getPeriod());
+    }
+
+    private List<AlarmMessage> mockAlarmMessagesHasSingleElement() {
+        AlarmMessage msg = new AlarmMessage();
+        msg.setScopeId(DefaultScopeDefine.SERVICE);
+        msg.setScope("SERVICE");
+        msg.setName("test-skywalking");
+        msg.setId0("dGVzdC1za3l3YWxraW5n.1");
+        msg.setAlarmMessage("Alarm caused by Rule service_resp_time_rule");
+        msg.setPeriod(3);
+        return Arrays.asList(msg);
+    }
+
+    private List<AlarmMessage> mockAlarmMessagesHasSourceAndDest() {
+        AlarmMessage msg = new AlarmMessage();
+        msg.setScopeId(DefaultScopeDefine.SERVICE_RELATION);
+        msg.setScope("");
+        msg.setName("test-skywalking");
+        msg.setId0(IDManager.ServiceID.buildId("sourceIdStr", true));
+        msg.setId1(IDManager.ServiceID.buildId("destIdStr", true));
+        msg.setAlarmMessage("Alarm caused by Rule service_resp_time_rule");
+        msg.setPeriod(5);
+        return Arrays.asList(msg);
+    }
+
+    class MockEventAnalyzerService implements EventAnalyzerService {
+
+        @Override
+        public void analyze(Event event) {
+            //ignore current mock process.
+        }
+    }
+}

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

@@ -281,7 +281,9 @@ public class NotifyHandlerTest {
 
         Rules rules = new Rules();
 
-        notifyHandler = new NotifyHandler(new AlarmRulesWatcher(rules, null));
+        moduleManager = mock(ModuleManager.class);
+
+        notifyHandler = new NotifyHandler(new AlarmRulesWatcher(rules, null), moduleManager);
 
         notifyHandler.init(alarmMessageList -> {
             for (AlarmMessage message : alarmMessageList) {
@@ -289,8 +291,6 @@ public class NotifyHandlerTest {
             }
         });
 
-        moduleManager = mock(ModuleManager.class);
-
         moduleProviderHolder = mock(ModuleProviderHolder.class);
 
         moduleServiceHolder = mock(ModuleServiceHolder.class);

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

@@ -35,5 +35,6 @@ public class AlarmMessage {
     private String ruleName;
     private String alarmMessage;
     private long startTime;
+    private transient int period;
     private transient boolean onlyAsCondition;
 }