From 82d4cb538112a98ed6425cf56c63e27cceb6731f Mon Sep 17 00:00:00 2001
From: ViacheslavKlimov <viacheslavklimov11@gmail.com>
Date: Tue, 15 Apr 2025 15:12:18 +0300
Subject: [PATCH] Add monitoring for calculated fields

---
 .../monitoring/client/WsClient.java           | 69 +++++++++++--------
 .../config/transport/TransportInfo.java       | 11 ++-
 .../transport/TransportMonitoringTarget.java  |  1 +
 .../monitoring/data/MonitoredServiceKey.java  |  1 +
 .../data/ServiceFailureException.java         | 11 ++-
 .../monitoring/data/cmd/EntityDataUpdate.java | 18 +++--
 .../monitoring/service/BaseHealthChecker.java | 44 ++++++++----
 .../service/BaseMonitoringService.java        | 37 +++++-----
 .../service/MonitoringReporter.java           |  2 +-
 .../transport/TransportHealthChecker.java     | 44 +++++++++++-
 .../src/main/resources/tb-monitoring.yml      | 11 +++
 .../thingsboard/rest/client/RestClient.java   | 20 +++++-
 12 files changed, 189 insertions(+), 80 deletions(-)

diff --git a/monitoring/src/main/java/org/thingsboard/monitoring/client/WsClient.java b/monitoring/src/main/java/org/thingsboard/monitoring/client/WsClient.java
index 9246feac27..9bacbdfd45 100644
--- a/monitoring/src/main/java/org/thingsboard/monitoring/client/WsClient.java
+++ b/monitoring/src/main/java/org/thingsboard/monitoring/client/WsClient.java
@@ -36,8 +36,11 @@ import org.thingsboard.server.common.data.query.EntityListFilter;
 import javax.net.ssl.SSLParameters;
 import java.net.URI;
 import java.nio.channels.NotYetConnectedException;
+import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -48,13 +51,13 @@ import java.util.stream.Collectors;
 @Slf4j
 public class WsClient extends WebSocketClient implements AutoCloseable {
 
-    public volatile JsonNode lastMsg;
+    public final List<JsonNode> lastMsgs = new ArrayList<>();
     private CountDownLatch reply;
     private CountDownLatch update;
 
     private final Lock updateLock = new ReentrantLock();
 
-    private long requestTimeoutMs;
+    private final long requestTimeoutMs;
 
     public WsClient(URI serverUri, long requestTimeoutMs) {
         super(serverUri);
@@ -63,7 +66,6 @@ public class WsClient extends WebSocketClient implements AutoCloseable {
 
     @Override
     public void onOpen(ServerHandshake serverHandshake) {
-
     }
 
     @Override
@@ -73,8 +75,9 @@ public class WsClient extends WebSocketClient implements AutoCloseable {
         }
         updateLock.lock();
         try {
-            lastMsg = JacksonUtil.toJsonNode(s);
-            log.trace("Received new msg: {}", lastMsg.toPrettyString());
+            JsonNode msg = JacksonUtil.toJsonNode(s);
+            lastMsgs.add(msg);
+            log.trace("Received new msg: {}", msg.toPrettyString());
             if (update != null) {
                 update.countDown();
             }
@@ -96,11 +99,11 @@ public class WsClient extends WebSocketClient implements AutoCloseable {
         log.error("WebSocket client error:", e);
     }
 
-    public void registerWaitForUpdate() {
+    public void registerWaitForUpdates(int count) {
         updateLock.lock();
         try {
-            lastMsg = null;
-            update = new CountDownLatch(1);
+            lastMsgs.clear();
+            update = new CountDownLatch(count);
         } finally {
             updateLock.unlock();
         }
@@ -111,6 +114,7 @@ public class WsClient extends WebSocketClient implements AutoCloseable {
     public void send(String text) throws NotYetConnectedException {
         updateLock.lock();
         try {
+            lastMsgs.clear();
             reply = new CountDownLatch(1);
         } finally {
             updateLock.unlock();
@@ -118,19 +122,19 @@ public class WsClient extends WebSocketClient implements AutoCloseable {
         super.send(text);
     }
 
-    public WsClient subscribeForTelemetry(List<UUID> devices, String key) {
+    public WsClient subscribeForTelemetry(List<UUID> devices, List<String> keys) {
         EntityDataCmd cmd = new EntityDataCmd();
         cmd.setCmdId(RandomUtils.nextInt(0, 1000));
 
         EntityListFilter devicesFilter = new EntityListFilter();
         devicesFilter.setEntityType(EntityType.DEVICE);
         devicesFilter.setEntityList(devices.stream().map(UUID::toString).collect(Collectors.toList()));
-        EntityDataPageLink pageLink = new EntityDataPageLink(100,0, null, null);
+        EntityDataPageLink pageLink = new EntityDataPageLink(100, 0, null, null);
         EntityDataQuery devicesQuery = new EntityDataQuery(devicesFilter, pageLink, Collections.emptyList(), Collections.emptyList(), Collections.emptyList());
         cmd.setQuery(devicesQuery);
 
         LatestValueCmd latestCmd = new LatestValueCmd();
-        latestCmd.setKeys(List.of(new EntityKey(EntityKeyType.TIME_SERIES, key)));
+        latestCmd.setKeys(keys.stream().map(key -> new EntityKey(EntityKeyType.TIME_SERIES, key)).toList());
         cmd.setLatestCmd(latestCmd);
 
         CmdsWrapper wrapper = new CmdsWrapper();
@@ -139,12 +143,12 @@ public class WsClient extends WebSocketClient implements AutoCloseable {
         return this;
     }
 
-    public JsonNode waitForUpdate(long ms) {
+    public List<JsonNode> waitForUpdates(long ms) {
         log.trace("update latch count: {}", update.getCount());
         try {
             if (update.await(ms, TimeUnit.MILLISECONDS)) {
                 log.trace("Waited for update");
-                return getLastMsg();
+                return getLastMsgs();
             }
         } catch (InterruptedException e) {
             log.debug("Failed to await reply", e);
@@ -157,7 +161,8 @@ public class WsClient extends WebSocketClient implements AutoCloseable {
         try {
             if (reply.await(requestTimeoutMs, TimeUnit.MILLISECONDS)) {
                 log.trace("Waited for reply");
-                return getLastMsg();
+                List<JsonNode> lastMsgs = getLastMsgs();
+                return lastMsgs.isEmpty() ? null : lastMsgs.get(0);
             }
         } catch (InterruptedException e) {
             log.debug("Failed to await reply", e);
@@ -166,24 +171,30 @@ public class WsClient extends WebSocketClient implements AutoCloseable {
         throw new IllegalStateException("No WS reply arrived within " + requestTimeoutMs + " ms");
     }
 
-    private JsonNode getLastMsg() {
-        if (lastMsg != null) {
-            JsonNode errorMsg = lastMsg.get("errorMsg");
-            if (errorMsg != null && !errorMsg.isNull() && StringUtils.isNotEmpty(errorMsg.asText())) {
-                throw new RuntimeException("WS error from server: " + errorMsg.asText());
-            } else {
-                return lastMsg;
-            }
-        } else {
-            return null;
+    private List<JsonNode> getLastMsgs() {
+        if (lastMsgs.isEmpty()) {
+            return lastMsgs;
+        }
+        List<JsonNode> errors = lastMsgs.stream()
+                .map(msg -> msg.get("errorMsg"))
+                .filter(errorMsg -> errorMsg != null && !errorMsg.isNull() && StringUtils.isNotEmpty(errorMsg.asText()))
+                .toList();
+        if (!errors.isEmpty()) {
+            throw new RuntimeException("WS error from server: " + errors.stream()
+                    .map(JsonNode::asText)
+                    .collect(Collectors.joining(", ")));
         }
+        return lastMsgs;
     }
 
-    public Object getTelemetryUpdate(UUID deviceId, String key) {
-        JsonNode lastMsg = getLastMsg();
-        if (lastMsg == null || lastMsg.isNull()) return null;
-        EntityDataUpdate update = JacksonUtil.treeToValue(lastMsg, EntityDataUpdate.class);
-        return update.getLatest(deviceId, key);
+    public Map<String, String> getLatest(UUID deviceId) {
+        Map<String, String> updates = new HashMap<>();
+        getLastMsgs().forEach(msg -> {
+            EntityDataUpdate update = JacksonUtil.treeToValue(msg, EntityDataUpdate.class);
+            Map<String, String> latest = update.getLatest(deviceId);
+            updates.putAll(latest);
+        });
+        return updates;
     }
 
     @Override
diff --git a/monitoring/src/main/java/org/thingsboard/monitoring/config/transport/TransportInfo.java b/monitoring/src/main/java/org/thingsboard/monitoring/config/transport/TransportInfo.java
index 208d28ffee..6b77e1e268 100644
--- a/monitoring/src/main/java/org/thingsboard/monitoring/config/transport/TransportInfo.java
+++ b/monitoring/src/main/java/org/thingsboard/monitoring/config/transport/TransportInfo.java
@@ -20,16 +20,15 @@ import lombok.Data;
 @Data
 public class TransportInfo {
 
-    private final TransportType transportType;
-    private final String baseUrl;
-    private final String queue;
+    private final TransportType type;
+    private final TransportMonitoringTarget target;
 
     @Override
     public String toString() {
-        if (queue.equals("Main")) {
-            return String.format("*%s* (%s)", transportType.getName(), baseUrl);
+        if (target.getQueue().equals("Main")) {
+            return String.format("*%s* (%s)", type.getName(), target.getBaseUrl());
         } else {
-            return String.format("*%s* (%s) _%s_", transportType.getName(), baseUrl, queue);
+            return String.format("*%s* (%s) _%s_", type.getName(), target.getBaseUrl(), target.getQueue());
         }
     }
 
diff --git a/monitoring/src/main/java/org/thingsboard/monitoring/config/transport/TransportMonitoringTarget.java b/monitoring/src/main/java/org/thingsboard/monitoring/config/transport/TransportMonitoringTarget.java
index 1e6a3c8509..e8a9ab03fa 100644
--- a/monitoring/src/main/java/org/thingsboard/monitoring/config/transport/TransportMonitoringTarget.java
+++ b/monitoring/src/main/java/org/thingsboard/monitoring/config/transport/TransportMonitoringTarget.java
@@ -28,6 +28,7 @@ public class TransportMonitoringTarget implements MonitoringTarget {
     private DeviceConfig device; // set manually during initialization
     private String queue;
     private boolean checkDomainIps;
+    private String namePrefix;
 
     @Override
     public UUID getDeviceId() {
diff --git a/monitoring/src/main/java/org/thingsboard/monitoring/data/MonitoredServiceKey.java b/monitoring/src/main/java/org/thingsboard/monitoring/data/MonitoredServiceKey.java
index 9c3ee5b786..7579d75231 100644
--- a/monitoring/src/main/java/org/thingsboard/monitoring/data/MonitoredServiceKey.java
+++ b/monitoring/src/main/java/org/thingsboard/monitoring/data/MonitoredServiceKey.java
@@ -19,5 +19,6 @@ public class MonitoredServiceKey {
 
     public static final String GENERAL = "Monitoring";
     public static final String EDQS = "*EDQS*";
+    public static final String CF = "*CF*";
 
 }
diff --git a/monitoring/src/main/java/org/thingsboard/monitoring/data/ServiceFailureException.java b/monitoring/src/main/java/org/thingsboard/monitoring/data/ServiceFailureException.java
index 5f46514bd1..b2592b8719 100644
--- a/monitoring/src/main/java/org/thingsboard/monitoring/data/ServiceFailureException.java
+++ b/monitoring/src/main/java/org/thingsboard/monitoring/data/ServiceFailureException.java
@@ -15,14 +15,21 @@
  */
 package org.thingsboard.monitoring.data;
 
+import lombok.Getter;
+
+@Getter
 public class ServiceFailureException extends RuntimeException {
 
-    public ServiceFailureException(Throwable cause) {
+    private final Object serviceKey;
+
+    public ServiceFailureException(Object serviceKey, Throwable cause) {
         super(cause.getMessage(), cause);
+        this.serviceKey = serviceKey;
     }
 
-    public ServiceFailureException(String message) {
+    public ServiceFailureException(Object serviceKey, String message) {
         super(message);
+        this.serviceKey = serviceKey;
     }
 
 }
diff --git a/monitoring/src/main/java/org/thingsboard/monitoring/data/cmd/EntityDataUpdate.java b/monitoring/src/main/java/org/thingsboard/monitoring/data/cmd/EntityDataUpdate.java
index b82dcbf54d..0706dedcf3 100644
--- a/monitoring/src/main/java/org/thingsboard/monitoring/data/cmd/EntityDataUpdate.java
+++ b/monitoring/src/main/java/org/thingsboard/monitoring/data/cmd/EntityDataUpdate.java
@@ -19,9 +19,11 @@ 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.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.UUID;
 
 @Data
@@ -31,14 +33,16 @@ 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()
+    public Map<String, String> getLatest(UUID entityId) {
+        if (update == null || update.isEmpty()) {
+            return Collections.emptyMap();
+        }
+        Map<String, String> result = new HashMap<>();
+        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);
+                .ifPresent(latest -> latest.forEach((key, tsValue) -> result.put(key, tsValue.getValue())));
+        return result;
     }
 
 }
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 482b3a30fb..34e0ebc4c5 100644
--- a/monitoring/src/main/java/org/thingsboard/monitoring/service/BaseHealthChecker.java
+++ b/monitoring/src/main/java/org/thingsboard/monitoring/service/BaseHealthChecker.java
@@ -52,11 +52,14 @@ public abstract class BaseHealthChecker<C extends MonitoringConfig, T extends Mo
     private TbStopWatch stopWatch;
     @Value("${monitoring.check_timeout_ms}")
     private int resultCheckTimeoutMs;
+    @Value("${monitoring.calculated_fields.enabled:true}")
+    protected boolean checkCalculatedFields;
 
     @Getter
     private final Map<String, BaseHealthChecker<C, T>> associates = new HashMap<>();
 
     public static final String TEST_TELEMETRY_KEY = "testData";
+    public static final String TEST_CF_TELEMETRY_KEY = "testDataCf";
 
     @PostConstruct
     private void init() {
@@ -68,7 +71,8 @@ public abstract class BaseHealthChecker<C extends MonitoringConfig, T extends Mo
     public final void check(WsClient wsClient) {
         log.debug("[{}] Checking", info);
         try {
-            wsClient.registerWaitForUpdate();
+            int expectedUpdatesCount = checkCalculatedFields ? 2 : 1;
+            wsClient.registerWaitForUpdates(expectedUpdatesCount);
 
             String testValue = UUID.randomUUID().toString();
             String testPayload = createTestPayload(testValue);
@@ -79,16 +83,16 @@ public abstract class BaseHealthChecker<C extends MonitoringConfig, T extends Mo
                 reporter.reportLatency(Latencies.request(getKey()), stopWatch.getTime());
                 log.trace("[{}] Sent test payload ({})", info, testPayload);
             } catch (Throwable e) {
-                throw new ServiceFailureException(e);
+                throw new ServiceFailureException(info, e);
             }
 
             log.trace("[{}] Waiting for WS update", info);
-            checkWsUpdate(wsClient, testValue);
+            checkWsUpdates(wsClient, testValue);
 
             reporter.serviceIsOk(info);
             reporter.serviceIsOk(MonitoredServiceKey.GENERAL);
-        } catch (ServiceFailureException serviceFailureException) {
-            reporter.serviceFailure(info, serviceFailureException);
+        } catch (ServiceFailureException e) {
+            reporter.serviceFailure(e.getServiceKey(), e);
         } catch (Exception e) {
             reporter.serviceFailure(MonitoredServiceKey.GENERAL, e);
         }
@@ -98,15 +102,28 @@ public abstract class BaseHealthChecker<C extends MonitoringConfig, T extends Mo
         });
     }
 
-    private void checkWsUpdate(WsClient wsClient, String testValue) {
+    private void checkWsUpdates(WsClient wsClient, String testValue) {
         stopWatch.start();
-        wsClient.waitForUpdate(resultCheckTimeoutMs);
-        log.trace("[{}] Waited for WS update. Last WS msg: {}", info, wsClient.lastMsg);
-        Object update = wsClient.getTelemetryUpdate(target.getDeviceId(), TEST_TELEMETRY_KEY);
-        if (update == null) {
-            throw new ServiceFailureException("No WS update arrived within " + resultCheckTimeoutMs + " ms");
-        } else if (!update.toString().equals(testValue)) {
-            throw new ServiceFailureException("Was expecting value " + testValue + " but got " + update);
+        wsClient.waitForUpdates(resultCheckTimeoutMs);
+        log.trace("[{}] Waited for WS update. Last WS msgs: {}", info, wsClient.lastMsgs);
+        Map<String, String> latest = wsClient.getLatest(target.getDeviceId());
+        if (latest.isEmpty()) {
+            throw new ServiceFailureException(info, "No WS update arrived within " + resultCheckTimeoutMs + " ms");
+        }
+        String actualValue = latest.get(TEST_TELEMETRY_KEY);
+        if (!testValue.equals(actualValue)) {
+            throw new ServiceFailureException(info, "Was expecting value " + testValue + " but got " + actualValue);
+        }
+        if (checkCalculatedFields) {
+            String cfTestValue = testValue + "-cf";
+            String actualCfValue = latest.get(TEST_CF_TELEMETRY_KEY);
+            if (actualCfValue == null) {
+                throw new ServiceFailureException(MonitoredServiceKey.CF, "No CF value arrived");
+            } else if (!cfTestValue.equals(actualCfValue)) {
+                throw new ServiceFailureException(MonitoredServiceKey.CF, "Was expecting CF value " + cfTestValue + " but got " + actualCfValue);
+            } else {
+                reporter.serviceIsOk(MonitoredServiceKey.CF);
+            }
         }
         reporter.reportLatency(Latencies.wsUpdate(getKey()), stopWatch.getTime());
     }
@@ -121,6 +138,7 @@ public abstract class BaseHealthChecker<C extends MonitoringConfig, T extends Mo
     protected abstract void destroyClient() throws Exception;
 
     protected abstract Object getInfo();
+
     protected abstract String getKey();
 
 }
diff --git a/monitoring/src/main/java/org/thingsboard/monitoring/service/BaseMonitoringService.java b/monitoring/src/main/java/org/thingsboard/monitoring/service/BaseMonitoringService.java
index 7219e4f9aa..f68600cc61 100644
--- a/monitoring/src/main/java/org/thingsboard/monitoring/service/BaseMonitoringService.java
+++ b/monitoring/src/main/java/org/thingsboard/monitoring/service/BaseMonitoringService.java
@@ -31,7 +31,6 @@ import org.thingsboard.monitoring.config.MonitoringTarget;
 import org.thingsboard.monitoring.data.Latencies;
 import org.thingsboard.monitoring.data.MonitoredServiceKey;
 import org.thingsboard.monitoring.data.ServiceFailureException;
-import org.thingsboard.monitoring.service.transport.TransportHealthChecker;
 import org.thingsboard.monitoring.util.TbStopWatch;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.page.PageData;
@@ -59,6 +58,9 @@ import java.util.UUID;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static org.thingsboard.monitoring.service.BaseHealthChecker.TEST_CF_TELEMETRY_KEY;
+import static org.thingsboard.monitoring.service.BaseHealthChecker.TEST_TELEMETRY_KEY;
+
 @Slf4j
 public abstract class BaseMonitoringService<C extends MonitoringConfig<T>, T extends MonitoringTarget> {
 
@@ -79,7 +81,9 @@ public abstract class BaseMonitoringService<C extends MonitoringConfig<T>, T ext
     protected ApplicationContext applicationContext;
 
     @Value("${monitoring.edqs.enabled:false}")
-    private boolean edqsMonitoringEnabled;
+    private boolean checkEdqs;
+    @Value("${monitoring.calculated_fields.enabled:true}")
+    protected boolean checkCalculatedFields;
 
     @PostConstruct
     private void init() {
@@ -121,7 +125,7 @@ public abstract class BaseMonitoringService<C extends MonitoringConfig<T>, T ext
 
             try (WsClient wsClient = wsClientFactory.createClient(accessToken)) {
                 stopWatch.start();
-                wsClient.subscribeForTelemetry(devices, TransportHealthChecker.TEST_TELEMETRY_KEY).waitForReply();
+                wsClient.subscribeForTelemetry(devices, getTestTelemetryKeys()).waitForReply();
                 reporter.reportLatency(Latencies.WS_SUBSCRIBE, stopWatch.getTime());
 
                 for (BaseHealthChecker<C, T> healthChecker : healthCheckers) {
@@ -129,22 +133,17 @@ public abstract class BaseMonitoringService<C extends MonitoringConfig<T>, T ext
                 }
             }
 
-            if (edqsMonitoringEnabled) {
-                try {
-                    stopWatch.start();
-                    checkEdqs();
-                    reporter.reportLatency(Latencies.EDQS_QUERY, stopWatch.getTime());
-
-                    reporter.serviceIsOk(MonitoredServiceKey.EDQS);
-                } catch (ServiceFailureException e) {
-                    reporter.serviceFailure(MonitoredServiceKey.EDQS, e);
-                } catch (Exception e) {
-                    reporter.serviceFailure(MonitoredServiceKey.GENERAL, e);
-                }
+            if (checkEdqs) {
+                stopWatch.start();
+                checkEdqs();
+                reporter.reportLatency(Latencies.EDQS_QUERY, stopWatch.getTime());
+                reporter.serviceIsOk(MonitoredServiceKey.EDQS);
             }
 
             reporter.reportLatencies(tbClient);
             log.debug("Finished {}", getName());
+        } catch (ServiceFailureException e) {
+            reporter.serviceFailure(e.getServiceKey(), e);
         } catch (Throwable error) {
             try {
                 reporter.serviceFailure(MonitoredServiceKey.GENERAL, error);
@@ -199,7 +198,7 @@ public abstract class BaseMonitoringService<C extends MonitoringConfig<T>, T ext
                 .collect(Collectors.toSet());
         Set<UUID> missing = Sets.difference(new HashSet<>(this.devices), devices);
         if (!missing.isEmpty()) {
-            throw new ServiceFailureException("Missing devices in the response: " + missing);
+            throw new ServiceFailureException(MonitoredServiceKey.EDQS, "Missing devices in the response: " + missing);
         }
 
         result.getData().stream()
@@ -211,7 +210,7 @@ public abstract class BaseMonitoringService<C extends MonitoringConfig<T>, T ext
                     Stream.of("name", "type", "testData").forEach(key -> {
                         TsValue value = values.get(key);
                         if (value == null || StringUtils.isBlank(value.getValue())) {
-                            throw new ServiceFailureException("Missing " + key + " for device " + entityData.getEntityId());
+                            throw new ServiceFailureException(MonitoredServiceKey.EDQS, "Missing " + key + " for device " + entityData.getEntityId());
                         }
                     });
                 });
@@ -232,6 +231,10 @@ public abstract class BaseMonitoringService<C extends MonitoringConfig<T>, T ext
                 .collect(Collectors.toSet());
     }
 
+    private List<String> getTestTelemetryKeys() {
+        return checkCalculatedFields ? List.of(TEST_TELEMETRY_KEY, TEST_CF_TELEMETRY_KEY) : List.of(TEST_TELEMETRY_KEY);
+    }
+
     private void stopHealthChecker(BaseHealthChecker<C, T> healthChecker) throws Exception {
         healthChecker.destroyClient();
         devices.remove(healthChecker.getTarget().getDeviceId());
diff --git a/monitoring/src/main/java/org/thingsboard/monitoring/service/MonitoringReporter.java b/monitoring/src/main/java/org/thingsboard/monitoring/service/MonitoringReporter.java
index 3554731e58..62ed0d74aa 100644
--- a/monitoring/src/main/java/org/thingsboard/monitoring/service/MonitoringReporter.java
+++ b/monitoring/src/main/java/org/thingsboard/monitoring/service/MonitoringReporter.java
@@ -113,7 +113,7 @@ public class MonitoringReporter {
 
     public void serviceFailure(Object serviceKey, Throwable error) {
         if (log.isDebugEnabled()) {
-            log.error("Error occurred", error);
+            log.error("[{}] Error occurred", serviceKey, error);
         }
         int failuresCount = failuresCounters.computeIfAbsent(serviceKey, k -> new AtomicInteger()).incrementAndGet();
         ServiceFailureNotification notification = new ServiceFailureNotification(serviceKey, error, failuresCount);
diff --git a/monitoring/src/main/java/org/thingsboard/monitoring/service/transport/TransportHealthChecker.java b/monitoring/src/main/java/org/thingsboard/monitoring/service/transport/TransportHealthChecker.java
index f5e6cb5469..b892f5d609 100644
--- a/monitoring/src/main/java/org/thingsboard/monitoring/service/transport/TransportHealthChecker.java
+++ b/monitoring/src/main/java/org/thingsboard/monitoring/service/transport/TransportHealthChecker.java
@@ -32,6 +32,14 @@ import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.DeviceProfileType;
 import org.thingsboard.server.common.data.DeviceTransportType;
 import org.thingsboard.server.common.data.TbResource;
+import org.thingsboard.server.common.data.cf.CalculatedField;
+import org.thingsboard.server.common.data.cf.CalculatedFieldType;
+import org.thingsboard.server.common.data.cf.configuration.Argument;
+import org.thingsboard.server.common.data.cf.configuration.ArgumentType;
+import org.thingsboard.server.common.data.cf.configuration.Output;
+import org.thingsboard.server.common.data.cf.configuration.OutputType;
+import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey;
+import org.thingsboard.server.common.data.cf.configuration.ScriptCalculatedFieldConfiguration;
 import org.thingsboard.server.common.data.device.credentials.lwm2m.LwM2MBootstrapClientCredentials;
 import org.thingsboard.server.common.data.device.credentials.lwm2m.LwM2MDeviceCredentials;
 import org.thingsboard.server.common.data.device.credentials.lwm2m.NoSecBootstrapClientCredential;
@@ -47,6 +55,8 @@ import org.thingsboard.server.common.data.page.PageLink;
 import org.thingsboard.server.common.data.security.DeviceCredentials;
 import org.thingsboard.server.common.data.security.DeviceCredentialsType;
 
+import java.util.Map;
+
 @Slf4j
 public abstract class TransportHealthChecker<C extends TransportMonitoringConfig> extends BaseHealthChecker<C, TransportMonitoringTarget> {
 
@@ -74,7 +84,7 @@ public abstract class TransportHealthChecker<C extends TransportMonitoringConfig
 
     @Override
     protected Object getInfo() {
-        return new TransportInfo(getTransportType(), target.getBaseUrl(), target.getQueue());
+        return new TransportInfo(getTransportType(), target);
     }
 
     @Override
@@ -87,9 +97,16 @@ public abstract class TransportHealthChecker<C extends TransportMonitoringConfig
 
     private Device getOrCreateDevice(TbClient tbClient) {
         TransportType transportType = config.getTransportType();
-        String deviceName = String.format("%s (%s) - %s", transportType.getName(), target.getQueue(), target.getBaseUrl());
+        String deviceName = String.format("%s %s (%s) - %s", target.getNamePrefix(), transportType.getName(), target.getQueue(), target.getBaseUrl());
         Device device = tbClient.getTenantDevice(deviceName).orElse(null);
         if (device != null) {
+            if (checkCalculatedFields) {
+                CalculatedField calculatedField = tbClient.getCalculatedFieldsByEntityId(device.getId(), new PageLink(1, 0, TEST_CF_TELEMETRY_KEY))
+                        .getData().stream().findFirst().orElse(null);
+                if (calculatedField == null) {
+                    createCalculatedField(tbClient, device);
+                }
+            }
             return device;
         }
 
@@ -128,7 +145,7 @@ public abstract class TransportHealthChecker<C extends TransportMonitoringConfig
 
     private DeviceProfile getOrCreateDeviceProfile(TbClient tbClient) {
         TransportType transportType = config.getTransportType();
-        String profileName = String.format("%s (%s)", transportType.getName(), target.getQueue());
+        String profileName = String.format("%s %s (%s)", target.getNamePrefix(), transportType.getName(), target.getQueue());
         DeviceProfile deviceProfile = tbClient.getDeviceProfiles(new PageLink(1, 0, profileName)).getData()
                 .stream().findFirst().orElse(null);
         if (deviceProfile != null) {
@@ -160,4 +177,25 @@ public abstract class TransportHealthChecker<C extends TransportMonitoringConfig
         return tbClient.saveDeviceProfile(deviceProfile);
     }
 
+    private void createCalculatedField(TbClient tbClient, Device device) {
+        log.info("Creating calculated field for device '{}'", device.getName());
+        CalculatedField calculatedField = new CalculatedField();
+        calculatedField.setName(TEST_CF_TELEMETRY_KEY);
+        calculatedField.setEntityId(device.getId());
+        calculatedField.setType(CalculatedFieldType.SCRIPT);
+        ScriptCalculatedFieldConfiguration configuration = new ScriptCalculatedFieldConfiguration();
+        Argument testDataArgument = new Argument();
+        testDataArgument.setRefEntityKey(new ReferencedEntityKey(TEST_TELEMETRY_KEY, ArgumentType.TS_LATEST, null));
+        configuration.setArguments(Map.of(
+                TEST_TELEMETRY_KEY, testDataArgument
+        ));
+        configuration.setExpression("return { \"" + TEST_CF_TELEMETRY_KEY + "\": "+ TEST_TELEMETRY_KEY + " + \"-cf\" };");
+        Output output = new Output();
+        output.setType(OutputType.TIME_SERIES);
+        configuration.setOutput(output);
+        calculatedField.setConfiguration(configuration);
+        calculatedField.setDebugMode(true);
+        tbClient.saveCalculatedField(calculatedField);
+    }
+
 }
diff --git a/monitoring/src/main/resources/tb-monitoring.yml b/monitoring/src/main/resources/tb-monitoring.yml
index 6cc2e79cb4..0955862f46 100644
--- a/monitoring/src/main/resources/tb-monitoring.yml
+++ b/monitoring/src/main/resources/tb-monitoring.yml
@@ -57,6 +57,8 @@ monitoring:
           queue: '${MQTT_TRANSPORT_USED_QUEUE:Main}'
           # Whether to monitor IPs associated with the domain from base url
           check_domain_ips: '${MQTT_TRANSPORT_CHECK_DOMAIN_IPS:false}'
+          # Prefix for the target device name
+          name_prefix: '${MQTT_TRANSPORT_TARGET_NAME_PREFIX:}'
         # To add more targets, use following environment variables:
         # monitoring.transports.mqtt.targets[1].base_url, monitoring.transports.mqtt.targets[2].base_url, etc.
 
@@ -72,6 +74,8 @@ monitoring:
           queue: '${COAP_TRANSPORT_USED_QUEUE:Main}'
           # Whether to monitor IPs associated with the domain from base url
           check_domain_ips: '${COAP_TRANSPORT_CHECK_DOMAIN_IPS:false}'
+          # Prefix for the target device name
+          name_prefix: '${COAP_TRANSPORT_TARGET_NAME_PREFIX:}'
         # To add more targets, use following environment variables:
         # monitoring.transports.coap.targets[1].base_url, monitoring.transports.coap.targets[2].base_url, etc.
 
@@ -87,6 +91,8 @@ monitoring:
           queue: '${HTTP_TRANSPORT_USED_QUEUE:Main}'
           # Whether to monitor IPs associated with the domain from base url
           check_domain_ips: '${HTTP_TRANSPORT_CHECK_DOMAIN_IPS:false}'
+          # Prefix for the target device name
+          name_prefix: '${HTTP_TRANSPORT_TARGET_NAME_PREFIX:}'
         # To add more targets, use following environment variables:
         # monitoring.transports.http.targets[1].base_url, monitoring.transports.http.targets[2].base_url, etc.
 
@@ -102,12 +108,17 @@ monitoring:
           queue: '${LWM2M_TRANSPORT_USED_QUEUE:Main}'
           # Whether to monitor IPs associated with the domain from base url
           check_domain_ips: '${LWM2M_TRANSPORT_CHECK_DOMAIN_IPS:false}'
+          # Prefix for the target device name
+          name_prefix: '${LWM2M_TRANSPORT_TARGET_NAME_PREFIX:}'
         # To add more targets, use following environment variables:
         # monitoring.transports.lwm2m.targets[1].base_url, monitoring.transports.lwm2m.targets[2].base_url, etc.
 
   edqs:
     enabled: "${EDQS_MONITORING_ENABLED:false}"
 
+  calculated_fields:
+    enabled: "${CALCULATED_FIELDS_MONITORING_ENABLED:true}"
+
   notifications:
     message_prefix: '${NOTIFICATION_MESSAGE_PREFIX:}'
     slack:
diff --git a/rest-client/src/main/java/org/thingsboard/rest/client/RestClient.java b/rest-client/src/main/java/org/thingsboard/rest/client/RestClient.java
index 4f7538d205..633dbe9c57 100644
--- a/rest-client/src/main/java/org/thingsboard/rest/client/RestClient.java
+++ b/rest-client/src/main/java/org/thingsboard/rest/client/RestClient.java
@@ -55,9 +55,9 @@ import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.EntityView;
 import org.thingsboard.server.common.data.EntityViewInfo;
 import org.thingsboard.server.common.data.EventInfo;
-import org.thingsboard.server.common.data.ResourceExportData;
 import org.thingsboard.server.common.data.OtaPackage;
 import org.thingsboard.server.common.data.OtaPackageInfo;
+import org.thingsboard.server.common.data.ResourceExportData;
 import org.thingsboard.server.common.data.ResourceSubType;
 import org.thingsboard.server.common.data.SaveDeviceWithCredentialsRequest;
 import org.thingsboard.server.common.data.StringUtils;
@@ -86,6 +86,7 @@ import org.thingsboard.server.common.data.asset.AssetProfileInfo;
 import org.thingsboard.server.common.data.asset.AssetSearchQuery;
 import org.thingsboard.server.common.data.audit.ActionType;
 import org.thingsboard.server.common.data.audit.AuditLog;
+import org.thingsboard.server.common.data.cf.CalculatedField;
 import org.thingsboard.server.common.data.device.DeviceSearchQuery;
 import org.thingsboard.server.common.data.domain.Domain;
 import org.thingsboard.server.common.data.domain.DomainInfo;
@@ -3765,7 +3766,7 @@ public class RestClient implements Closeable {
     }
 
     public PageData<TbResourceInfo> getImages(PageLink pageLink, boolean includeSystemImages) {
-       return this.getImages(pageLink, null, includeSystemImages);
+        return this.getImages(pageLink, null, includeSystemImages);
     }
 
     public PageData<TbResourceInfo> getImages(PageLink pageLink, ResourceSubType imageSubType, boolean includeSystemImages) {
@@ -4056,6 +4057,21 @@ public class RestClient implements Closeable {
                 timeout).getBody();
     }
 
+    public CalculatedField saveCalculatedField(CalculatedField calculatedField) {
+        return restTemplate.postForEntity(baseURL + "/api/calculatedField", calculatedField, CalculatedField.class).getBody();
+    }
+
+    public PageData<CalculatedField> getCalculatedFieldsByEntityId(EntityId entityId, PageLink pageLink) {
+        Map<String, String> params = new HashMap<>();
+        addPageLinkToParam(params, pageLink);
+        return restTemplate.exchange(
+                baseURL + "/api/" + entityId.getEntityType() + "/" + entityId.getId() + "/calculatedFields?" + getUrlParams(pageLink),
+                HttpMethod.GET, HttpEntity.EMPTY,
+                new ParameterizedTypeReference<PageData<CalculatedField>>() {
+                }, params).getBody();
+
+    }
+
     private String getTimeUrlParams(TimePageLink pageLink) {
         String urlParams = getUrlParams(pageLink);
         if (pageLink.getStartTime() != null) {