From 6264a8c490eefec872a8a327a504be0de65e7324 Mon Sep 17 00:00:00 2001
From: Mike Hansen <mike.hansen@connectus.ai>
Date: Mon, 19 Jul 2021 11:01:13 -0400
Subject: [PATCH] [WIFI-3165] Keep last stats received timestamp in gateway
 client session map

Signed-off-by: Mike Hansen <mike.hansen@connectus.ai>
---
 .../OpensyncExternalIntegrationCloud.java     | 10 +--
 .../integration/utils/MqttStatsPublisher.java | 10 ++-
 .../external/integration/OvsdbSession.java    | 33 +++++++-
 .../src/main/docker/app/opensync/logback.xml  |  4 -
 .../src/main/resources/logback.xml            |  1 -
 .../opensync/mqtt/OpensyncMqttClient.java     | 81 +++++++++----------
 6 files changed, 80 insertions(+), 59 deletions(-)

diff --git a/opensync-ext-cloud/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/OpensyncExternalIntegrationCloud.java b/opensync-ext-cloud/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/OpensyncExternalIntegrationCloud.java
index b02ece2..cea45e3 100644
--- a/opensync-ext-cloud/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/OpensyncExternalIntegrationCloud.java
+++ b/opensync-ext-cloud/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/OpensyncExternalIntegrationCloud.java
@@ -210,10 +210,7 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
                 }
 
                 long locationId = autoProvisionedLocationId;
-                if ((customer.getDetails() != null) && (customer.getDetails().getAutoProvisioning() != null)
-                        && customer.getDetails().getAutoProvisioning().isEnabled()) {
-                    locationId = customer.getDetails().getAutoProvisioning().getLocationId();
-                }
+                locationId = customer.getDetails().getAutoProvisioning().getLocationId();
 
                 try {
                     Location location = locationServiceInterface.get(locationId);
@@ -221,7 +218,6 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
                 } catch (Exception e) {
                     LOG.error("Cannot auto-provision equipment because customer location with id {} cannot be found", locationId);
                     throw new IllegalStateException("Cannot auto-provision equipment because customer location cannot be found : " + locationId);
-
                 }
 
                 ce.setSerial(connectNodeInfo.serialNumber);
@@ -250,7 +246,6 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
                     }
                     if (!radioType.equals(RadioType.UNSUPPORTED)) {
                         advancedRadioConfiguration = RadioConfiguration.createWithDefaults(radioType);
-
                         advancedRadioMap.put(radioType, advancedRadioConfiguration);
                         radioConfiguration = ElementRadioConfiguration.createWithDefaults(radioType);
                         radioMap.put(radioType, radioConfiguration);
@@ -386,7 +381,7 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
                 }
 
             }
-
+            
             EquipmentRoutingRecord equipmentRoutingRecord = gatewayController.registerCustomerEquipment(ce.getName(), ce.getCustomerId(), ce.getId());
 
             // Status and client cleanup, when AP reconnects or has been
@@ -497,7 +492,6 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
     private void updateApStatus(Equipment ce, ConnectNodeInfo connectNodeInfo) {
 
         try {
-
             Status statusRecord = statusServiceInterface.getOrNull(ce.getCustomerId(), ce.getId(), StatusDataType.EQUIPMENT_ADMIN);
             if (statusRecord == null) {
                 statusRecord = new Status();
diff --git a/opensync-ext-cloud/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/utils/MqttStatsPublisher.java b/opensync-ext-cloud/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/utils/MqttStatsPublisher.java
index 6e41cd2..1bb0090 100644
--- a/opensync-ext-cloud/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/utils/MqttStatsPublisher.java
+++ b/opensync-ext-cloud/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/utils/MqttStatsPublisher.java
@@ -51,6 +51,7 @@ import com.telecominfraproject.wlan.equipment.EquipmentServiceInterface;
 import com.telecominfraproject.wlan.equipment.models.Equipment;
 import com.telecominfraproject.wlan.opensync.external.integration.OvsdbSession;
 import com.telecominfraproject.wlan.opensync.external.integration.OvsdbSessionMapInterface;
+import com.telecominfraproject.wlan.opensync.external.integration.controller.OpensyncCloudGatewayController;
 import com.telecominfraproject.wlan.opensync.util.OvsdbToWlanCloudTypeMappingUtility;
 import com.telecominfraproject.wlan.profile.ProfileServiceInterface;
 import com.telecominfraproject.wlan.profile.models.Profile;
@@ -138,7 +139,9 @@ public class MqttStatsPublisher implements StatsPublisherInterface {
     private RealtimeEventPublisher realtimeEventPublisher;
     @Autowired
     private AlarmServiceInterface alarmServiceInterface;
-
+    @Autowired
+    private OpensyncCloudGatewayController gatewayController;
+    
     @Value("${tip.wlan.mqttStatsPublisher.temperatureThresholdInC:80}")
     private int temperatureThresholdInC;
 
@@ -170,7 +173,9 @@ public class MqttStatsPublisher implements StatsPublisherInterface {
         long equipmentId = ce.getId();
         long locationId = ce.getLocationId();
         long profileId = ce.getProfileId();
-
+        
+        // update timestamp for active customer equipment
+        gatewayController.updateActiveCustomer(customerId);
         List<ServiceMetric> metricRecordList = new ArrayList<>();
 
         try {
@@ -260,6 +265,7 @@ public class MqttStatsPublisher implements StatsPublisherInterface {
     @Override
     public void publishSystemEventFromTableStateMonitor(SystemEvent event) {
         LOG.info("Publishing SystemEvent received by TableStateMonitor {}", event);
+        gatewayController.updateActiveCustomer(event.getCustomerId());
         cloudEventDispatcherInterface.publishEvent(event);
     }
 
diff --git a/opensync-ext-interface/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/OvsdbSession.java b/opensync-ext-interface/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/OvsdbSession.java
index cd0d516..a263ec7 100644
--- a/opensync-ext-interface/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/OvsdbSession.java
+++ b/opensync-ext-interface/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/OvsdbSession.java
@@ -1,5 +1,7 @@
 package com.telecominfraproject.wlan.opensync.external.integration;
 
+import java.util.Objects;
+
 import com.vmware.ovsdb.service.OvsdbClient;
 
 public class OvsdbSession {
@@ -7,7 +9,7 @@ public class OvsdbSession {
     private String apId;
     private long routingId;
     private long equipmentId;
-    private int customerId;
+    private long mostRecentStatsTimestamp;
     
     public OvsdbClient getOvsdbClient() {
         return ovsdbClient;
@@ -33,5 +35,34 @@ public class OvsdbSession {
     public void setEquipmentId(long equipmentId) {
         this.equipmentId = equipmentId;
     }
+    public long getMostRecentStatsTimestamp() {
+        return mostRecentStatsTimestamp;
+    }
+    public void setMostRecentStatsTimestamp(long mostRecentStatsTimestamp) {
+        this.mostRecentStatsTimestamp = mostRecentStatsTimestamp;
+    }
+    @Override
+    public int hashCode() {
+        return Objects.hash(apId, equipmentId, mostRecentStatsTimestamp, ovsdbClient, routingId);
+    }
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj)
+            return true;
+        if (obj == null)
+            return false;
+        if (getClass() != obj.getClass())
+            return false;
+        OvsdbSession other = (OvsdbSession) obj;
+        return Objects.equals(apId, other.apId) && equipmentId == other.equipmentId && mostRecentStatsTimestamp == other.mostRecentStatsTimestamp
+                && Objects.equals(ovsdbClient, other.ovsdbClient) && routingId == other.routingId;
+    }
+    
+    @Override
+    public String toString() {
+        return "OvsdbSession [ovsdbClient=" + ovsdbClient + ", apId=" + apId + ", routingId=" + routingId + ", equipmentId=" + equipmentId
+                + ", mostRecentStatsTimestamp=" + mostRecentStatsTimestamp + "]";
+    }
+
     
 }
diff --git a/opensync-gateway-cloud-docker/src/main/docker/app/opensync/logback.xml b/opensync-gateway-cloud-docker/src/main/docker/app/opensync/logback.xml
index 1d6cd3c..a2d333a 100644
--- a/opensync-gateway-cloud-docker/src/main/docker/app/opensync/logback.xml
+++ b/opensync-gateway-cloud-docker/src/main/docker/app/opensync/logback.xml
@@ -87,10 +87,6 @@
     <appender-ref ref="mqttDataFile"/>
   </logger>
 
-  <logger name="MQTT_CLIENT_TRACER" level="INFO" additivity="false">
-    <appender-ref ref="mqttMsgTracer"/>
-  </logger>
-  
   <root level="WARN">
     <appender-ref ref="logfile"/>
   </root>
diff --git a/opensync-gateway-cloud-process/src/main/resources/logback.xml b/opensync-gateway-cloud-process/src/main/resources/logback.xml
index 89c241e..30c956f 100644
--- a/opensync-gateway-cloud-process/src/main/resources/logback.xml
+++ b/opensync-gateway-cloud-process/src/main/resources/logback.xml
@@ -64,7 +64,6 @@
   <logger name="com.vmware.ovsdb.netty.OvsdbConnectionHandler" level="ERROR"/>
   
   <logger name="MQTT_DATA" level="INFO"/>
-  <logger name="MQTT_CLIENT_TRACER" level="DEBUG"/>
 
   <!--   
   <logger name="org.springframework.security.web.authentication.preauth" level="DEBUG"/>
diff --git a/opensync-gateway/src/main/java/com/telecominfraproject/wlan/opensync/mqtt/OpensyncMqttClient.java b/opensync-gateway/src/main/java/com/telecominfraproject/wlan/opensync/mqtt/OpensyncMqttClient.java
index fb274f7..2c13460 100644
--- a/opensync-gateway/src/main/java/com/telecominfraproject/wlan/opensync/mqtt/OpensyncMqttClient.java
+++ b/opensync-gateway/src/main/java/com/telecominfraproject/wlan/opensync/mqtt/OpensyncMqttClient.java
@@ -36,6 +36,8 @@ import com.netflix.servo.monitor.Stopwatch;
 import com.netflix.servo.monitor.Timer;
 import com.netflix.servo.tag.TagList;
 import com.telecominfraproject.wlan.cloudmetrics.CloudMetricsTags;
+import com.telecominfraproject.wlan.opensync.external.integration.OvsdbSession;
+import com.telecominfraproject.wlan.opensync.external.integration.OvsdbSessionMapInterface;
 import com.telecominfraproject.wlan.opensync.external.integration.utils.StatsPublisherInterface;
 import com.telecominfraproject.wlan.opensync.util.ZlibUtil;
 
@@ -46,19 +48,10 @@ import sts.OpensyncStats.Report;
 @Component
 public class OpensyncMqttClient implements ApplicationListener<ContextClosedEvent> {
 
-    // private static final String METRICS_WKR_PFX = "metrics-wkr-";
-
-    // public static final int REPORT_PROCESSING_THRESHOLD_SEC = 30;
-
     private static final Logger LOG = LoggerFactory.getLogger(OpensyncMqttClient.class);
 
     private static final Logger MQTT_LOG = LoggerFactory.getLogger("MQTT_DATA");
 
-    private static final Logger MQTT_TRACER_LOG = LoggerFactory.getLogger("MQTT_CLIENT_TRACER");
-
-    // private static final ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newCachedThreadPool(new
-    // CustomizableThreadFactory(METRICS_WKR_PFX));
-
     public static Charset utf8 = Charset.forName("UTF-8");
 
     private final TagList tags = CloudMetricsTags.commonTags;
@@ -71,6 +64,9 @@ public class OpensyncMqttClient implements ApplicationListener<ContextClosedEven
 
     @Autowired
     private StatsPublisherInterface statsPublisher;
+    
+    @Autowired
+    private OvsdbSessionMapInterface ovsdbSessionMapInterface;
 
     // dtop: use anonymous constructor to ensure that the following code always
     // get executed,
@@ -153,37 +149,6 @@ public class OpensyncMqttClient implements ApplicationListener<ContextClosedEven
                         mqtt.setClientId("opensync_mqtt");
                         mqtt.setUserName(username);
                         mqtt.setPassword(password);
-                        mqtt.setTracer(new Tracer() {
-                            @Override
-                            public void onReceive(MQTTFrame frame) {
-                                switch (frame.messageType()) {
-                                    case PINGRESP.TYPE:
-                                        // PINGs don't want to fill log
-                                        MQTT_TRACER_LOG.trace("OpensyncMqttClient Rx: {}", frame);
-                                        break;
-                                    default:
-                                        MQTT_TRACER_LOG.info("OpensyncMqttClient Rx: {}", frame);
-                                }
-                            }
-
-                            @Override
-                            public void onSend(MQTTFrame frame) {
-                                switch (frame.messageType()) {
-                                    case PINGREQ.TYPE:
-                                        // PINGs don't want to fill log
-                                        MQTT_TRACER_LOG.trace("OpensyncMqttClient Tx: {}", frame);
-                                        break;                                
-                                    default:
-                                        MQTT_TRACER_LOG.info("OpensyncMqttClient Tx: {}", frame);
-                                }
-                            }
-
-                            @Override
-                            public void debug(String message, Object... args) {
-                                MQTT_TRACER_LOG.info(String.format(message, args));
-                            }
-                        });
-
                         blockingConnection = mqtt.blockingConnection();
                         blockingConnection.connect();
 
@@ -228,14 +193,22 @@ public class OpensyncMqttClient implements ApplicationListener<ContextClosedEven
                                 // Only supported protobuf on the TIP opensync APs is Report
                                 Report statsReport = Report.parseFrom(payload);
                                 mqttMsg.ack();
-
+                                String apId = extractApIdFromTopic(mqttMsg.getTopic());
+                                if (apId != null) {
+                                    OvsdbSession ovsdbSession = ovsdbSessionMapInterface.getSession(extractApIdFromTopic(mqttMsg.getTopic()));
+                                    if (ovsdbSession != null) {
+                                        ovsdbSession.setMostRecentStatsTimestamp(System.currentTimeMillis());
+                                        LOG.debug("Last metrics received from AP updated to {}",ovsdbSession.toString());
+                                    } else {
+                                        LOG.debug("No ovsdb session exists for this AP {}",apId);
+                                    }
+                                }
                                 MQTT_LOG.info("Topic {}\n{}", mqttMsg.getTopic(), jsonPrinter.print(statsReport));
                                 statsPublisher.processMqttMessage(mqttMsg.getTopic(), statsReport);                           
                                 LOG.debug("Dispatched report for topic {} to backend for processing", mqttMsg.getTopic());
 
                             } catch (Exception e) {
-                                String msgStr = new String(mqttMsg.getPayload(), utf8);
-                                LOG.warn("Could not process message topic = {}\nmessage = {}", mqttMsg.getTopic(), msgStr);
+                                LOG.error("Exception processing topic for message {}",mqttMsg, e);
                             } finally {
                                 stopwatchTimerMessageProcess.stop();
                             }
@@ -271,4 +244,26 @@ public class OpensyncMqttClient implements ApplicationListener<ContextClosedEven
             mqttClientThread.interrupt();
         }
     }
+    
+    /**
+     * @param topic
+     * @return apId extracted from the topic name, or null if it cannot be
+     *         extracted
+     */
+    static String extractApIdFromTopic(String topic) {
+        // Topic is formatted as
+        // "/ap/"+clientCn+"_"+ret.serialNumber+"/opensync"
+        if (topic == null) {
+            return null;
+        }
+
+        String[] parts = topic.split("/");
+        if (parts.length < 3) {
+            return null;
+        }
+
+        // apId is the third element in the topic
+        return parts[2];
+    }
+    
 }