#include #include #include #include #include #include #include #include #include #include #include #include #include #include extern ProcessType glob_process_type; AlgBase::AlgBase(const string &name, const mix_cc::json &rule_json, const string &ruleId) : rule_id_(ruleId), rule_name_(name), rule_json_(rule_json), gb_logger_(std::make_unique("Logger:" + name)), logger_(std::make_unique("AlgBase", AUTO_CATCH_PID)), last_alarm_time_( system_clock::now() - (minutes(std::max(5, CMemVar::Const()->AlarmIntervalHours * 60) - 1) + 50s)), con_monitor_(1), delay_time_(50ms) {} AlgBase::~AlgBase() { } void AlgBase::logReset(int task_seq) { task_seq = task_seq; logger_.reset(new LOG("task:" + rule_name_, AUTO_CATCH_PID)); } int AlgBase::init() { this->refresh_now_time(); last_save_start_time_ = this->now_time_; last_rule_state_updae_start_time_ = this->now_time_; last_heart_beat_start_time_ = this->now_time_; // 使用修正后的XorShift128Plus XorShift128Plus xsrng(rule_id_); // 生成随机偏移 int r1 = xsrng.get_int_fast(0, 101); int r2 = xsrng.get_int_fast(0, 1001); int r3 = xsrng.get_int_fast(0, 11); // 计算总偏移 auto offset_ms = std::chrono::milliseconds(r1 * 10 + r2 * 5); auto offset_ms2 = std::chrono::milliseconds(r1 * 2 + r3 * 10); // 更新保存间隔 save_interval_ms_ += offset_ms; rule_state_update_interval_ms_ += offset_ms2; logger_->Debug() << "ruleid:" << rule_id_ << "save_interval_ms_:" << save_interval_ms_.count() << ";rule_state_update_interval_ms_:" << rule_state_update_interval_ms_.count() << std::endl; /*初始化保存正常数据的数据--使规则开始运行,不保存----start*/ data_info_.update(mix_cc::mix_time_t(now_time_).to_milliseconds(), mix_cc::mix_time_t(now_time_).to_milliseconds()); SingletonTemplate::GetInstance().insert(rule_id_, &data_info_); /*初始化保存正常数据的数据--使规则开始运行,不保存----end*/ int res = 0; is_usable_ = false; if (rule_json_.contains("before_exec")) { try { this->prr_ = std::stoi(this->rule_json_.at("before_exec").get()); } catch (const std::exception &e) { logger_->Error() << e.what() << "location:" << BOOST_CURRENT_LOCATION << endl; this->prr_ = this->rule_json_.at("before_exec").get(); } logger_->Debug() << rule_name_ << "的执行前置条件[0-无;1-表达式]:" << prr_ << endl; } try { res += this->reload_config_tag(); data_info_.init(m_tags); try { int ms = std::stoi(rule_json_.at("trigger").at("value").get()); if (ms < 20) { ms = 20; } this->delay_time_ = milliseconds(ms); exp_mpdule_ptr_ = std::make_unique(mm_vars, m_tags, is_exp_alg_); if (this->prr_ == 1) { string exp_str = ""; if (rule_json_.at("function").contains("pre_result")) { auto tmp_exp = rule_json_.at("function") .at("pre_result") .at("value") .get(); logger_->Info() << "tmp_exp:" << tmp_exp << std::endl; exp_str = get_macro_replaced_exp(tmp_exp); logger_->Info() << "exp_str:" << exp_str << std::endl; exp_mpdule_ptr_->add_exp("pre_result", exp_str); logger_->Info() << "pre_result:" << exp_mpdule_ptr_->get_exp_str("pre_result") << std::endl; } logger_->Debug() << "ruleid:" << this->rule_id_ << ",rulename:" << this->rule_name_ << ",执行前提表达式:" << exp_str << std::endl; } } catch (const std::exception &e) { this->error_code_list_.push_back( {ErrorType::Empty, ErrorLocation::ExeCyc}); logger_->Error() << e.what() << "location:" << BOOST_CURRENT_LOCATION << endl; res += -1; } last_run_start_time_ = system_clock::now() - this->delay_time_; } catch (const std::exception &e) { logger_->Error() << e.what() << "location:" << BOOST_CURRENT_LOCATION << endl; res += -1; } logger_->Debug() << "|" << rule_id_ << "|alg init 完成 90%! " << ",last_alarm_time_:" << mix_cc::mix_time_t(last_alarm_time_).to_formatted_time() << std::endl; this->rule_stat_.limit_down = 0; this->rule_stat_.limit_up = 0; this->rule_stat_.current_value = 0; if (glob_process_type == ProcessType::kMon) { this->update_map_rule(); } logger_->Debug() << "|" << rule_id_ << "|alg init 完成! " << std::endl; return res; } /** * @brief 保证间隔delay_time_以上执行一次 * @return true * @return false */ bool AlgBase::get_cycled() { auto now = system_clock::now(); if ((now - last_run_start_time_) > this->delay_time_) { last_run_start_time_ = now; } else { return false; } return true; } // 载入tag点配置信息 int AlgBase::reload_config_tag() { const auto tag_dict = rule_json_["tags"]; m_tags.resize(tag_dict.size()); this->rule_stat_.items.resize(tag_dict.size()); for (auto x : tag_dict.items()) { int item_index = stoi(x.key().substr(3)); logger_->Debug() << "item_index:" << item_index << ",x.key():" << x.key() << std::endl; if (item_index > tag_dict.size()) { logger_->Error() << "|tag序号异常|" << ",ruleid:" << rule_id_ << ",rule_name:" << rule_name_ << ",tag_dict.size():" << tag_dict.size() << ",item_index:" << item_index << std::endl; return -1; } auto itemi = x.value().at("value").get(); logger_->Debug() << "tag" << item_index << ":" << itemi << std::endl; m_tags[item_index - 1] = string(CMemVar::Const()->UnitNo) + "_" + itemi; this->rule_stat_.items[item_index - 1] = itemi; } var_cache_.init(m_tags.size(), pv_num_); if (m_tags.empty()) { error_code_list_.push_back({ErrorType::Empty, ErrorLocation::Tags}); logger_->Debug() << "m_tags为空!" << endl; return -1; } return 0; } // 获取数据源 int AlgBase::get_data_source() { return this->data_source_; } // 设置上次报警时间,这个时间是由执行进行类决定的, // 因为最终报警是否被发送也是由该类决定的 void AlgBase::set_last_alarm_time(TimePoint time_point) { this->last_alarm_time_ = time_point; } void AlgBase::exec_mon_call() { if (this->get_cycled()) { if (this->get_usable() && get_prr()) { auto alarm = this->exec_mon(); this->con_monitor_.update(alarm.alarmed); if (alarm.alarmed) { logger_->Debug() << alarm.content << endl; alarm_poster_.alarm(alarm, &last_alarm_time_); exp_mpdule_ptr_->fun_reset(); } else { if (get_save_data_cycled()) { this->save_rule_norm_data(); } } } else { this->con_monitor_.update(true); } if (get_update_rule_stat_cycled()) { this->update_map_rule(); } } if (get_heart_beat_log_cycled()) { logger_->Debug() << "heart_beat_log----active!" << std::endl; } } void AlgBase::save_rule_norm_data() { data_info_.update(mix_cc::mix_time_t(now_time_).to_milliseconds() - 1000 * 5, mix_cc::mix_time_t(now_time_).to_milliseconds()); SingletonTemplate::GetInstance().insert(rule_id_, &data_info_); } void AlgBase::exec_task_call(const mix_cc::time_range_t &time_range) { gb_logger_->log_info("task-开始执行:" + rule_name_); this->task_time_range_ = time_range; auto alarms = this->exec_task(time_range); for (auto alarm : alarms) { if (alarm.alarmed) { alarm_poster_.alarm(alarm, "task"); } } gb_logger_->log_info("task-执行完成:" + rule_name_); } /** * @brief 区别需要统计样本的模型 */ void AlgBase::exec_normal_task_call(const mix_cc::time_range_t &time_range) { gb_logger_->log_info("task-开始执行:" + rule_name_); this->task_time_range_ = time_range; auto alarm = this->exec_normal_task(time_range); if (alarm.alarmed) { this->set_last_alarm_time(time_range.get_left()); alarm_poster_.alarm(alarm, &last_alarm_time_); } gb_logger_->log_info("task-执行完成:" + rule_name_); } void AlgBase::exec_cron_call() { this->exec_cron(); } void AlgBase::set_usable(bool usable) { is_usable_ = usable; logger_->Debug() << rule_name_ << " usable:" << usable << endl; } mix_cc::json AlgBase::exec_cron() { return mix_cc::json{}; } // 刷新ihyperDB缓存,把ihyperDB中的数据转移到当前程序缓存中 // 注意该函数是先更改query_time_range_,再查询ihd的数据 int AlgBase::refresh_ihd_cache() { try { // 如果数据时间包含空时间 if (query_time_range_.get_left() == TimePoint() || query_time_range_.get_right() == TimePoint()) { // 则设置时间为当前时间前5秒 query_time_range_.set_left(now_time_ - 5s); query_time_range_.set_right(now_time_); } else { // 否则把开始时间设置为上次查询的结束时间 query_time_range_.set_left(query_time_range_.get_right()); // 结束时间设置为当前时间 query_time_range_.set_right(now_time_); } // 进行查询 auto queried_batch_maybe = mix_cc::ihd::make_query_batch_maybe( m_tags, query_time_range_, 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 -1; } 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 -1; } std::tie(queried_time_, queried_data_) = result_maybe.unsafe_get_just(); } catch (const std::exception &e) { logger_->Debug() << e.what() << ",locations:" << BOOST_CURRENT_LOCATION << endl; gb_logger_->log_exception(e); } return 0; } // query_time_range_ 会被刷新 l0 = r1 ,r0 = now int AlgBase::refresh_ihd_cache(TimeDur delay_time) { try { // 如果数据时间包含空时间 if (query_time_range_.get_left() == TimePoint() || query_time_range_.get_right() == TimePoint()) { query_time_range_.set_left(now_time_ - 10 * delay_time); query_time_range_.set_right(now_time_); } else { // 否则把开始时间设置为上次查询的结束时间 query_time_range_.set_left(query_time_range_.get_right()); // 结束时间设置为当前时间 query_time_range_.set_right(now_time_); } { gb_logger_->log_info( "ihd函数内查询开始时间:" + mix_cc::mix_time_t(query_time_range_.get_left()).to_formatted_time()); gb_logger_->log_info("ihd函数内查询结束时间:" + mix_cc::mix_time_t(query_time_range_.get_right()) .to_formatted_time()); } // 进行查询 auto queried_batch_maybe = mix_cc::ihd::make_query_batch_maybe( m_tags, query_time_range_, delay_time); if (queried_batch_maybe.is_nothing()) { queried_time_.clear(); queried_data_ = typename decltype(queried_batch_maybe)::type::Mat2d{}; gb_logger_->log_error("Tag点没有查到相应的信息"); return -1; } 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 -1; } std::tie(queried_time_, queried_data_) = result_maybe.unsafe_get_just(); } catch (const std::exception &e) { gb_logger_->log_exception(e); } return 0; } int AlgBase::refresh_now_time() { if (data_source_ == DataSource::IHDB) { // 因为ihdb存储的耗时比较长,直接查询当前的时间数据 // 可能不存在对应的存储的数据 // 所以需要把数据向前找5s now_time_ = system_clock::now() - 5s; } else { now_time_ = system_clock::now(); } return 0; } bool AlgBase::update_map_rule() { try { std::lock_guard guard(lm); SingletonTemp::GetInstance().update_display( this->rule_id_, this->rule_stat_); SingletonTemp::GetInstance().update_cold( this->rule_id_, this->rule_stat_); return true; } catch (...) { gb_logger_->log_error(this->rule_name_ + "update_map_rule()"); return false; } } bool AlgBase::get_prr() { if (this->prr_ == 1) { exp_mpdule_ptr_->update(); bool prr_result = (bool)exp_mpdule_ptr_->get_value("pre_result"); this->now_prr_ = prr_result; return prr_result; } now_prr_ = true; return true; } int AlgBase::ihd_get_by_tag(string tag, mix_cc::time_range_t time_range) { std::vector tags; tags.push_back(tag); try { auto queried_batch_maybe = mix_cc::ihd::make_query_batch_maybe( tags, time_range, 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 -1; } 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 -1; } std::tie(queried_time_, queried_data_) = result_maybe.unsafe_get_just(); if (queried_data_.rows() == 0) { gb_logger_->log_error("Tag点没有查到相应的数据"); return -1; } } catch (const std::exception &e) { gb_logger_->log_exception(e); return -1; } return 0; } int AlgBase::refresh_ihd_cache(mix_cc::time_range_t time_range) { try { auto queried_batch_maybe = mix_cc::ihd::make_query_batch_maybe( m_tags, time_range, 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 -1; } 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 -1; } std::tie(queried_time_, queried_data_) = result_maybe.unsafe_get_just(); if (queried_data_.rows() == 0) { gb_logger_->log_error("Tag点没有查到相应的数据"); return -1; } } catch (const std::exception &e) { gb_logger_->log_exception(e); return -1; } return 0; } void AlgBase::update_limit_alarm(double lb, double ub, double va, int64_t stime, int64_t etime) { try { std::lock_guard guard(lm); rule_stat_.limit_down = lb; rule_stat_.limit_up = ub; rule_stat_.current_value = va; if (rule_stat_.current_value == 0 || va <= ub) { this->query_time_range_.set_range( std::chrono::time_point( std::chrono::milliseconds(stime)), std::chrono::time_point( std::chrono::milliseconds(etime))); } } catch (const std::exception &e) { logger_->Error() << e.what() << std::endl; } } mix_cc::time_range_t AlgBase::get_alarm_time() { try { if ((this->query_time_range_.get_right() - this->query_time_range_.get_left()) < seconds(5)) { return mix_cc::time_range_t(this->query_time_range_.get_right() - seconds(5), this->query_time_range_.get_right()); } else if (std::chrono::duration_cast( this->query_time_range_.get_left().time_since_epoch()) .count() == 0) { return mix_cc::time_range_t(this->query_time_range_.get_right() - seconds(5), this->query_time_range_.get_right()); } else if (std::chrono::duration_cast( this->query_time_range_.get_left().time_since_epoch()) .count() > 300) { return mix_cc::time_range_t(this->query_time_range_.get_right() - seconds(300), this->query_time_range_.get_right()); } return this->query_time_range_; } catch (const std::exception &e) { logger_->Error() << e.what() << std::endl; return mix_cc::time_range_t(this->now_time_ - seconds(5), this->now_time_); } } bool AlgBase::get_save_data_cycled() { auto now = system_clock::now(); if ((now - last_save_start_time_) > save_interval_ms_) { last_save_start_time_ = now; } else { return false; } return true; } bool AlgBase::get_update_rule_stat_cycled() { auto now = system_clock::now(); if ((now - last_rule_state_updae_start_time_) > rule_state_update_interval_ms_) { last_rule_state_updae_start_time_ = now; } else { return false; } return true; } bool AlgBase::get_heart_beat_log_cycled() { auto now = system_clock::now(); if ((now - last_heart_beat_start_time_) > std::chrono::minutes(5)) { last_heart_beat_start_time_ = now; } else { return false; } return true; }