Analytics: adding smoothing plugin

- Allows to perform system-wide smoothing using only one accumulator per sensor
- Basic functionality works, more thorough testing to follow
- Metadata updating functionality still to be implemented

analytics/Makefile

@@ -45,6 +45,9 @@ operators/%.o: CXXFLAGS+= $(PLUGINFLAGS)
 libdcdboperator_aggregator.$(LIBEXT): operators/aggregator/AggregatorOperator.o operators/aggregator/AggregatorConfigurator.o ../common/src/sensornavigator.o
 	$(CXX) $(LIBFLAGS)$@ -o $@ $^ -L$(DCDBDEPLOYPATH)/lib/ -lboost_log -lboost_system -lboost_regex
 	
+libdcdboperator_smoothing.$(LIBEXT): operators/smoothing/SmoothingOperator.o operators/smoothing/SmoothingConfigurator.o ../common/src/sensornavigator.o
+	$(CXX) $(LIBFLAGS)$@ -o $@ $^ -L$(DCDBDEPLOYPATH)/lib/ -lboost_log -lboost_system -lboost_regex
+	
 libdcdboperator_regressor.$(LIBEXT): operators/regressor/RegressorOperator.o operators/regressor/RegressorConfigurator.o ../common/src/sensornavigator.o
 	$(CXX) $(LIBFLAGS)$@ -o $@ $^ -L$(DCDBDEPLOYPATH)/lib/ -lboost_log -lboost_system -lboost_regex -lopencv_core -lopencv_ml
 	

analytics/README.md

@@ -786,7 +786,7 @@ The configuration parameters specific to the _Aggregator_ plugin are the followi
 |:----- |:----------- |
 | window | Length in milliseconds of the time window that is used to retrieve recent readings for the input sensors, starting from the latest one.
 
-Additionally, output sensors in operators of the Aggregator plugin accept the following parameters:
+Additionally, output sensors in operators of the _Aggregator_ plugin accept the following parameters:
 
 | Value | Explanation |
 |:----- |:----------- |
@@ -803,6 +803,26 @@ to the corresponding [section](#joboperators) for more details.
 
 > NOTE       The Job Aggregator plugin does not support the _relative_ option supported by the Aggregator plugin, and always uses the _absolute_ sensor query mode.
 
+## Smoothing Plugin <a name="smoothingPlugin"></a>
+The _Smoothing_ plugin performs similar functions as the _Aggregator_ plugin, but is optimized for performing running averages. It uses a single accumulator to compute approximate running averages, improving memory usage and reducing the amount of queried data.
+
+Units ins the _Smoothing_ plugin are also instantiated differently compared to other plugins. As it is meant to perform running average on most or all sensors present in a system, there is no need to specify the input sensors of an operator: the plugin will automatically fetch all instantiated sensors in the system, and create separate units for them, each with their separate average output sensors. 
+For this reason, pattern expressions specified on output sensors (e.g., _<bottomup 1, filter cpu>avg60_) are ignored and only the MQTT suffixes of the output sensors are used to construct units. If required, users can still specify input sensors using the Unit System syntax, to select subsets of the available sensors in the system. Operators of the _Smoothing_ plugin accept the following configuration parameters: 
+
+| Value | Explanation |
+|:----- |:----------- |
+| separator | Character used to separate the MQTT prefix of output sensors (which is the input sensor's name) from the suffix (which is the average's identifier). Default is "#". 
+| updateMetadata | Boolean. If _true_, the operator will automatically update the metadata entries of all sensors for which averages are computed, to expose the new operations. Enabled by default.
+| exclude | String containing a regular expression, defining which sensors must be excluded from the smoothing process.
+
+Additionally, output sensors in operators of the _Smoothing_ plugin accept the following parameters:
+
+| Value | Explanation |
+|:----- |:----------- |
+| range | Range in milliseconds of the average to be stored in this output sensor.
+
+> NOTE &ensp;&ensp;&ensp;&ensp;&ensp; The _Smoothing_ plugin will automatically update the metadata table of the input sensors in the units to expose the computed averages. This behavior can be altered by setting the _updateMetadata_ operator parameter to _false_. The sensors of the averages themselves are not published by default.  
+
 ## Regressor Plugin <a name="regressorPlugin"></a>
 
 The _Regressor_ plugin is able to perform regression of sensors, by using _random forest_ machine learning predictors. The algorithmic backend is provided by the OpenCV library.

analytics/config/smoothing.conf

+smoother smt1 {
+	interval   		10000
+	minValues  		10
+	duplicate  		false
+	separator  	   	"#"
+	updateMetadata 	true
+	exclude			"/cpu[0-9]*/"
+
+	output {
+
+		sensor avg300 {
+    		mqttsuffix  /avg300
+    		range		300000
+    	}
+    	
+		sensor avg3600 {
+			mqttsuffix  /avg3600
+			range		3600000
+		}
+	}
+}

analytics/includes/JobOperatorConfiguratorTemplate.h

@@ -94,7 +94,10 @@ protected:
     virtual bool readUnits(Operator& op, std::vector<shared_ptr<SBase>>& protoInputs, std::vector<shared_ptr<SBase>>& protoOutputs,
             std::vector<shared_ptr<SBase>>& protoGlobalOutputs, inputMode_t inputMode) {
         // Forcing the job operator to not be duplicated
-        op.setDuplicate(false);
+        if(op.getDuplicate()) {
+            LOG(warning) << this->_operatorName << " " << op.getName() << ": The units of this operator cannot be duplicated.";
+            op.setDuplicate(false);
+        }
         shared_ptr<UnitTemplate<SBase>> jobUnit;
         try {
             jobUnit = this->_unitGen.generateHierarchicalUnit(SensorNavigator::rootKey, std::list<std::string>(), protoGlobalOutputs,

analytics/includes/UnitGenerator.h

@@ -82,6 +82,87 @@ public:
     * @param navi       Shared pointer to a SensorNavigator object
     */
     void setNavigator(shared_ptr<SensorNavigator> navi) { _navi = navi; }
+
+    /**
+    * @brief            Parses a string encoding a tree level
+    *
+    *                   This method serves to parse strings that are used to express hierarchy levels in config files
+    *                   of the data analytics framework. These strings are in the format "<topdown X>.*" or
+    *                   "<bottomup X>.*", and signify "sensors that are in nodes X levels down from level 0 in the
+    *                   sensor tree", or "sensors that are in nodes X levels up from the deepest level in the sensor
+    *                   tree" respectively. As such, the method returns the depth level of sensors represented by the
+    *                   input string.
+    *
+    * @param s          String to be parsed
+    * @return           Absolute depth level in the sensor tree that is encoded in the string
+    */
+    int  parseNodeLevelString(const string& s) {
+        if(!_navi || !_navi->treeExists())
+            throw runtime_error("UnitGenerator: SensorNavigator tree not initialized!");
+
+        int _treeDepth = _navi->getTreeDepth();
+        bool topDown = false;
+        if(boost::regex_search(s.c_str(), _match, _blockRx)) {
+            string blockMatch = _match.str(0);
+            if(boost::regex_search(blockMatch.c_str(), _match, _topRx))
+                topDown = true;
+            else if(boost::regex_search(blockMatch.c_str(), _match, _bottomRx))
+                topDown = false;
+            else
+                throw runtime_error("UnitGenerator: Syntax error in configuration!");
+            blockMatch = _match.str(0);
+            int lv;
+            if(topDown)
+                lv = !boost::regex_search(blockMatch.c_str(), _match, _numRx) ? 0 : (int)stoi(_match.str(0));
+            else
+                lv = !boost::regex_search(blockMatch.c_str(), _match, _numRx) ? _treeDepth : _treeDepth - (int)stoi(_match.str(0));
+
+            if( lv < 0 ) lv = 0;
+            else if( lv > _treeDepth ) lv = _treeDepth;
+
+            return lv;
+        }
+        else
+            return -1;
+    }
+
+    /**
+    * @brief            Resolves a string encoding a tree level starting from a given node
+    *
+    *                   This method takes as input strings in the format specified for parseNodeLevelString(). It then
+    *                   takes as input also the name of a node in the sensor tree. The method will then return the set
+    *                   of sensors expressed by "s", that belong to nodes encoded in its hierarchy level and that are
+    *                   related to "node", either as ancestors or descendants. If a filter was included in the unit
+    *                   clause, e.g. <bottomup 1, filter cpu>freq, then only sensors in nodes matching the filter
+    *                   regular expression will be returned.
+    *
+    * @param s          String to be parsed
+    * @param node       Name of the target node
+    * @param replace    If False only the names of the resolved units will be returned, without the sensor name
+    * @return           Set of sensors encoded in "s" that are associated with "node"
+    */
+    set<string> *resolveNodeLevelString(const string& s, const string& node, const bool replace=true) {
+        int level = parseNodeLevelString(s);
+        if(!_navi->nodeExists(node))
+            throw invalid_argument("UnitGenerator: Node " + node + " does not exist!");
+
+        set<string> *sensors = new set<string>();
+        if( level <= -1 )
+            sensors->insert(replace ? s : SensorNavigator::rootKey);
+        else {
+            //Ensuring that only the "filter" clause matches is enough, since we already checked for the entire
+            // < > configuration block in parseNodeLevelString
+            boost::regex filter = boost::regex(boost::trim_copy(!boost::regex_search(s.c_str(), _match, _filterRx) ? ".*" : _match.str(0)));
+            set<string> *nodes = _navi->navigate(node, level - _navi->getNodeDepth(node));
+            // Filtering is performed here, as node names always include all upper levels of the hierarchy
+            for(const auto& n : *nodes) {
+                if (boost::regex_search(n.c_str(), _match, filter))
+                    sensors->insert(replace ? boost::regex_replace(s, _blockRx, n) : n);
+            }
+            delete nodes;
+        }
+        return sensors;
+    }
     
     /**
     * @brief                   Computes and instantiates a single unit
@@ -424,87 +505,6 @@ public:
     }
     
 protected:
-
-    /**
-    * @brief            Parses a string encoding a tree level
-    *
-    *                   This method serves to parse strings that are used to express hierarchy levels in config files
-    *                   of the data analytics framework. These strings are in the format "<topdown X>.*" or
-    *                   "<bottomup X>.*", and signify "sensors that are in nodes X levels down from level 0 in the
-    *                   sensor tree", or "sensors that are in nodes X levels up from the deepest level in the sensor
-    *                   tree" respectively. As such, the method returns the depth level of sensors represented by the
-    *                   input string.
-    *
-    * @param s          String to be parsed
-    * @return           Absolute depth level in the sensor tree that is encoded in the string
-    */
-    int  parseNodeLevelString(const string& s) {
-        if(!_navi || !_navi->treeExists())
-            throw runtime_error("UnitGenerator: SensorNavigator tree not initialized!");
-
-        int _treeDepth = _navi->getTreeDepth();
-        bool topDown = false;
-        if(boost::regex_search(s.c_str(), _match, _blockRx)) {
-            string blockMatch = _match.str(0);
-            if(boost::regex_search(blockMatch.c_str(), _match, _topRx))
-                topDown = true;
-            else if(boost::regex_search(blockMatch.c_str(), _match, _bottomRx))
-                topDown = false;
-            else
-                throw runtime_error("UnitGenerator: Syntax error in configuration!");
-            blockMatch = _match.str(0);
-            int lv;
-            if(topDown)
-                lv = !boost::regex_search(blockMatch.c_str(), _match, _numRx) ? 0 : (int)stoi(_match.str(0));
-            else
-                lv = !boost::regex_search(blockMatch.c_str(), _match, _numRx) ? _treeDepth : _treeDepth - (int)stoi(_match.str(0));
-
-            if( lv < 0 ) lv = 0;
-            else if( lv > _treeDepth ) lv = _treeDepth;
-
-            return lv;
-        }
-        else
-            return -1;
-    }
-
-    /**
-    * @brief            Resolves a string encoding a tree level starting from a given node
-    *
-    *                   This method takes as input strings in the format specified for parseNodeLevelString(). It then
-    *                   takes as input also the name of a node in the sensor tree. The method will then return the set
-    *                   of sensors expressed by "s", that belong to nodes encoded in its hierarchy level and that are
-    *                   related to "node", either as ancestors or descendants. If a filter was included in the unit
-    *                   clause, e.g. <bottomup 1, filter cpu>freq, then only sensors in nodes matching the filter
-    *                   regular expression will be returned.
-    *
-    * @param s          String to be parsed
-    * @param node       Name of the target node
-    * @param replace    If False only the names of the resolved units will be returned, without the sensor name
-    * @return           Set of sensors encoded in "s" that are associated with "node"
-    */
-    set<string> *resolveNodeLevelString(const string& s, const string& node, const bool replace=true) {
-        int level = parseNodeLevelString(s);
-        if(!_navi->nodeExists(node))
-            throw invalid_argument("UnitGenerator: Node " + node + " does not exist!");
-
-        set<string> *sensors = new set<string>();
-        if( level <= -1 )
-            sensors->insert(replace ? s : SensorNavigator::rootKey);
-        else {
-            //Ensuring that only the "filter" clause matches is enough, since we already checked for the entire
-            // < > configuration block in parseNodeLevelString
-            boost::regex filter = boost::regex(boost::trim_copy(!boost::regex_search(s.c_str(), _match, _filterRx) ? ".*" : _match.str(0)));
-            set<string> *nodes = _navi->navigate(node, level - _navi->getNodeDepth(node));
-            // Filtering is performed here, as node names always include all upper levels of the hierarchy
-            for(const auto& n : *nodes) {
-                if (boost::regex_search(n.c_str(), _match, filter))
-                    sensors->insert(replace ? boost::regex_replace(s, _blockRx, n) : n);
-            }
-            delete nodes;
-        }
-        return sensors;
-    }
     
     /**
     *                       This private method will resolve all nodes in the current sensor tree that satisfy 

analytics/operators/smoothing/SmoothingConfigurator.cpp

+//================================================================================
+// Name        : SmoothingConfigurator.cpp
+// Author      : Alessio Netti
+// Contact     : info@dcdb.it
+// Copyright   : Leibniz Supercomputing Centre
+// Description :
+//================================================================================
+
+//================================================================================
+// This file is part of DCDB (DataCenter DataBase)
+// Copyright (C) 2019-2019 Leibniz Supercomputing Centre
+//
+// This program is free software; you can redistribute it and/or
+// modify it under the terms of the GNU General Public License
+// as published by the Free Software Foundation; either version 2
+// of the License, or (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the Free Software
+// Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+//================================================================================
+
+#include "SmoothingConfigurator.h"
+
+SmoothingConfigurator::SmoothingConfigurator() : OperatorConfiguratorTemplate() {
+    _operatorName = "smoother";
+    _baseName     = "sensor";
+}
+
+SmoothingConfigurator::~SmoothingConfigurator() {}
+
+void SmoothingConfigurator::operatorAttributes(SmoothingOperator& op, CFG_VAL config) {
+    BOOST_FOREACH(boost::property_tree::iptree::value_type &val, config)
+    {
+        if (boost::iequals(val.first, "separator")) {
+            op.setSeparator(val.second.data());
+        } else if (boost::iequals(val.first, "updateMetadata")) {
+            op.setUpdateMetadata(to_bool(val.second.data()));
+        } else if (boost::iequals(val.first, "exclude")) {
+            op.setExclude(val.second.data());
+        }
+    }
+}
+
+void SmoothingConfigurator::sensorBase(SmoothingSensorBase& s, CFG_VAL config) {
+    BOOST_FOREACH(boost::property_tree::iptree::value_type &val, config)
+    {
+        if (boost::iequals(val.first, "range")) {
+            s.setRange(std::stoull(val.second.data()));
+        }
+    }
+}
+
+bool SmoothingConfigurator::unit(UnitTemplate<SmoothingSensorBase>& u) {
+    if(u.isTopUnit()) {
+        LOG(error) << "    " << _operatorName << ": This operator type only supports flat units!";
+        return false;
+    }
+    if(u.getInputs().size() != 1) {
+        LOG(error) << "    " << _operatorName << ": Exactly one input sensor per unit must be defined!";
+        return false;
+    }
+    if(u.getOutputs().empty()) {
+        LOG(error) << "    " << _operatorName << ": At least one output sensor per unit must be defined!";
+        return false;
+    }
+    return true;
+}
+
+bool SmoothingConfigurator::readUnits(SmoothingOperator& op, std::vector<SmoothingSBPtr>& protoInputs, std::vector<SmoothingSBPtr>& protoOutputs,
+               std::vector<SmoothingSBPtr>& protoGlobalOutputs, inputMode_t inputMode) {
+    vector<shared_ptr<UnitTemplate<SmoothingSensorBase>>> *units = NULL;
+    vector<SmoothingSBPtr> realInputs, realOutputs;
+
+    // Printing a warning if declaring global outputs or selecting on-demand mode
+    if(!protoGlobalOutputs.empty()) {
+        LOG(warning) << _operatorName << " " << op.getName() << ": Global outputs will be ignored.";
+        protoGlobalOutputs.clear();
+    }
+    if(!op.getStreaming()) {
+        LOG(warning) << _operatorName << " " << op.getName() << ": This operator does not support on-demand mode.";
+        op.setStreaming(true);
+    }
+
+    // If no inputs are specified, we pick all sensors present in the sensor tree
+    if(protoInputs.empty() && inputMode==SELECTIVE) {
+        set<string> *inputNames = _queryEngine.getNavigator()->getSensors(SensorNavigator::rootKey, true);
+        for(const auto& n : *inputNames) {
+            SmoothingSensorBase ssb(n);
+            ssb.setMqtt(n);
+            protoInputs.push_back(std::make_shared<SmoothingSensorBase>(ssb));
+        }
+        delete inputNames;
+    // Resolving all input sensors beforehand if some were defined
+    } else if(!protoInputs.empty()) {
+        vector<SmoothingSBPtr> tempInputs;
+        for(const auto & sIn : protoInputs) {
+            set<string> *inputNames = _unitGen.resolveNodeLevelString(sIn->getName(), SensorNavigator::rootKey);
+            for (const auto &n : *inputNames) {
+                SmoothingSensorBase ssb(n);
+                ssb.setMqtt(n);
+                tempInputs.push_back(std::make_shared<SmoothingSensorBase>(ssb));
+            }
+            delete inputNames;
+        }
+        // Replacing the original inputs
+        protoInputs.clear();
+        protoInputs.insert(protoInputs.end(), tempInputs.begin(), tempInputs.end());
+        tempInputs.clear();
+    }
+
+    boost::cmatch match;
+    boost::regex excludeReg(op.getExclude());
+    // Generating one separate unit for each input sensor
+    for(const auto& pIn : protoInputs) {
+        if (op.getExclude()=="" || !boost::regex_search(pIn->getMqtt().c_str(), match, excludeReg)) {
+            realInputs.push_back(pIn);
+            for (const auto &sOut : protoOutputs) {
+                // Removing patterns from the output prototype sensors
+                SmoothingSensorBase ssb(*sOut);
+                ssb.setMqtt(MQTTChecker::formatTopic(pIn->getMqtt()) + _stripTopic(ssb.getMqtt(), op.getSeparator()));
+                ssb.setName(ssb.getMqtt());
+                realOutputs.push_back(std::make_shared<SmoothingSensorBase>(ssb));
+            }
+            try {
+                units = _unitGen.generateAutoUnit(SensorNavigator::rootKey, std::list<std::string>(), protoGlobalOutputs,
+                        realInputs, realOutputs, inputMode, "", !op.getStreaming(), op.getEnforceTopics(), op.getRelaxed());
+            }
+            catch (const std::exception &e) {
+                LOG(error) << _operatorName << " " << op.getName() << ": Error when creating units: " << e.what();
+                delete units;
+                return false;
+            }
+
+            if(units->size() > 1) {
+                LOG(error) << _operatorName << " " << op.getName() << ": Unexpected number of units created.";
+                delete units;
+                return false;
+            }
+            
+            for (auto &u: *units) {
+                u->setName(pIn->getMqtt());
+                if (!constructSensorTopics(*u, op)) {
+                    op.clearUnits();
+                    delete units;
+                    return false;
+                }
+                if (!unit(*u)) {
+                    LOG(error) << "    Unit " << u->getName() << " did not pass the final check!";
+                    op.clearUnits();
+                    delete units;
+                    return false;
+                } else {
+                    LOG(debug) << "    Unit " << u->getName() << " generated.";
+                    op.addUnit(u);
+                }
+            }
+            realInputs.clear();
+            realOutputs.clear();
+            delete units;
+        }
+    }
+    return true;
+}
+
+std::string SmoothingConfigurator::_stripTopic(const std::string& topic, const std::string& separator) {
+    if(topic.empty()) return topic;
+    std::string newTopic = topic;
+    // Stripping off all forward slashes
+    while(!newTopic.empty() && newTopic.front()==MQTT_SEP) newTopic.erase(0, 1);
+    while(!newTopic.empty() && newTopic.back()==MQTT_SEP)  newTopic.erase(newTopic.size()-1, 1);
+    // Adding a front separator
+    newTopic = separator + newTopic;
+    return newTopic;
+}

analytics/operators/smoothing/SmoothingConfigurator.h

+//================================================================================
+// Name        : SmoothingConfigurator.h
+// Author      : Alessio Netti
+// Contact     : info@dcdb.it
+// Copyright   : Leibniz Supercomputing Centre
+// Description :
+//================================================================================
+
+//================================================================================
+// This file is part of DCDB (DataCenter DataBase)
+// Copyright (C) 2019-2019 Leibniz Supercomputing Centre
+//
+// This program is free software; you can redistribute it and/or
+// modify it under the terms of the GNU General Public License
+// as published by the Free Software Foundation; either version 2
+// of the License, or (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the Free Software
+// Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+//================================================================================
+
+#ifndef PROJECT_SMOOTHINGCONFIGURATOR_H
+#define PROJECT_SMOOTHINGCONFIGURATOR_H
+
+#include "../../includes/OperatorConfiguratorTemplate.h"
+#include "SmoothingOperator.h"
+
+/**
+ * @brief Configurator for the smoothing plugin.
+ *
+ * @ingroup smoothing
+ */
+class SmoothingConfigurator : virtual public OperatorConfiguratorTemplate<SmoothingOperator, SmoothingSensorBase> {
+
+public:
+    SmoothingConfigurator();
+    ~SmoothingConfigurator();
+
+private:
+    void sensorBase(SmoothingSensorBase& s, CFG_VAL config) override;
+    void operatorAttributes(SmoothingOperator& op, CFG_VAL config) override;
+    bool unit(UnitTemplate<SmoothingSensorBase>& u) override;
+
+    bool readUnits(SmoothingOperator& op, std::vector<SmoothingSBPtr>& protoInputs, std::vector<SmoothingSBPtr>& protoOutputs, 
+            std::vector<SmoothingSBPtr>& protoGlobalOutputs, inputMode_t inputMode) override;
+
+    std::string _stripTopic(const std::string& topic, const std::string& separator);
+
+};
+
+extern "C" OperatorConfiguratorInterface* create() {
+    return new SmoothingConfigurator;
+}
+
+extern "C" void destroy(OperatorConfiguratorInterface* c) {
+    delete c;
+}
+
+#endif //PROJECT_SMOOTHINGCONFIGURATOR_H

analytics/operators/smoothing/SmoothingOperator.cpp

+//================================================================================
+// Name        : SmoothingOperator.cpp
+// Author      : Alessio Netti
+// Contact     : info@dcdb.it
+// Copyright   : Leibniz Supercomputing Centre
+// Description :
+//================================================================================
+
+//================================================================================
+// This file is part of DCDB (DataCenter DataBase)
+// Copyright (C) 2019-2019 Leibniz Supercomputing Centre
+//
+// This program is free software; you can redistribute it and/or
+// modify it under the terms of the GNU General Public License
+// as published by the Free Software Foundation; either version 2
+// of the License, or (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the Free Software
+// Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+//================================================================================
+
+#include "SmoothingOperator.h"
+
+SmoothingOperator::SmoothingOperator(const std::string& name) : OperatorTemplate(name) {
+    _separator = "#";
+    _exclude = "";
+    _updateMetadata = true;
+}
+
+SmoothingOperator::SmoothingOperator(const SmoothingOperator &other) : OperatorTemplate(other) {
+    _separator = other._separator;
+    _exclude = "";
+    _updateMetadata = other._updateMetadata;
+}
+
+SmoothingOperator::~SmoothingOperator() {}
+
+void SmoothingOperator::printConfig(LOG_LEVEL ll) {
+    OperatorTemplate<SmoothingSensorBase>::printConfig(ll);
+}
+
+void SmoothingOperator::compute(U_Ptr unit) {
+    // Clearing the buffer
+    _buffer.clear();
+    SmoothingSBPtr sIn=unit->getInputs()[0], sOut=unit->getOutputs()[0];
+    uint64_t startTs = sOut->getTimestamp() ? getTimestamp()-sOut->getTimestamp() : 0;
+    
+    if(!_queryEngine.querySensor(sIn->getName(), startTs, 0, _buffer, true))
+        throw std::runtime_error("Operator " + _name + ": cannot read from sensor " + sIn->getName() + "!");
+    
+    for(const auto& v : _buffer) {
+        if (v.timestamp > sOut->getTimestamp()) {
+            for(const auto& s : unit->getOutputs())
+                s->smoothAndStore(v);
+        } else {
+            continue;
+        }
+    }
+}
+
+bool SmoothingOperator::execOnStart() {
+    // Setting the reference initial timestamp
+    for(const auto& u : _units)
+        for(const auto& s : u->getOutputs()) {
+            s->setTimestamp(0);
+            s->setValue(0.0);
+        }
+        
+    return true;
+}

analytics/operators/smoothing/SmoothingOperator.h

+//================================================================================
+// Name        : SmoothingOperator.h
+// Author      : Alessio Netti
+// Contact     : info@dcdb.it
+// Copyright   : Leibniz Supercomputing Centre
+// Description :
+//================================================================================