Fix PG loading - now it works, at least once
parent
2a8e40835e
commit
9126ffb0f9
|
@ -3,7 +3,8 @@
|
||||||
std::string base64_encode(const std::string &in)
|
std::string base64_encode(const std::string &in)
|
||||||
{
|
{
|
||||||
std::string out;
|
std::string out;
|
||||||
unsigned val = 0, valb = -6;
|
unsigned val = 0;
|
||||||
|
int valb = -6;
|
||||||
for (unsigned char c: in)
|
for (unsigned char c: in)
|
||||||
{
|
{
|
||||||
val = (val << 8) + c;
|
val = (val << 8) + c;
|
||||||
|
@ -33,7 +34,8 @@ std::string base64_decode(const std::string &in)
|
||||||
for (int i = 0; i < 64; i++)
|
for (int i = 0; i < 64; i++)
|
||||||
T[(unsigned char)("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"[i])] = i;
|
T[(unsigned char)("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"[i])] = i;
|
||||||
}
|
}
|
||||||
unsigned val = 0, valb = -8;
|
unsigned val = 0;
|
||||||
|
int valb = -8;
|
||||||
for (unsigned char c: in)
|
for (unsigned char c: in)
|
||||||
{
|
{
|
||||||
if (T[c] == -1)
|
if (T[c] == -1)
|
||||||
|
|
|
@ -159,30 +159,25 @@ void osd_t::parse_pgs(json11::Json data)
|
||||||
throw std::runtime_error("Bad key in PG hash: "+pg_item.first);
|
throw std::runtime_error("Bad key in PG hash: "+pg_item.first);
|
||||||
}
|
}
|
||||||
auto & pg_json = pg_item.second;
|
auto & pg_json = pg_item.second;
|
||||||
osd_num_t primary_osd = 0;
|
osd_num_t primary_osd = pg_json["primary"].uint64_value();
|
||||||
|
if (primary_osd == this->osd_num)
|
||||||
|
{
|
||||||
|
// Take this PG
|
||||||
std::vector<osd_num_t> target_set;
|
std::vector<osd_num_t> target_set;
|
||||||
for (auto pg_osd_num: pg_json["osd_set"].array_items())
|
for (auto pg_osd_num: pg_json["osd_set"].array_items())
|
||||||
{
|
{
|
||||||
osd_num_t pg_osd = pg_osd_num.uint64_value();
|
osd_num_t pg_osd = pg_osd_num.uint64_value();
|
||||||
target_set.push_back(pg_osd);
|
target_set.push_back(pg_osd);
|
||||||
if (pg_osd != 0 && primary_osd == 0)
|
|
||||||
{
|
|
||||||
primary_osd = pg_osd;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
if (target_set.size() != 3)
|
if (target_set.size() != 3)
|
||||||
{
|
{
|
||||||
throw std::runtime_error("Bad PG "+std::to_string(pg_num)+" config format: incorrect osd_set");
|
throw std::runtime_error("Bad PG "+std::to_string(pg_num)+" config format: incorrect osd_set");
|
||||||
}
|
}
|
||||||
if (primary_osd == this->osd_num)
|
|
||||||
{
|
|
||||||
// Take this PG
|
|
||||||
this->pgs[pg_num] = (pg_t){
|
this->pgs[pg_num] = (pg_t){
|
||||||
.state = PG_PEERING,
|
.state = PG_PEERING,
|
||||||
.pg_cursize = 0,
|
.pg_cursize = 0,
|
||||||
.pg_num = pg_num,
|
.pg_num = pg_num,
|
||||||
.target_set = target_set,
|
.target_set = target_set,
|
||||||
.cur_set = target_set,
|
|
||||||
};
|
};
|
||||||
this->pgs[pg_num].print_state();
|
this->pgs[pg_num].print_state();
|
||||||
// Add peers
|
// Add peers
|
||||||
|
@ -216,10 +211,9 @@ void osd_t::load_and_connect_peers()
|
||||||
peering_state = peering_state & ~OSD_CONNECTING_PEERS;
|
peering_state = peering_state & ~OSD_CONNECTING_PEERS;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
else if (peer_states.find(osd_num) == peer_states.end() &&
|
else if (peer_states.find(osd_num) == peer_states.end())
|
||||||
time(NULL) - wp_it->second.last_load_attempt >= peer_connect_interval)
|
|
||||||
{
|
{
|
||||||
if (!loading_peer_config)
|
if (!loading_peer_config && (time(NULL) - wp_it->second.last_load_attempt >= peer_connect_interval))
|
||||||
{
|
{
|
||||||
// (Re)load OSD state from Consul
|
// (Re)load OSD state from Consul
|
||||||
wp_it->second.last_load_attempt = time(NULL);
|
wp_it->second.last_load_attempt = time(NULL);
|
||||||
|
@ -237,7 +231,7 @@ void osd_t::load_and_connect_peers()
|
||||||
{
|
{
|
||||||
// Try to connect
|
// Try to connect
|
||||||
wp_it->second.connecting = true;
|
wp_it->second.connecting = true;
|
||||||
const std::string & addr = peer_states[osd_num]["addresses"][wp_it->second.address_index].string_value();
|
const std::string addr = peer_states[osd_num]["addresses"][wp_it->second.address_index].string_value();
|
||||||
int64_t port = peer_states[osd_num]["port"].int64_value();
|
int64_t port = peer_states[osd_num]["port"].int64_value();
|
||||||
wp_it++;
|
wp_it++;
|
||||||
connect_peer(osd_num, addr.c_str(), port, [this](osd_num_t osd_num, int peer_fd)
|
connect_peer(osd_num, addr.c_str(), port, [this](osd_num_t osd_num, int peer_fd)
|
||||||
|
@ -266,6 +260,7 @@ void osd_t::load_and_connect_peers()
|
||||||
if (!wanted_peers.size())
|
if (!wanted_peers.size())
|
||||||
{
|
{
|
||||||
// Connected to all peers
|
// Connected to all peers
|
||||||
|
printf("Connected to all peers\n");
|
||||||
peering_state = peering_state & ~OSD_CONNECTING_PEERS;
|
peering_state = peering_state & ~OSD_CONNECTING_PEERS;
|
||||||
}
|
}
|
||||||
repeer_pgs(osd_num, true);
|
repeer_pgs(osd_num, true);
|
||||||
|
@ -292,20 +287,24 @@ void osd_t::load_and_connect_peers()
|
||||||
loading_peer_config = false;
|
loading_peer_config = false;
|
||||||
if (err != "")
|
if (err != "")
|
||||||
{
|
{
|
||||||
printf("Failed to load peer configuration from Consul");
|
printf("Failed to load peer configuration from Consul: %s\n", err.c_str());
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
for (auto & res: data["Results"].array_items())
|
for (auto & res: data["Results"].array_items())
|
||||||
{
|
{
|
||||||
std::string key = res["KV"]["Key"].string_value();
|
std::string key = res["KV"]["Key"].string_value();
|
||||||
// <consul_prefix>/osd/state/<osd_num>.
|
// <consul_prefix>/osd/state/<osd_num>.
|
||||||
osd_num_t osd_num = std::stoull(key.substr(consul_prefix.length()+11, key.length()-consul_prefix.length()-12));
|
osd_num_t peer_osd = std::stoull(key.substr(consul_prefix.length()+11, key.length()-consul_prefix.length()-12));
|
||||||
std::string json_err;
|
std::string json_err;
|
||||||
json11::Json data = json11::Json::parse(base64_decode(res["KV"]["Value"].string_value()), json_err);
|
json11::Json st = json11::Json::parse(base64_decode(res["KV"]["Value"].string_value()), json_err);
|
||||||
if (osd_num > 0 && data.is_object() && data["state"] == "up" &&
|
if (json_err != "")
|
||||||
data["addresses"].is_array() && data["port"].is_number())
|
|
||||||
{
|
{
|
||||||
peer_states[osd_num] = data;
|
printf("Bad JSON in Consul key %s: %s\n", key.c_str(), json_err.c_str());
|
||||||
|
}
|
||||||
|
if (peer_osd > 0 && st.is_object() && st["state"] == "up" &&
|
||||||
|
st["addresses"].is_array() && st["port"].is_number())
|
||||||
|
{
|
||||||
|
peer_states[peer_osd] = st;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
21
osd_http.cpp
21
osd_http.cpp
|
@ -119,7 +119,7 @@ void osd_t::http_request_json(std::string host, std::string request,
|
||||||
json11::Json data = json11::Json::parse(res->body, json_err);
|
json11::Json data = json11::Json::parse(res->body, json_err);
|
||||||
if (json_err != "")
|
if (json_err != "")
|
||||||
{
|
{
|
||||||
callback("Bad JSON: "+json_err+" (response: "+res->body+")", json11::Json());
|
callback("Bad JSON: "+json_err+" (response: "+(res->body == "" ? txt : res->body)+")", json11::Json());
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
callback(std::string(), data);
|
callback(std::string(), data);
|
||||||
|
@ -142,14 +142,27 @@ http_response_t *parse_http_response(std::string res)
|
||||||
status_text = NULL;
|
status_text = NULL;
|
||||||
}
|
}
|
||||||
int prev = pos;
|
int prev = pos;
|
||||||
while ((pos = res.find("\r\n", prev)) > prev)
|
while ((pos = res.find("\r\n", prev)) >= prev)
|
||||||
{
|
{
|
||||||
if (pos == prev+2)
|
if (pos == prev)
|
||||||
|
{
|
||||||
|
if (parsed->headers["transfer-encoding"] == "chunked")
|
||||||
|
{
|
||||||
|
prev = pos+2;
|
||||||
|
while ((pos = res.find("\r\n", prev)) >= prev)
|
||||||
|
{
|
||||||
|
uint64_t len = strtoull(res.c_str()+prev, NULL, 16);
|
||||||
|
parsed->body += res.substr(pos+2, len);
|
||||||
|
prev = pos+2+len+2;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
else
|
||||||
{
|
{
|
||||||
parsed->body = res.substr(pos+2);
|
parsed->body = res.substr(pos+2);
|
||||||
|
}
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
std::string header = res.substr(prev, pos);
|
std::string header = res.substr(prev, pos-prev);
|
||||||
int p2 = header.find(":");
|
int p2 = header.find(":");
|
||||||
if (p2 >= 0)
|
if (p2 >= 0)
|
||||||
{
|
{
|
||||||
|
|
|
@ -230,9 +230,8 @@ void osd_t::repeer_pgs(osd_num_t osd_num, bool is_connected)
|
||||||
for (int r = 0; r < p.second.target_set.size(); r++)
|
for (int r = 0; r < p.second.target_set.size(); r++)
|
||||||
{
|
{
|
||||||
if (p.second.target_set[r] == osd_num &&
|
if (p.second.target_set[r] == osd_num &&
|
||||||
p.second.cur_set[r] != real_osd)
|
(p.second.cur_set.size() < r || p.second.cur_set[r] != real_osd))
|
||||||
{
|
{
|
||||||
p.second.cur_set[r] = real_osd;
|
|
||||||
repeer = true;
|
repeer = true;
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
@ -287,8 +286,11 @@ void osd_t::start_pg_peering(pg_num_t pg_num)
|
||||||
dirty_pgs.erase(pg.pg_num);
|
dirty_pgs.erase(pg.pg_num);
|
||||||
// Start peering
|
// Start peering
|
||||||
pg.pg_cursize = 0;
|
pg.pg_cursize = 0;
|
||||||
for (int role = 0; role < pg.cur_set.size(); role++)
|
pg.cur_set.resize(pg.target_set.size());
|
||||||
|
for (int role = 0; role < pg.target_set.size(); role++)
|
||||||
{
|
{
|
||||||
|
pg.cur_set[role] = pg.target_set[role] == this->osd_num ||
|
||||||
|
osd_peer_fds.find(pg.target_set[role]) != osd_peer_fds.end() ? pg.target_set[role] : 0;
|
||||||
if (pg.cur_set[role] != 0)
|
if (pg.cur_set[role] != 0)
|
||||||
{
|
{
|
||||||
pg.pg_cursize++;
|
pg.pg_cursize++;
|
||||||
|
@ -308,8 +310,10 @@ void osd_t::start_pg_peering(pg_num_t pg_num)
|
||||||
for (role = 0; role < pg.cur_set.size(); role++)
|
for (role = 0; role < pg.cur_set.size(); role++)
|
||||||
{
|
{
|
||||||
if (pg.cur_set[role] == it->first)
|
if (pg.cur_set[role] == it->first)
|
||||||
|
{
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
}
|
||||||
if (pg.state == PG_INCOMPLETE || role >= pg.cur_set.size())
|
if (pg.state == PG_INCOMPLETE || role >= pg.cur_set.size())
|
||||||
{
|
{
|
||||||
// Discard the result after completion, which, chances are, will be unsuccessful
|
// Discard the result after completion, which, chances are, will be unsuccessful
|
||||||
|
@ -342,8 +346,10 @@ void osd_t::start_pg_peering(pg_num_t pg_num)
|
||||||
for (role = 0; role < pg.cur_set.size(); role++)
|
for (role = 0; role < pg.cur_set.size(); role++)
|
||||||
{
|
{
|
||||||
if (pg.cur_set[role] == it->first)
|
if (pg.cur_set[role] == it->first)
|
||||||
|
{
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
}
|
||||||
if (pg.state == PG_INCOMPLETE || role >= pg.cur_set.size())
|
if (pg.state == PG_INCOMPLETE || role >= pg.cur_set.size())
|
||||||
{
|
{
|
||||||
if (it->second.buf)
|
if (it->second.buf)
|
||||||
|
|
Loading…
Reference in New Issue