ae7834adaa
将 ExpBase::mon_proc() 中的 exp_type_ 分支逻辑提取为 4 个子类，通过 doMonProc() 虚函数实现多态分发： - LogicAlg (exp_type 1)：实时逻辑判断 - BoundAlg (exp_type 2)：监控变量上下限检测 - BoundHoldAlg (exp_type 5)：持续超限检测（继承 BoundAlg） - FeedbackAlg (exp_type 3/4)：动作反馈处理 ExpBase 新增纯虚函数 doMonProc() 和钩子函数 doInitExtend()， init() 和 mon_proc() 中所有类型分支替换为虚函数委托调用。
837 lines
29 KiB
C++
837 lines
29 KiB
C++
|
||
#include <eqpalg/algs/exp_base.h>
|
||
#include <eqpalg/exp_macro/get_macro_replaced_exp.h>
|
||
#include <eqpalg/table_struct/t_sample_mag.h>
|
||
#include <eqpalg/utility/build_alarm_info.h>
|
||
#include <float.h>
|
||
#include <memory>
|
||
#include <mix_cc/ihyper_db/utility.h>
|
||
#include <mix_cc/sql.h>
|
||
#include <mix_cc/sql/database/db2_t.h>
|
||
#include <mix_cc/type/mix_time.h>
|
||
#include <shm/TaskData.h>
|
||
#include <unordered_map>
|
||
#include <vector>
|
||
extern ProcessType glob_process_type;
|
||
ExpBase::ExpBase(const string &name, const mix_cc::json &rule_json,
|
||
const string &ruleId, size_t exp_type)
|
||
: AlgBase(name, rule_json, ruleId), exp_type_(exp_type) {
|
||
logger_.reset(
|
||
new LOG("ExpBase-" + std::to_string(exp_type) + ":" + rule_name_,
|
||
AUTO_CATCH_PID));
|
||
}
|
||
ExpBase::~ExpBase() {
|
||
}
|
||
|
||
int ExpBase::init() {
|
||
|
||
int ret = 0;
|
||
try {
|
||
ret += AlgBase::init(); /*1.tag点；2.执行周期*/
|
||
this->con_monitor_.setThreshold(100);
|
||
// 重新载入数据源配置信息
|
||
ret += this->reload_config_data_source(); /*3.数据源*/
|
||
// 在载入数据源信息完成后，载入表达式配置之前，必须刷新变量，把变量信息初始化到mm_vars内
|
||
if (glob_process_type == ProcessType::kMon ||
|
||
glob_process_type == ProcessType::kTask) {
|
||
ret += expr_engine_->firstFill(data_source_, now_time_, query_time_range_); /*4.数据项*/
|
||
}
|
||
|
||
// 必须在刷新变量后，才可以初始化表达式
|
||
ret += this->reload_config_exp_act(); /*5.动作表达式*/
|
||
if (feedback_mode_) {
|
||
ret += this->reload_config_exp_feedback(); /*6.反馈表达式*/
|
||
fb_fsm_.configure(keep_mode_, time_out_);
|
||
}
|
||
|
||
doInitExtend();
|
||
} catch (const std::exception &e) {
|
||
logger_->Debug() << e.what() << ".location:" << BOOST_CURRENT_LOCATION
|
||
<< endl;
|
||
}
|
||
if (ret == 0) {
|
||
this->exp_is_wrong_ = false;
|
||
}
|
||
return ret;
|
||
}
|
||
|
||
AlarmInfo ExpBase::exec_mon() {
|
||
AlarmInfo out_alarm{};
|
||
/*----规则配置参数检查----start*/
|
||
if (this->exp_is_wrong_) {
|
||
if (!exp_wrong_is_alarmed_) {
|
||
std::string msg = rule_name_ + "配置参数加载异常，请检规则配置！";
|
||
mix_cc::time_range_t time_range(chrono::system_clock::now() - 20s,
|
||
chrono::system_clock::now());
|
||
if (!this->error_code_list_.empty()) {
|
||
for (auto item : this->error_code_list_) {
|
||
msg +=
|
||
ErrorCode::ErrorLocationDescription.at(int(item.error_location));
|
||
msg += ErrorCode::ErrorTypeDescription.at(int(item.error_type));
|
||
msg += "。";
|
||
}
|
||
}
|
||
msg += this->error_message_str_;
|
||
|
||
auto alarm_task = utility::build_alarm_info(
|
||
MsgLevel::ERROR, rule_id_, rule_name_, "EXPACT", msg, time_range);
|
||
|
||
if (alarm_task.alarmed) {
|
||
last_alarm_time_ -= minutes(5);
|
||
alarm_poster_.alarm(alarm_task, &last_alarm_time_);
|
||
}
|
||
exp_wrong_is_alarmed_ = true;
|
||
}
|
||
return out_alarm;
|
||
}
|
||
/*----规则配置参数检查----end*/
|
||
|
||
/*----定时载入置信区间---错峰调用--start*/
|
||
if (now_time_ - last_load_time_ >
|
||
(minutes(30) + save_interval_ms_ + rule_state_update_interval_ms_)) {
|
||
this->reload_ci_dist();
|
||
}
|
||
/*----定时载入置信区间-----end*/
|
||
|
||
try {
|
||
this->refresh_now_time();
|
||
// 根据数据来源种类，决定最后的执行过程
|
||
switch (data_source_) {
|
||
// 如果是共享内存，只需要执行当前周期的数据和表达式
|
||
// 并判断是否报警
|
||
case DataSource::MEMORY: {
|
||
try {
|
||
expr_engine_->refreshFromMemory(now_time_, query_time_range_);
|
||
} catch (const std::exception &e) {
|
||
logger_->Error() << rule_id_ << ":refreshFromMemory()异常！"
|
||
<< e.what() << ",location:" << BOOST_CURRENT_LOCATION
|
||
<< endl;
|
||
return out_alarm;
|
||
}
|
||
if (this->refresh_counts_ < 11) {
|
||
/*首次载入，刷mm_vars 11次 刷新mm_vars[pv1_10]*/
|
||
this->refresh_counts_++;
|
||
} else {
|
||
out_alarm = mon_proc();
|
||
}
|
||
} break;
|
||
// 如果是ihyperDB
|
||
case DataSource::IHDB:
|
||
// 需要先重新载入ihyperDB的缓存数据
|
||
refresh_ihd_cache();
|
||
// 然后刷以此把缓存中的数据取出
|
||
for (auto i = 0; i < queried_data_.rows(); i++) {
|
||
expr_engine_->refreshFromIhdRow(i, queried_data_, queried_time_, now_time_, query_time_range_);
|
||
if (!out_alarm.alarmed) {
|
||
auto tmp = mon_proc();
|
||
if (tmp.alarmed) {
|
||
out_alarm = (tmp);
|
||
}
|
||
}
|
||
}
|
||
break;
|
||
default:
|
||
break;
|
||
}
|
||
} catch (const std::exception &e) {
|
||
gb_logger_->log_exception(e);
|
||
}
|
||
return out_alarm;
|
||
}
|
||
|
||
std::vector<AlarmInfo> ExpBase::exec_task(mix_cc::time_range_t time_range) {
|
||
|
||
std::vector<AlarmInfo> out_alarms;
|
||
/*统计类，需要单独处理数据*/
|
||
if (this->exp_type_ == ExpType::Bound ||
|
||
this->exp_type_ == ExpType::CondBound ||
|
||
this->exp_type_ == ExpType::BoundHoldTime) {
|
||
this->refresh_counts_ = 0;
|
||
try {
|
||
/**
|
||
* 1.获取样本id，数据查询时间范围，查数据
|
||
* 2.计算参数
|
||
* 3.存数据至 db2 T_SAMPLE_STAT表
|
||
* 4.存样本管理 db2 T_SAMPLE_MAG
|
||
*/
|
||
logger_->Debug()
|
||
<< "stime:"
|
||
<< mix_cc::mix_time_t(time_range.get_left()).to_formatted_time()
|
||
<< ",endtime:"
|
||
<< mix_cc::mix_time_t(time_range.get_right()).to_formatted_time()
|
||
<< endl;
|
||
std::string sample_id = this->get_id(time_range);
|
||
stat_collector_.ensureInitialized();
|
||
|
||
logger_->Debug() << "reset_data()------1" << endl;
|
||
if ((time_range.get_right() - time_range.get_left()) <
|
||
this->query_interval_time_) {
|
||
this->query_time_range_ = time_range;
|
||
task_mon_pro(); /*累计数据*/
|
||
} else {
|
||
for (auto t = time_range.get_left(); t < time_range.get_right();
|
||
t += query_interval_time_) {
|
||
this->query_time_range_.set_left(t);
|
||
if (time_range.get_right() - t > query_interval_time_) {
|
||
this->query_time_range_.set_right(t + query_interval_time_);
|
||
} else {
|
||
this->query_time_range_.set_right(time_range.get_right());
|
||
}
|
||
task_mon_pro(); /*累计数据*/
|
||
}
|
||
}
|
||
logger_->Debug() << "reset_data()------1" << endl;
|
||
stat_collector_.resetData(); /*参数重置*/
|
||
logger_->Debug() << "reset_data()------2" << endl;
|
||
logger_->Debug() << "task_seq:" << task_seq << std::endl;
|
||
auto &data_record = TaskShm::TaskRecordPtr.get()
|
||
->
|
||
operator[](exp_type_ * 1000 + task_seq)
|
||
.data_record;
|
||
logger_->Debug() << "dataSize:" << data_record.size() << std::endl;
|
||
for (size_t j = 0; j < data_record.size(); j++) {
|
||
double dataJ = data_record[j];
|
||
stat_collector_.distAdd(dataJ);
|
||
}
|
||
logger_->Debug() << "reset_data()------3" << endl;
|
||
auto store_res = stat_collector_.taskStoreDb2(sample_id);
|
||
logger_->Debug() << "reset_data()------4" << endl;
|
||
this->sample_result_ = stat_collector_.getSampleStatStr();
|
||
this->update_t_sample_mag(store_res);
|
||
this->alarm_poster_.zmqp_send(912, this->sample_result_);
|
||
TaskShm::TaskRecordPtr.get()->erase(exp_type_ * 1000 + task_seq);
|
||
logger_->Debug() << "|" << rule_id_ << "|" << rule_name_ << "|"
|
||
<< "统计完成！info:" << sample_result_ << std::endl;
|
||
} catch (const std::exception &e) {
|
||
logger_->Error() << "ExpBase::exec_task:" << e.what()
|
||
<< ",location:" << BOOST_CURRENT_LOCATION << endl;
|
||
return out_alarms;
|
||
}
|
||
return out_alarms;
|
||
}
|
||
|
||
/*其他算法*/
|
||
try {
|
||
// 刷一遍 mmvar 防止p1_3数据错误带来的问题
|
||
now_time_ = time_range.get_left() - query_interval_time_;
|
||
if (this->delay_time_ > this->ihd_min_time_particles_ &&
|
||
(this->delay_time_.count() % this->ihd_min_time_particles_.count() ==
|
||
0)) {
|
||
refresh_ihd_cache(this->delay_time_);
|
||
} else {
|
||
refresh_ihd_cache();
|
||
}
|
||
for (auto i = 0; i < queried_data_.rows(); i++) {
|
||
expr_engine_->refreshFromIhdRow(i, queried_data_, queried_time_, now_time_, query_time_range_);
|
||
}
|
||
// 对每个ihdb 查询周期的数据，进行处理
|
||
for (auto now_time = time_range.get_left();
|
||
now_time <= time_range.get_right(); now_time += query_interval_time_) {
|
||
this->now_time_ = now_time;
|
||
if (this->delay_time_ > this->ihd_min_time_particles_ &&
|
||
(this->delay_time_.count() % this->ihd_min_time_particles_.count() ==
|
||
0)) {
|
||
refresh_ihd_cache(this->delay_time_);
|
||
} else {
|
||
refresh_ihd_cache();
|
||
}
|
||
for (auto i = 0; i < queried_data_.rows(); i++) {
|
||
expr_engine_->refreshFromIhdRow(i, queried_data_, queried_time_, now_time_, query_time_range_);
|
||
auto tmp = mon_proc();
|
||
if (tmp.alarmed) {
|
||
out_alarms.push_back(tmp);
|
||
}
|
||
}
|
||
}
|
||
gb_logger_->log_info("本次测试报警数量：" +
|
||
std::to_string(out_alarms.size()));
|
||
auto msg =
|
||
rule_name_ + "本次测试报警数量：" + std::to_string(out_alarms.size());
|
||
auto alarm_task = utility::build_alarm_info(
|
||
MsgLevel::INFO, rule_id_, rule_name_, "EXPACT", msg, time_range);
|
||
out_alarms.clear();
|
||
out_alarms.push_back(alarm_task);
|
||
} catch (const std::exception &e) {
|
||
gb_logger_->log_exception(e);
|
||
}
|
||
|
||
return out_alarms;
|
||
}
|
||
|
||
AlarmInfo ExpBase::mon_proc() {
|
||
expr_engine_->autoResetFunVars();
|
||
try {
|
||
return doMonProc();
|
||
} catch (const std::exception &e) {
|
||
gb_logger_->log_error("rule_name:" + rule_name_ + ",error:" + e.what());
|
||
return AlarmInfo{};
|
||
}
|
||
}
|
||
bool ExpBase::get_cycled_cron() {
|
||
auto now = system_clock::now();
|
||
if ((now - last_run_start_time_) > std::chrono::minutes(1)) {
|
||
last_run_start_time_ = now;
|
||
} else {
|
||
return false;
|
||
}
|
||
return true;
|
||
}
|
||
|
||
mix_cc::json ExpBase::exec_cron() {
|
||
if (this->get_cycled_cron()) {
|
||
if (this->exp_type_ == ExpType::Bound ||
|
||
this->exp_type_ == ExpType::CondBound ||
|
||
this->exp_type_ == ExpType::BoundHoldTime) {
|
||
int stat_size = this->cron_proc();
|
||
logger_->Debug() << this->rule_name_ << ",cron 统计数据量：" << stat_size
|
||
<< std::endl;
|
||
}
|
||
}
|
||
return {};
|
||
}
|
||
|
||
int ExpBase::cron_proc() {
|
||
int size_data = 0;
|
||
this->refresh_now_time();
|
||
if (exp_type_ == ExpType::Bound || exp_type_ == ExpType::CondBound ||
|
||
exp_type_ == ExpType::BoundHoldTime) {
|
||
/*只保存有上下限的,2-监控变量-上下限；4-动作反馈-上下限；5-监控变量-上下限-持续*/
|
||
this->rule_stat_.stat_values.clear();
|
||
SingletonTemp<EqpStat>::GetInstance().get_stat_values(this->rule_id_,
|
||
this->rule_stat_);
|
||
if (!this->rule_stat_.stat_values.empty()) {
|
||
size_data = stat_collector_.processCron(
|
||
this->rule_stat_.stat_values, now_time_,
|
||
CronUpdateDelay, last_load_time_);
|
||
}
|
||
this->rule_stat_.stat_values.clear();
|
||
this->rule_stat_.stat_values.shrink_to_fit();
|
||
}
|
||
return size_data;
|
||
}
|
||
|
||
int ExpBase::reload_config_exp_feedback() {
|
||
int res = 0;
|
||
// 获取feedback信息
|
||
if (rule_json_.at("function").at("action_start").contains("param")) {
|
||
//是否保持
|
||
keep_mode_ = std::stoi(rule_json_.at("function")
|
||
.at("action_start")
|
||
.at("param")
|
||
.at("hold")
|
||
.at("value")
|
||
.get<std::string>());
|
||
logger_->Debug() << "keep:" << keep_mode_ << endl;
|
||
//超时时间
|
||
time_out_ = milliseconds(std::stoi(rule_json_.at("function")
|
||
.at("action_end")
|
||
.at("param")
|
||
.at("timeout")
|
||
.at("value")
|
||
.get<std::string>()));
|
||
logger_->Debug() << "init timeout:" << time_out_.count() << endl;
|
||
// 如果超时时间小于3分钟，则默认为超时时间为3分钟
|
||
if (time_out_ < minutes(3) && time_out_ != milliseconds(-32768)) {
|
||
time_out_ = minutes(3);
|
||
}
|
||
logger_->Debug() << "timeout:" << time_out_.count() << endl;
|
||
|
||
auto tmp_exp = string(rule_json_.at("function")
|
||
.at("action_end")
|
||
.at("value")
|
||
.get<std::string>());
|
||
exp_str_ = get_macro_replaced_exp(tmp_exp);
|
||
|
||
if (exp_str_ != "") {
|
||
int reg_ret = expr_engine_->registerExpression("feedback", exp_str_);
|
||
if (reg_ret != 0) {
|
||
this->error_code_list_.push_back(
|
||
{ErrorType::CalError, ErrorLocation::FBExp});
|
||
return -1;
|
||
}
|
||
}
|
||
|
||
} else {
|
||
time_out_ = milliseconds(600000);
|
||
}
|
||
// 获取监控变量信息
|
||
auto tmp_exp =
|
||
rule_json_.at("function").at("result").at("value").get<std::string>();
|
||
exp_str_ = get_macro_replaced_exp(tmp_exp);
|
||
|
||
if (exp_str_ != "") {
|
||
int reg_ret = expr_engine_->registerExpression("result", exp_str_);
|
||
if (reg_ret != 0) {
|
||
this->error_code_list_.push_back(
|
||
{ErrorType::CalError, ErrorLocation::ResultExp});
|
||
res += reg_ret;
|
||
}
|
||
}
|
||
|
||
if (exp_str_.find("time", 0) != string::npos &&
|
||
exp_type_ == ExpType::CondBound) {
|
||
fb_fsm_.setTimeMode(true);
|
||
rule_stat_.unit = "ms";
|
||
} else {
|
||
fb_fsm_.setTimeMode(false);
|
||
}
|
||
return res;
|
||
}
|
||
|
||
// 重新载入数据源配置
|
||
int ExpBase::reload_config_data_source() {
|
||
int res = 0;
|
||
if (!rule_json_.contains("datasource")) {
|
||
data_source_ = 0;
|
||
logger_->Debug()
|
||
<< "默认数据源为iHyerDB！ data source[0:iHyerDB，1:memory]："
|
||
<< data_source_ << endl;
|
||
return 0;
|
||
}
|
||
try {
|
||
data_source_ =
|
||
std::stoi(rule_json_.at("datasource").at("value").get<std::string>());
|
||
logger_->Debug() << "data source[0:iHyerDB，1:memory]:" << data_source_
|
||
<< endl;
|
||
} catch (const std::exception &e) {
|
||
gb_logger_->log_error(std::string("ExpBase::reload_config_data_source()") +
|
||
e.what());
|
||
this->error_code_list_.push_back(
|
||
{ErrorType::Empty, ErrorLocation::DataSource});
|
||
return -1;
|
||
}
|
||
|
||
return res;
|
||
}
|
||
|
||
int ExpBase::reload_config_exp_act() {
|
||
// 根据key，对不同版本的算法都进行取值
|
||
int res = 0;
|
||
if (rule_json_.at("function").contains("action_start")) {
|
||
auto tmp_exp = rule_json_.at("function")
|
||
.at("action_start")
|
||
.at("value")
|
||
.get<std::string>();
|
||
exp_str_ = get_macro_replaced_exp(tmp_exp);
|
||
feedback_mode_ = true;
|
||
|
||
} else if (rule_json_.at("function").contains("result")) {
|
||
if (rule_json_.at("function").contains("filter_exp")) {
|
||
auto tmp_exp = rule_json_.at("function")
|
||
.at("filter_exp")
|
||
.at("value")
|
||
.get<std::string>();
|
||
exp_str_ = get_macro_replaced_exp(tmp_exp);
|
||
feedback_mode_ = false;
|
||
if (exp_str_ != "") {
|
||
int reg_ret = expr_engine_->registerExpression("feedback", exp_str_);
|
||
if (reg_ret != 0) {
|
||
this->error_code_list_.push_back(
|
||
{ErrorType::CalError, ErrorLocation::FBExp});
|
||
return -1;
|
||
}
|
||
} else {
|
||
logger_->Error() << "filter_exp exp_feedback_:"
|
||
<< "为空"
|
||
<< ",location:" << BOOST_CURRENT_LOCATION << endl;
|
||
this->error_code_list_.push_back(
|
||
{ErrorType::CalError, ErrorLocation::FBExp});
|
||
return -1;
|
||
}
|
||
}
|
||
auto tmp_exp =
|
||
rule_json_.at("function").at("result").at("value").get<std::string>();
|
||
exp_str_ = get_macro_replaced_exp(tmp_exp);
|
||
|
||
feedback_mode_ = false;
|
||
}
|
||
|
||
if (exp_str_ != "") {
|
||
int reg_ret = expr_engine_->registerExpression("act", exp_str_);
|
||
if (reg_ret != 0) {
|
||
this->error_code_list_.push_back(
|
||
{ErrorType::CalError, ErrorLocation::ActExp});
|
||
return -1;
|
||
}
|
||
}
|
||
if (rule_json_.at("output").contains("error")) {
|
||
try {
|
||
error_str_ =
|
||
rule_json_.at("output").at("error").at("value").get<std::string>();
|
||
} catch (const std::exception &e) {
|
||
logger_->Error() << "output,error出错：" << e.what()
|
||
<< ",location:" << BOOST_CURRENT_LOCATION << endl;
|
||
error_message_str_ += "报警内容格式异常！";
|
||
return -1;
|
||
}
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
int ExpBase::reload_config_up_down() {
|
||
try {
|
||
if (rule_json_.at("function").at("result").contains("param")) {
|
||
if (rule_json_.at("function").at("result").at("param").contains("unit")) {
|
||
unit_ = rule_json_.at("function")
|
||
.at("result")
|
||
.at("param")
|
||
.at("unit")
|
||
.at("value")
|
||
.get<std::string>();
|
||
}
|
||
rule_stat_.unit = unit_;
|
||
limit_down_ = std::stod(rule_json_.at("function")
|
||
.at("result")
|
||
.at("param")
|
||
.at("limit_down")
|
||
.at("value")
|
||
.get<std::string>());
|
||
limit_up_ = std::stod(rule_json_.at("function")
|
||
.at("result")
|
||
.at("param")
|
||
.at("limit_up")
|
||
.at("value")
|
||
.get<std::string>());
|
||
this->rule_stat_.current_value = (limit_down_ + limit_up_) / 2;
|
||
if (-32768 == (int)limit_down_ && (int)limit_up_ != (int)limit_down_) {
|
||
this->rule_stat_.current_value = 0;
|
||
this->detect_mode_ = DetectMode::OnlyRight;
|
||
} else if ((-32768 == (int)limit_up_ || 32768 == (int)limit_up_ ||
|
||
32767 == (int)limit_up_) &&
|
||
(int)limit_up_ != (int)limit_down_) {
|
||
this->detect_mode_ = DetectMode::OnlyLeft;
|
||
this->rule_stat_.current_value = limit_down_ + 1;
|
||
} else if (-32768 == (int)limit_up_ &&
|
||
(int)limit_up_ == (int)limit_down_) {
|
||
this->detect_mode_ = DetectMode::ErrorMode;
|
||
}
|
||
bound_checker_.setLimits(limit_down_, limit_up_);
|
||
///共享内存参数
|
||
logger_->Info() << rule_name_
|
||
<< ",detect_mode_[0-双侧，1-仅left；2-仅right；3-错误]:"
|
||
<< static_cast<int>(bound_checker_.detectMode()) << ",limit_down:" << limit_down_ << ","
|
||
<< "limit_up:" << limit_up_ << std::endl;
|
||
this->rule_stat_.limit_down = limit_down_;
|
||
this->rule_stat_.limit_up = limit_up_;
|
||
}
|
||
if (rule_json_.contains("self_learning")) {
|
||
this->dist_mode_ = rule_json_.at("self_learning").at("mode").get<int>();
|
||
this->is_learning_ =
|
||
rule_json_.at("self_learning").at("is_learning").get<bool>();
|
||
logger_->Debug() << "is_learning_:" << is_learning_ << std::endl;
|
||
}
|
||
} catch (const std::exception &e) {
|
||
logger_->Error() << e.what() << ",location:" << BOOST_CURRENT_LOCATION
|
||
<< endl;
|
||
return -1;
|
||
}
|
||
|
||
return 0;
|
||
}
|
||
int ExpBase::reload_config_up_down_hold_time() {
|
||
try {
|
||
if (rule_json_.at("function")
|
||
.at("result")
|
||
.at("param")
|
||
.contains("hold_time")) {
|
||
hold_time_ = milliseconds(std::stoi(rule_json_.at("function")
|
||
.at("result")
|
||
.at("param")
|
||
.at("hold_time")
|
||
.at("value")
|
||
.get<std::string>()));
|
||
logger_->Info() << rule_name_ << "hold_time:" << hold_time_.count()
|
||
<< std::endl;
|
||
}
|
||
} catch (const std::exception &e) {
|
||
logger_->Error() << e.what() << ",location:" << BOOST_CURRENT_LOCATION
|
||
<< endl;
|
||
return -1;
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
void ExpBase::set_last_alarm_time(TimePoint time_point) {
|
||
this->refresh_counts_ = 0;
|
||
if (this->is_usable_) {
|
||
expr_engine_->firstFill(data_source_, now_time_, query_time_range_);
|
||
}
|
||
AlgBase::set_last_alarm_time(time_point);
|
||
}
|
||
|
||
void ExpBase::set_usable(bool usable) {
|
||
this->refresh_counts_ = 0;
|
||
if (this->is_usable_) {
|
||
expr_engine_->firstFill(data_source_, now_time_, query_time_range_);
|
||
}
|
||
AlgBase::set_usable(usable);
|
||
}
|
||
|
||
int ExpBase::reload_ci_dist() {
|
||
if (this->exp_type_ != ExpType::Bound &&
|
||
this->exp_type_ != ExpType::CondBound &&
|
||
this->exp_type_ != ExpType::BoundHoldTime) {
|
||
return 0;
|
||
}
|
||
double newDown, newUp;
|
||
if (stat_collector_.reloadCiDist(newDown, newUp, now_time_, last_load_time_)) {
|
||
this->limit_down_ = newDown;
|
||
this->limit_up_ = newUp;
|
||
bound_checker_.setLimits(newDown, newUp);
|
||
this->rule_stat_.limit_down = limit_down_;
|
||
this->rule_stat_.limit_up = limit_up_;
|
||
this->rule_stat_.current_value = (limit_down_ + limit_up_) / 2;
|
||
logger_->Info()
|
||
<< "更新置信区间，[" << newDown << ","
|
||
<< newUp << "]"
|
||
<< ",type[0-手动设置的区间;1-在线更新的区间；2-离线分析的区间]:"
|
||
<< this->dist_mode_ << endl;
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
void ExpBase::query_ihd_data() {
|
||
vector<string> ihd_tags = m_tags;
|
||
auto queried_batch_maybe = mix_cc::ihd::make_query_batch_maybe(
|
||
ihd_tags, query_time_range_, this->ihd_min_time_particles_);
|
||
if (queried_batch_maybe.is_nothing()) {
|
||
queried_time_.clear();
|
||
queried_data_ = typename decltype(queried_batch_maybe)::type::Mat2d{};
|
||
gb_logger_->log_error("Tag点没有查到相应的信息");
|
||
return;
|
||
}
|
||
auto queried_bath = queried_batch_maybe.unsafe_get_just();
|
||
auto result_maybe = mix_cc::ihd::read_data_with_time_maybe(&queried_bath);
|
||
if (result_maybe.is_nothing()) {
|
||
queried_time_.clear();
|
||
queried_data_ = typename decltype(queried_batch_maybe)::type::Mat2d{};
|
||
gb_logger_->log_error("Tag点没有查到相应的数据");
|
||
return;
|
||
}
|
||
std::tie(queried_time_, queried_data_) = result_maybe.unsafe_get_just();
|
||
}
|
||
|
||
void ExpBase::task_mon_pro() {
|
||
/*1.查数据
|
||
*2.执行
|
||
*/
|
||
query_ihd_data(); /*查询m_tags+运行前提条件(如果有，放置在最后一列)*/
|
||
logger_->Debug() << rule_name_ << ",ihd size:" << queried_data_.size()
|
||
<< endl;
|
||
for (auto i = 0; i < queried_data_.rows(); i++) {
|
||
expr_engine_->refreshFromIhdRow(i, queried_data_, queried_time_, now_time_, query_time_range_);
|
||
if (this->refresh_counts_ < 3) {
|
||
this->refresh_counts_++;
|
||
} else {
|
||
auto res = task_base_proc();
|
||
if (res.is_valid && this->task_prr(i)) {
|
||
TaskShm::TaskRecordPtr.get()
|
||
->
|
||
operator[](exp_type_ * 1000 + task_seq)
|
||
.data_record.push_back(res.value);
|
||
stat_collector_.runningStatAdd(res.value);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
string ExpBase::get_id(mix_cc::time_range_t time_range) {
|
||
this->sample_id_ =
|
||
this->rule_id_.substr(3, 21) + "_" +
|
||
std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(
|
||
time_range.get_left().time_since_epoch())
|
||
.count()) +
|
||
"_" +
|
||
std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(
|
||
time_range.get_right().time_since_epoch())
|
||
.count());
|
||
return this->sample_id_;
|
||
}
|
||
|
||
TaskReturnType ExpBase::task_base_proc() {
|
||
TaskReturnType task_return_data;
|
||
double result_value = 0;
|
||
try {
|
||
// 获得是否满足前提条件表达式
|
||
result_value = expr_engine_->evaluate("act");
|
||
bool act_triggered = static_cast<bool>(result_value);
|
||
// 检测是否是表达式-反馈模式
|
||
if (feedback_mode_) {
|
||
// 使用 FbStateMachine 替代旧的手动状态管理
|
||
auto [fbState, needFunReset] = fb_fsm_.update(
|
||
act_triggered, now_time_, expr_engine_->vars(), m_tags.size());
|
||
|
||
if (needFunReset) {
|
||
expr_engine_->markFunVarsNeedReset();
|
||
}
|
||
|
||
if (fbState == FbState::Started || fbState == FbState::NotHold) {
|
||
return task_return_data;
|
||
}
|
||
|
||
if (fbState == FbState::InProgress) {
|
||
bool fbCond = expr_engine_->evaluateBool("feedback");
|
||
bool done =
|
||
fb_fsm_.checkFeedback(fbCond, now_time_, expr_engine_->vars());
|
||
if (done) {
|
||
result_value = expr_engine_->evaluate("result");
|
||
expr_engine_->printVars();
|
||
// 并且数据合法
|
||
if (!std::isnan(result_value)) {
|
||
task_return_data.is_valid = true;
|
||
task_return_data.value = result_value;
|
||
return task_return_data;
|
||
}
|
||
}
|
||
return task_return_data;
|
||
}
|
||
|
||
if (fbState == FbState::Timeout) {
|
||
return task_return_data;
|
||
}
|
||
|
||
// Idle or Done (terminal states that auto-reset) — nothing to do
|
||
return task_return_data;
|
||
}
|
||
// 表达式-样本，无需反馈
|
||
else if (!std::isnan(result_value)) {
|
||
try {
|
||
if (expr_engine_->evaluateBool("feedback")) {
|
||
task_return_data.is_valid = true;
|
||
task_return_data.value = result_value;
|
||
logger_->Debug() << "实时值：" << result_value << endl;
|
||
expr_engine_->printVars();
|
||
}
|
||
} catch (...) {
|
||
task_return_data.is_valid = true;
|
||
task_return_data.value = result_value;
|
||
}
|
||
|
||
return task_return_data;
|
||
}
|
||
|
||
} catch (const std::exception &e) {
|
||
this->logger_->Error() << "ExpBase::task_base_proc():" << e.what()
|
||
<< ",location:" << BOOST_CURRENT_LOCATION << endl;
|
||
return task_return_data;
|
||
}
|
||
return task_return_data;
|
||
}
|
||
|
||
int ExpBase::update_t_sample_mag(bool vlid) {
|
||
T_SAMPLE_MAG tsm;
|
||
auto update_ret = exec<db2_t, size_t>(
|
||
update(tsm)
|
||
.set(tsm.result() = this->sample_result_,
|
||
tsm.type() = this->sample_type_, tsm.usable() = 0,
|
||
tsm.starttime() =
|
||
mix_cc::mix_time_t(this->task_time_range_.get_left()),
|
||
tsm.endtime() =
|
||
mix_cc::mix_time_t(this->task_time_range_.get_right()),
|
||
tsm.verify() = (vlid ? 1 : 3))
|
||
.where(tsm.sampleid() = this->sample_id_,
|
||
tsm.ruleId() = this->rule_id_));
|
||
if (!update_ret.is_nothing()) {
|
||
auto res = update_ret.unsafe_get_just();
|
||
if (res == 0) {
|
||
auto inset_ret = exec<db2_t, size_t>(insert_into(tsm).set(
|
||
tsm.ruleId() = this->rule_id_, tsm.result() = this->sample_result_,
|
||
tsm.type() = this->sample_type_, tsm.sampleid() = this->sample_id_,
|
||
tsm.usable() = 0,
|
||
tsm.starttime() =
|
||
mix_cc::mix_time_t(this->task_time_range_.get_left()),
|
||
tsm.endtime() =
|
||
mix_cc::mix_time_t(this->task_time_range_.get_right()),
|
||
tsm.verify() = (vlid ? 1 : 3)));
|
||
if (inset_ret.is_nothing()) {
|
||
logger_->Error() << "T_SAMPLE_MAG,插入数据失败" << std::endl;
|
||
return -2;
|
||
}
|
||
}
|
||
} else {
|
||
gb_logger_->log_error("T_SAMPLE_MAG 表更新失败:" + rule_name_);
|
||
return -1;
|
||
}
|
||
|
||
gb_logger_->log_info(
|
||
"T_SAMPLE_MAG更新:" + rule_name_ +
|
||
",vlid[true-计算成功；false-计算失败]:" + std::to_string(vlid)) +
|
||
",结果:" + this->sample_result_;
|
||
return 0;
|
||
}
|
||
|
||
bool ExpBase::detect_up_down(const double &value) {
|
||
switch (this->detect_mode_) {
|
||
case DetectMode::Default:
|
||
return value < this->limit_down_ || value > limit_up_;
|
||
break;
|
||
case DetectMode::OnlyLeft:
|
||
return value < limit_down_;
|
||
break;
|
||
case DetectMode::OnlyRight:
|
||
return value > limit_up_;
|
||
break;
|
||
default:
|
||
return false;
|
||
break;
|
||
}
|
||
return false;
|
||
}
|
||
|
||
bool ExpBase::task_prr(int row) {
|
||
if (this->prr_ == 1) {
|
||
expr_engine_->refreshFromIhdRow(row, queried_data_, queried_time_, now_time_, query_time_range_);
|
||
bool prr_result = expr_engine_->evaluateBool("pre_result");
|
||
return prr_result;
|
||
}
|
||
return true;
|
||
}
|
||
|
||
void ExpBase::reset_dev_data() {
|
||
if (glob_process_type == ProcessType::kMon) {
|
||
if (exp_type_ == ExpType::Bound || exp_type_ == ExpType::CondBound ||
|
||
exp_type_ == ExpType::BoundHoldTime) {
|
||
stat_collector_.reset(rule_id_);
|
||
this->reload_config_up_down(); /*7.上下限*/
|
||
this->reload_ci_dist();
|
||
this->last_load_time_ = chrono::system_clock::now();
|
||
SingletonTemp<EqpStat>::GetInstance().update_static(this->rule_id_,
|
||
false);
|
||
}
|
||
}
|
||
}
|
||
|
||
void ExpBase::save_rule_norm_data() {
|
||
if (feedback_mode_ && fb_fsm_.currentState() == FbState::Done ||
|
||
((exp_type_ == ExpType::Bound || exp_type_ == ExpType::BoundHoldTime) &&
|
||
filter_flag_ == true)) {
|
||
int64_t default_lt =
|
||
mix_cc::mix_time_t(now_time_).to_milliseconds() - 1000 * 5;
|
||
int64_t stime = mm_vars["stime"];
|
||
int64_t lt = std::min(default_lt, stime);
|
||
data_info_.update(lt, mix_cc::mix_time_t(now_time_).to_milliseconds());
|
||
} else {
|
||
if (con_monitor_.getCurrentState()) {
|
||
AlgBase::save_rule_norm_data();
|
||
} else {
|
||
/*查不到数据，无法保存*/
|
||
data_info_.update(mix_cc::mix_time_t(now_time_).to_milliseconds(),
|
||
mix_cc::mix_time_t(now_time_).to_milliseconds());
|
||
}
|
||
}
|
||
}
|
||
|
||
bool ExpBase::get_prr() {
|
||
if (this->prr_ == 1) {
|
||
// 变量刷新已由 exec_mon() 中的 refreshFromMemory 完成，此处只需求值
|
||
bool prr_result = expr_engine_->evaluateBool("pre_result");
|
||
this->now_prr_ = prr_result;
|
||
|
||
if (!this->now_prr_) {
|
||
fb_fsm_.forceReset();
|
||
}
|
||
return prr_result;
|
||
}
|
||
now_prr_ = true;
|
||
return true;
|
||
}
|