thebestvibedev/tromcho.net
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Duplicate all data in /index/image/ and support reading it from there

Browse Source
This commit is contained in:
Vitaliy Filippov
2026-03-20 21:00:32 +03:00
parent 3a0455877a
commit ccbde3ff91
8 changed files with 122 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/client/etcd_state_client.cpp
+64 -5
View File
@@ -304,6 +304,7 @@ void etcd_state_client_t::parse_config(const json11::Json & config)
{
this->etcd_min_reload_interval = 50;
}
this->use_image_index_for_auth = config["use_image_index_for_auth"].bool_value();
if (this->etcd_ws_keepalive_interval != old_etcd_ws_keepalive_interval && ws_keepalive_timer >= 0)
{
#ifndef __MOCK__
@@ -435,6 +436,7 @@ void etcd_state_client_t::start_etcd_watcher_selected()
etcd_watch_ws = http_init(get_http_ctx());
else
http_close(etcd_watch_ws);
etcd_total_watches = (use_image_index_for_auth ? 4 : 3);
open_websocket(etcd_watch_ws, url.addr, url.hostname, url.path+"/watch", { .timeout = etcd_slow_timeout, .ssl = url.ssl },
[this, cur_addr = url.addr](http_message_t *msg)
{
@@ -454,11 +456,12 @@ void etcd_state_client_t::start_etcd_watcher_selected()
{
if (watch_id == ETCD_CONFIG_WATCH_ID ||
watch_id == ETCD_PG_STATE_WATCH_ID ||
watch_id == ETCD_OSD_STATE_WATCH_ID)
watch_id == ETCD_OSD_STATE_WATCH_ID ||
watch_id == ETCD_INDEX_WATCH_ID)
{
etcd_watches_initialised++;
}
if (etcd_watches_initialised == ETCD_TOTAL_WATCHES && this->log_level > 0)
if (etcd_watches_initialised == etcd_total_watches && this->log_level > 0)
{
fprintf(stderr, "Successfully subscribed to etcd at %s, revision %ju/%ju/%ju\n", cur_addr.c_str(),
etcd_watch_revision_config, etcd_watch_revision_osd, etcd_watch_revision_pg);
@@ -479,7 +482,7 @@ void etcd_state_client_t::start_etcd_watcher_selected()
fprintf(stderr, "Revisions before %ju were compacted by etcd, reloading state\n",
data["result"]["compact_revision"].uint64_value());
http_close(etcd_watch_ws);
etcd_watch_revision_config = etcd_watch_revision_osd = etcd_watch_revision_pg = 0;
etcd_watch_revision_config = etcd_watch_revision_osd = etcd_watch_revision_pg = etcd_watch_revision_index = 0;
on_reload_hook();
}
return;
@@ -500,7 +503,7 @@ void etcd_state_client_t::start_etcd_watcher_selected()
// Save revision only if it's present in the message - because sometimes etcd sends something without a header, like:
// {"error": {"grpc_code": 14, "http_code": 503, "http_status": "Service Unavailable", "message": "error reading from server: EOF"}}
// Also don't save revision from the initial created: true messages because they always contain the latest revision
if (etcd_watches_initialised == ETCD_TOTAL_WATCHES &&
if (etcd_watches_initialised == etcd_total_watches &&
!data["result"]["header"]["revision"].is_null() &&
!data["result"]["created"].bool_value())
{
@@ -523,6 +526,8 @@ void etcd_state_client_t::start_etcd_watcher_selected()
etcd_watch_revision_pg = watch_rev;
else if (watch_id == ETCD_OSD_STATE_WATCH_ID)
etcd_watch_revision_osd = watch_rev;
else if (watch_id == ETCD_INDEX_WATCH_ID)
etcd_watch_revision_index = watch_rev;
etcd_urls_to_try.clear();
}
// First gather all changes into a hash to remove multiple overwrites
@@ -598,6 +603,18 @@ void etcd_state_client_t::start_etcd_watcher_selected()
{ "progress_notify", true },
} }
}).dump());
if (use_image_index_for_auth)
{
http_post_message(etcd_watch_ws, WS_TEXT, json11::Json(json11::Json::object {
{ "create_request", json11::Json::object {
{ "key", base64_encode(etcd_prefix+"/index/image/") },
{ "range_end", base64_encode(etcd_prefix+"/index/image0") },
{ "start_revision", etcd_watch_revision_index },
{ "watch_id", ETCD_INDEX_WATCH_ID },
{ "progress_notify", true },
} }
}).dump());
}
// FIXME: Do not watch /pg/history/ at all in client code (not in OSD)
if (on_start_watcher_hook)
{
@@ -621,7 +638,7 @@ void etcd_state_client_t::start_ws_keepalive()
{
ws_keepalive_timer = tfd->set_timer(etcd_ws_keepalive_interval*1000, true, [this](int)
{
if (!etcd_watch_ws || etcd_watches_initialised < ETCD_TOTAL_WATCHES)
if (!etcd_watch_ws || etcd_watches_initialised < etcd_total_watches)
{
// Do nothing
}
@@ -767,6 +784,15 @@ void etcd_state_client_t::load_pgs()
} }
},
};
if (use_image_index_for_auth)
{
txn.push_back(json11::Json::object {
{ "request_range", json11::Json::object {
{ "key", base64_encode(etcd_prefix+"/index/image/") },
{ "range_end", base64_encode(etcd_prefix+"/index/image0") },
} }
});
}
json11::Json::object req = { { "success", txn } };
json11::Json checks = load_pgs_checks_hook != NULL ? load_pgs_checks_hook() : json11::Json();
if (checks.array_items().size() > 0)
@@ -1344,6 +1370,39 @@ void etcd_state_client_t::parse_state(const etcd_kv_t & kv)
}
}
}
else if (key.substr(0, etcd_prefix.length()+13) == etcd_prefix+"/index/image/")
{
// <etcd_prefix>/index/image/%s
// Used for environments based on etcd authentication to grant client image access
// by names - granting by ID is inconvenient because of snapshots
std::string inode_name = key.substr(etcd_prefix.length()+13);
if (!value.is_object())
{
auto n_it = this->inode_by_name.find(inode_name);
if (n_it != this->inode_by_name.end())
{
uint64_t inode_num = n_it->second;
if (on_inode_change_hook != NULL)
{
on_inode_change_hook(inode_num, true);
}
this->inode_config.erase(inode_num);
this->inode_by_name.erase(n_it);
}
}
else if (value["name"] == inode_name)
{
uint64_t inode_num = INODE_WITH_POOL(value["pool_id"].uint64_value(), value["id"].uint64_value());
if (!INODE_POOL(inode_num) || !INODE_NO_POOL(inode_num))
{
fprintf(stderr, "Invalid inode ID in key %s: %ju\n", key.c_str(), inode_num);
}
else
{
insert_inode_config(deserialize_inode_cfg(inode_num, value, kv.mod_revision));
}
}
}
else if (key == etcd_prefix+"/config/node_placement")
{
// <etcd_prefix>/config/node_placement
src/client/etcd_state_client.h
+4 -1
View File
@@ -13,7 +13,7 @@
#define ETCD_CONFIG_WATCH_ID 1
#define ETCD_OSD_STATE_WATCH_ID 2
#define ETCD_PG_STATE_WATCH_ID 3
#define ETCD_TOTAL_WATCHES 3
#define ETCD_INDEX_WATCH_ID 4
#define DEFAULT_BLOCK_SIZE 128*1024
#define MIN_DATA_BLOCK_SIZE 4*1024
@@ -126,6 +126,7 @@ protected:
std::vector<inode_watch_t*> watches;
std::vector<std::function<void()>> on_resolve_queue;
bool new_pg_config = false;
bool use_image_index_for_auth = false;
int ws_keepalive_timer = -1;
int ws_alive = 0;
bool rand_initialized = false;
@@ -157,9 +158,11 @@ public:
http_context_t *http_ctx = NULL;
http_co_t *etcd_watch_ws = NULL, *keepalive_client = NULL;
int etcd_watches_initialised = 0;
int etcd_total_watches = 0;
uint64_t etcd_watch_revision_config = 0;
uint64_t etcd_watch_revision_osd = 0;
uint64_t etcd_watch_revision_pg = 0;
uint64_t etcd_watch_revision_index = 0;
timespec etcd_last_reload = {};
int load_pgs_timer_id = -1;
std::map<pool_id_t, pool_config_t> pool_config;
src/cmd/cli.h
+2
View File
@@ -101,3 +101,5 @@ std::string format_lat(uint64_t lat);
std::string format_q(double depth);
bool stupid_glob(const std::string str, const std::string glob);
json11::Json::object merge_json_objects(json11::Json::object obj1, const json11::Json::object & obj2);
src/cmd/cli_common.cpp
+7
View File
@@ -244,3 +244,10 @@ void cli_tool_t::iterate_kvs_2(json11::Json kvs, const std::string & prefix, std
cb(pool_id, num, kv.value);
}
}
json11::Json::object merge_json_objects(json11::Json::object obj1, const json11::Json::object & obj2)
{
for (auto & kv: obj2)
obj1[kv.first] = kv.second;
return obj1;
}
src/cmd/cli_create.cpp
+14 -8
View File
@@ -480,10 +480,13 @@ resume_3:
json11::Json::object {
{ "request_put", json11::Json::object {
{ "key", base64_encode(parent->cli->st_cli.etcd_prefix+"/index/image/"+image_name) },
{ "value", base64_encode(json11::Json(json11::Json::object{
{ "id", new_id },
{ "pool_id", (uint64_t)new_pool_id },
}).dump()) },
{ "value", base64_encode(json11::Json(merge_json_objects(
parent->cli->st_cli.serialize_inode_cfg(&new_cfg),
json11::Json::object{
{ "id", new_id },
{ "pool_id", (uint64_t)new_pool_id },
}
)).dump()) },
} },
},
json11::Json::object {
@@ -538,10 +541,13 @@ resume_3:
success.push_back(json11::Json::object {
{ "request_put", json11::Json::object {
{ "key", base64_encode(parent->cli->st_cli.etcd_prefix+"/index/image/"+image_name+"@"+new_snap) },
{ "value", base64_encode(json11::Json(json11::Json::object{
{ "id", old_id },
{ "pool_id", (uint64_t)old_pool_id },
}).dump()) },
{ "value", base64_encode(json11::Json(merge_json_objects(
parent->cli->st_cli.serialize_inode_cfg(&snap_cfg),
json11::Json::object{
{ "id", old_id },
{ "pool_id", (uint64_t)old_pool_id },
}
)).dump()) },
} },
});
};
src/cmd/cli_modify.cpp
+21 -16
View File
@@ -188,7 +188,6 @@ resume_1:
} }
});
}
if (new_name != "")
{
std::string old_idx_key = base64_encode(
parent->cli->st_cli.etcd_prefix+"/index/image/"+image_name
@@ -202,23 +201,29 @@ resume_1:
{ "result", "LESS" },
{ "mod_revision", cfg.mod_revision+1 },
});
checks.push_back(json11::Json::object {
{ "target", "VERSION" },
{ "version", 0 },
{ "key", new_idx_key },
});
success.push_back(json11::Json::object {
{ "request_delete_range", json11::Json::object {
{ "key", old_idx_key },
} }
});
if (new_name != "")
{
checks.push_back(json11::Json::object {
{ "target", "VERSION" },
{ "version", 0 },
{ "key", new_idx_key },
});
success.push_back(json11::Json::object {
{ "request_delete_range", json11::Json::object {
{ "key", old_idx_key },
} }
});
}
success.push_back(json11::Json::object {
{ "request_put", json11::Json::object {
{ "key", new_idx_key },
{ "value", base64_encode(json11::Json(json11::Json::object{
{ "id", INODE_NO_POOL(inode_num) },
{ "pool_id", (uint64_t)INODE_POOL(inode_num) },
}).dump()) },
{ "key", new_name.empty() ? old_idx_key : new_idx_key },
{ "value", base64_encode(json11::Json(merge_json_objects(
parent->cli->st_cli.serialize_inode_cfg(&cfg),
json11::Json::object{
{ "id", INODE_NO_POOL(inode_num) },
{ "pool_id", (uint64_t)INODE_POOL(inode_num) },
}
)).dump()) },
} }
});
}
src/cmd/cli_rm.cpp
+7 -4
View File
@@ -487,10 +487,13 @@ resume_100:
json11::Json::object {
{ "request_put", json11::Json::object {
{ "key", base64_encode(parent->cli->st_cli.etcd_prefix+"/index/image/"+child_cfg->name) },
{ "value", base64_encode(json11::Json({
{ "id", INODE_NO_POOL(inverse_parent) },
{ "pool_id", (uint64_t)INODE_POOL(inverse_parent) },
}).dump()) },
{ "value", base64_encode(json11::Json(merge_json_objects(
parent->cli->st_cli.serialize_inode_cfg(&new_cfg),
json11::Json::object{
{ "id", INODE_NO_POOL(inverse_parent) },
{ "pool_id", (uint64_t)INODE_POOL(inverse_parent) },
}
)).dump()) },
} },
},
};