[WIFI-3166] Session ID is a -ve number in opensyncgw logs and on UI logs

Signed-off-by: Mike Hansen <mike.hansen@connectus.ai>

opensync-ext-cloud/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/controller/OpensyncCloudGatewayController.java

@@ -209,9 +209,9 @@ public class OpensyncCloudGatewayController {
                     case CellSizeAttributesRequest:
                         ret.add(sendCellSizeRequest(session, (CEGWCellSizeAttributesRequest) command));
                         break;
-//                    case MostRecentStatsTimestamp:
-//                        ret.add(sendGetMostRecentStatsTimestampRequest(command, inventoryId));
-//                        break;
+                    case MostRecentStatsTimestamp:
+                        ret.add(sendGetMostRecentStatsTimestampRequest(command, inventoryId));
+                        break;
                     default:
                         LOG.warn("[{}] Failed to deliver command {}, unsupported command type", inventoryId, command);
                         ret.add(new EquipmentCommandResponse(CEGWCommandResultCode.UnsupportedCommand,
@@ -250,7 +250,32 @@ public class OpensyncCloudGatewayController {
         return new GatewayDefaults();
     }
     
-
+    private EquipmentCommandResponse sendGetMostRecentStatsTimestampRequest(CEGWBaseCommand command, String inventoryId) {
+        Long ts = lastReceivedStatsTimestamp(inventoryId);
+        if (ts == null) {
+            return new EquipmentCommandResponse(CEGWCommandResultCode.NoRouteToCE,
+                    null, command,
+                    registeredGateway == null ? null : registeredGateway.getHostname(),
+                            registeredGateway == null ? -1 : registeredGateway.getPort());
+        } else {
+            return new EquipmentCommandResponse(CEGWCommandResultCode.Success,
+                    ts.toString(), command,
+                    registeredGateway == null ? null : registeredGateway.getHostname(),
+                            registeredGateway == null ? -1 : registeredGateway.getPort());
+        }
+    }
+    
+    @RequestMapping(value = "/lastReceivedStatsTimestamp", method = RequestMethod.GET)
+    public Long lastReceivedStatsTimestamp(@RequestParam String apId) {
+        Long ret = null;
+        if (ovsdbSessionMapInterface.getSession(apId) != null) {
+            ret = ovsdbSessionMapInterface.getSession(apId).getMostRecentStatsTimestamp();
+            LOG.debug("lastReceivedStatsTimestamp for apId {} {}",apId,ret);
+        } else {
+            LOG.warn("lastReceivedStatsTimestamp found no session for {}, cannot get timestamp",apId);
+        }
+        return ret;
+    }
 
     /**
      * Verify a route to customer equipment

opensync-ext-cloud/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/utils/MqttStatsPublisher.java

@@ -5,6 +5,7 @@ import java.net.Inet4Address;
 import java.net.Inet6Address;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.text.DecimalFormat;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Date;
@@ -51,6 +52,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,6 +140,8 @@ 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;
@@ -157,7 +161,6 @@ public class MqttStatsPublisher implements StatsPublisherInterface {
     @Override
     @Async
     public void processMqttMessage(String topic, Report report) {
-
         long startTime = System.nanoTime();
         String apId = extractApIdFromTopic(topic);
         LOG.info("Received report on topic {} for ap {}", topic, report.getNodeID());
@@ -173,6 +176,7 @@ public class MqttStatsPublisher implements StatsPublisherInterface {
         long profileId = ce.getProfileId();
         
         // update timestamp for active customer equipment
+        gatewayController.updateActiveCustomer(customerId);
         List<ServiceMetric> metricRecordList = new ArrayList<>();
 
         try {
@@ -208,11 +212,8 @@ public class MqttStatsPublisher implements StatsPublisherInterface {
 
             if (!metricRecordList.isEmpty()) {
                 long serviceMetricTimestamp = System.currentTimeMillis();
-                metricRecordList.stream().forEach(smr -> {  
-                	// TODO use serviceMetricTimestamp rather than 0. This is done for now since there are some
-                	// channel metrics that have overlapping keys which messes up Cassandra if the same time stamp is used
-                	// and setting it to 0 allows the CloudEventDispatcherController to assign unique time stamps. 
-                    smr.setCreatedTimestamp(0);
+                metricRecordList.stream().forEach(smr -> {              
+                    smr.setCreatedTimestamp(serviceMetricTimestamp);
                     if (smr.getLocationId() == 0)
                         smr.setLocationId(locationId);
                     if (smr.getCustomerId() == 0)
@@ -265,6 +266,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);
     }
 

opensync-ext-cloud/src/test/java/com/telecominfraproject/wlan/opensync/external/integration/OpensyncExternalIntegrationCloudTest.java

@@ -176,7 +176,7 @@ public class OpensyncExternalIntegrationCloudTest {
         location.setDetails(details);
         location.setName("Location-UT");
         location.setLocationType(LocationType.BUILDING);
-        Mockito.when(locationServiceInterface.getOrNull(Mockito.anyLong())).thenReturn(location);
+        Mockito.when(locationServiceInterface.get(8L)).thenReturn(location);
         Customer customer = new Customer();
         customer.setId(2);
         CustomerDetails customerDetails = new CustomerDetails();
@@ -232,7 +232,7 @@ public class OpensyncExternalIntegrationCloudTest {
         opensyncExternalIntegrationCloud.apConnected("Test_Client_21P10C68818122", createConnectNodeInfo());
 
         Mockito.verify(firmwareServiceInterface).getDefaultCustomerTrackSetting();
-        Mockito.verify(locationServiceInterface).getOrNull(ArgumentMatchers.anyLong());
+        Mockito.verify(locationServiceInterface).get(8L);
 
     }
 
@@ -246,7 +246,7 @@ public class OpensyncExternalIntegrationCloudTest {
         location.setDetails(details);
         location.setName("Location-UT");
         location.setLocationType(LocationType.BUILDING);
-        Mockito.when(locationServiceInterface.getOrNull(8L)).thenReturn(location);
+        Mockito.when(locationServiceInterface.get(8L)).thenReturn(location);
 
         Customer customer = new Customer();
         customer.setId(2);
@@ -326,7 +326,7 @@ public class OpensyncExternalIntegrationCloudTest {
         Mockito.verify(customerServiceInterface).getOrNull(ArgumentMatchers.anyInt());
         Mockito.verify(equipmentServiceInterface).getByInventoryIdOrNull("Test_Client_21P10C68818122");
         Mockito.verify(firmwareServiceInterface).getDefaultCustomerTrackSetting();
-        Mockito.verify(locationServiceInterface, Mockito.times(2)).getOrNull(ArgumentMatchers.anyLong());
+        Mockito.verify(locationServiceInterface, Mockito.times(2)).get(ArgumentMatchers.anyLong());
 
     }
 

opensync-ext-interface/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/OpensyncExternalIntegrationInterface.java

@@ -54,6 +54,4 @@ public interface OpensyncExternalIntegrationInterface {
     void nodeStateDbTableUpdate(List<Map<String, String>> nodeStateAttributes, String apId);
 
     void clearEquipmentStatus(String apId);
-    
-    void processMqttMessage(String topic, Report report);
 }

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,6 +9,7 @@ public class OvsdbSession {
     private String apId;
     private long routingId;
     private long equipmentId;
+    private long mostRecentStatsTimestamp;
     
     public OvsdbClient getOvsdbClient() {
         return ovsdbClient;
@@ -32,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 + "]";
+    }
+
+    
 }

opensync-gateway/src/main/java/com/telecominfraproject/wlan/opensync/mqtt/OpensyncMqttClient.java

@@ -11,6 +11,10 @@ import org.fusesource.mqtt.client.MQTT;
 import org.fusesource.mqtt.client.Message;
 import org.fusesource.mqtt.client.QoS;
 import org.fusesource.mqtt.client.Topic;
+import org.fusesource.mqtt.client.Tracer;
+import org.fusesource.mqtt.codec.MQTTFrame;
+import org.fusesource.mqtt.codec.PINGREQ;
+import org.fusesource.mqtt.codec.PINGRESP;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -32,7 +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.OpensyncExternalIntegrationInterface;
+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;
 
@@ -58,7 +63,10 @@ public class OpensyncMqttClient implements ApplicationListener<ContextClosedEven
     private final Timer timerMessageProcess = new BasicTimer(MonitorConfig.builder("osgw-mqtt-messageProcessTimer").withTags(tags).build());
 
     @Autowired
-    private OpensyncExternalIntegrationInterface opensyncExternalIntegrationInterface;
+    private StatsPublisherInterface statsPublisher;
+    
+    @Autowired
+    private OvsdbSessionMapInterface ovsdbSessionMapInterface;
 
     // dtop: use anonymous constructor to ensure that the following code always
     // get executed,
@@ -185,8 +193,18 @@ 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));
-                                opensyncExternalIntegrationInterface.processMqttMessage(mqttMsg.getTopic(), statsReport);                           
+                                statsPublisher.processMqttMessage(mqttMsg.getTopic(), statsReport);                           
                                 LOG.debug("Dispatched report for topic {} to backend for processing", mqttMsg.getTopic());
 
                             } catch (Exception e) {
@@ -226,5 +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];
+    }
+    
 }