| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| #include "available_detector.h" |
| |
| #include <algorithm> |
| #include <dsn/dist/replication/duplication_common.h> |
| #include <dsn/utils/time_utils.h> |
| #include <iomanip> |
| #include <sstream> |
| |
| #include "base/pegasus_key_schema.h" |
| #include "result_writer.h" |
| |
| namespace pegasus { |
| namespace server { |
| |
| DEFINE_TASK_CODE(LPC_DETECT_AVAILABLE, TASK_PRIORITY_COMMON, ::dsn::THREAD_POOL_DEFAULT) |
| |
| available_detector::available_detector() |
| : _client(nullptr), |
| _ddl_client(nullptr), |
| _app_id(-1), |
| _partition_count(0), |
| _recent_day_detect_times(0), |
| _recent_day_fail_times(0), |
| _recent_hour_detect_times(0), |
| _recent_hour_fail_times(0), |
| _recent_minute_detect_times(0), |
| _recent_minute_fail_times(0) |
| { |
| // initialize information for available_detector. |
| _cluster_name = dsn::replication::get_current_cluster_name(); |
| _app_name = dsn_config_get_value_string( |
| "pegasus.collector", "available_detect_app", "", "available detector app name"); |
| dassert(_app_name.size() > 0, ""); |
| _alert_script_dir = dsn_config_get_value_string("pegasus.collector", |
| "available_detect_alert_script_dir", |
| ".", |
| "available detect alert script dir"); |
| dassert(_alert_script_dir.size() > 0, ""); |
| _alert_email_address = dsn_config_get_value_string( |
| "pegasus.collector", |
| "available_detect_alert_email_address", |
| "", |
| "available detect alert email address, empty means not send email"); |
| _meta_list.clear(); |
| dsn::replication::replica_helper::load_meta_servers(_meta_list); |
| dassert(_meta_list.size() > 0, ""); |
| _detect_interval_seconds = |
| (uint32_t)dsn_config_get_value_uint64("pegasus.collector", |
| "available_detect_interval_seconds", |
| 3, // default value 3s |
| "detect interval seconds"); |
| _alert_fail_count = (int32_t)dsn_config_get_value_uint64("pegasus.collector", |
| "available_detect_alert_fail_count", |
| 30, |
| "available detect alert fail count"); |
| _detect_timeout = |
| (uint32_t)dsn_config_get_value_uint64("pegasus.collector", |
| "available_detect_timeout", |
| 1000, // unit is millisecond,default is 1s = 1000ms |
| "available detect timeout"); |
| // initialize the _client. |
| if (!pegasus_client_factory::initialize(nullptr)) { |
| dassert(false, "Initialize the pegasus client failed"); |
| } |
| _client = pegasus_client_factory::get_client(_cluster_name.c_str(), _app_name.c_str()); |
| dassert(_client != nullptr, "Initialize the _client failed"); |
| _result_writer = dsn::make_unique<result_writer>(_client); |
| _ddl_client.reset(new replication_ddl_client(_meta_list)); |
| dassert(_ddl_client != nullptr, "Initialize the _ddl_client failed"); |
| if (!_alert_email_address.empty()) { |
| _send_alert_email_cmd = "cd " + _alert_script_dir + "; bash sendmail.sh alert " + |
| _alert_email_address + " " + _cluster_name + " " + _app_name + " "; |
| _send_availability_info_email_cmd = "cd " + _alert_script_dir + |
| "; bash sendmail.sh availability_info " + |
| _alert_email_address + " " + _cluster_name + " "; |
| } |
| |
| _pfc_detect_times_day.init_app_counter("app.pegasus", |
| "cluster.available.detect.times.day", |
| COUNTER_TYPE_NUMBER, |
| "statistic the available detect total times every day"); |
| _pfc_fail_times_day.init_app_counter("app.pegasus", |
| "cluster.available.fail.times.day", |
| COUNTER_TYPE_NUMBER, |
| "statistic the available detect fail times every day"); |
| _pfc_available_day.init_app_counter("app.pegasus", |
| "cluster.available.day", |
| COUNTER_TYPE_NUMBER, |
| "statistic the availability of cluster every day"); |
| _pfc_detect_times_hour.init_app_counter( |
| "app.pegasus", |
| "cluster.available.detect.times.hour", |
| COUNTER_TYPE_NUMBER, |
| "statistic the available detect total times every hour"); |
| _pfc_fail_times_hour.init_app_counter("app.pegasus", |
| "cluster.available.fail.times.hour", |
| COUNTER_TYPE_NUMBER, |
| "statistic the available detect fail times every hour"); |
| _pfc_available_hour.init_app_counter("app.pegasus", |
| "cluster.available.hour", |
| COUNTER_TYPE_NUMBER, |
| "statistic the availability of cluster every hour"); |
| _pfc_detect_times_minute.init_app_counter( |
| "app.pegasus", |
| "cluster.available.detect.times.minute", |
| COUNTER_TYPE_NUMBER, |
| "statistic the available detect total times every minute"); |
| _pfc_fail_times_minute.init_app_counter( |
| "app.pegasus", |
| "cluster.available.fail.times.minute", |
| COUNTER_TYPE_NUMBER, |
| "statistic the available detect fail times every minute"); |
| _pfc_available_minute.init_app_counter("app.pegasus", |
| "cluster.available.minute", |
| COUNTER_TYPE_NUMBER, |
| "statistic the availability of cluster every minute"); |
| _pfc_available_day->set(1000000); // init to 100% |
| _pfc_available_hour->set(1000000); // init to 100% |
| _pfc_available_minute->set(1000000); // init to 100% |
| } |
| |
| available_detector::~available_detector() = default; |
| |
| void available_detector::start() |
| { |
| // available detector delay 60s to wait the pegasus finishing the initialization. |
| _detect_timer = ::dsn::tasking::enqueue(LPC_DETECT_AVAILABLE, |
| &_tracker, |
| std::bind(&available_detector::detect_available, this), |
| 0, |
| std::chrono::minutes(1)); |
| report_availability_info(); |
| } |
| |
| void available_detector::stop() { _tracker.cancel_outstanding_tasks(); } |
| |
| void available_detector::detect_available() |
| { |
| if (!generate_hash_keys()) { |
| derror("initialize hash_keys failed, do not detect available, retry after 60 seconds"); |
| _detect_timer = |
| ::dsn::tasking::enqueue(LPC_DETECT_AVAILABLE, |
| &_tracker, |
| std::bind(&available_detector::detect_available, this), |
| 0, |
| std::chrono::minutes(1)); |
| return; |
| } |
| _detect_tasks.clear(); |
| _detect_tasks.resize(_partition_count); |
| _fail_count.resize(_partition_count); |
| // initialize detect_times to zero, this is used to generate distinct value. |
| for (auto i = 0; i < _partition_count; i++) { |
| _fail_count[i].reset(new std::atomic<int32_t>(0)); |
| _fail_count[i]->store(0); |
| auto call_func = std::bind(&available_detector::on_detect, this, i); |
| _detect_tasks[i] = |
| ::dsn::tasking::enqueue_timer(LPC_DETECT_AVAILABLE, |
| &_tracker, |
| std::move(call_func), |
| std::chrono::seconds(_detect_interval_seconds)); |
| } |
| } |
| |
| std::string get_date_to_string(uint64_t now_ms) |
| { |
| char buf[50] = {'\0'}; |
| ::dsn::utils::time_ms_to_date(now_ms, buf, 50); |
| return std::string(buf); |
| } |
| |
| std::string get_hour_to_string(uint64_t now_ms) |
| { |
| char buf[50] = {'\0'}; |
| ::dsn::utils::time_ms_to_date_time(now_ms, buf, 50); |
| return std::string(buf, 13); |
| } |
| |
| std::string get_minute_to_string(uint64_t now_ms) |
| { |
| char buf[50] = {'\0'}; |
| ::dsn::utils::time_ms_to_date_time(now_ms, buf, 50); |
| return std::string(buf, 16); |
| } |
| |
| void available_detector::report_availability_info() |
| { |
| uint64_t now_ms = dsn_now_ms(); |
| _old_day = get_date_to_string(now_ms); |
| _old_hour = get_hour_to_string(now_ms); |
| _old_minute = get_minute_to_string(now_ms); |
| auto call_func = [this]() { |
| uint64_t now_ms = dsn_now_ms(); |
| std::string new_day = get_date_to_string(now_ms); |
| std::string new_hour = get_hour_to_string(now_ms); |
| std::string new_minute = get_minute_to_string(now_ms); |
| if (new_day != _old_day) { |
| this->on_day_report(); |
| _old_day = new_day; |
| } |
| if (new_hour != _old_hour) { |
| this->on_hour_report(); |
| _old_hour = new_hour; |
| } |
| if (new_minute != _old_minute) { |
| this->on_minute_report(); |
| _old_minute = new_minute; |
| } |
| }; |
| _report_task = ::dsn::tasking::enqueue_timer( |
| LPC_DETECT_AVAILABLE, |
| &_tracker, |
| std::move(call_func), |
| std::chrono::minutes(1), |
| 0, |
| std::chrono::minutes(2) // waiting for pegasus finishing start. |
| ); |
| } |
| |
| bool available_detector::generate_hash_keys() |
| { |
| // get app_id and partition_count. |
| auto err = _ddl_client->list_app(_app_name, _app_id, _partition_count, partitions); |
| if (err == ::dsn::ERR_OK && _app_id >= 0) { |
| _hash_keys.clear(); |
| for (auto pidx = 0; pidx < _partition_count; pidx++) { |
| std::string base_key = "detect_available_p" + std::to_string(pidx) + "_"; |
| // loop to generate the key that can fall into pidx-th partition. |
| int32_t base_idx = 0; |
| while (true) { |
| ++base_idx; |
| std::string key = base_key + std::to_string(base_idx); |
| ::dsn::blob my_hash_key(key.c_str(), 0, key.length()); |
| ::dsn::blob tmp_key; |
| pegasus_generate_key(tmp_key, my_hash_key, ::dsn::blob()); |
| if (pegasus_key_hash(tmp_key) % _partition_count == pidx) { |
| _hash_keys.emplace_back(key); |
| break; |
| } |
| } |
| } |
| return true; |
| } else { |
| dwarn("Get partition count of table '%s' on cluster '%s' failed", |
| _app_name.c_str(), |
| _cluster_name.c_str()); |
| return false; |
| } |
| } |
| |
| void available_detector::on_detect(int32_t idx) |
| { |
| if (idx == 0) { |
| ddebug("detecting table[%s] with app_id[%d] and partition_count[%d] on cluster[%s], " |
| "recent_day_detect_times(%" PRId64 "), recent_day_fail_times(%" PRId64 "), " |
| "recent_hour_detect_times(%" PRId64 "), recent_hour_fail_times(%" PRId64 ") " |
| "recent_minute_detect_times(%" PRId64 "), recent_minute_fail_times(%" PRId64 ")", |
| _app_name.c_str(), |
| _app_id, |
| _partition_count, |
| _cluster_name.c_str(), |
| _recent_day_detect_times.load(), |
| _recent_day_fail_times.load(), |
| _recent_hour_detect_times.load(), |
| _recent_hour_fail_times.load(), |
| _recent_minute_detect_times.load(), |
| _recent_minute_fail_times.load()); |
| } |
| dinfo("available_detector begin to detect partition[%d] of table[%s] with id[%d] on the " |
| "cluster[%s]", |
| idx, |
| _app_name.c_str(), |
| _app_id, |
| _cluster_name.c_str()); |
| auto time = dsn_now_ms(); |
| std::string value = "detect_value_" + std::to_string((time / 1000)); |
| _recent_day_detect_times.fetch_add(1); |
| _recent_hour_detect_times.fetch_add(1); |
| _recent_minute_detect_times.fetch_add(1); |
| |
| // define async_get callback function. |
| auto async_get_callback = |
| [this, idx](int err, std::string &&_value, pegasus_client::internal_info &&info) { |
| std::atomic<int> &cnt = (*_fail_count[idx]); |
| if (err != PERR_OK) { |
| int prev = cnt.fetch_add(1); |
| _recent_day_fail_times.fetch_add(1); |
| _recent_hour_fail_times.fetch_add(1); |
| _recent_minute_fail_times.fetch_add(1); |
| derror("async_get partition[%d] fail, fail_count = %d, hash_key = %s, error = %s", |
| idx, |
| prev + 1, |
| _hash_keys[idx].c_str(), |
| _client->get_error_string(err)); |
| check_and_send_email(&cnt, idx); |
| } else { |
| cnt.store(0); |
| dinfo("async_get partition[%d] ok, hash_key = %s, value = %s", |
| idx, |
| _hash_keys[idx].c_str(), |
| _value.c_str()); |
| } |
| }; |
| |
| // define async_set callback function. |
| auto async_set_callback = |
| [ this, idx, user_async_get_callback = std::move(async_get_callback) ]( |
| int err, pegasus_client::internal_info &&info) |
| { |
| std::atomic<int> &cnt = (*_fail_count[idx]); |
| if (err != PERR_OK) { |
| int prev = cnt.fetch_add(1); |
| _recent_day_fail_times.fetch_add(1); |
| _recent_hour_fail_times.fetch_add(1); |
| _recent_minute_fail_times.fetch_add(1); |
| derror("async_set partition[%d] fail, fail_count = %d, hash_key = %s , error = %s", |
| idx, |
| prev + 1, |
| _hash_keys[idx].c_str(), |
| _client->get_error_string(err)); |
| check_and_send_email(&cnt, idx); |
| } else { |
| dinfo("async_set partition[%d] ok, hash_key = %s", idx, _hash_keys[idx].c_str()); |
| _client->async_get( |
| _hash_keys[idx], "", std::move(user_async_get_callback), _detect_timeout); |
| } |
| }; |
| |
| _client->async_set(_hash_keys[idx], "", value, std::move(async_set_callback), _detect_timeout); |
| } |
| |
| void available_detector::check_and_send_email(std::atomic<int> *cnt, int32_t idx) |
| { |
| bool send_email = false; |
| if (cnt->load() >= _alert_fail_count) { |
| ::dsn::utils::auto_lock<::dsn::utils::ex_lock_nr> l(_alert_lock); |
| if (cnt->load() >= _alert_fail_count) { |
| for (auto i = 0; i < _partition_count; i++) { |
| std::atomic<int> &c = (*_fail_count[i]); |
| c.store(0); |
| } |
| send_email = true; |
| } |
| } |
| if (send_email) { |
| ddebug("start to send alert email, partition_index = %d", idx); |
| if (_send_alert_email_cmd.empty()) { |
| ddebug("ignore sending alert email because email address is not set, " |
| "partition_index = %d", |
| idx); |
| } else { |
| int r = system((_send_alert_email_cmd + std::to_string(idx)).c_str()); |
| if (r == 0) { |
| ddebug("send alert email done, partition_index = %d", idx); |
| } else { |
| derror("send alert email failed, partition_index = %d, " |
| "command_return = %d", |
| idx, |
| r); |
| } |
| } |
| } |
| } |
| |
| void available_detector::on_day_report() |
| { |
| ddebug("start to report on new day, last_day = %s", _old_day.c_str()); |
| int64_t detect_times = _recent_day_detect_times.fetch_and(0); |
| int64_t fail_times = _recent_day_fail_times.fetch_and(0); |
| int64_t succ_times = std::max(0L, detect_times - fail_times); |
| int64_t available = 0; |
| std::string hash_key("detect_available_day"); |
| std::string sort_key(_old_day); |
| std::string value("0,0,0"); |
| if (detect_times > 0) { |
| available = (int64_t)((double)succ_times / detect_times * 1000000); |
| std::ostringstream oss; |
| oss << detect_times << ',' << succ_times << ',' << available; |
| value = oss.str(); |
| } |
| |
| _pfc_detect_times_day->set(detect_times); |
| _pfc_fail_times_day->set(fail_times); |
| _pfc_available_day->set(available); |
| |
| ddebug("start to send availability email, date = %s", _old_day.c_str()); |
| if (_send_availability_info_email_cmd.empty()) { |
| ddebug("ignore sending availability email because email address is not set, " |
| "date = %s, total_detect_times = %u, total_fail_times = %u", |
| _old_day.c_str(), |
| detect_times, |
| fail_times); |
| } else { |
| int r = system((_send_availability_info_email_cmd + std::to_string(detect_times) + " " + |
| std::to_string(fail_times) + " " + _old_day) |
| .c_str()); |
| if (r == 0) { |
| ddebug("send availability email done, date = %s, " |
| "total_detect_times = %u, total_fail_times = %u", |
| _old_day.c_str(), |
| detect_times, |
| fail_times); |
| } else { |
| derror("send availability email fail, date = %s, " |
| "total_detect_times = %u, total_fail_times = %u, command_return = %d", |
| _old_day.c_str(), |
| detect_times, |
| fail_times, |
| r); |
| } |
| } |
| |
| _result_writer->set_result(hash_key, sort_key, value); |
| } |
| |
| void available_detector::on_hour_report() |
| { |
| ddebug("start to report on new hour, last_hour = %s", _old_hour.c_str()); |
| int64_t detect_times = _recent_hour_detect_times.fetch_and(0); |
| int64_t fail_times = _recent_hour_fail_times.fetch_and(0); |
| int64_t succ_times = std::max(0L, detect_times - fail_times); |
| int64_t available = 0; |
| std::string hash_key("detect_available_hour"); |
| std::string sort_key(_old_hour); |
| std::string value("0,0,0"); |
| if (detect_times > 0) { |
| available = (int64_t)((double)succ_times / detect_times * 1000000); |
| std::ostringstream oss; |
| oss << detect_times << ',' << succ_times << ',' << available; |
| value = oss.str(); |
| } |
| |
| _pfc_detect_times_hour->set(detect_times); |
| _pfc_fail_times_hour->set(fail_times); |
| _pfc_available_hour->set(available); |
| |
| _result_writer->set_result(hash_key, sort_key, value); |
| } |
| |
| void available_detector::on_minute_report() |
| { |
| ddebug("start to report on new minute, last_minute = %s", _old_minute.c_str()); |
| int64_t detect_times = _recent_minute_detect_times.fetch_and(0); |
| int64_t fail_times = _recent_minute_fail_times.fetch_and(0); |
| int64_t succ_times = std::max(0L, detect_times - fail_times); |
| int64_t available = 0; |
| std::string hash_key("detect_available_minute"); |
| std::string sort_key(_old_minute); |
| std::string value("0,0,0"); |
| if (detect_times > 0) { |
| available = (int64_t)((double)succ_times / detect_times * 1000000); |
| std::ostringstream oss; |
| oss << detect_times << ',' << succ_times << ',' << available; |
| value = oss.str(); |
| } |
| |
| _pfc_detect_times_minute->set(detect_times); |
| _pfc_fail_times_minute->set(fail_times); |
| _pfc_available_minute->set(available); |
| |
| _result_writer->set_result(hash_key, sort_key, value); |
| } |
| } // namespace server |
| } // namespace pegasus |
