Monitoring: ability to specify used queue

monitoring/src/main/java/org/thingsboard/monitoring/config/MonitoringTarget.java

@@ -23,6 +23,10 @@ public interface MonitoringTarget {
 
     String getBaseUrl();
 
+    default String getQueue() {
+        return "Main";
+    }
+
     boolean isCheckDomainIps();
 
 }

monitoring/src/main/java/org/thingsboard/monitoring/config/transport/TransportInfo.java

@@ -22,10 +22,15 @@ public class TransportInfo {
 
     private final TransportType transportType;
     private final String baseUrl;
+    private final String queue;
 
     @Override
     public String toString() {
-        return String.format("*%s* (%s)", transportType.getName(), baseUrl);
+        if (queue.equals("Main")) {
+            return String.format("*%s* (%s)", transportType.getName(), baseUrl);
+        } else {
+            return String.format("*%s* (%s) _%s_", transportType.getName(), baseUrl, queue);
+        }
     }
 
 }

monitoring/src/main/java/org/thingsboard/monitoring/config/transport/TransportMonitoringTarget.java

@@ -25,6 +25,7 @@ public class TransportMonitoringTarget implements MonitoringTarget {
 
     private String baseUrl;
     private DeviceConfig device; // set manually during initialization
+    private String queue;
     private boolean checkDomainIps;
 
     @Override

monitoring/src/main/java/org/thingsboard/monitoring/data/notification/ServiceFailureNotification.java

@@ -43,7 +43,7 @@ public class ServiceFailureNotification implements Notification {
         if (errorMsg == null) {
             errorMsg = error.getClass().getSimpleName();
         }
-        return String.format("[%s] Failure: %s (number of subsequent failures: %s)", serviceKey, errorMsg, failuresCount);
+        return String.format("%s - Failure: %s (number of subsequent failures: %s)", serviceKey, errorMsg, failuresCount);
     }
 
 }

monitoring/src/main/java/org/thingsboard/monitoring/data/notification/ServiceRecoveryNotification.java

@@ -25,7 +25,7 @@ public class ServiceRecoveryNotification implements Notification {
 
     @Override
     public String getText() {
-        return String.format("[%s] is OK", serviceKey);
+        return String.format("%s is OK", serviceKey);
     }
 
 }

monitoring/src/main/java/org/thingsboard/monitoring/service/transport/TransportHealthChecker.java

@@ -29,6 +29,8 @@ import org.thingsboard.monitoring.service.BaseHealthChecker;
 import org.thingsboard.monitoring.util.ResourceUtils;
 import org.thingsboard.server.common.data.Device;
 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.device.credentials.lwm2m.LwM2MBootstrapClientCredentials;
 import org.thingsboard.server.common.data.device.credentials.lwm2m.LwM2MDeviceCredentials;
@@ -38,6 +40,9 @@ 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.device.data.Lwm2mDeviceTransportConfiguration;
+import org.thingsboard.server.common.data.device.profile.DefaultDeviceProfileConfiguration;
+import org.thingsboard.server.common.data.device.profile.DefaultDeviceProfileTransportConfiguration;
+import org.thingsboard.server.common.data.device.profile.DeviceProfileData;
 import org.thingsboard.server.common.data.page.PageLink;
 import org.thingsboard.server.common.data.security.DeviceCredentials;
 import org.thingsboard.server.common.data.security.DeviceCredentialsType;
@@ -45,27 +50,19 @@ import org.thingsboard.server.common.data.security.DeviceCredentialsType;
 @Slf4j
 public abstract class TransportHealthChecker<C extends TransportMonitoringConfig> extends BaseHealthChecker<C, TransportMonitoringTarget> {
 
-    private static final String DEFAULT_DEVICE_NAME = "[Monitoring] %s transport (%s)";
-    private static final String DEFAULT_PROFILE_NAME = "[Monitoring] %s";
-
     public TransportHealthChecker(C config, TransportMonitoringTarget target) {
         super(config, target);
     }
 
     @Override
     protected void initialize(TbClient tbClient) {
-        String deviceName = String.format(DEFAULT_DEVICE_NAME, config.getTransportType(), target.getBaseUrl());
+        Device device = getOrCreateDevice(tbClient);
-        Device device = tbClient.getTenantDevice(deviceName)
-                .orElseGet(() -> {
-                    log.info("Creating new device '{}'", deviceName);
-                    return createDevice(config.getTransportType(), deviceName, tbClient);
-                });
         DeviceCredentials credentials = tbClient.getDeviceCredentialsByDeviceId(device.getId())
                 .orElseThrow(() -> new IllegalArgumentException("No credentials found for device " + device.getId()));
 
         DeviceConfig deviceConfig = new DeviceConfig();
         deviceConfig.setId(device.getId().toString());
-        deviceConfig.setName(deviceName);
+        deviceConfig.setName(device.getName());
         deviceConfig.setCredentials(credentials);
         target.setDevice(deviceConfig);
     }
@@ -77,51 +74,43 @@ public abstract class TransportHealthChecker<C extends TransportMonitoringConfig
 
     @Override
     protected Object getInfo() {
-        return new TransportInfo(getTransportType(), target.getBaseUrl());
+        return new TransportInfo(getTransportType(), target.getBaseUrl(), target.getQueue());
     }
 
     @Override
     protected String getKey() {
-        return getTransportType().name().toLowerCase() + "Transport";
+        return getTransportType().name().toLowerCase() + (target.getQueue().equals("Main") ? "" : target.getQueue()) + "Transport";
     }
 
     protected abstract TransportType getTransportType();
 
 
-    private Device createDevice(TransportType transportType, String name, TbClient tbClient) {
+    private Device getOrCreateDevice(TbClient tbClient) {
-        Device device = new Device();
+        TransportType transportType = config.getTransportType();
-        device.setName(name);
+        String deviceName = String.format("%s (%s) - %s", transportType.getName(), target.getQueue(), target.getBaseUrl());
+        Device device = tbClient.getTenantDevice(deviceName).orElse(null);
+        if (device != null) {
+            return device;
+        }
+
+        log.info("Creating new device '{}'", deviceName);
+        device = new Device();
+        device.setName(deviceName);
 
         DeviceCredentials credentials = new DeviceCredentials();
         credentials.setCredentialsId(RandomStringUtils.randomAlphabetic(20));
-
         DeviceData deviceData = new DeviceData();
         deviceData.setConfiguration(new DefaultDeviceConfiguration());
+
+        DeviceProfile deviceProfile = getOrCreateDeviceProfile(tbClient);
+        device.setType(deviceProfile.getName());
+        device.setDeviceProfileId(deviceProfile.getId());
+
         if (transportType != TransportType.LWM2M) {
-            device.setType("default");
             deviceData.setTransportConfiguration(new DefaultDeviceTransportConfiguration());
             credentials.setCredentialsType(DeviceCredentialsType.ACCESS_TOKEN);
         } else {
-            tbClient.getResources(new PageLink(1, 0, "lwm2m monitoring")).getData()
-                    .stream().findFirst()
-                    .orElseGet(() -> {
-                        TbResource newResource = ResourceUtils.getResource("lwm2m/resource.json", TbResource.class);
-                        log.info("Creating LwM2M resource");
-                        return tbClient.saveResource(newResource);
-                    });
-            String profileName = String.format(DEFAULT_PROFILE_NAME, transportType);
-            DeviceProfile profile = tbClient.getDeviceProfiles(new PageLink(1, 0, profileName)).getData()
-                    .stream().findFirst()
-                    .orElseGet(() -> {
-                        DeviceProfile newProfile = ResourceUtils.getResource("lwm2m/device_profile.json", DeviceProfile.class);
-                        newProfile.setName(profileName);
-                        log.info("Creating LwM2M device profile");
-                        return tbClient.saveDeviceProfile(newProfile);
-                    });
-            device.setType(profileName);
-            device.setDeviceProfileId(profile.getId());
             deviceData.setTransportConfiguration(new Lwm2mDeviceTransportConfiguration());
-
             credentials.setCredentialsType(DeviceCredentialsType.LWM2M_CREDENTIALS);
             LwM2MDeviceCredentials lwm2mCreds = new LwM2MDeviceCredentials();
             NoSecClientCredential client = new NoSecClientCredential();
@@ -133,7 +122,42 @@ public abstract class TransportHealthChecker<C extends TransportMonitoringConfig
             lwm2mCreds.setBootstrap(bootstrap);
             credentials.setCredentialsValue(JacksonUtil.toString(lwm2mCreds));
         }
+
         return tbClient.saveDeviceWithCredentials(device, credentials).get();
     }
 
+    private DeviceProfile getOrCreateDeviceProfile(TbClient tbClient) {
+        TransportType transportType = config.getTransportType();
+        String profileName = String.format("%s (%s)", transportType.getName(), target.getQueue());
+        DeviceProfile deviceProfile = tbClient.getDeviceProfiles(new PageLink(1, 0, profileName)).getData()
+                .stream().findFirst().orElse(null);
+        if (deviceProfile != null) {
+            return deviceProfile;
+        }
+
+        log.info("Creating new device profile '{}'", profileName);
+        if (transportType != TransportType.LWM2M) {
+            deviceProfile = new DeviceProfile();
+            deviceProfile.setType(DeviceProfileType.DEFAULT);
+            deviceProfile.setTransportType(DeviceTransportType.DEFAULT);
+            DeviceProfileData profileData = new DeviceProfileData();
+            profileData.setConfiguration(new DefaultDeviceProfileConfiguration());
+            profileData.setTransportConfiguration(new DefaultDeviceProfileTransportConfiguration());
+            deviceProfile.setProfileData(profileData);
+        } else {
+            tbClient.getResources(new PageLink(1, 0, "lwm2m monitoring")).getData()
+                    .stream().findFirst()
+                    .orElseGet(() -> {
+                        TbResource newResource = ResourceUtils.getResource("lwm2m/resource.json", TbResource.class);
+                        log.info("Creating LwM2M resource");
+                        return tbClient.saveResource(newResource);
+                    });
+            deviceProfile = ResourceUtils.getResource("lwm2m/device_profile.json", DeviceProfile.class);
+        }
+
+        deviceProfile.setName(profileName);
+        deviceProfile.setDefaultQueueName(target.getQueue());
+        return tbClient.saveDeviceProfile(deviceProfile);
+    }
+
 }

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

@@ -53,6 +53,8 @@ monitoring:
       targets:
           # MQTT transport base url, tcp://DOMAIN:1883 by default
         - base_url: '${MQTT_TRANSPORT_BASE_URL:tcp://${monitoring.domain}:1883}'
+          # Queue to use for target device
+          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}'
         # To add more targets, use following environment variables:
@@ -66,6 +68,8 @@ monitoring:
       targets:
           # CoAP transport base url, coap://DOMAIN by default
         - base_url: '${COAP_TRANSPORT_BASE_URL:coap://${monitoring.domain}}'
+          # Queue to use for target device
+          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}'
         # To add more targets, use following environment variables:
@@ -79,6 +83,8 @@ monitoring:
       targets:
           # HTTP transport base url, http://DOMAIN by default
         - base_url: '${HTTP_TRANSPORT_BASE_URL:http://${monitoring.domain}}'
+          # Queue to use for target device
+          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}'
         # To add more targets, use following environment variables:
@@ -92,6 +98,8 @@ monitoring:
       targets:
           # LwM2M transport base url, coap://DOMAIN:5685 by default
         - base_url: '${LWM2M_TRANSPORT_BASE_URL:coap://${monitoring.domain}:5685}'
+          # Queue to use for target device
+          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}'
         # To add more targets, use following environment variables: