1. All the inventories add last_update_time attribute, delete the... (16b82365) · Commits · SUMMER2020 / students / proj-2005254

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

+1 −0

Original line number	Diff line number	Diff line
		@@ -36,5 +36,6 @@ public class Const {
		public static final String SEGMENT_SPAN_SPLIT = "S";
		public static final String UNKNOWN = "Unknown";
		public static final String EMPTY_STRING = "";
		public static final String EMPTY_JSON_OBJECT_STRING = "{}";
		public static final String DOMAIN_OPERATION_NAME = "{domain}";
		}

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cache/CacheUpdateTimer.java

+40 −14

Original line number	Diff line number	Diff line
		@@ -22,10 +22,10 @@ import java.util.*;
		import java.util.concurrent.*;
		import org.apache.skywalking.apm.util.RunnableWithExceptionProtection;
		import org.apache.skywalking.oap.server.core.CoreModule;
		import org.apache.skywalking.oap.server.core.register.ServiceInventory;
		import org.apache.skywalking.oap.server.core.register.*;
		import org.apache.skywalking.oap.server.core.storage.StorageModule;
		import org.apache.skywalking.oap.server.core.storage.cache.IServiceInventoryCacheDAO;
		import org.apache.skywalking.oap.server.library.module.ModuleManager;
		import org.apache.skywalking.oap.server.core.storage.cache.*;
		import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
		import org.slf4j.*;

		/**
		@@ -38,34 +38,60 @@ public enum CacheUpdateTimer {

		private Boolean isStarted = false;

		public void start(ModuleManager moduleManager) {
		logger.info("Cache update timer start");
		public void start(ModuleDefineHolder moduleDefineHolder) {
		logger.info("Cache updateServiceInventory timer start");

		final long timeInterval = 3;
		final long timeInterval = 10;

		if (!isStarted) {
		Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(
		new RunnableWithExceptionProtection(() -> update(moduleManager),
		new RunnableWithExceptionProtection(() -> update(moduleDefineHolder),
		t -> logger.error("Cache update failure.", t)), 1, timeInterval, TimeUnit.SECONDS);

		this.isStarted = true;
		}
		}

		private void update(ModuleManager moduleManager) {
		IServiceInventoryCacheDAO serviceInventoryCacheDAO = moduleManager.find(StorageModule.NAME).provider().getService(IServiceInventoryCacheDAO.class);
		ServiceInventoryCache serviceInventoryCache = moduleManager.find(CoreModule.NAME).provider().getService(ServiceInventoryCache.class);
		List<ServiceInventory> serviceInventories = serviceInventoryCacheDAO.loadLastMappingUpdate();
		private void update(ModuleDefineHolder moduleDefineHolder) {
		updateServiceInventory(moduleDefineHolder);
		updateNetAddressInventory(moduleDefineHolder);
		}

		private void updateServiceInventory(ModuleDefineHolder moduleDefineHolder) {
		IServiceInventoryCacheDAO serviceInventoryCacheDAO = moduleDefineHolder.find(StorageModule.NAME).provider().getService(IServiceInventoryCacheDAO.class);
		ServiceInventoryCache serviceInventoryCache = moduleDefineHolder.find(CoreModule.NAME).provider().getService(ServiceInventoryCache.class);
		List<ServiceInventory> serviceInventories = serviceInventoryCacheDAO.loadLastUpdate(System.currentTimeMillis() - 60000);

		serviceInventories.forEach(serviceInventory -> {
		logger.info("Update mapping service id in the cache of service inventory, service id: {}, mapping service id: {}", serviceInventory.getSequence(), serviceInventory.getMappingServiceId());
		ServiceInventory cache = serviceInventoryCache.get(serviceInventory.getSequence());
		if (Objects.nonNull(cache)) {
		if (cache.getMappingServiceId() != serviceInventory.getMappingServiceId()) {
		cache.setMappingServiceId(serviceInventory.getMappingServiceId());
		cache.setMappingLastUpdateTime(serviceInventory.getMappingLastUpdateTime());
		cache.setServiceNodeType(serviceInventory.getServiceNodeType());
		cache.setProperties(serviceInventory.getProperties());
		logger.info("Update the cache of service inventory, service id: {}", serviceInventory.getSequence());
		}
		} else {
		logger.warn("Unable to found the id of {} in service inventory cache.", serviceInventory.getSequence());
		}
		});
		}

		private void updateNetAddressInventory(ModuleDefineHolder moduleDefineHolder) {
		INetworkAddressInventoryCacheDAO addressInventoryCacheDAO = moduleDefineHolder.find(StorageModule.NAME).provider().getService(INetworkAddressInventoryCacheDAO.class);
		NetworkAddressInventoryCache addressInventoryCache = moduleDefineHolder.find(CoreModule.NAME).provider().getService(NetworkAddressInventoryCache.class);
		List<NetworkAddressInventory> addressInventories = addressInventoryCacheDAO.loadLastUpdate(System.currentTimeMillis() - 60000);

		addressInventories.forEach(addressInventory -> {
		NetworkAddressInventory cache = addressInventoryCache.get(addressInventory.getSequence());
		if (Objects.nonNull(cache)) {
		if (!cache.getNetworkAddressNodeType().equals(addressInventory.getNetworkAddressNodeType())) {
		cache.setNetworkAddressNodeType(addressInventory.getNetworkAddressNodeType());
		logger.info("Update the cache of net address inventory, address id: {}", addressInventory.getSequence());
		}
		} else {
		logger.warn("Unable to found the id of {} in net address inventory cache.", addressInventory.getSequence());
		}
		});
		}
		}
		No newline at end of file

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/EndpointInventory.java

+9 −5

Original line number	Diff line number	Diff line
		@@ -91,6 +91,7 @@ public class EndpointInventory extends RegisterSource {

		remoteBuilder.addDataLongs(getRegisterTime());
		remoteBuilder.addDataLongs(getHeartbeatTime());
		remoteBuilder.addDataLongs(getLastUpdateTime());

		remoteBuilder.addDataStrings(Strings.isNullOrEmpty(name) ? Const.EMPTY_STRING : name);
		return remoteBuilder;
		@@ -103,6 +104,7 @@ public class EndpointInventory extends RegisterSource {

		setRegisterTime(remoteData.getDataLongs(0));
		setHeartbeatTime(remoteData.getDataLongs(1));
		setLastUpdateTime(remoteData.getDataLongs(2));

		setName(remoteData.getDataStrings(0));
		}
		@@ -115,12 +117,13 @@ public class EndpointInventory extends RegisterSource {

		@Override public EndpointInventory map2Data(Map<String, Object> dbMap) {
		EndpointInventory inventory = new EndpointInventory();
		inventory.setSequence((Integer)dbMap.get(SEQUENCE));
		inventory.setServiceId((Integer)dbMap.get(SERVICE_ID));
		inventory.setSequence(((Number)dbMap.get(SEQUENCE)).intValue());
		inventory.setServiceId(((Number)dbMap.get(SERVICE_ID)).intValue());
		inventory.setName((String)dbMap.get(NAME));
		inventory.setDetectPoint((Integer)dbMap.get(DETECT_POINT));
		inventory.setRegisterTime((Long)dbMap.get(REGISTER_TIME));
		inventory.setHeartbeatTime((Long)dbMap.get(HEARTBEAT_TIME));
		inventory.setDetectPoint(((Number)dbMap.get(DETECT_POINT)).intValue());
		inventory.setRegisterTime(((Number)dbMap.get(REGISTER_TIME)).longValue());
		inventory.setHeartbeatTime(((Number)dbMap.get(HEARTBEAT_TIME)).longValue());
		inventory.setLastUpdateTime(((Number)dbMap.get(LAST_UPDATE_TIME)).longValue());
		return inventory;
		}

		@@ -132,6 +135,7 @@ public class EndpointInventory extends RegisterSource {
		map.put(DETECT_POINT, storageData.getDetectPoint());
		map.put(REGISTER_TIME, storageData.getRegisterTime());
		map.put(HEARTBEAT_TIME, storageData.getHeartbeatTime());
		map.put(LAST_UPDATE_TIME, storageData.getLastUpdateTime());
		return map;
		}
		}

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/NetworkAddressInventory.java

+13 −8

Original line number	Diff line number	Diff line
		@@ -88,6 +88,7 @@ public class NetworkAddressInventory extends RegisterSource {
		inventory.setSequence(getSequence());
		inventory.setRegisterTime(getRegisterTime());
		inventory.setHeartbeatTime(getHeartbeatTime());
		inventory.setLastUpdateTime(getLastUpdateTime());
		inventory.setName(name);
		inventory.setNodeType(nodeType);

		@@ -95,14 +96,14 @@ public class NetworkAddressInventory extends RegisterSource {
		}

		@Override public boolean combine(RegisterSource registerSource) {
		boolean isCombine = super.combine(registerSource);
		boolean isChanged = super.combine(registerSource);
		NetworkAddressInventory inventory = (NetworkAddressInventory)registerSource;

		if (nodeType != inventory.nodeType) {
		setNodeType(inventory.nodeType);
		if (this.nodeType != inventory.getNodeType() && inventory.getLastUpdateTime() >= this.getLastUpdateTime()) {
		setNodeType(inventory.getNodeType());
		return true;
		} else {
		return isCombine;
		return isChanged;
		}
		}

		@@ -113,6 +114,7 @@ public class NetworkAddressInventory extends RegisterSource {

		remoteBuilder.addDataLongs(getRegisterTime());
		remoteBuilder.addDataLongs(getHeartbeatTime());
		remoteBuilder.addDataLongs(getLastUpdateTime());

		remoteBuilder.addDataStrings(Strings.isNullOrEmpty(name) ? Const.EMPTY_STRING : name);
		return remoteBuilder;
		@@ -124,6 +126,7 @@ public class NetworkAddressInventory extends RegisterSource {

		setRegisterTime(remoteData.getDataLongs(0));
		setHeartbeatTime(remoteData.getDataLongs(1));
		setLastUpdateTime(remoteData.getDataLongs(2));

		setName(remoteData.getDataStrings(0));
		}
		@@ -136,11 +139,12 @@ public class NetworkAddressInventory extends RegisterSource {

		@Override public NetworkAddressInventory map2Data(Map<String, Object> dbMap) {
		NetworkAddressInventory inventory = new NetworkAddressInventory();
		inventory.setSequence((Integer)dbMap.get(SEQUENCE));
		inventory.setSequence(((Number)dbMap.get(SEQUENCE)).intValue());
		inventory.setName((String)dbMap.get(NAME));
		inventory.setNodeType((Integer)dbMap.get(NODE_TYPE));
		inventory.setRegisterTime((Long)dbMap.get(REGISTER_TIME));
		inventory.setHeartbeatTime((Long)dbMap.get(HEARTBEAT_TIME));
		inventory.setNodeType(((Number)dbMap.get(NODE_TYPE)).intValue());
		inventory.setRegisterTime(((Number)dbMap.get(REGISTER_TIME)).longValue());
		inventory.setHeartbeatTime(((Number)dbMap.get(HEARTBEAT_TIME)).longValue());
		inventory.setLastUpdateTime(((Number)dbMap.get(LAST_UPDATE_TIME)).longValue());
		return inventory;
		}

		@@ -151,6 +155,7 @@ public class NetworkAddressInventory extends RegisterSource {
		map.put(NODE_TYPE, storageData.getNodeType());
		map.put(REGISTER_TIME, storageData.getRegisterTime());
		map.put(HEARTBEAT_TIME, storageData.getHeartbeatTime());
		map.put(LAST_UPDATE_TIME, storageData.getLastUpdateTime());
		return map;
		}
		}

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/RegisterSource.java

+12 −5

Original line number	Diff line number	Diff line
		@@ -31,17 +31,24 @@ public abstract class RegisterSource extends StreamData implements StorageData {
		public static final String SEQUENCE = "sequence";
		public static final String REGISTER_TIME = "register_time";
		public static final String HEARTBEAT_TIME = "heartbeat_time";
		public static final String LAST_UPDATE_TIME = "last_update_time";

		@Getter @Setter @Column(columnName = SEQUENCE) private int sequence;
		@Getter @Setter @Column(columnName = REGISTER_TIME) private long registerTime;
		@Getter @Setter @Column(columnName = HEARTBEAT_TIME) private long heartbeatTime;
		@Getter @Setter @Column(columnName = REGISTER_TIME) private long registerTime = 0L;
		@Getter @Setter @Column(columnName = HEARTBEAT_TIME) private long heartbeatTime = 0L;
		@Setter @Getter @Column(columnName = LAST_UPDATE_TIME) private long lastUpdateTime = 0L;

		public boolean combine(RegisterSource registerSource) {
		boolean isChanged = false;
		if (heartbeatTime < registerSource.getHeartbeatTime()) {
		heartbeatTime = registerSource.getHeartbeatTime();
		return true;
		} else {
		return false;
		isChanged = true;
		}

		if (lastUpdateTime < registerSource.getLastUpdateTime()) {
		lastUpdateTime = registerSource.getLastUpdateTime();
		isChanged = true;
		}
		return isChanged;
		}
		}