Fix `EventHookCallback` build event: build the layer from `Service's Layer` (#11413)

docs/en/changes/changes.md

@@ -21,6 +21,8 @@
 * Fix table exists check in the JDBC Storage Plugin.
 * Enhance extensibility of HTTP Server library.
 * Adjust `AlarmRecord` alarmMessage column length to 512.
+* Fix `EventHookCallback` build event: build the layer from `Service's Layer`.
+* Fix `AlarmCore` doAlarm: catch exception for each callback to avoid interruption.
 
 #### UI
 

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

@@ -93,7 +93,11 @@ public class AlarmCore {
 
                 if (!alarmMessageList.isEmpty()) {
                     for (AlarmCallback callback : allCallbacks) {
-                        callback.doAlarm(alarmMessageList);
+                        try {
+                            callback.doAlarm(alarmMessageList);
+                        } catch (Exception e) {
+                            LOGGER.error(e.getMessage(), e);
+                        }
                     }
                 }
             } catch (Exception e) {

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

@@ -18,15 +18,17 @@
 
 package org.apache.skywalking.oap.server.core.alarm.provider;
 
+import java.io.IOException;
 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.CoreModule;
 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.analysis.Layer;
+import org.apache.skywalking.oap.server.core.query.MetadataQueryService;
 import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
 import org.apache.skywalking.oap.server.library.module.ModuleManager;
 
@@ -41,22 +43,36 @@ import java.util.UUID;
 public class EventHookCallback implements AlarmCallback {
 
     private final ModuleManager manager;
+    private MetadataQueryService metadataQueryService;
+
+    private MetadataQueryService getMetadataQueryService() {
+        if (metadataQueryService == null) {
+            this.metadataQueryService = manager.find(CoreModule.NAME)
+                                                     .provider()
+                                                     .getService(MetadataQueryService.class);
+        }
+        return metadataQueryService;
+    }
 
     public EventHookCallback(ModuleManager manager) {
         this.manager = manager;
     }
 
     @Override
-    public void doAlarm(List<AlarmMessage> alarmMessage) {
+    public void doAlarm(List<AlarmMessage> alarmMessage) throws Exception {
         EventAnalyzerService analyzerService = manager.find(EventAnalyzerModule.NAME).provider().getService(EventAnalyzerService.class);
-        alarmMessage.forEach(a -> {
+        for (AlarmMessage a : alarmMessage) {
             for (Event event : constructCurrentEvent(a)) {
                 analyzerService.analyze(event);
             }
-        });
+        }
     }
 
-    private List<Event> constructCurrentEvent(AlarmMessage msg) {
+    private String getLayer(String serviceId) throws IOException {
+        return getMetadataQueryService().getService(serviceId).getLayers().iterator().next();
+    }
+
+    private List<Event> constructCurrentEvent(AlarmMessage msg) throws IOException {
         List<Event> events = new ArrayList<>(2);
         long now = System.currentTimeMillis();
         Event.Builder builder = Event.newBuilder()
@@ -65,8 +81,7 @@ public class EventHookCallback implements AlarmCallback {
                 .setStartTime(now - (msg.getPeriod() * 60 * 1000))
                 .setMessage(msg.getAlarmMessage())
                 .setType(Type.Error)
-                .setEndTime(now)
+                .setEndTime(now);
-                .setLayer(Layer.GENERAL.name());
         switch (msg.getScopeId()) {
             case DefaultScopeDefine.SERVICE :
                 IDManager.ServiceID.ServiceIDDefinition serviceIdDef = IDManager.ServiceID.analysisId(msg.getId0());
@@ -75,6 +90,7 @@ public class EventHookCallback implements AlarmCallback {
                             .setService(serviceIdDef.getName())
                             .build()
                 );
+                builder.setLayer(getLayer(msg.getId0()));
                 events.add(builder.build());
                 break;
             case DefaultScopeDefine.SERVICE_RELATION :
@@ -91,6 +107,7 @@ public class EventHookCallback implements AlarmCallback {
                                 .setService(destServiceIdDef.getName())
                                 .build()
                 ).setUuid(UUID.randomUUID().toString());
+                builder.setLayer(getLayer(sourceServiceIdDef.isReal() ? msg.getId0() : msg.getId1()));
                 events.add(builder.build());
                 break;
             case DefaultScopeDefine.SERVICE_INSTANCE :
@@ -101,6 +118,7 @@ public class EventHookCallback implements AlarmCallback {
                                 .setService(IDManager.ServiceID.analysisId(instanceIdDef.getServiceId()).getName())
                                 .build()
                 );
+                builder.setLayer(getLayer(instanceIdDef.getServiceId()));
                 events.add(builder.build());
                 break;
             case DefaultScopeDefine.SERVICE_INSTANCE_RELATION :
@@ -119,6 +137,8 @@ public class EventHookCallback implements AlarmCallback {
                                 .setService(IDManager.ServiceID.analysisId(destInstanceIdDef.getServiceId()).getName())
                                 .build()
                 ).setUuid(UUID.randomUUID().toString());
+                IDManager.ServiceID.ServiceIDDefinition srcInstanceServiceIdDef = IDManager.ServiceID.analysisId(sourceInstanceIdDef.getServiceId());
+                builder.setLayer(getLayer(srcInstanceServiceIdDef.isReal() ? sourceInstanceIdDef.getServiceId() : destInstanceIdDef.getServiceId()));
                 events.add(builder.build());
                 break;
             case DefaultScopeDefine.ENDPOINT :
@@ -129,6 +149,7 @@ public class EventHookCallback implements AlarmCallback {
                                 .setService(IDManager.ServiceID.analysisId(endpointIDDef.getServiceId()).getName())
                                 .build()
                 );
+                builder.setLayer(getLayer(endpointIDDef.getServiceId()));
                 events.add(builder.build());
                 break;
             case DefaultScopeDefine.ENDPOINT_RELATION :
@@ -147,6 +168,8 @@ public class EventHookCallback implements AlarmCallback {
                                 .setService(IDManager.ServiceID.analysisId(destEndpointIDDef.getServiceId()).getName())
                                 .build()
                 ).setUuid(UUID.randomUUID().toString());
+                IDManager.ServiceID.ServiceIDDefinition srcEndpointServiceIdDef = IDManager.ServiceID.analysisId(sourceEndpointIDDef.getServiceId());
+                builder.setLayer(getLayer(srcEndpointServiceIdDef.isReal() ? sourceEndpointIDDef.getServiceId() : destEndpointIDDef.getServiceId()));
                 events.add(builder.build());
                 break;
         }

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

@@ -21,8 +21,12 @@ 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.CoreModule;
 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.analysis.Layer;
+import org.apache.skywalking.oap.server.core.query.MetadataQueryService;
+import org.apache.skywalking.oap.server.core.query.type.Service;
 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;
@@ -52,6 +56,8 @@ public class EventHookCallbackTest {
 
     private EventAnalyzerService eventAnalyzerService = mock(EventAnalyzerServiceImpl.class);
 
+    private MetadataQueryService metadataQueryService = mock(MetadataQueryService.class);
+
     @Test
     public void testEventCallbackHasRightFlow() throws Exception {
         List<AlarmMessage> msgs = mockAlarmMessagesHasSingleElement();
@@ -59,7 +65,9 @@ public class EventHookCallbackTest {
         when(moduleManager.find("event-analyzer")).thenReturn(moduleProviderHolder);
         when(moduleProviderHolder.provider()).thenReturn(moduleServiceHolder);
         when(moduleServiceHolder.getService(EventAnalyzerService.class)).thenReturn(mockEventAnalyzerService);
-
+        when(moduleManager.find(CoreModule.NAME)).thenReturn(moduleProviderHolder);
+        when(moduleServiceHolder.getService(MetadataQueryService.class)).thenReturn(metadataQueryService);
+        when(metadataQueryService.getService(any())).thenReturn(mockService());
         //make sure current service be called.
         callback.doAlarm(msgs);
         verify(mockEventAnalyzerService).analyze(any(Event.class));
@@ -79,12 +87,15 @@ public class EventHookCallbackTest {
     }
 
     @Test
-    public void testRelationEventBeProperlyConstructed() {
+    public void testRelationEventBeProperlyConstructed() throws Exception {
         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);
+        when(moduleManager.find(CoreModule.NAME)).thenReturn(moduleProviderHolder);
+        when(moduleServiceHolder.getService(MetadataQueryService.class)).thenReturn(metadataQueryService);
+        when(metadataQueryService.getService(any())).thenReturn(mockService());
         callback.doAlarm(msgs);
 
         ArgumentCaptor<Event> argument = ArgumentCaptor.forClass(Event.class);
@@ -123,6 +134,13 @@ public class EventHookCallbackTest {
         return Arrays.asList(msg);
     }
 
+    private Service mockService() {
+        Service service = new Service();
+        service.setName("test-skywalking");
+        service.getLayers().add(Layer.GENERAL.name());
+        return service;
+    }
+
     class MockEventAnalyzerService implements EventAnalyzerService {
 
         @Override