10.12., 9:00 - 11:00: Due to updates GitLab may be unavailable for some minutes between 09:00 and 11:00.

Commit 88f3104f authored by Alessio Netti's avatar Alessio Netti

PrintConfig functionality added to Data Analytics Framework

- Added some more indentation to make the configurations more readable
- Fixed one more warning in dcdbquery
parent abf6af19
...@@ -11,6 +11,11 @@ AverageAnalyzer::~AverageAnalyzer() { ...@@ -11,6 +11,11 @@ AverageAnalyzer::~AverageAnalyzer() {
delete _buffer; delete _buffer;
} }
void AverageAnalyzer::printConfig(LOG_LEVEL ll) {
LOG_VAR(ll) << " Window: " << _window;
AnalyzerTemplate<SensorBase>::printConfig(ll);
}
void AverageAnalyzer::compute(int unitID) { void AverageAnalyzer::compute(int unitID) {
long long max=0, sum=0, avg=0; long long max=0, sum=0, avg=0;
......
...@@ -15,6 +15,7 @@ public: ...@@ -15,6 +15,7 @@ public:
void setWindow(unsigned long long w) { _window = w; } void setWindow(unsigned long long w) { _window = w; }
unsigned long long getWindow() { return _window; } unsigned long long getWindow() { return _window; }
void printConfig(LOG_LEVEL ll) override;
private: private:
......
...@@ -82,6 +82,13 @@ public: ...@@ -82,6 +82,13 @@ public:
*/ */
virtual std::vector<AnalyzerPtr>& getAnalyzers() = 0; virtual std::vector<AnalyzerPtr>& getAnalyzers() = 0;
/**
* @brief Prints the current plugin configuration
*
* @param ll Logging level at which the configuration is printed
*/
virtual void printConfig(LOG_LEVEL ll) = 0;
protected: protected:
// Logger object // Logger object
......
...@@ -105,6 +105,27 @@ public: ...@@ -105,6 +105,27 @@ public:
_cacheInterval = pluginSettings.cacheInterval; _cacheInterval = pluginSettings.cacheInterval;
} }
/**
* @brief Print configuration as read in.
*
* @param ll Logging level to log with
*/
void printConfig(LOG_LEVEL ll) final {
LOG_VAR(ll) << " General: ";
LOG_VAR(ll) << " MQTT-Prefix: " << (_mqttPrefix != "" ? _mqttPrefix : "DEFAULT");
LOG_VAR(ll) << " Sensor Pattern: " << (_sensorPattern != "" ? _sensorPattern : "DEFAULT");
LOG_VAR(ll) << " Cache interval: " << _cacheInterval << " ms";
//prints plugin specific configurator attributes and entities if present
printConfiguratorConfig(ll);
LOG_VAR(ll) << " Analyzers: ";
for(auto a : _analyzers) {
LOG_VAR(ll) << " Analyzer: " << a->getName();
a->printConfig(ll);
}
}
/** /**
* @brief Read a config file and instantiate analyzers accordingly * @brief Read a config file and instantiate analyzers accordingly
* *
...@@ -265,7 +286,16 @@ protected: ...@@ -265,7 +286,16 @@ protected:
*/ */
virtual void global(CFG_VAL config) {} virtual void global(CFG_VAL config) {}
/**
* @brief Print information about configurable configurator attributes.
*
* This method is virtual and can be overridden on a per-plugin basis.
*
* @param ll Severity level to log with
*/
virtual void printConfiguratorConfig(LOG_LEVEL ll) {
LOG_VAR(ll) << " No other plugin-specific general parameters defined";
}
/** /**
* @brief Store an analyzer in the internal vectors * @brief Store an analyzer in the internal vectors
...@@ -368,23 +398,19 @@ protected: ...@@ -368,23 +398,19 @@ protected:
for (auto &u: *units) { for (auto &u: *units) {
constructSensorNames(*u, an); constructSensorNames(*u, an);
if (an.getStreaming()) { if (an.getStreaming()) {
LOG(debug) << " Unit \"" << u->getName() << "\"";
for (const auto &i : u->getInputs())
LOG(debug) << " Input " << i->getName();
for (const auto &o : u->getOutputs())
LOG(debug) << " Output " << o->getName() << " using MQTT-topic \"" << o->getMqtt() << "\"";
if (!unit(*u)) { if (!unit(*u)) {
LOG(error) << " Unit " << u->getName() << " did not pass the final check!"; LOG(error) << " Unit " << u->getName() << " did not pass the final check!";
an.clearUnits(); an.clearUnits();
delete units; delete units;
return false; return false;
} else } else {
LOG(debug) << " Unit " << u->getName() << " generated.";
an.addUnit(u); an.addUnit(u);
}
} else { } else {
if (unit(*u)) { if (unit(*u)) {
an.addToOndemandCache(u); an.addToOndemandCache(u);
LOG(debug) << " Template unit for on-demand operation " + u->getName() + " generated!"; LOG(debug) << " Template unit for on-demand operation " + u->getName() + " generated.";
} else { } else {
LOG(error) << " Template unit " << u->getName() << " did not pass the final check!"; LOG(error) << " Template unit " << u->getName() << " did not pass the final check!";
an.clearUnits(); an.clearUnits();
......
...@@ -193,6 +193,13 @@ public: ...@@ -193,6 +193,13 @@ public:
*/ */
virtual map<string, reading_t> computeOnDemand(const string& node="") = 0; virtual map<string, reading_t> computeOnDemand(const string& node="") = 0;
/**
* @brief Prints the current analyzer configuration
*
* @param ll Logging level at which the configuration is printed
*/
virtual void printConfig(LOG_LEVEL ll) = 0;
// Getter methods // Getter methods
const string& getName() const { return _name; } const string& getName() const { return _name; }
const string& getMqttPart() const { return _mqttPart; } const string& getMqttPart() const { return _mqttPart; }
......
...@@ -102,6 +102,27 @@ public: ...@@ -102,6 +102,27 @@ public:
} }
} }
/**
* @brief Prints the current analyzer configuration
*
* @param ll Logging level at which the configuration is printed
*/
virtual void printConfig(LOG_LEVEL ll) override {
LOG_VAR(ll) << " MQTT part: " << (_mqttPart != "" ? _mqttPart : "DEFAULT");
LOG_VAR(ll) << " Sync readings: " << (_sync ? "enabled" : "disabled");
LOG_VAR(ll) << " Streaming mode: " << (_streaming ? "enabled" : "disabled");
LOG_VAR(ll) << " Duplicated mode: " << (_duplicate ? "enabled" : "disabled");
LOG_VAR(ll) << " MinValues: " << _minValues;
LOG_VAR(ll) << " Interval: " << _interval;
LOG_VAR(ll) << " Start delay: " << _delayInterval;
if(!_units.empty()) {
LOG_VAR(ll) << " Units:";
for (auto u : _units)
u->printConfig(ll, lg);
} else
LOG_VAR(ll) << " Units: none";
}
/** /**
* @brief Adds an unit to this analyzer * @brief Adds an unit to this analyzer
* *
......
...@@ -76,6 +76,14 @@ public: ...@@ -76,6 +76,14 @@ public:
*/ */
virtual std::vector<SBasePtr>& getBaseOutputs() = 0; virtual std::vector<SBasePtr>& getBaseOutputs() = 0;
/**
* @brief Prints the current unit configuration
*
* @param ll Logging level at which the configuration is printed
* @param lg Logger object to be used
*/
virtual void printConfig(LOG_LEVEL ll, LOGGER& lg) = 0;
}; };
//for better readability //for better readability
......
...@@ -187,6 +187,23 @@ public: ...@@ -187,6 +187,23 @@ public:
*/ */
void addOutput(const S_Ptr output) { _outputs.push_back(output); _baseOutputs.push_back(output); } void addOutput(const S_Ptr output) { _outputs.push_back(output); _baseOutputs.push_back(output); }
/**
* @brief Prints the current unit configuration
*
* @param ll Logging level at which the configuration is printed
* @param lg Logger object to be used
*/
virtual void printConfig(LOG_LEVEL ll, LOGGER& lg) override {
LOG_VAR(ll) << " Unit: " << _name;
LOG_VAR(ll) << " Inputs: ";
for (const auto &i : _inputs)
LOG_VAR(ll) << " " << i->getName();
LOG_VAR(ll) << " Outputs: ";
for (const auto &o : _outputs)
o->printConfig(ll, lg, 20);
}
protected: protected:
// Name corresponds to the output unit we are addressing // Name corresponds to the output unit we are addressing
......
...@@ -257,22 +257,14 @@ public: ...@@ -257,22 +257,14 @@ public:
static std::string formatName(const std::string& name, int cpuID=-1) {return cpuID<0 ? name : "cpu" + std::to_string(cpuID) + "." + name;} static std::string formatName(const std::string& name, int cpuID=-1) {return cpuID<0 ? name : "cpu" + std::to_string(cpuID) + "." + name;}
virtual void printConfig(LOG_LEVEL ll, LOGGER& lg) { virtual void printConfig(LOG_LEVEL ll, LOGGER& lg, unsigned leadingSpaces=16) {
LOG_VAR(ll) << " Sensor: " << _name; std::string leading(leadingSpaces, ' ');
LOG_VAR(ll) << " MQTT Topic: " << _mqtt; LOG_VAR(ll) << leading << "Sensor: " << _name;
LOG_VAR(ll) << " sink: " << getSinkPath(); LOG_VAR(ll) << leading << " MQTT Topic: " << _mqtt;
LOG_VAR(ll) << " subSampling: " << getSubsampling(); LOG_VAR(ll) << leading << " Sink: " << (getSinkPath() != "" ? getSinkPath() : "none");
if(_skipConstVal) { LOG_VAR(ll) << leading << " SubSampling: " << getSubsampling();
LOG_VAR(ll) << " Skipping constant values"; LOG_VAR(ll) << leading << (_skipConstVal ? " Skipping constant values" : " No skipping of constant values");
} else { LOG_VAR(ll) << leading << (_delta ? " Storing delta readings" : " Storing absolute readings");
LOG_VAR(ll) << " No skipping of constant values";
}
if(_delta) {
LOG_VAR(ll) << " Storing delta readings";
} else {
LOG_VAR(ll) << " Storing absolute readings";
}
} }
protected: protected:
......
...@@ -254,56 +254,92 @@ int main(int argc, char** argv) { ...@@ -254,56 +254,92 @@ int main(int argc, char** argv) {
return 1; return 1;
} }
_analyticsManager = new AnalyticsManager();
// Preparing the SensorNavigator
bool failedTree = false;
std::shared_ptr<SensorNavigator> navigator = std::make_shared<SensorNavigator>();
vector<std::string> names, topics;
for(const auto& p : _configuration->getPlugins())
for(const auto& g : p.configurator->getSensorGroups())
for(const auto& s : g->getSensors()) {
names.push_back(s->getName());
topics.push_back(s->getMqtt());
}
try {
navigator->buildTree(globalSettings.hierarchy, &names, &topics);
} catch(const std::invalid_argument& e) {
LOG(error) << e.what();
LOG(error) << "Failed to build sensor hierarchy tree, data analytics manager will not be initialized!";
failedTree = true;
}
if(!failedTree) {
_queryEngine.setNavigator(navigator);
_queryEngine.triggerUpdate();
_queryEngine.setQueryCallback(sensorQueryCallback);
if(!_analyticsManager->load(argv[argc-1], "dcdbpusher.conf", pluginSettings)) {
LOG(fatal) << "Failed to load data analytics manager!";
return 1;
}
}
//print configuration to give some feedback //print configuration to give some feedback
//config of plugins is only printed if the config shall be validated or to debug level otherwise //config of plugins is only printed if the config shall be validated or to debug level otherwise
LOG_LEVEL vLogLevel = LOG_LEVEL::debug; LOG_LEVEL vLogLevel = LOG_LEVEL::debug;
if (globalSettings.validateConfig) { if (globalSettings.validateConfig) {
vLogLevel = boost::log::trivial::info; vLogLevel = boost::log::trivial::info;
} }
LOG_VAR(vLogLevel) << "----- Configuration: -----"; LOG_VAR(vLogLevel) << "----- Configuration -----";
//print global settings in either case //print global settings in either case
LOG(info) << "Global Settings:"; LOG(info) << "Global Settings:";
LOG(info) << " Broker: " << globalSettings.brokerHost << ":" << globalSettings.brokerPort; LOG(info) << " Broker: " << globalSettings.brokerHost << ":" << globalSettings.brokerPort;
LOG(info) << " Threads: " << globalSettings.threads; LOG(info) << " Threads: " << globalSettings.threads;
LOG(info) << " Daemonize: " << (globalSettings.daemonize ? "Enabled" : "Disabled"); LOG(info) << " Daemonize: " << (globalSettings.daemonize ? "Enabled" : "Disabled");
LOG(info) << " MaxMsgNum: " << globalSettings.maxMsgNum; LOG(info) << " MaxMsgNum: " << globalSettings.maxMsgNum;
LOG(info) << " MaxInflightMsgNum: " << globalSettings.maxInflightMsgNum; LOG(info) << " MaxInflightMsgNum: " << globalSettings.maxInflightMsgNum;
LOG(info) << " MaxQueuedMsgNum: " << globalSettings.maxQueuedMsgNum; LOG(info) << " MaxQueuedMsgNum: " << globalSettings.maxQueuedMsgNum;
LOG(info) << " MQTT-QoS: " << globalSettings.qosLevel; LOG(info) << " MQTT-QoS: " << globalSettings.qosLevel;
LOG(info) << " MQTT-prefix: " << pluginSettings.mqttPrefix; LOG(info) << " MQTT-prefix: " << pluginSettings.mqttPrefix;
LOG(info) << " Write-Dir: " << pluginSettings.tempdir; LOG(info) << " Write-Dir: " << pluginSettings.tempdir;
LOG(info) << " Hierarchy: " << globalSettings.hierarchy; LOG(info) << " Hierarchy: " << globalSettings.hierarchy;
LOG(info) << " CacheInterval: " << pluginSettings.cacheInterval / 1000 << " [s]"; LOG(info) << " CacheInterval: " << pluginSettings.cacheInterval / 1000 << " [s]";
if(globalSettings.validateConfig) { if(globalSettings.validateConfig) {
LOG(info) << " Only validating config files."; LOG(info) << " Only validating config files.";
} else { } else {
LOG(info) << " validateConfig: Disabled"; LOG(info) << " ValidateConfig: Disabled";
} }
LOG(info) << "RestAPI Settings:"; LOG(info) << "RestAPI Settings:";
LOG(info) << " REST Server: " << restAPISettings.restHost << ":" << restAPISettings.restPort; LOG(info) << " REST Server: " << restAPISettings.restHost << ":" << restAPISettings.restPort;
#ifdef DEBUG #ifdef DEBUG
LOG(info) << " Certificate: " << restAPISettings.certificate; LOG(info) << " Certificate: " << restAPISettings.certificate;
LOG(info) << " Private key file: " << restAPISettings.privateKey; LOG(info) << " Private key file: " << restAPISettings.privateKey;
LOG(info) << " DH params from: " << restAPISettings.dhFile; LOG(info) << " DH params from: " << restAPISettings.dhFile;
#else #else
LOG(info) << " Certificate, private key and DH-param file not printed."; LOG(info) << " Certificate, private key and DH-param file not printed.";
#endif #endif
LOG_VAR(vLogLevel) << "----- Sampling Configuration -----";
for(auto& p : _configuration->getPlugins()) { for(auto& p : _configuration->getPlugins()) {
LOG_VAR(vLogLevel) << "Plugin \"" << p.id << "\""; LOG_VAR(vLogLevel) << "Sampling Plugin \"" << p.id << "\"";
p.configurator->printConfig(vLogLevel); p.configurator->printConfig(vLogLevel);
} }
LOG_VAR(vLogLevel) << "----- End Config -----"; LOG_VAR(vLogLevel) << "----- Analytics Configuration -----";
for(auto& p : _analyticsManager->getPlugins()) {
LOG_VAR(vLogLevel) << "Analytics Plugin \"" << p.id << "\"";
p.configurator->printConfig(vLogLevel);
}
LOG_VAR(vLogLevel) << "----- End Configuration -----";
if (globalSettings.validateConfig) { if (globalSettings.validateConfig) {
return 0; return 0;
} }
//MQTTPusher and Https server get their own threads //MQTTPusher and Https server get their own threads
_analyticsManager = new AnalyticsManager();
_mqttPusher = new MQTTPusher(globalSettings.brokerPort, globalSettings.brokerHost, pluginSettings.sensorPattern, globalSettings.qosLevel, _mqttPusher = new MQTTPusher(globalSettings.brokerPort, globalSettings.brokerHost, pluginSettings.sensorPattern, globalSettings.qosLevel,
_configuration->getPlugins(), _analyticsManager->getPlugins(), globalSettings.maxMsgNum, globalSettings.maxInflightMsgNum, globalSettings.maxQueuedMsgNum); _configuration->getPlugins(), _analyticsManager->getPlugins(), globalSettings.maxMsgNum, globalSettings.maxInflightMsgNum, globalSettings.maxQueuedMsgNum);
_httpsServer = new HttpsServer(restAPISettings, _configuration->getPlugins(), _mqttPusher, _analyticsManager, io); _httpsServer = new HttpsServer(restAPISettings, _configuration->getPlugins(), _mqttPusher, _analyticsManager, io);
...@@ -328,34 +364,7 @@ int main(int argc, char** argv) { ...@@ -328,34 +364,7 @@ int main(int argc, char** argv) {
} }
} }
// Preparing the SensorNavigator
bool failedTree = false;
std::shared_ptr<SensorNavigator> navigator = std::make_shared<SensorNavigator>();
vector<std::string> names, topics;
for(const auto& p : _configuration->getPlugins())
for(const auto& g : p.configurator->getSensorGroups())
for(const auto& s : g->getSensors()) {
names.push_back(s->getName());
topics.push_back(s->getMqtt());
}
try {
navigator->buildTree(globalSettings.hierarchy, &names, &topics);
} catch(const std::invalid_argument& e) {
LOG(error) << e.what();
LOG(error) << "Failed to build sensor hierarchy tree, data analytics manager will not be initialized!";
failedTree = true;
}
if(!failedTree) { if(!failedTree) {
_queryEngine.setNavigator(navigator);
_queryEngine.triggerUpdate();
_queryEngine.setQueryCallback(sensorQueryCallback);
if(!_analyticsManager->load(argv[argc-1], "dcdbpusher.conf", pluginSettings)) {
LOG(fatal) << "Failed to load data analytics manager!";
return 1;
}
if(!_queryEngine.updated.is_lock_free()) if(!_queryEngine.updated.is_lock_free())
LOG(warning) << "This machine does not support lock-free atomics. Performance may be degraded."; LOG(warning) << "This machine does not support lock-free atomics. Performance may be degraded.";
......
...@@ -316,25 +316,25 @@ public: ...@@ -316,25 +316,25 @@ public:
* @param ll Logging level to log with * @param ll Logging level to log with
*/ */
void printConfig(LOG_LEVEL ll) final { void printConfig(LOG_LEVEL ll) final {
LOG_VAR(ll) << " General: "; LOG_VAR(ll) << " General: ";
if (_mqttPrefix != "") { if (_mqttPrefix != "") {
LOG_VAR(ll) << " MQTT-Prefix: " << _mqttPrefix; LOG_VAR(ll) << " MQTT-Prefix: " << _mqttPrefix;
} else { } else {
LOG_VAR(ll) << " MQTT-Prefix: DEFAULT"; LOG_VAR(ll) << " MQTT-Prefix: DEFAULT";
} }
if (_sensorPattern != "") { if (_sensorPattern != "") {
LOG_VAR(ll) << " Sensor Pattern: " << _sensorPattern; LOG_VAR(ll) << " Sensor Pattern: " << _sensorPattern;
} }
if (_cacheInterval != DEFAULT_CACHE_INTERVAL) { if (_cacheInterval != DEFAULT_CACHE_INTERVAL) {
LOG_VAR(ll) << " Cache interval: " << _cacheInterval << " ms"; LOG_VAR(ll) << " Cache interval: " << _cacheInterval << " ms";
} else { } else {
LOG_VAR(ll) << " Cache interval: DEFAULT"; LOG_VAR(ll) << " Cache interval: DEFAULT";
} }
//prints plugin specific configurator attributes and entities if present //prints plugin specific configurator attributes and entities if present
printConfiguratorConfig(ll); printConfiguratorConfig(ll);
LOG_VAR(ll) << " Groups:"; LOG_VAR(ll) << " Groups:";
for(auto g : _sensorGroups) { for(auto g : _sensorGroups) {
g->SensorGroupInterface::printConfig(ll); g->SensorGroupInterface::printConfig(ll);
g->printConfig(ll); g->printConfig(ll);
...@@ -722,7 +722,7 @@ protected: ...@@ -722,7 +722,7 @@ protected:
*/ */
virtual void printConfiguratorConfig(LOG_LEVEL ll) { virtual void printConfiguratorConfig(LOG_LEVEL ll) {
//Overwrite if necessary //Overwrite if necessary
LOG_VAR(ll) << " No other plugin specific general parameters or entities defined"; LOG_VAR(ll) << " No other plugin-specific general parameters or entities defined";
} }
/** /**
......
...@@ -81,19 +81,19 @@ public: ...@@ -81,19 +81,19 @@ public:
} }
virtual void printConfig(LOG_LEVEL ll) { virtual void printConfig(LOG_LEVEL ll) {
LOG_VAR(ll) << " Sensor Group: " << _groupName; LOG_VAR(ll) << " Sensor Group: " << _groupName;
if (_mqttPart != "") { if (_mqttPart != "") {
LOG_VAR(ll) << " MQTT part: " << _mqttPart; LOG_VAR(ll) << " MQTT part: " << _mqttPart;
} }
if (_sync) { if (_sync) {
LOG_VAR(ll) << " Synchronized readings enabled"; LOG_VAR(ll) << " Synchronized readings enabled";
} else { } else {
LOG_VAR(ll) << " Synchronized readings disabled"; LOG_VAR(ll) << " Synchronized readings disabled";
} }
LOG_VAR(ll) << " minValues: " << _minValues; LOG_VAR(ll) << " minValues: " << _minValues;
LOG_VAR(ll) << " interval: " << _interval; LOG_VAR(ll) << " interval: " << _interval;
} }
//have to be overwritten //have to be overwritten
......
...@@ -76,7 +76,7 @@ public: ...@@ -76,7 +76,7 @@ public:
} }
virtual void printConfig(LOG_LEVEL ll) override { virtual void printConfig(LOG_LEVEL ll) override {
LOG_VAR(ll) << " Sensors:"; LOG_VAR(ll) << " Sensors:";
for(auto s : _sensors) { for(auto s : _sensors) {
s->SensorBase::printConfig(ll, lg); s->SensorBase::printConfig(ll, lg);
s->printConfig(ll, lg); s->printConfig(ll, lg);
......
...@@ -57,12 +57,13 @@ public: ...@@ -57,12 +57,13 @@ public:
void setPropertyId(const std::string& property) { _propertyId = static_cast<BACNET_PROPERTY_ID>(stoul(property)); } void setPropertyId(const std::string& property) { _propertyId = static_cast<BACNET_PROPERTY_ID>(stoul(property)); }
void setObjectIndex(int32_t objectIndex) { _objectIndex = objectIndex; } void setObjectIndex(int32_t objectIndex) { _objectIndex = objectIndex; }
void printConfig(LOG_LEVEL ll, LOGGER& lg) override { void printConfig(LOG_LEVEL ll, LOGGER& lg, unsigned leadingSpaces=16) override {
LOG_VAR(ll) << " Factor: " << _factor; std::string leading(leadingSpaces, ' ');
LOG_VAR(ll) << " objectInstance: " << _objectInstance; LOG_VAR(ll) << leading << " Factor: " << _factor;
LOG_VAR(ll) << " objectType: " << _objectType; LOG_VAR(ll) << leading << " objectInstance: " << _objectInstance;
LOG_VAR(ll) << " propertyId: " << _propertyId; LOG_VAR(ll) << leading << " objectType: " << _objectType;
LOG_VAR(ll) << " objectIndex: " << _objectIndex; LOG_VAR(ll) << leading << " propertyId: " << _propertyId;
LOG_VAR(ll) << leading << " objectIndex: " << _objectIndex;
} }
protected: protected:
......
...@@ -90,10 +90,10 @@ void BACnetSensorGroup::readAsync() { ...@@ -90,10 +90,10 @@ void BACnetSensorGroup::readAsync() {
} }
void BACnetSensorGroup::printConfig(LOG_LEVE