Fixed assignment of file sink paths to sensors

- The MQTT topic of a sensor is now used to identify the file name of the
sink and is appended to the path given as input

analytics/includes/JobAnalyzerConfiguratorTemplate.h

@@ -100,6 +100,12 @@ protected:
             delete units;
             return false;
         }
+        if(*units.size() > 1) {
+            LOG(error) << this->_analyzerName << " " << an.getName() << ": Invalid job template unit, please check your configuration!";
+            delete units;
+            return false;
+        }
+        
         shared_ptr<UnitTemplate<SBase>> jobUnit = *units[0];
         delete units;
         an.clearUnits();

analytics/includes/UnitGenerator.h

@@ -329,11 +329,10 @@ public:
         for(const auto& out : outputs) {
             SBase uOut(*out);
             uOut.setName(resolveJobString(uOut.getName(), j));
-                uOut.setMqtt(_navi->buildTopicForNode(j, uOut.getMqtt()));
-                // Duplicating the file sink adding the name of each unit to the path
-                string sPath = uOut.getSinkPath();
-                if(sPath != "")
-                    uOut.setSinkPath(adjustSinkPath(sPath, j));
+            uOut.setMqtt(_navi->buildTopicForNode(j, uOut.getMqtt()));
+            // Duplicating the file sink adding the name of each unit to the path
+            if(uOut.getSinkPath()!="")
+                uOut.setSinkPath(MQTTChecker::topicToFile(uOut.getMqtt(), uOut.getSinkPath()));
             jobUnit->addOutput(uOut);
         }
         return jobUnit;
@@ -363,17 +362,6 @@ public:
 
 protected:
     
-    // Adjust the path of a sensor sink such that it will include the unit name as well
-    string adjustSinkPath(const string& sPath, const string& node) {
-        string newSPath = sPath;
-        size_t idx = newSPath.find_last_of("/\\");
-        if(idx != string::npos)
-            newSPath = newSPath.substr(0, idx+1) + MQTTChecker::topicToName(node) + "_" + newSPath.substr(idx+1, string::npos);
-        else
-            newSPath = MQTTChecker::topicToName(node) + "_" + newSPath;
-        return newSPath;
-    }
-
     // This private method will resolve all nodes in the current sensor tree that satisfy the "unit" pattern,
     // and then verify whether "node" belongs to this set or not.
     bool nodeBelongsToPattern(const string& node, const string& unit) {
@@ -458,16 +446,14 @@ protected:
             delete sensors;
             // If we are instantiating output sensors by unit, we generate mqtt topics by using the prefix
             // associated to the respective node in the sensor tree, and the sensor suffix itself
-            if(u != SensorNavigator::rootKey) {
+            if(u != SensorNavigator::rootKey)
                 uOut.setMqtt(_navi->buildTopicForNode(u, uOut.getMqtt()));
-                // Duplicating the file sink adding the name of each unit to the path
-                string sPath = uOut.getSinkPath();
-                if(sPath != "")
-                    uOut.setSinkPath(adjustSinkPath(sPath, u));
-            }
             // If we are not using units (only unit is root, out of the hierarchy) we build sensors like in samplers
             else
                 uOut.setMqtt(MQTTChecker::formatTopic(mqttPrefix) + MQTTChecker::formatTopic(uOut.getMqtt()));
+            // Duplicating the file sink adding the name of each unit to the path
+            if(uOut.getSinkPath()!="")
+                uOut.setSinkPath(MQTTChecker::topicToFile(uOut.getMqtt(), uOut.getSinkPath()));
             unitOutputs.push_back(make_shared<SBase>(uOut));
         }
         shared_ptr<UnitTemplate<SBase>> unPtr = make_shared<UnitTemplate<SBase>>(u, unitInputs, unitOutputs);

common/include/mqttchecker.h

@@ -33,6 +33,7 @@
 
 #define MQTT_SEP '/'
 #define NAME_SEP '.'
+#define JOB_STR "job"
 
 /**
  * Class that manages constraint for MQTT topic formatting
@@ -74,7 +75,28 @@ public:
      * @return          The numerical job ID
      */
     static uint32_t topicToJob(const std::string& topic) {
-        return 0;
+        std::string jobKey(JOB_STR), jobId = topic;
+        jobId.erase(std::remove(jobId.begin(), jobId.end(), MQTT_SEP), jobId.end());
+        size_t pos = jobId.find(jobKey);
+        if (pos != std::string::npos) 
+            jobId.erase(pos, jobKey.length());
+        return std::stoull(jobId);
+    }
+    
+    /**
+     * @brief           Converts an MQTT topic to a filesystem path representation
+     * 
+     * @param topic     The MQTT topic to be processed 
+     * @param path      The base destination path
+     * @return          A processed file system path
+     */
+    static std::string topicToFile(const std::string& topic, const std::string& path="") {
+        std::string tPath = path;
+        if(tPath.empty())
+            tPath = "./";
+        else if(tPath[tPath.size()-1]!='/')
+            tPath += '/';
+        return tPath + topicToName(topic);
     }
 
     /**

common/include/sensorbase.h

@@ -68,6 +68,7 @@ public:
 	SensorBase(const SensorBase& other) :
 		_name(other._name),
 		_mqtt(other._mqtt),
+		_sinkPath(other._sinkPath),
 		_skipConstVal(other._skipConstVal),
 		_cacheInterval(other._cacheInterval),
 		_subsamplingFactor(other._subsamplingFactor),
@@ -88,6 +89,7 @@ public:
 	SensorBase& operator=(const SensorBase& other) {
 		_name = other._name;
 		_mqtt = other._mqtt;
+		_sinkPath = other._sinkPath;
 		_skipConstVal = other._skipConstVal;
 		_cacheInterval = other._cacheInterval;
 		_subsamplingFactor = other._subsamplingFactor;

dcdbpusher/README.md

@@ -303,7 +303,7 @@ All the different plugins share some same general principles in common regarding
     3. Sensors hold only those attributes which are necessary to uniquely identify the target sensor. Common base attributes:
         * __mqttsuffix__  (to make its [mqtt-topic](#mqttTopic) unique)
         * __delta__ (identifies a monotonic sensor. If set to "on", differences between successive readings are collected)
-        * __sink__ (a path to a file to which sensor readings should be written, disabled by default)
+        * __sink__ (a path to a directory to which sensor readings should be written, disabled by default)
         * __subSampling__ (subsampling factor S. If > 1, only one reading every S is sent over MQTT, and the others are kept locally)
 5. Be aware that naming of sensor/group/entity is not fixed. A plugin developer can name them as he likes, e.g. counter/multicounter/host.
 6. It is possible to define template groups or entities in the config file, but not template sensors (as a sensor should only consists of attributs which make him unique this would not be too useful). To specify a template group/entity simply prefix its definition with `template_` (see the example below). You can reference them later by using the `default` attribute. A template entity can consist of groups and these in turn can consist of sensors. When using a template, all of its attribute values are copied to the actual sensor. Copied attributes can be overwritten in the actual entity/sensor (some of them even should be overwritten, e.g. the mqttPart). However, groups/sensors associated with a template are copied to the actual entity/group and can NOT be overwritten. One can specify further groups/sensors which are then added to those copied from the template. Template entitys/groups itself or sensors within them are never used in live operation of the plugin. They are purely cosmetic for convenient configuration.

dcdbpusher/includes/ConfiguratorTemplate.h

@@ -814,6 +814,8 @@ protected:
 			for(auto& s: g->getSensors()) {
 				s->setMqtt(MQTTChecker::formatTopic(_mqttPrefix) + MQTTChecker::formatTopic(g->getMqttPart()) + MQTTChecker::formatTopic(s->getMqtt()));
 				s->setName(s->getMqtt());
+				if(s->getSinkPath()!="")
+					s->setSinkPath(MQTTChecker::topicToFile(s->getMqtt(), s->getSinkPath()));
 			}
 		return true;
 	}