Add EDQS monitoring

monitoring/src/main/java/org/thingsboard/monitoring/data/Latencies.java

@@ -20,6 +20,7 @@ public class Latencies {
     public static final String WS_CONNECT = "wsConnect";
     public static final String WS_SUBSCRIBE = "wsSubscribe";
     public static final String LOG_IN = "logIn";
+    public static final String EDQS_QUERY = "edqsQuery";
 
     public static String request(String key) {
         return String.format("%sRequest", key);

monitoring/src/main/java/org/thingsboard/monitoring/data/MonitoredServiceKey.java

@@ -18,5 +18,6 @@ package org.thingsboard.monitoring.data;
 public class MonitoredServiceKey {
 
     public static final String GENERAL = "Monitoring";
+    public static final String EDQS = "*EDQS*";
 
 }

monitoring/src/main/java/org/thingsboard/monitoring/service/BaseMonitoringService.java

@@ -15,10 +15,13 @@
  */
 package org.thingsboard.monitoring.service;
 
+import com.google.common.collect.Sets;
 import jakarta.annotation.PostConstruct;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.ApplicationContext;
 import org.thingsboard.monitoring.client.TbClient;
 import org.thingsboard.monitoring.client.WsClient;
@@ -27,13 +30,26 @@ import org.thingsboard.monitoring.config.MonitoringConfig;
 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;
+import org.thingsboard.server.common.data.query.EntityData;
+import org.thingsboard.server.common.data.query.EntityDataPageLink;
+import org.thingsboard.server.common.data.query.EntityDataQuery;
+import org.thingsboard.server.common.data.query.EntityDataSortOrder;
+import org.thingsboard.server.common.data.query.EntityKey;
+import org.thingsboard.server.common.data.query.EntityKeyType;
+import org.thingsboard.server.common.data.query.EntityTypeFilter;
+import org.thingsboard.server.common.data.query.TsValue;
 
 import java.net.InetAddress;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
@@ -41,6 +57,7 @@ import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 @Slf4j
 public abstract class BaseMonitoringService<C extends MonitoringConfig<T>, T extends MonitoringTarget> {
@@ -61,6 +78,9 @@ public abstract class BaseMonitoringService<C extends MonitoringConfig<T>, T ext
     @Autowired
     protected ApplicationContext applicationContext;
 
+    @Value("${monitoring.edqs.enabled:false}")
+    private boolean edqsMonitoringEnabled;
+
     @PostConstruct
     private void init() {
         if (configs == null || configs.isEmpty()) {
@@ -108,6 +128,21 @@ public abstract class BaseMonitoringService<C extends MonitoringConfig<T>, T ext
                     check(healthChecker, wsClient);
                 }
             }
+
+            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);
+                }
+            }
+
             reporter.reportLatencies(tbClient);
             log.debug("Finished {}", getName());
         } catch (Throwable error) {
@@ -149,6 +184,39 @@ public abstract class BaseMonitoringService<C extends MonitoringConfig<T>, T ext
         }
     }
 
+    private void checkEdqs() {
+        EntityTypeFilter entityTypeFilter = new EntityTypeFilter();
+        entityTypeFilter.setEntityType(EntityType.DEVICE);
+        EntityDataPageLink pageLink = new EntityDataPageLink(100, 0, null, new EntityDataSortOrder(new EntityKey(EntityKeyType.ENTITY_FIELD, "name")));
+        EntityDataQuery entityDataQuery = new EntityDataQuery(entityTypeFilter, pageLink,
+                List.of(new EntityKey(EntityKeyType.ENTITY_FIELD, "name"), new EntityKey(EntityKeyType.ENTITY_FIELD, "type")),
+                List.of(new EntityKey(EntityKeyType.TIME_SERIES, "testData")),
+                Collections.emptyList());
+
+        PageData<EntityData> result = tbClient.findEntityDataByQuery(entityDataQuery);
+        Set<UUID> devices = result.getData().stream()
+                .map(entityData -> entityData.getEntityId().getId())
+                .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);
+        }
+
+        result.getData().stream()
+                .filter(entityData -> this.devices.contains(entityData.getEntityId().getId()))
+                .forEach(entityData -> {
+                    Map<String, TsValue> values = new HashMap<>(entityData.getLatest().get(EntityKeyType.ENTITY_FIELD));
+                    values.putAll(entityData.getLatest().get(EntityKeyType.TIME_SERIES));
+
+                    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());
+                        }
+                    });
+                });
+    }
+
     @SneakyThrows
     private Set<String> getAssociatedUrls(String baseUrl) {
         URI url = new URI(baseUrl);

monitoring/src/main/resources/tb-monitoring.yml

@@ -105,6 +105,9 @@ monitoring:
         # 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}"
+
   notifications:
     message_prefix: '${NOTIFICATION_MESSAGE_PREFIX:}'
     slack: