diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/Rules.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/Rules.java index bfbced09b..e7a2e34db 100644 --- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/Rules.java +++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/Rules.java @@ -18,14 +18,17 @@ package org.apache.skywalking.oap.server.core.alarm.provider; -import java.util.List; -import lombok.AccessLevel; -import lombok.Getter; -import lombok.Setter; +import java.util.*; +import lombok.*; @Setter(AccessLevel.PUBLIC) @Getter(AccessLevel.PUBLIC) public class Rules { private List rules; private List webhooks; + + public Rules() { + this.rules = new ArrayList<>(); + this.webhooks = new ArrayList<>(); + } } diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/RulesReader.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/RulesReader.java index e2a28736c..897d892fb 100644 --- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/RulesReader.java +++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/RulesReader.java @@ -18,11 +18,8 @@ package org.apache.skywalking.oap.server.core.alarm.provider; -import java.io.InputStream; -import java.io.Reader; -import java.util.ArrayList; -import java.util.List; -import java.util.Map; +import java.io.*; +import java.util.*; import org.yaml.snakeyaml.Yaml; /** @@ -46,39 +43,42 @@ public class RulesReader { public Rules readRules() { Rules rules = new Rules(); - Map rulesData = (Map)yamlData.get("rules"); - if (rulesData != null) { - rules.setRules(new ArrayList<>()); - rulesData.forEach((k, v) -> { - if (((String)k).endsWith("_rule")) { - AlarmRule alarmRule = new AlarmRule(); - alarmRule.setAlarmRuleName((String)k); - Map settings = (Map)v; - Object indicatorName = settings.get("indicator-name"); - if (indicatorName == null) { - throw new IllegalArgumentException("indicator-name can't be null"); + if (Objects.nonNull(yamlData)) { + Map rulesData = (Map)yamlData.get("rules"); + if (rulesData != null) { + rules.setRules(new ArrayList<>()); + rulesData.forEach((k, v) -> { + if (((String)k).endsWith("_rule")) { + AlarmRule alarmRule = new AlarmRule(); + alarmRule.setAlarmRuleName((String)k); + Map settings = (Map)v; + Object indicatorName = settings.get("indicator-name"); + if (indicatorName == null) { + throw new IllegalArgumentException("indicator-name can't be null"); + } + + alarmRule.setIndicatorName((String)indicatorName); + alarmRule.setIncludeNames((ArrayList)settings.getOrDefault("include-names", new ArrayList(0))); + alarmRule.setThreshold(settings.get("threshold").toString()); + alarmRule.setOp((String)settings.get("op")); + alarmRule.setPeriod((Integer)settings.getOrDefault("period", 1)); + alarmRule.setCount((Integer)settings.getOrDefault("count", 1)); + alarmRule.setSilencePeriod((Integer)settings.getOrDefault("silence-period", alarmRule.getPeriod())); + alarmRule.setMessage((String)settings.getOrDefault("message", "Alarm caused by Rule " + alarmRule.getAlarmRuleName())); + + rules.getRules().add(alarmRule); } - - alarmRule.setIndicatorName((String)indicatorName); - alarmRule.setIncludeNames((ArrayList)settings.getOrDefault("include-names", new ArrayList(0))); - alarmRule.setThreshold(settings.get("threshold").toString()); - alarmRule.setOp((String)settings.get("op")); - alarmRule.setPeriod((Integer)settings.getOrDefault("period", 1)); - alarmRule.setCount((Integer)settings.getOrDefault("count", 1)); - alarmRule.setSilencePeriod((Integer)settings.getOrDefault("silence-period", alarmRule.getPeriod())); - alarmRule.setMessage((String)settings.getOrDefault("message", "Alarm caused by Rule " + alarmRule.getAlarmRuleName())); - - rules.getRules().add(alarmRule); - } - }); - } - List webhooks = (List)yamlData.get("webhooks"); - if (webhooks != null) { - rules.setWebhooks(new ArrayList<>()); - webhooks.forEach(url -> { - rules.getWebhooks().add((String)url); - }); + }); + } + List webhooks = (List)yamlData.get("webhooks"); + if (webhooks != null) { + rules.setWebhooks(new ArrayList<>()); + webhooks.forEach(url -> { + rules.getWebhooks().add((String)url); + }); + } } + return rules; } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/EndpointInventoryRegister.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/EndpointInventoryRegister.java index 895f421b7..eb361f5c0 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/EndpointInventoryRegister.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/EndpointInventoryRegister.java @@ -18,6 +18,7 @@ package org.apache.skywalking.oap.server.core.register.service; +import java.util.Objects; import org.apache.skywalking.oap.server.core.*; import org.apache.skywalking.oap.server.core.cache.EndpointInventoryCache; import org.apache.skywalking.oap.server.core.register.EndpointInventory; @@ -70,4 +71,15 @@ public class EndpointInventoryRegister implements IEndpointInventoryRegister { @Override public int get(int serviceId, String endpointName) { return getCacheService().getEndpointId(serviceId, endpointName); } + + @Override public void heartbeat(int endpointId, long heartBeatTime) { + EndpointInventory endpointInventory = getCacheService().get(endpointId); + if (Objects.nonNull(endpointInventory)) { + endpointInventory.setHeartbeatTime(heartBeatTime); + + InventoryProcess.INSTANCE.in(endpointInventory); + } else { + logger.warn("Endpoint {} heartbeat, but not found in storage."); + } + } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/IEndpointInventoryRegister.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/IEndpointInventoryRegister.java index 2258aac9f..2a7146145 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/IEndpointInventoryRegister.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/IEndpointInventoryRegister.java @@ -29,4 +29,6 @@ public interface IEndpointInventoryRegister extends Service { int getOrCreate(int serviceId, String endpointName, DetectPoint detectPoint); int get(int serviceId, String endpointName); + + void heartbeat(int endpointId, long heartBeatTime); } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/INetworkAddressInventoryRegister.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/INetworkAddressInventoryRegister.java index eb3b37560..f812f580b 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/INetworkAddressInventoryRegister.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/INetworkAddressInventoryRegister.java @@ -29,4 +29,6 @@ public interface INetworkAddressInventoryRegister extends Service { int get(String networkAddress); void update(int addressId, int srcLayer, int serverType); + + void heartbeat(int addressId, long heartBeatTime); } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/IServiceInstanceInventoryRegister.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/IServiceInstanceInventoryRegister.java index acee9a8d9..7ed7b4e1c 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/IServiceInstanceInventoryRegister.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/IServiceInstanceInventoryRegister.java @@ -30,4 +30,6 @@ public interface IServiceInstanceInventoryRegister extends Service { ServiceInstanceInventory.AgentOsInfo osInfo); int getOrCreate(int serviceId, int addressId, long registerTime); + + void heartbeat(int serviceInstanceId, long heartBeatTime); } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/IServiceInventoryRegister.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/IServiceInventoryRegister.java index 1667c61d4..a31b270d8 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/IServiceInventoryRegister.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/IServiceInventoryRegister.java @@ -28,4 +28,6 @@ public interface IServiceInventoryRegister extends Service { int getOrCreate(String serviceName); int getOrCreate(int addressId); + + void heartbeat(int serviceId, long heartBeatTime); } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/NetworkAddressInventoryRegister.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/NetworkAddressInventoryRegister.java index d4ea42918..e702d673c 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/NetworkAddressInventoryRegister.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/NetworkAddressInventoryRegister.java @@ -18,11 +18,13 @@ package org.apache.skywalking.oap.server.core.register.service; +import java.util.Objects; import org.apache.skywalking.oap.server.core.*; import org.apache.skywalking.oap.server.core.cache.NetworkAddressInventoryCache; import org.apache.skywalking.oap.server.core.register.NetworkAddressInventory; import org.apache.skywalking.oap.server.core.register.worker.InventoryProcess; import org.apache.skywalking.oap.server.library.module.ModuleManager; +import org.slf4j.*; import static java.util.Objects.*; @@ -31,6 +33,8 @@ import static java.util.Objects.*; */ public class NetworkAddressInventoryRegister implements INetworkAddressInventoryRegister { + private static final Logger logger = LoggerFactory.getLogger(NetworkAddressInventoryRegister.class); + private final ModuleManager moduleManager; private NetworkAddressInventoryCache networkAddressInventoryCache; private IServiceInventoryRegister serviceInventoryRegister; @@ -111,4 +115,15 @@ public class NetworkAddressInventoryRegister implements INetworkAddressInventory } return true; } + + @Override public void heartbeat(int addressId, long heartBeatTime) { + NetworkAddressInventory networkAddress = getNetworkAddressInventoryCache().get(addressId); + if (Objects.nonNull(networkAddress)) { + networkAddress.setHeartbeatTime(heartBeatTime); + + InventoryProcess.INSTANCE.in(networkAddress); + } else { + logger.warn("Network address {} heartbeat, but not found in storage."); + } + } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInstanceInventoryRegister.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInstanceInventoryRegister.java index bf6762755..af0299226 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInstanceInventoryRegister.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInstanceInventoryRegister.java @@ -18,11 +18,11 @@ package org.apache.skywalking.oap.server.core.register.service; -import org.apache.skywalking.oap.server.core.Const; +import java.util.Objects; +import org.apache.skywalking.oap.server.core.*; +import org.apache.skywalking.oap.server.core.cache.ServiceInstanceInventoryCache; import org.apache.skywalking.oap.server.core.register.ServiceInstanceInventory; import org.apache.skywalking.oap.server.core.register.worker.InventoryProcess; -import org.apache.skywalking.oap.server.core.storage.StorageModule; -import org.apache.skywalking.oap.server.core.storage.cache.IServiceInstanceInventoryCacheDAO; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.library.util.BooleanUtils; import org.slf4j.*; @@ -37,17 +37,17 @@ public class ServiceInstanceInventoryRegister implements IServiceInstanceInvento private static final Logger logger = LoggerFactory.getLogger(ServiceInstanceInventoryRegister.class); private final ModuleManager moduleManager; - private IServiceInstanceInventoryCacheDAO cacheDAO; + private ServiceInstanceInventoryCache serviceInstanceInventoryCache; public ServiceInstanceInventoryRegister(ModuleManager moduleManager) { this.moduleManager = moduleManager; } - private IServiceInstanceInventoryCacheDAO getCacheDAO() { - if (isNull(cacheDAO)) { - cacheDAO = moduleManager.find(StorageModule.NAME).getService(IServiceInstanceInventoryCacheDAO.class); + private ServiceInstanceInventoryCache getServiceInstanceInventoryCache() { + if (isNull(serviceInstanceInventoryCache)) { + serviceInstanceInventoryCache = moduleManager.find(CoreModule.NAME).getService(ServiceInstanceInventoryCache.class); } - return cacheDAO; + return serviceInstanceInventoryCache; } @Override public int getOrCreate(int serviceId, String serviceInstanceName, long registerTime, @@ -56,7 +56,7 @@ public class ServiceInstanceInventoryRegister implements IServiceInstanceInvento logger.debug("Get or create service instance by service instance name, service id: {}, service instance name: {}, registerTime: {}", serviceId, serviceInstanceName, registerTime); } - int serviceInstanceId = getCacheDAO().getServiceInstanceId(serviceId, serviceInstanceName); + int serviceInstanceId = getServiceInstanceInventoryCache().getServiceInstanceId(serviceId, serviceInstanceName); if (serviceInstanceId == Const.NONE) { ServiceInstanceInventory serviceInstanceInventory = new ServiceInstanceInventory(); @@ -83,7 +83,7 @@ public class ServiceInstanceInventoryRegister implements IServiceInstanceInvento logger.debug("get or create service instance by address id, service id: {}, address id: {}, registerTime: {}", serviceId, addressId, registerTime); } - int serviceInstanceId = getCacheDAO().getServiceInstanceId(serviceId, addressId); + int serviceInstanceId = getServiceInstanceInventoryCache().getServiceInstanceId(serviceId, addressId); if (serviceInstanceId == Const.NONE) { ServiceInstanceInventory serviceInstanceInventory = new ServiceInstanceInventory(); @@ -99,4 +99,14 @@ public class ServiceInstanceInventoryRegister implements IServiceInstanceInvento } return serviceInstanceId; } + + @Override public void heartbeat(int serviceInstanceId, long heartBeatTime) { + ServiceInstanceInventory serviceInstanceInventory = getServiceInstanceInventoryCache().get(serviceInstanceId); + if (Objects.nonNull(serviceInstanceInventory)) { + serviceInstanceInventory.setHeartbeatTime(heartBeatTime); + InventoryProcess.INSTANCE.in(serviceInstanceInventory); + } else { + logger.warn("Service instance {} heartbeat, but not found in storage."); + } + } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInventoryRegister.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInventoryRegister.java index 6b22eb425..62e8ab6a4 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInventoryRegister.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInventoryRegister.java @@ -18,12 +18,14 @@ package org.apache.skywalking.oap.server.core.register.service; +import java.util.Objects; import org.apache.skywalking.oap.server.core.*; import org.apache.skywalking.oap.server.core.cache.ServiceInventoryCache; import org.apache.skywalking.oap.server.core.register.ServiceInventory; import org.apache.skywalking.oap.server.core.register.worker.InventoryProcess; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.library.util.BooleanUtils; +import org.slf4j.*; import static java.util.Objects.isNull; @@ -32,6 +34,8 @@ import static java.util.Objects.isNull; */ public class ServiceInventoryRegister implements IServiceInventoryRegister { + private static final Logger logger = LoggerFactory.getLogger(ServiceInventoryRegister.class); + private final ModuleManager moduleManager; private ServiceInventoryCache serviceInventoryCache; @@ -81,4 +85,15 @@ public class ServiceInventoryRegister implements IServiceInventoryRegister { } return serviceId; } + + @Override public void heartbeat(int serviceId, long heartBeatTime) { + ServiceInventory serviceInventory = getServiceInventoryCache().get(serviceId); + if (Objects.nonNull(serviceInventory)) { + serviceInventory.setHeartbeatTime(heartBeatTime); + + InventoryProcess.INSTANCE.in(serviceInventory); + } else { + logger.warn("Service {} heartbeat, but not found in storage."); + } + } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterPersistentWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterPersistentWorker.java index c4283f0d4..cfed8c306 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterPersistentWorker.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterPersistentWorker.java @@ -62,7 +62,7 @@ public class RegisterPersistentWorker extends AbstractWorker { try { RegisterSource newSource = registerDAO.get(modelName, source.id()); if (Objects.nonNull(newSource)) { - newSource.combine(newSource); + newSource.combine(source); registerDAO.forceUpdate(modelName, newSource); } else { int sequence = registerDAO.max(modelName); diff --git a/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v5/InstanceDiscoveryServiceHandler.java b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v5/InstanceDiscoveryServiceHandler.java index 4295b0912..11681f145 100644 --- a/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v5/InstanceDiscoveryServiceHandler.java +++ b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v5/InstanceDiscoveryServiceHandler.java @@ -35,11 +35,9 @@ public class InstanceDiscoveryServiceHandler extends InstanceDiscoveryServiceGrp private static final Logger logger = LoggerFactory.getLogger(InstanceDiscoveryServiceHandler.class); private final IServiceInstanceInventoryRegister serviceInstanceInventoryRegister; -// private final IInstanceHeartBeatService instanceHeartBeatService; public InstanceDiscoveryServiceHandler(ModuleManager moduleManager) { this.serviceInstanceInventoryRegister = moduleManager.find(CoreModule.NAME).getService(IServiceInstanceInventoryRegister.class); -// this.instanceHeartBeatService = moduleManager.find(CoreModule.NAME).getService(IInstanceHeartBeatService.class); } @Override @@ -61,10 +59,10 @@ public class InstanceDiscoveryServiceHandler extends InstanceDiscoveryServiceGrp } @Override public void heartbeat(ApplicationInstanceHeartbeat request, StreamObserver responseObserver) { -// int instanceId = request.getApplicationInstanceId(); -// long heartBeatTime = request.getHeartbeatTime(); -// this.instanceHeartBeatService.heartBeat(instanceId, heartBeatTime); -// responseObserver.onNext(Downstream.getDefaultInstance()); -// responseObserver.onCompleted(); + int serviceInstanceId = request.getApplicationInstanceId(); + long heartBeatTime = request.getHeartbeatTime(); + serviceInstanceInventoryRegister.heartbeat(serviceInstanceId, heartBeatTime); + responseObserver.onNext(Downstream.getDefaultInstance()); + responseObserver.onCompleted(); } } diff --git a/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v5/InstanceHeartBeatTestCase.java b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v5/InstanceHeartBeatTestCase.java new file mode 100644 index 000000000..4f31c0a62 --- /dev/null +++ b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v5/InstanceHeartBeatTestCase.java @@ -0,0 +1,39 @@ +/* + * 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.receiver.register.provider.handler.v5; + +import io.grpc.*; +import org.apache.skywalking.apm.network.language.agent.*; + +/** + * @author peng-yongsheng + */ +public class InstanceHeartBeatTestCase { + + public static void main(String[] args) { + ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 11800).usePlaintext(true).build(); + + InstanceDiscoveryServiceGrpc.InstanceDiscoveryServiceBlockingStub stub = InstanceDiscoveryServiceGrpc.newBlockingStub(channel); + + ApplicationInstanceHeartbeat.Builder builder = ApplicationInstanceHeartbeat.newBuilder(); + builder.setApplicationInstanceId(2); + builder.setHeartbeatTime(System.currentTimeMillis() + 5 * 1000 * 60); + Downstream heartbeat = stub.heartbeat(builder.build()); + } +}