From 8a12c63400b0433ca2f30ea68afd1f14f9a12f30 Mon Sep 17 00:00:00 2001
From: ViacheslavKlimov <viacheslavklimov11@gmail.com>
Date: Mon, 9 Jan 2023 13:08:15 +0200
Subject: [PATCH] Add more logging; fix race condition in WS client

---
 monitoring/src/main/conf/logback.xml          |  8 ++--
 .../monitoring/client/WsClient.java           | 48 +++++++++++++++----
 .../monitoring/data/MonitoredServiceKey.java  |  1 -
 .../service/TransportMonitoringService.java   |  5 ++
 .../impl/CoapTransportMonitoringService.java  |  4 ++
 .../impl/HttpTransportMonitoringService.java  |  3 ++
 .../impl/MqttTransportMonitoringService.java  |  4 ++
 7 files changed, 58 insertions(+), 15 deletions(-)

diff --git a/monitoring/src/main/conf/logback.xml b/monitoring/src/main/conf/logback.xml
index ea90a898cd..5d37a17110 100644
--- a/monitoring/src/main/conf/logback.xml
+++ b/monitoring/src/main/conf/logback.xml
@@ -17,7 +17,7 @@
 
 -->
 <!DOCTYPE configuration>
-<configuration>
+<configuration scan="true" scanPeriod="30 seconds">
 
     <appender name="fileLogAppender"
               class="ch.qos.logback.core.rolling.RollingFileAppender">
@@ -41,10 +41,8 @@
 
     <logger name="org" level="WARN"/>
     <logger name="org.thingsboard.server" level="INFO"/>
-    <logger name="org.thingsboard.monitoring" level="TRACE"/>
-    <logger name="org.thingsboard.monitoring.client" level="WARN"/>
-    <logger name="org.openqa.selenium" level="WARN"/>
-    <logger name="io.github.bonigarcia" level="WARN"/>
+    <logger name="org.thingsboard.monitoring" level="DEBUG"/>
+    <logger name="org.thingsboard.monitoring.client" level="INFO"/>
 
     <root level="INFO">
         <appender-ref ref="fileLogAppender"/>
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 1c33da0338..56d29de5fa 100644
--- a/monitoring/src/main/java/org/thingsboard/monitoring/client/WsClient.java
+++ b/monitoring/src/main/java/org/thingsboard/monitoring/client/WsClient.java
@@ -33,14 +33,18 @@ import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 @Slf4j
 public class WsClient extends WebSocketClient {
 
-    private volatile String lastMsg;
+    public volatile String lastMsg;
     private CountDownLatch reply;
     private CountDownLatch update;
 
+    private final Lock updateLock = new ReentrantLock();
+
     public WsClient(URI serverUri) {
         super(serverUri);
     }
@@ -52,12 +56,21 @@ public class WsClient extends WebSocketClient {
 
     @Override
     public void onMessage(String s) {
-        lastMsg = s;
-        if (reply != null) {
-            reply.countDown();
+        log.trace("Received new msg: {}", s);
+        if (s == null) {
+            return;
         }
-        if (update != null) {
-            update.countDown();
+        updateLock.lock();
+        try {
+            lastMsg = s;
+            if (update != null) {
+                update.countDown();
+            }
+            if (reply != null) {
+                reply.countDown();
+            }
+        } finally {
+            updateLock.unlock();
         }
     }
 
@@ -72,13 +85,24 @@ public class WsClient extends WebSocketClient {
     }
 
     public void registerWaitForUpdate() {
-        lastMsg = null;
-        update = new CountDownLatch(1);
+        updateLock.lock();
+        try {
+            lastMsg = null;
+            update = new CountDownLatch(1);
+        } finally {
+            updateLock.unlock();
+        }
+        log.trace("Registered wait for update");
     }
 
     @Override
     public void send(String text) throws NotYetConnectedException {
-        reply = new CountDownLatch(1);
+        updateLock.lock();
+        try {
+            reply = new CountDownLatch(1);
+        } finally {
+            updateLock.unlock();
+        }
         super.send(text);
     }
 
@@ -93,27 +117,33 @@ public class WsClient extends WebSocketClient {
         CmdsWrapper wrapper = new CmdsWrapper();
         wrapper.setTsSubCmds(List.of(subCmd));
         send(JacksonUtil.toString(wrapper));
+        log.trace("Subscribed for telemetry (key: {})", telemetryKey);
     }
 
     public JsonNode waitForUpdate(long ms) {
+        log.trace("update latch count: {}", update.getCount());
         try {
             if (update.await(ms, TimeUnit.MILLISECONDS)) {
+                log.trace("Waited for update");
                 return getLastMsg();
             }
         } catch (InterruptedException e) {
             log.debug("Failed to await reply", e);
         }
+        log.trace("No update arrived within {} ms", ms);
         return null;
     }
 
     public JsonNode waitForReply(int ms) {
         try {
             if (reply.await(ms, TimeUnit.MILLISECONDS)) {
+                log.trace("Waited for reply");
                 return getLastMsg();
             }
         } catch (InterruptedException e) {
             log.debug("Failed to await reply", e);
         }
+        log.trace("No reply arrived within {} ms", ms);
         return null;
     }
 
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 f2227978ae..57609d609a 100644
--- a/monitoring/src/main/java/org/thingsboard/monitoring/data/MonitoredServiceKey.java
+++ b/monitoring/src/main/java/org/thingsboard/monitoring/data/MonitoredServiceKey.java
@@ -18,6 +18,5 @@ package org.thingsboard.monitoring.data;
 public class MonitoredServiceKey {
 
     public static final String GENERAL = "Monitoring";
-    public static final String WEB_UI = "Web UI";
 
 }
diff --git a/monitoring/src/main/java/org/thingsboard/monitoring/service/TransportMonitoringService.java b/monitoring/src/main/java/org/thingsboard/monitoring/service/TransportMonitoringService.java
index e27002449e..6cddf05469 100644
--- a/monitoring/src/main/java/org/thingsboard/monitoring/service/TransportMonitoringService.java
+++ b/monitoring/src/main/java/org/thingsboard/monitoring/service/TransportMonitoringService.java
@@ -81,6 +81,7 @@ public abstract class TransportMonitoringService<C extends TransportMonitoringSe
         monitoringExecutor.scheduleWithFixedDelay(() -> {
             WsClient wsClient = null;
             try {
+                log.trace("[{}] Checking", transportInfo);
                 wsClient = establishWsClient();
                 wsClient.registerWaitForUpdate();
 
@@ -88,13 +89,16 @@ public abstract class TransportMonitoringService<C extends TransportMonitoringSe
                 String testPayload = JacksonUtil.newObjectNode().set(TEST_TELEMETRY_KEY, new TextNode(testValue)).toString();
                 try {
                     initClientAndSendPayload(testPayload);
+                    log.trace("[{}] Sent test payload ({})", transportInfo, testPayload);
                 } catch (Throwable e) {
                     throw new TransportFailureException(e);
                 }
 
+                log.trace("[{}] Waiting for WS update", transportInfo);
                 checkWsUpdate(wsClient, testValue);
 
                 monitoringReporter.serviceIsOk(transportInfo);
+                monitoringReporter.serviceIsOk(MonitoredServiceKey.GENERAL);
             } catch (TransportFailureException transportFailureException) {
                 monitoringReporter.serviceFailure(transportInfo, transportFailureException);
             } catch (Exception e) {
@@ -141,6 +145,7 @@ public abstract class TransportMonitoringService<C extends TransportMonitoringSe
     private void checkWsUpdate(WsClient wsClient, String testValue) {
         stopWatch.start();
         wsClient.waitForUpdate(wsConfig.getResultCheckTimeoutMs());
+        log.trace("[{}] Waited for WS update. Last WS msg: {}", transportInfo, wsClient.lastMsg);
         Object update = wsClient.getTelemetryKeyUpdate(TEST_TELEMETRY_KEY);
         if (update == null) {
             throw new TransportFailureException("No WS update arrived within " + wsConfig.getResultCheckTimeoutMs() + " ms");
diff --git a/monitoring/src/main/java/org/thingsboard/monitoring/service/impl/CoapTransportMonitoringService.java b/monitoring/src/main/java/org/thingsboard/monitoring/service/impl/CoapTransportMonitoringService.java
index 30315fbd50..43558a6d49 100644
--- a/monitoring/src/main/java/org/thingsboard/monitoring/service/impl/CoapTransportMonitoringService.java
+++ b/monitoring/src/main/java/org/thingsboard/monitoring/service/impl/CoapTransportMonitoringService.java
@@ -15,6 +15,7 @@
  */
 package org.thingsboard.monitoring.service.impl;
 
+import lombok.extern.slf4j.Slf4j;
 import org.eclipse.californium.core.CoapClient;
 import org.eclipse.californium.core.CoapResponse;
 import org.eclipse.californium.core.coap.CoAP;
@@ -31,6 +32,7 @@ import java.io.IOException;
 
 @Component
 @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
+@Slf4j
 public class CoapTransportMonitoringService extends TransportMonitoringService<CoapTransportMonitoringServiceConfig> {
 
     private CoapClient coapClient;
@@ -46,6 +48,7 @@ public class CoapTransportMonitoringService extends TransportMonitoringService<C
             String uri = target.getBaseUrl() + "/api/v1/" + accessToken + "/telemetry";
             coapClient = new CoapClient(uri);
             coapClient.setTimeout((long) config.getRequestTimeoutMs());
+            log.debug("Initialized CoAP client for URI {}", uri);
         }
     }
 
@@ -63,6 +66,7 @@ public class CoapTransportMonitoringService extends TransportMonitoringService<C
         if (coapClient != null) {
             coapClient.shutdown();
             coapClient = null;
+            log.info("Disconnected CoAP client");
         }
     }
 
diff --git a/monitoring/src/main/java/org/thingsboard/monitoring/service/impl/HttpTransportMonitoringService.java b/monitoring/src/main/java/org/thingsboard/monitoring/service/impl/HttpTransportMonitoringService.java
index 99bd5936fd..1062d675d6 100644
--- a/monitoring/src/main/java/org/thingsboard/monitoring/service/impl/HttpTransportMonitoringService.java
+++ b/monitoring/src/main/java/org/thingsboard/monitoring/service/impl/HttpTransportMonitoringService.java
@@ -15,6 +15,7 @@
  */
 package org.thingsboard.monitoring.service.impl;
 
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.boot.web.client.RestTemplateBuilder;
 import org.springframework.context.annotation.Scope;
@@ -29,6 +30,7 @@ import java.time.Duration;
 
 @Component
 @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
+@Slf4j
 public class HttpTransportMonitoringService extends TransportMonitoringService<HttpTransportMonitoringServiceConfig> {
 
     private RestTemplate restTemplate;
@@ -44,6 +46,7 @@ public class HttpTransportMonitoringService extends TransportMonitoringService<H
                     .setConnectTimeout(Duration.ofMillis(config.getRequestTimeoutMs()))
                     .setReadTimeout(Duration.ofMillis(config.getRequestTimeoutMs()))
                     .build();
+            log.debug("Initialized HTTP client");
         }
     }
 
diff --git a/monitoring/src/main/java/org/thingsboard/monitoring/service/impl/MqttTransportMonitoringService.java b/monitoring/src/main/java/org/thingsboard/monitoring/service/impl/MqttTransportMonitoringService.java
index 5e6542d29e..5cab9f146f 100644
--- a/monitoring/src/main/java/org/thingsboard/monitoring/service/impl/MqttTransportMonitoringService.java
+++ b/monitoring/src/main/java/org/thingsboard/monitoring/service/impl/MqttTransportMonitoringService.java
@@ -15,6 +15,7 @@
  */
 package org.thingsboard.monitoring.service.impl;
 
+import lombok.extern.slf4j.Slf4j;
 import org.eclipse.paho.client.mqttv3.IMqttToken;
 import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
 import org.eclipse.paho.client.mqttv3.MqttClient;
@@ -31,6 +32,7 @@ import org.thingsboard.monitoring.service.TransportMonitoringService;
 
 @Component
 @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
+@Slf4j
 public class MqttTransportMonitoringService extends TransportMonitoringService<MqttTransportMonitoringServiceConfig> {
 
     private MqttClient mqttClient;
@@ -56,6 +58,7 @@ public class MqttTransportMonitoringService extends TransportMonitoringService<M
             if (result.getException() != null) {
                 throw result.getException();
             }
+            log.debug("Initialized MQTT client for URI {}", mqttClient.getServerURI());
         }
     }
 
@@ -72,6 +75,7 @@ public class MqttTransportMonitoringService extends TransportMonitoringService<M
         if (mqttClient != null) {
             mqttClient.disconnect();
             mqttClient = null;
+            log.info("Disconnected MQTT client");
         }
     }