diff --git a/monitoring/src/main/java/org/thingsboard/monitoring/service/BaseHealthChecker.java b/monitoring/src/main/java/org/thingsboard/monitoring/service/BaseHealthChecker.java index 697573289d..290fd307c6 100644 --- a/monitoring/src/main/java/org/thingsboard/monitoring/service/BaseHealthChecker.java +++ b/monitoring/src/main/java/org/thingsboard/monitoring/service/BaseHealthChecker.java @@ -15,6 +15,7 @@ */ package org.thingsboard.monitoring.service; +import lombok.Getter; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; @@ -30,13 +31,17 @@ import org.thingsboard.monitoring.util.TbStopWatch; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; +import java.util.HashMap; +import java.util.Map; import java.util.UUID; @RequiredArgsConstructor @Slf4j public abstract class BaseHealthChecker { + @Getter protected final C config; + @Getter protected final T target; private Object info; @@ -48,6 +53,9 @@ public abstract class BaseHealthChecker> associates = new HashMap<>(); + public static final String TEST_TELEMETRY_KEY = "testData"; @PostConstruct @@ -84,6 +92,10 @@ public abstract class BaseHealthChecker { + healthChecker.check(wsClient); + }); } private void checkWsUpdate(WsClient wsClient, String testValue) { @@ -99,7 +111,6 @@ public abstract class BaseHealthChecker, T ext tbClient.logIn(); configs.forEach(config -> { config.getTargets().forEach(target -> { - initHealthChecker(target, config); + BaseHealthChecker healthChecker = initHealthChecker(target, config); + healthCheckers.add(healthChecker); + if (target.isCheckDomainIps()) { - initIpsHealthCheckers(target, config); + getAssociatedUrls(target.getBaseUrl()).forEach(url -> { + healthChecker.getAssociates().put(url, initHealthChecker(createTarget(url), config)); + }); } }); }); } - private void initHealthChecker(T target, C config) { + private BaseHealthChecker initHealthChecker(T target, C config) { BaseHealthChecker healthChecker = (BaseHealthChecker) createHealthChecker(config, target); log.info("Initializing {} for {}", healthChecker.getClass().getSimpleName(), target.getBaseUrl()); healthChecker.initialize(tbClient); devices.add(target.getDeviceId()); - healthCheckers.add(healthChecker); - } - - @SneakyThrows - private void initIpsHealthCheckers(T target, C config) { - URI baseUrl = new URI(target.getBaseUrl()); - String domain = baseUrl.getHost(); - - Set ips = Arrays.stream(InetAddress.getAllByName(domain)) - .map(InetAddress::getHostAddress) - .collect(Collectors.toSet()); - for (String ip : ips) { - String url = new URI(baseUrl.getScheme(), null, ip, baseUrl.getPort(), "", null, null).toString(); - initHealthChecker(createTarget(url), config); - } + return healthChecker; } public final void runChecks() { @@ -108,9 +101,8 @@ public abstract class BaseMonitoringService, T ext try (WsClient wsClient = wsClientFactory.createClient(accessToken)) { wsClient.subscribeForTelemetry(devices, TransportHealthChecker.TEST_TELEMETRY_KEY).waitForReply(); - for (BaseHealthChecker healthChecker : healthCheckers) { - healthChecker.check(wsClient); + check(healthChecker, wsClient); } } reporter.reportLatencies(tbClient); @@ -124,6 +116,57 @@ public abstract class BaseMonitoringService, T ext } } + private void check(BaseHealthChecker healthChecker, WsClient wsClient) throws Exception { + healthChecker.check(wsClient); + + T target = healthChecker.getTarget(); + if (target.isCheckDomainIps()) { + Set associatedUrls = getAssociatedUrls(target.getBaseUrl()); + Map> associates = healthChecker.getAssociates(); + Set prevAssociatedUrls = new HashSet<>(associates.keySet()); + + boolean changed = false; + for (String url : associatedUrls) { + if (!prevAssociatedUrls.contains(url)) { + BaseHealthChecker associate = initHealthChecker(createTarget(url), healthChecker.getConfig()); + associates.put(url, associate); + changed = true; + } + } + for (String url : prevAssociatedUrls) { + if (!associatedUrls.contains(url)) { + stopHealthChecker(healthChecker); + associates.remove(url); + changed = true; + } + } + if (changed) { + log.info("Updated IPs for {}: {} (old list: {})", target.getBaseUrl(), associatedUrls, prevAssociatedUrls); + } + } + } + + @SneakyThrows + private Set getAssociatedUrls(String baseUrl) { + URI url = new URI(baseUrl); + return Arrays.stream(InetAddress.getAllByName(url.getHost())) + .map(InetAddress::getHostAddress) + .map(ip -> { + try { + return new URI(url.getScheme(), null, ip, url.getPort(), "", null, null).toString(); + } catch (URISyntaxException e) { + throw new RuntimeException(e); + } + }) + .collect(Collectors.toSet()); + } + + private void stopHealthChecker(BaseHealthChecker healthChecker) throws Exception { + healthChecker.destroyClient(); + devices.remove(healthChecker.getTarget().getDeviceId()); + log.info("Stopped {} for {}", healthChecker.getClass().getSimpleName(), healthChecker.getTarget().getBaseUrl()); + } + protected abstract BaseHealthChecker createHealthChecker(C config, T target); protected abstract T createTarget(String baseUrl);