Analytics: added relative mode switch for Aggregator plugin

analytics/README.md

@@ -21,9 +21,10 @@
 	1. [Aggregator Plugin](#averagePlugin)
 	2. [Job Aggregator Plugin](#jobaveragePlugin)
 	3. [Regressor Plugin](#regressorPlugin)
-	4. [File Sink Plugin](#filesinkPlugin)
-	5. [Tester Plugin](#testerPlugin)
-	6. [Writing Plugins](#writingPlugins)
+	4. [Tester Plugin](#testerPlugin)
+4. [Sink Plugins](#sinkplugins)
+	1. [File Sink Plugin](#filesinkPlugin)
+	2. [Writing Plugins](#writingPlugins)
 
 # Introduction <a name="introduction"></a>
 In this Readme we describe the DCDB Data Analytics framework, and all data abstractions that are associated with it. 
@@ -630,6 +631,8 @@ Additionally, output sensors in operators of the Aggregator plugin accept the fo
 |:----- |:----------- |
 | operation | Operation to be performed over the input sensors. Can be "sum", "average", "maximum", "minimum", "std", "percentiles" or "observations".
 | percentile |  Specific percentile to be computed when using the "percentiles" operation. Can be an integer in the (0,100) range.
+| relative | If true, the _relative_ query mode will be used. Otherwise the _absolute_ mode is used.
+
 
 ## Job Aggregator Plugin <a name="jobaveragePlugin"></a>
 
@@ -637,6 +640,8 @@ The _Job Aggregator_ plugin offers the same functionality as the _Aggregator_ pl
 it performs aggregation of the specified input sensors across all nodes on which each job is running. Please refer
 to the corresponding [section](#joboperators) for more details.
 
+> NOTE &ensp;&ensp;&ensp;&ensp;&ensp; The Job Aggregator plugin does not support the _relative_ option supported by the Aggregator plugin, and always uses the _absolute_ sensor query mode.
+
 ## 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.
@@ -671,6 +676,18 @@ Finally, the Regressor plugin supports the following additional REST API action:
 | train | Triggers a new training phase for the random forest model. Feature vectors are temporarily collected in-memory until _trainingSamples_ vectors are obtained. Until this moment, the old random forest model is still used to perform prediction.
 | importances | Returns the sorted importance values for the input features, together with the respective labels, if available.
 
+## Tester Plugin <a name="testerPlugin"></a>
+The _Tester_ plugin can be used to test the functionality and performance of the query engine, as well as of the unit system. It will perform a specified number of queries over the set of input sensors for each unit, and then output as a sensor the total number of retrieved readings. The following are the configuration parameters for operators in the _Tester_ plugin:
+
+| Value | Explanation |
+|:----- |:----------- |
+| window | Length in milliseconds of the time window that is used to retrieve recent readings for the input sensors, starting from the latest one.
+| queries | Number of queries to be performed at each computation interval. If more than the number of input sensors per unit, these will be looped over multiple times.
+| relative | If true, the _relative_ query mode will be used. Otherwise the _absolute_ mode is used.
+
+# Sink Plugins <a name="sinkplugins"></a>
+Here we describe available plugins in DCDBAnalytics that are devoted to the output of sensor data (_sinks_), and that do not perform any analysis.
+
 ## File Sink Plugin <a name="filesinkPlugin"></a>
 The _File Sink_ plugin allows to write the output of any other sensor to the local file system. As such, it does not produce output sensors by itself, and only reads from input sensors.
 The input sensors can either be fully qualified, or can be described through the unit system. In this case, multiple input sensors can be generated automatically, and the respective output paths need to be adjusted by enabling the _autoName_ attribute described below, to prevent multiple sensors from being written to the same file. The file sink operators (named sinks) support the following attributes:
@@ -685,15 +702,6 @@ Additionally, input sensors in sinks accept the following parameters:
 |:----- |:----------- |
 | path | The path to which the sensors's readings should be written. It is interpreted as described above for the _autoName_ attribute.
 
-## Tester Plugin <a name="testerPlugin"></a>
-The _Tester_ plugin can be used to test the functionality and performance of the query engine, as well as of the unit system. It will perform a specified number of queries over the set of input sensors for each unit, and then output as a sensor the total number of retrieved readings. The following are the configuration parameters for operators in the _Tester_ plugin:
-
-| Value | Explanation |
-|:----- |:----------- |
-| window | Length in milliseconds of the time window that is used to retrieve recent readings for the input sensors, starting from the latest one.
-| queries | Number of queries to be performed at each computation interval. If more than the number of input sensors per unit, these will be looped over multiple times.
-| relative | If true, the _relative_ query mode will be used. Otherwise the _absolute_ mode is used.
-
 ## Writing DCDB Analytics Plugins <a name="writingPlugins"></a>
 Generating a DCDBAnalytics plugin requires implementing a _Operator_ and _Configurator_ class which contain all logic
 tied to the specific plugin. Such classes should be derived from _OperatorTemplate_ and _OperatorConfiguratorTemplate_

analytics/operators/aggregator/AggregatorConfigurator.cpp

@@ -66,6 +66,8 @@ void AggregatorConfigurator::operatorAttributes(AggregatorOperator& op, CFG_VAL
         {
                 if (boost::iequals(val.first, "window"))
                         op.setWindow(stoull(val.second.data()) * 1000000);
+                else if (boost::iequals(val.first, "relative"))
+                        op.setRelative(to_bool(val.second.data()));
         }
 }
 

analytics/operators/aggregator/AggregatorOperator.cpp

@@ -29,11 +29,13 @@
 
 AggregatorOperator::AggregatorOperator(const std::string& name) : OperatorTemplate(name) { 
     _window = 0;
+    _relative = true;
     _buffer = nullptr;
 }
 
 AggregatorOperator::AggregatorOperator(const AggregatorOperator& other) : OperatorTemplate(other) {
     _window = other._window;
+    _relative = other._relative;
     _buffer = nullptr;
 }
 
@@ -44,6 +46,7 @@ AggregatorOperator::~AggregatorOperator() {
 
 void AggregatorOperator::printConfig(LOG_LEVEL ll) {
     LOG_VAR(ll) << "            Window:          " << _window;
+    LOG_VAR(ll) << "            Relative mode:   " << (_relative ? "enabled" : "disabled");
     OperatorTemplate<AggregatorSensorBase>::printConfig(ll);
 }
 
@@ -52,11 +55,14 @@ void AggregatorOperator::compute(U_Ptr unit) {
     if(_buffer) 
         _buffer->clear();
     size_t elCtr=0;
+    uint64_t startTs=0, endTs=0, now=getTimestamp();
+    startTs = _relative ? _window : now - _window;
+    endTs   = _relative ? 0 : now;
     for(const auto& in : unit->getInputs()) {
         // Getting the most recent values as specified in _window
         // Since we do not clear the internal buffer, all sensor readings will be accumulated in the same vector
         elCtr = _buffer==nullptr ? 0 : _buffer->size();
-        _buffer = _queryEngine.querySensor(in->getName(), _window, 0, _buffer);
+        _buffer = _queryEngine.querySensor(in->getName(), startTs, endTs, _buffer, _relative);
         if(!_buffer || _buffer->size()<=elCtr)
             throw std::runtime_error("Operator " + _name + ": cannot read from sensor " + in->getName() + "!");
     }

analytics/operators/aggregator/AggregatorOperator.h

@@ -48,8 +48,10 @@ public:
     virtual ~AggregatorOperator();
 
     void setWindow(unsigned long long w)        { _window = w; }
+    void setRelative(bool r)                    { _relative = r; }
     
     unsigned long long getWindow()              { return _window; }
+    bool getRelative()                          { return _relative; }
     
     void printConfig(LOG_LEVEL ll) override;
 
@@ -71,6 +73,7 @@ protected:
     vector<reading_t> *_buffer;
     vector<AggregatorSBPtr> _quantileSensors;
     unsigned long long _window;
+    bool _relative;
     
 };
 

analytics/operators/testeroperator/TesterOperator.cpp

@@ -49,7 +49,7 @@ TesterOperator::~TesterOperator() {
 void TesterOperator::printConfig(LOG_LEVEL ll) {
     LOG_VAR(ll) << "            Window:          " << _window;
     LOG_VAR(ll) << "            Queries:         " << _numQueries;
-    LOG_VAR(ll) << "            Relative mode:   " << (_relative ? "enabled" : "disabled");;
+    LOG_VAR(ll) << "            Relative mode:   " << (_relative ? "enabled" : "disabled");
     OperatorTemplate<SensorBase>::printConfig(ll);
 }