Commit 78d94c06 authored by Carla Guillen's avatar Carla Guillen
Browse files

adding batch_domain

parent fa63da91
...@@ -150,7 +150,7 @@ MariaDB::~MariaDB(){ ...@@ -150,7 +150,7 @@ MariaDB::~MariaDB(){
} }
bool MariaDB::getDBJobID(const std::string & job_id_string, std::string& job_db_id, const std::string & user, int number_nodes) { bool MariaDB::getDBJobID(const std::string & job_id_string, std::string& job_db_id, const std::string & user, int number_nodes, int batch_domain) {
std::lock_guard<std::mutex> lock(mut); std::lock_guard<std::mutex> lock(mut);
Job_info_t *job_info=_jobCache.find(user, number_nodes, job_id_string); Job_info_t *job_info=_jobCache.find(user, number_nodes, job_id_string);
if(job_info){ //found if(job_info){ //found
...@@ -160,7 +160,7 @@ bool MariaDB::getDBJobID(const std::string & job_id_string, std::string& job_db_ ...@@ -160,7 +160,7 @@ bool MariaDB::getDBJobID(const std::string & job_id_string, std::string& job_db_
std::stringstream build_query; std::stringstream build_query;
build_query << "SELECT job_id, job_id_string FROM Accounting WHERE job_id_string='" << job_id_string << "' AND user='" << user << "'"; build_query << "SELECT job_id, job_id_string FROM Accounting WHERE job_id_string='" << job_id_string << "' AND user='" << user << "'";
build_query << " AND nodes=" << number_nodes; build_query << " AND nodes=" << number_nodes << " AND batch_domain=" << batch_domain;
auto query = build_query.str(); auto query = build_query.str();
LOG(debug)<< query; LOG(debug)<< query;
if (mysql_real_query(_mysql, query.c_str(), query.size())) { if (mysql_real_query(_mysql, query.c_str(), query.size())) {
...@@ -224,7 +224,7 @@ bool MariaDB::getCurrentSuffixAggregateTable(std::string & suffix){ ...@@ -224,7 +224,7 @@ bool MariaDB::getCurrentSuffixAggregateTable(std::string & suffix){
} }
bool MariaDB::insertIntoJob(const std::string& job_id_string, const std::string& uid, std::string & job_id_db, const std::string & suffix, int number_nodes){ bool MariaDB::insertIntoJob(const std::string& job_id_string, const std::string& uid, std::string & job_id_db, const std::string & suffix, int number_nodes, int batch_domain){
std::lock_guard<std::mutex> lock(mut); std::lock_guard<std::mutex> lock(mut);
//maybe another thread did this for us //maybe another thread did this for us
Job_info_t *job_info=_jobCache.find(uid,number_nodes, job_id_string); Job_info_t *job_info=_jobCache.find(uid,number_nodes, job_id_string);
...@@ -236,7 +236,7 @@ bool MariaDB::insertIntoJob(const std::string& job_id_string, const std::string& ...@@ -236,7 +236,7 @@ bool MariaDB::insertIntoJob(const std::string& job_id_string, const std::string&
//Also check that job was not inserted by another collector (shouldn't really happen but "sicher ist sicher" //Also check that job was not inserted by another collector (shouldn't really happen but "sicher ist sicher"
std::stringstream build_query; std::stringstream build_query;
build_query << "SELECT job_id, job_id_string FROM Accounting WHERE job_id_string ='"; build_query << "SELECT job_id, job_id_string FROM Accounting WHERE job_id_string ='";
build_query << job_id_string << "' AND user='" << uid <<"' AND nodes=" << number_nodes; build_query << job_id_string << "' AND user='" << uid <<"' AND nodes=" << number_nodes << " AND batch_domain=" << batch_domain;
auto select_query = build_query.str(); auto select_query = build_query.str();
LOG(debug) << select_query; LOG(debug) << select_query;
...@@ -260,10 +260,11 @@ bool MariaDB::insertIntoJob(const std::string& job_id_string, const std::string& ...@@ -260,10 +260,11 @@ bool MariaDB::insertIntoJob(const std::string& job_id_string, const std::string&
if(!job_found_in_db) { if(!job_found_in_db) {
std::stringstream build_insert; std::stringstream build_insert;
build_insert << "INSERT IGNORE INTO Accounting (job_id_string, user, nodes, aggregate_first_suffix, aggregate_last_suffix) VALUES (\'" << job_id_string << "\',\'"; build_insert << "INSERT IGNORE INTO Accounting (job_id_string, user, nodes, aggregate_first_suffix, aggregate_last_suffix, batch_domain) VALUES (\'" << job_id_string << "\',\'";
build_insert << uid << "\',\'"; build_insert << uid << "\',\'";
build_insert << number_nodes << "\',\'"; build_insert << number_nodes << "\',\'";
build_insert << suffix << "\',\'" << suffix << "\')"; build_insert << suffix << "\',\'" << suffix << "\',\'";
build_insert << batch_domain << "\')";
std::string query = build_insert.str(); std::string query = build_insert.str();
LOG(debug)<< query; LOG(debug)<< query;
......
...@@ -117,15 +117,15 @@ public: ...@@ -117,15 +117,15 @@ public:
* @param job_id_strings job id strings including array jobid. * @param job_id_strings job id strings including array jobid.
* @param job_id_map job_id_string to job_id (db) map * @param job_id_map job_id_string to job_id (db) map
*/ */
bool getDBJobID(const std::string & job_id_string, std::string& job_db_id, const std::string & user, int number_nodes); bool getDBJobID(const std::string & job_id_string, std::string& job_db_id, const std::string & user, int number_nodes, int batch_domain);
/** /**
* Insert job in the accounting table. * Insert job in the accounting table.
*/ */
bool insertIntoJob(const std::string& job_id_string, const std::string& uid, std::string & job_id_db, const std::string & suffix, int number_nodes); bool insertIntoJob(const std::string& job_id_string, const std::string& uid, std::string & job_id_db, const std::string & suffix, int number_nodes, int batch_domain);
/** /**
* Insert performance data into the aggregate table (Aggregate_<suffix> * Insert performance data into the aggregate table (Aggregate_<suffix>)
*/ */
bool insertInAggregateTable(const std::string& suffix, Aggregate_info_t & agg_info); bool insertInAggregateTable(const std::string& suffix, Aggregate_info_t & agg_info);
......
...@@ -77,6 +77,13 @@ void PerSystSqlConfigurator::operatorAttributes(PerSystSqlOperator& op, CFG_VAL ...@@ -77,6 +77,13 @@ void PerSystSqlConfigurator::operatorAttributes(PerSystSqlOperator& op, CFG_VAL
} catch (const std::exception &e) { } catch (const std::exception &e) {
LOG(error) << " Error parsing number_quantiles \"" << val.second.data() << "\": " << e.what(); LOG(error) << " Error parsing number_quantiles \"" << val.second.data() << "\": " << e.what();
} }
} else if (boost::iequals(val.first, "batch_domain")){
try {
int batch_domain = std::stoi(val.second.data());
op.setBatchDomain(batch_domain);
} catch (const std::exception &e) {
LOG(error) << " Error parsing batch_domain \"" << val.second.data() << "\": " << e.what();
}
} else if (boost::iequals(val.first, "go_back_ms")) { } else if (boost::iequals(val.first, "go_back_ms")) {
try { try {
auto go_back_ms = std::stoi(val.second.data()); auto go_back_ms = std::stoi(val.second.data());
......
...@@ -49,7 +49,7 @@ ...@@ -49,7 +49,7 @@
PerSystSqlOperator::PerSystSqlOperator(const std::string& name) : PerSystSqlOperator::PerSystSqlOperator(const std::string& name) :
OperatorTemplate(name), JobOperatorTemplate(name), _number_of_even_quantiles( OperatorTemplate(name), JobOperatorTemplate(name), _number_of_even_quantiles(
0), _severity_formula(NOFORMULA), _severity_threshold(0), _severity_exponent( 0), _batch_domain(-1), _severity_formula(NOFORMULA), _severity_threshold(0), _severity_exponent(
0), _severity_max_memory(0), _go_back_ns(0), _backend(DEFAULT), _scaling_factor( 0), _severity_max_memory(0), _go_back_ns(0), _backend(DEFAULT), _scaling_factor(
1), _searchedOnceForMetaData(false), _property_id(0) { 1), _searchedOnceForMetaData(false), _property_id(0) {
_persystdb = MariaDB::getInstance(); _persystdb = MariaDB::getInstance();
...@@ -73,6 +73,7 @@ void PerSystSqlOperator::copy(const PerSystSqlOperator& other){ ...@@ -73,6 +73,7 @@ void PerSystSqlOperator::copy(const PerSystSqlOperator& other){
this->_buffer = other._buffer; this->_buffer = other._buffer;
this->_quantileSensors = other._quantileSensors; this->_quantileSensors = other._quantileSensors;
this->_number_of_even_quantiles = other._number_of_even_quantiles; this->_number_of_even_quantiles = other._number_of_even_quantiles;
this->_batch_domain = other._batch_domain;
this->_severity_formula = other._severity_formula; this->_severity_formula = other._severity_formula;
this->_severity_threshold = other._severity_threshold; this->_severity_threshold = other._severity_threshold;
this->_severity_exponent = other._severity_exponent; this->_severity_exponent = other._severity_exponent;
...@@ -114,6 +115,7 @@ void PerSystSqlOperator::printConfig(LOG_LEVEL ll) { ...@@ -114,6 +115,7 @@ void PerSystSqlOperator::printConfig(LOG_LEVEL ll) {
} }
LOG_VAR(ll) << "Property Configuration:"; LOG_VAR(ll) << "Property Configuration:";
LOG_VAR(ll) << "\tnumber_of_even_quantiles=" << _number_of_even_quantiles; LOG_VAR(ll) << "\tnumber_of_even_quantiles=" << _number_of_even_quantiles;
LOG_VAR(ll) << "\tbatch_domain=" << _batch_domain;
LOG_VAR(ll) << "\tproperty_id=" << _property_id; LOG_VAR(ll) << "\tproperty_id=" << _property_id;
LOG_VAR(ll) << "Severity Configuration:"; LOG_VAR(ll) << "Severity Configuration:";
LOG_VAR(ll) << "\tseverity_formula=" << _severity_formula; LOG_VAR(ll) << "\tseverity_formula=" << _severity_formula;
...@@ -172,7 +174,7 @@ void PerSystSqlOperator::compute(U_Ptr unit, qeJobData& jobData) { ...@@ -172,7 +174,7 @@ void PerSystSqlOperator::compute(U_Ptr unit, qeJobData& jobData) {
} }
uint64_t measurement_ts = 0; uint64_t measurement_ts = 0;
if(_buffer.size() == 0){ if(_buffer.size() == 0){
LOG(debug) << "PerSystSql Operator " << _name << ": no data in queryEngine found!"; LOG(error) << "PerSystSql Operator " << _name << ": no data in queryEngine found!";
return; return;
} else { } else {
measurement_ts = _buffer[0].timestamp; measurement_ts = _buffer[0].timestamp;
...@@ -194,8 +196,8 @@ void PerSystSqlOperator::compute(U_Ptr unit, qeJobData& jobData) { ...@@ -194,8 +196,8 @@ void PerSystSqlOperator::compute(U_Ptr unit, qeJobData& jobData) {
LOG(error) << "Failed to create Aggregate table!"; LOG(error) << "Failed to create Aggregate table!";
return; return;
} }
if(!_persystdb->getDBJobID(jobData.jobId, agg_info.job_id_db, jobData.userId, jobData.nodes.size()) && if(!_persystdb->getDBJobID(jobData.jobId, agg_info.job_id_db, jobData.userId, jobData.nodes.size(), _batch_domain) &&
!_persystdb->insertIntoJob(jobData.jobId, jobData.userId, agg_info.job_id_db, table_suffix, jobData.nodes.size()) ){ !_persystdb->insertIntoJob(jobData.jobId, jobData.userId, agg_info.job_id_db, table_suffix, jobData.nodes.size(), _batch_domain) ){
LOG(error) << "Job insertion not possible, no job id db available for slurm job id" << jobData.jobId; LOG(error) << "Job insertion not possible, no job id db available for slurm job id" << jobData.jobId;
return; return;
} }
......
...@@ -73,6 +73,10 @@ public: ...@@ -73,6 +73,10 @@ public:
_number_of_even_quantiles = numberOfEvenQuantiles; _number_of_even_quantiles = numberOfEvenQuantiles;
} }
void setBatchDomain(int batch_domain){
_batch_domain = batch_domain;
}
void pushbackQuantileSensor(AggregatorSBPtr qSensor){ void pushbackQuantileSensor(AggregatorSBPtr qSensor){
_quantileSensors.push_back(qSensor); _quantileSensors.push_back(qSensor);
} }
...@@ -137,6 +141,7 @@ private: ...@@ -137,6 +141,7 @@ private:
std::vector<reading_t> _buffer; std::vector<reading_t> _buffer;
std::vector<AggregatorSBPtr> _quantileSensors; std::vector<AggregatorSBPtr> _quantileSensors;
unsigned int _number_of_even_quantiles; unsigned int _number_of_even_quantiles;
int _batch_domain;
Formula _severity_formula; Formula _severity_formula;
double _severity_threshold; double _severity_threshold;
double _severity_exponent; double _severity_exponent;
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment