Properly handle malformed MQTT messages in CollectAgent.

CollectAgent/collectagent.cpp

@@ -37,7 +37,6 @@ void sigHandler(int sig)
 
 void mqttCallback(SimpleMQTTMessage *msg)
 {
-
   /*
    * Increment the msgCtr/vmsgCtr for statistics.
    */
@@ -49,29 +48,35 @@ void mqttCallback(SimpleMQTTMessage *msg)
    * Decode the message and put into the database.
    */
   if (msg->isPublish()) {
-	  
-	  uint64_t val;
-	  uint64_t ts;
-	  
-	  //In the 64 bit message case, the collect agent provides a timestamp
-	  if(msg->getPayloadLength()==sizeof(uint64_t)) {
-		  
-		  //printf("Providing timestamp...\n");
-		  val = *((uint64_t*)msg->getPayload());
-		  ts = Messaging::calculateTimestamp();
-		  //printf("Val = %" PRIu64 ", timestamp = %" PRIu64 "\n", val, ts);
-	   }
-	   
-	   //...otherwise it just retrieves it from the MQTT message payload. 
-	   else if((msg->getPayloadLength()%sizeof(mqttPayload)==0) && (msg->getPayloadLength()>0)){
-		   
-		   //printf("Retrieving timestamp...\n");
-		   mqttPayload *payload = (mqttPayload*)msg->getPayload();
-		   
-		   val = payload->value; // payload[n].value
-		   ts = payload->timestamp;
-		   //printf("Val = %" PRIu64 ", timestamp = %" PRIu64 "\n", val, ts);
-	   }
+
+      uint64_t val;
+      uint64_t ts;
+
+      //In the 64 bit message case, the collect agent provides a timestamp
+      if(msg->getPayloadLength()==sizeof(uint64_t)) {
+
+          //printf("Providing timestamp...\n");
+          val = *((uint64_t*)msg->getPayload());
+          ts = Messaging::calculateTimestamp();
+          //printf("Val = %" PRIu64 ", timestamp = %" PRIu64 "\n", val, ts);
+      }
+
+      //...otherwise it just retrieves it from the MQTT message payload.
+      else if((msg->getPayloadLength()%sizeof(mqttPayload)==0) && (msg->getPayloadLength()>0)){
+
+          //printf("Retrieving timestamp...\n");
+          mqttPayload *payload = (mqttPayload*)msg->getPayload();
+
+          val = payload->value; // payload[n].value
+          ts = payload->timestamp;
+          //printf("Val = %" PRIu64 ", timestamp = %" PRIu64 "\n", val, ts);
+      }
+
+      //...otherwise this message is malformed -> ignore...
+      else {
+          delete msg;
+          return;
+      }
 
       /*
        * Check if we can decode the message topic
@@ -82,22 +87,22 @@ void mqttCallback(SimpleMQTTMessage *msg)
       if (mySensorDataStore->topicToSid(&sid,msg->getTopic())) {
 #if 0
           cout << "Topic decode successful:"
-               << "\nRaw: " << hex << setw(16) << setfill('0') << sid.raw[0] << " " << hex << setw(16) << setfill('0') << sid.raw[1]
-               << "\ndatacenter_id: " << hex << ((sid.dl & 0xFF00000000000000) >> 56)
-               << "\ncluster_id:    " << hex << ((sid.dl & 0x00FF000000000000) >> 48)
-               << "\nrack_id:       " << hex << ((sid.dl & 0x0000FFFF00000000) >> 32)
-               << "\nchassis_id:    " << hex << ((sid.dl & 0x00000000FF000000) >> 24)
-               << "\nbic_id:        " << hex << ((sid.dl & 0x0000000000FF0000) >> 16)
-               << "\nbmc_id:        " << hex << ((sid.dl & 0x000000000000FF00) >> 8)
-               << "\nknc_id:        " << hex << ((sid.dl & 0x00000000000000FF))
-               << "\ndevice_id:     " << hex << sid.dsid.device_id
-               << "\nreserved:      " << hex << sid.dsid.rsvd
-               << "\nsensor_number: " << hex << sid.dsid.sensor_number
-               << "\n";
+              << "\nRaw: " << hex << setw(16) << setfill('0') << sid.raw[0] << " " << hex << setw(16) << setfill('0') << sid.raw[1]
+              << "\ndatacenter_id: " << hex << ((sid.dl & 0xFF00000000000000) >> 56)
+              << "\ncluster_id:    " << hex << ((sid.dl & 0x00FF000000000000) >> 48)
+              << "\nrack_id:       " << hex << ((sid.dl & 0x0000FFFF00000000) >> 32)
+              << "\nchassis_id:    " << hex << ((sid.dl & 0x00000000FF000000) >> 24)
+              << "\nbic_id:        " << hex << ((sid.dl & 0x0000000000FF0000) >> 16)
+              << "\nbmc_id:        " << hex << ((sid.dl & 0x000000000000FF00) >> 8)
+              << "\nknc_id:        " << hex << ((sid.dl & 0x00000000000000FF))
+              << "\ndevice_id:     " << hex << sid.dsid.device_id
+              << "\nreserved:      " << hex << sid.dsid.rsvd
+              << "\nsensor_number: " << hex << sid.dsid.sensor_number
+              << "\n";
 #endif
           mySensorDataStore->insert(&sid, ts, val);
       }
-#if 0
+#if 1
       else {
           cout << "Wrong topic format: " << msg->getTopic() << "\n";
       }