ViacheslavKlimov
3 years ago
21 changed files with 348 additions and 289 deletions
@ -0,0 +1,44 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed 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.thingsboard.monitoring.data.cmd; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonIgnoreProperties; |
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.query.EntityData; |
|||
import org.thingsboard.server.common.data.query.EntityKeyType; |
|||
import org.thingsboard.server.common.data.query.TsValue; |
|||
|
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
|
|||
@Data |
|||
@JsonIgnoreProperties(ignoreUnknown = true) |
|||
public class EntityDataUpdate { |
|||
|
|||
@JsonIgnoreProperties(ignoreUnknown = true) |
|||
private List<EntityData> update; |
|||
|
|||
public String getLatest(UUID entityId, String key) { |
|||
if (update == null) return null; |
|||
|
|||
return update.stream() |
|||
.filter(entityData -> entityData.getEntityId().getId().equals(entityId)).findFirst() |
|||
.map(EntityData::getLatest).map(latest -> latest.get(EntityKeyType.TIME_SERIES)) |
|||
.map(latest -> latest.get(key)).map(TsValue::getValue) |
|||
.orElse(null); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,28 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed 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.thingsboard.monitoring.data.cmd; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.query.EntityKey; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Data |
|||
public class LatestValueCmd { |
|||
|
|||
private List<EntityKey> keys; |
|||
|
|||
} |
|||
@ -1,52 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed 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.thingsboard.monitoring.data.cmd; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonIgnoreProperties; |
|||
import lombok.Data; |
|||
import org.springframework.data.util.Pair; |
|||
|
|||
import java.util.Comparator; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Objects; |
|||
|
|||
@Data |
|||
@JsonIgnoreProperties(ignoreUnknown = true) |
|||
public class TimeseriesUpdate { |
|||
|
|||
private final int subscriptionId; |
|||
private final int errorCode; |
|||
private final String errorMsg; |
|||
private Map<String, List<List<Object>>> data; |
|||
|
|||
public Object getLatest(String key) { |
|||
if (!data.containsKey(key)) return null; |
|||
return data.get(key).stream() |
|||
.map(tsAndValue -> { |
|||
if (tsAndValue == null || tsAndValue.size() != 2) { |
|||
return null; |
|||
} |
|||
long ts = Long.parseLong(tsAndValue.get(0).toString()); |
|||
Object value = tsAndValue.get(1); |
|||
return Pair.of(ts, value); |
|||
}) |
|||
.filter(Objects::nonNull) |
|||
.max(Comparator.comparing(Pair::getFirst)) |
|||
.map(Pair::getSecond).orElse(null); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,143 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed 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.thingsboard.monitoring.transport; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.apache.commons.lang3.StringUtils; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.boot.context.event.ApplicationReadyEvent; |
|||
import org.springframework.context.ApplicationContext; |
|||
import org.springframework.context.event.EventListener; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|||
import org.thingsboard.monitoring.client.TbClient; |
|||
import org.thingsboard.monitoring.client.WsClient; |
|||
import org.thingsboard.monitoring.client.WsClientFactory; |
|||
import org.thingsboard.monitoring.config.DeviceConfig; |
|||
import org.thingsboard.monitoring.config.MonitoringTargetConfig; |
|||
import org.thingsboard.monitoring.config.service.TransportMonitoringConfig; |
|||
import org.thingsboard.monitoring.data.Latencies; |
|||
import org.thingsboard.monitoring.data.MonitoredServiceKey; |
|||
import org.thingsboard.monitoring.service.MonitoringReporter; |
|||
import org.thingsboard.monitoring.util.TbStopWatch; |
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.data.device.data.DefaultDeviceConfiguration; |
|||
import org.thingsboard.server.common.data.device.data.DefaultDeviceTransportConfiguration; |
|||
import org.thingsboard.server.common.data.device.data.DeviceData; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.security.DeviceCredentials; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import java.util.LinkedList; |
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
@Service |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public final class TransportMonitoringService { |
|||
|
|||
private final List<TransportMonitoringConfig> configs; |
|||
private final List<TransportHealthChecker<?>> transportHealthCheckers = new LinkedList<>(); |
|||
private final List<UUID> devices = new LinkedList<>(); |
|||
|
|||
private final TbClient tbClient; |
|||
private final WsClientFactory wsClientFactory; |
|||
private final TbStopWatch stopWatch; |
|||
private final MonitoringReporter reporter; |
|||
private final ApplicationContext applicationContext; |
|||
private ScheduledExecutorService scheduler; |
|||
@Value("${monitoring.transports.monitoring_rate_ms}") |
|||
private int monitoringRateMs; |
|||
|
|||
@PostConstruct |
|||
private void init() { |
|||
configs.forEach(config -> { |
|||
config.getTargets().stream() |
|||
.filter(target -> StringUtils.isNotBlank(target.getBaseUrl())) |
|||
.peek(target -> checkMonitoringTarget(config, target, tbClient)) |
|||
.forEach(target -> { |
|||
TransportHealthChecker<?> transportHealthChecker = applicationContext.getBean(config.getTransportType().getServiceClass(), config, target); |
|||
transportHealthCheckers.add(transportHealthChecker); |
|||
devices.add(target.getDevice().getId()); |
|||
}); |
|||
}); |
|||
scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("monitoring-executor")); |
|||
} |
|||
|
|||
@EventListener(ApplicationReadyEvent.class) |
|||
public void startMonitoring() { |
|||
scheduleCheck(0); |
|||
} |
|||
|
|||
private void scheduleCheck(int delay) { |
|||
log.debug("Scheduling next check for {} ms", delay); |
|||
scheduler.schedule(() -> { |
|||
try { |
|||
stopWatch.start(); |
|||
String accessToken = tbClient.logIn(); |
|||
reporter.reportLatency(Latencies.LOG_IN, stopWatch.getTime()); |
|||
|
|||
try (WsClient wsClient = wsClientFactory.createClient(accessToken)) { |
|||
wsClient.subscribeForTelemetry(devices, TransportHealthChecker.TEST_TELEMETRY_KEY).waitForReply(); |
|||
|
|||
for (TransportHealthChecker<?> transportHealthChecker : transportHealthCheckers) { |
|||
transportHealthChecker.check(wsClient); |
|||
} |
|||
} |
|||
reporter.reportLatencies(tbClient); |
|||
} catch (Exception e) { |
|||
reporter.serviceFailure(MonitoredServiceKey.GENERAL, e); |
|||
} |
|||
scheduleCheck(monitoringRateMs); |
|||
}, delay, TimeUnit.MILLISECONDS); |
|||
} |
|||
|
|||
private void checkMonitoringTarget(TransportMonitoringConfig config, MonitoringTargetConfig target, TbClient tbClient) { |
|||
DeviceConfig deviceConfig = target.getDevice(); |
|||
tbClient.logIn(); |
|||
|
|||
DeviceId deviceId; |
|||
if (deviceConfig == null || deviceConfig.getId() == null) { |
|||
String deviceName = String.format("[%s] Monitoring device (%s)", config.getTransportType(), target.getBaseUrl()); |
|||
Device device = tbClient.getTenantDevice(deviceName) |
|||
.orElseGet(() -> { |
|||
log.info("Creating new device '{}'", deviceName); |
|||
Device monitoringDevice = new Device(); |
|||
monitoringDevice.setName(deviceName); |
|||
monitoringDevice.setType("default"); |
|||
DeviceData deviceData = new DeviceData(); |
|||
deviceData.setConfiguration(new DefaultDeviceConfiguration()); |
|||
deviceData.setTransportConfiguration(new DefaultDeviceTransportConfiguration()); |
|||
return tbClient.saveDevice(monitoringDevice); |
|||
}); |
|||
deviceId = device.getId(); |
|||
target.getDevice().setId(deviceId.toString()); |
|||
} else { |
|||
deviceId = new DeviceId(deviceConfig.getId()); |
|||
} |
|||
|
|||
log.debug("Loading credentials for device {}", deviceId); |
|||
DeviceCredentials credentials = tbClient.getDeviceCredentialsByDeviceId(deviceId) |
|||
.orElseThrow(() -> new IllegalArgumentException("No credentials found for device " + deviceId)); |
|||
target.getDevice().setCredentials(credentials); |
|||
} |
|||
|
|||
} |
|||
Loading…
Reference in new issue