Ceph Observer动态更新配置参数

ceph Observer(观察者模式)，可以动态的更改系统配置参数，而不需重启服务。

md_config_t:
//multimap 类型key为option，value为关注该可以的对象。
observers     (typedef std::multimap <std::string, md_config_obs_t*> obs_map_t)
//发生变动的option
changed        (typedef std::set < std::string >)

//将观察者与其观察的key放入observer中。

void md_config_t::add_observer(md_config_obs_t* observer_)
{
Mutex::Locker l(lock);
const char **keys = observer_->get_tracked_conf_keys();
for (const char ** k = keys; *k; ++k) {
obs_map_t::value_type val(*k, observer_);
observers.insert(val);
}
}

//将制定的观察者从observers集合中去除

void md_config_t::remove_observer(md_config_obs_t* observer_)
{
Mutex::Locker l(lock);
bool found_obs = false;
for (obs_map_t::iterator o = observers.begin(); o != observers.end(); ) {
if (o->second == observer_) {
observers.erase(o++);
found_obs = true;
}
else {
++o;
}
}
assert(found_obs);
}

//获取observer的配置参数集，不同的的子系统实现不同

get_tracked_conf_keys()

//更新全局配置参数，然后把发生变动的参数的key放入到changed集合中

//配置参数集合大致可以分为两类，一：与子系统相关的日志级别参数、二: 普通服务配置参数

//set_val()->set_val_impl()->set_val_impl()
int set_val(const char *key, const string& s, bool meta=true, bool safe=true) {
return set_val(key, s.c_str(), meta, safe);
}

int md_config_t::set_val(const char *key, const char *val, bool meta, bool safe)
{
Mutex::Locker l(lock);
if (!key)
return -EINVAL;
if (!val)
return -EINVAL;

std::string v(val);
if (meta)
expand_meta(v, &std::cerr);

string k(ConfFile::normalize_key_name(key));

// subsystems?
if (strncmp(k.c_str(), "debug_", 6) == 0) {
for (int o = 0; o < subsys.get_num(); o++) {
std::string as_option = "debug_" + subsys.get_name(o);
if (k == as_option) {
int log, gather;
int r = sscanf(v.c_str(), "%d/%d", &log, &gather);
if (r >= 1) {
if (r < 2)
gather = log;
//      cout << "subsys " << subsys.get_name(o) << " log " << log << " gather " << gather << std::endl;
subsys.set_log_level(o, log);
subsys.set_gather_level(o, gather);
return 0;
}
return -EINVAL;
}
}
}

for (int i = 0; i < NUM_CONFIG_OPTIONS; ++i) {
config_option *opt = &config_optionsp[i];
if (strcmp(opt->name, k.c_str()) == 0) {
if (safe && internal_safe_to_start_threads) {
// If threads have been started...
if ((opt->type == OPT_STR) || (opt->type == OPT_ADDR) ||
(opt->type == OPT_UUID)) {
// And this is NOT an integer valued variable....
if (observers.find(opt->name) == observers.end()) {
// And there is no observer to safely change it...
// You lose.
return -ENOSYS;
}
}
}
return set_val_impl(v.c_str(), opt);
}
}

// couldn't find a configuration option with key 'key'
return -ENOENT;
}

int md_config_t::set_val_impl(const char *val, const config_option *opt)
{
assert(lock.is_locked());
int ret = set_val_raw(val, opt);
if (ret)
return ret;
changed.insert(opt->name);
return 0;
}

int md_config_t::set_val_raw(const char *val, const config_option *opt)
{
assert(lock.is_locked());
switch (opt->type) {
case OPT_INT: {
std::string err;
int f = strict_si_cast<int>(val, &err);
if (!err.empty())
return -EINVAL;
*(int*)opt->conf_ptr(this) = f;
return 0;
}
case OPT_LONGLONG: {
std::string err;
long long f = strict_si_cast<long long>(val, &err);
if (!err.empty())
return -EINVAL;
*(long long*)opt->conf_ptr(this) = f;
return 0;
}
case OPT_STR:
*(std::string*)opt->conf_ptr(this) = val ? val : "";
return 0;
case OPT_FLOAT: {
std::string err;
float f = strict_strtof(val, &err);
if (!err.empty())
return -EINVAL;
*(float*)opt->conf_ptr(this) = f;
return 0;
}
....
case OPT_UUID: {
uuid_d *u = (uuid_d*)opt->conf_ptr(this);
if (!u->parse(val))
return -EINVAL;
return 0;
}
}
return -ENOSYS;
}

void md_config_t::apply_changes(std::ostream *oss)
{
Mutex::Locker l(lock);
_apply_changes(oss);
}

void md_config_t::_apply_changes(std::ostream *oss)
{
/* Maps observers to the configuration options that they care about which
* have changed. */
typedef std::map < md_config_obs_t*, std::set <std::string> > rev_obs_map_t;

expand_all_meta();

// create the reverse observer mapping, mapping observers to the set of
// changed keys that they'll get.
rev_obs_map_t robs;
std::set <std::string> empty_set;
char buf[128];
char *bufptr = (char*)buf;
//将更新后的配置参数与oberver相关联。
for (changed_set_t::const_iterator c = changed.begin();
c != changed.end(); ++c) {
const std::string &key(*c);
if ((oss) &&
(!_get_val(key.c_str(), &bufptr, sizeof(buf))) &&
!_internal_field(key)) {
(*oss) << key << " = '" << buf << "' ";
}
pair < obs_map_t::iterator, obs_map_t::iterator >
range(observers.equal_range(key));
for (obs_map_t::iterator r = range.first; r != range.second; ++r) {
rev_obs_map_t::value_type robs_val(r->second, empty_set);
pair < rev_obs_map_t::iterator, bool > robs_ret(robs.insert(robs_val));
std::set <std::string> &keys(robs_ret.first->second);
keys.insert(key);
}
}

// Make any pending observer callbacks
//调用observer的回调函数handle_conf_change() 更新系统参数。
for (rev_obs_map_t::const_iterator r = robs.begin(); r != robs.end(); ++r) {
md_config_obs_t *obs = r->first;
obs->handle_conf_change(this, r->second);
}

changed.clear();
}