diff --git a/src/client/cluster_client.cpp b/src/client/cluster_client.cpp index e62c3ebe..2f8559bf 100644 --- a/src/client/cluster_client.cpp +++ b/src/client/cluster_client.cpp @@ -73,6 +73,12 @@ cluster_client_t::~cluster_client_t() retry_timeout_duration = 0; retry_timeout_id = -1; } + if (list_retry_timeout_id >= 0) + { + tfd->clear_timer(list_retry_timeout_id); + list_retry_timeout_id = -1; + list_retry_time = {}; + } msgr.repeer_pgs = [](osd_num_t){}; if (ringloop) { @@ -403,6 +409,19 @@ void cluster_client_t::on_load_config_hook(json11::Json::object & etcd_global_co } // client_retry_enospc client_retry_enospc = config["client_retry_enospc"].is_null() ? true : config["client_retry_enospc"].bool_value(); + // peer_connect_timeout, wait_up_timeout + peer_connect_timeout = config["peer_connect_timeout"].uint64_value(); + if (!peer_connect_timeout) + peer_connect_timeout = 5; + if (!config["wait_up_timeout"].is_null()) + wait_up_timeout = config["wait_up_timeout"].uint64_value(); + else + { + auto etcd_report_interval = config["etcd_report_interval"].uint64_value(); + if (!etcd_report_interval) + etcd_report_interval = 5; + wait_up_timeout = 1+etcd_report_interval+(st_cli.max_etcd_attempts*(2*st_cli.etcd_quick_timeout)+999)/1000; + } // log_level log_level = config["log_level"].uint64_value(); msgr.parse_config(config); @@ -463,6 +482,7 @@ void cluster_client_t::on_change_pg_state_hook(pool_id_t pool_id, pg_num_t pg_nu } // Always continue to resume operations hung because of lack of the primary OSD continue_ops(); + continue_lists(); } bool cluster_client_t::get_immediate_commit(uint64_t inode) @@ -483,6 +503,7 @@ void cluster_client_t::on_change_osd_state_hook(uint64_t peer_osd) if (msgr.wanted_peers.find(peer_osd) != msgr.wanted_peers.end()) { msgr.connect_peer(peer_osd, st_cli.peer_states[peer_osd]); + continue_lists(); } } diff --git a/src/client/cluster_client.h b/src/client/cluster_client.h index a6c71ada..8027d359 100644 --- a/src/client/cluster_client.h +++ b/src/client/cluster_client.h @@ -96,6 +96,8 @@ class cluster_client_t int client_retry_interval = 50; // ms int client_eio_retry_interval = 1000; // ms bool client_retry_enospc = true; + int peer_connect_timeout = 5; // sec + int wait_up_timeout = 10; // sec (for listings) int retry_timeout_id = -1; int retry_timeout_duration = 0; @@ -111,6 +113,8 @@ class cluster_client_t bool pgs_loaded = false; ring_consumer_t consumer; std::vector> on_ready_hooks; + int list_retry_timeout_id = -1; + timespec list_retry_time; std::vector lists; std::multimap raw_ops; int continuing_ops = 0; @@ -136,13 +140,12 @@ public: bool get_immediate_commit(uint64_t inode); void continue_ops(int time_passed = 0); - // FIXME: list_inode_start/list_inode_next - not an ideal interface :) - inode_list_t *list_inode_start(inode_t inode, - std::function&& objects, pg_num_t pg_num, osd_num_t primary_osd, int errcode, int status)> callback); + + inode_list_t *list_inode_start(inode_t inode, int max_parallel_pgs, std::function&& objects, pg_num_t pg_num, std::vector && inactive_osds, int errcode, int status)> callback); + void list_inode_next(inode_list_t *lst); int list_pg_count(inode_list_t *lst); - const std::vector & list_inode_get_inactive_osds(inode_list_t *lst); - const std::vector & list_inode_get_inactive_pgs(inode_list_t *lst); - void list_inode_next(inode_list_t *lst, int next_pgs); + //inline uint32_t get_bs_bitmap_granularity() { return st_cli.global_bitmap_granularity; } //inline uint64_t get_bs_block_size() { return st_cli.global_block_size; } uint64_t next_op_id(); @@ -170,7 +173,10 @@ protected: void calc_wait(cluster_op_t *op); void inc_wait(uint64_t opcode, uint64_t flags, cluster_op_t *next, int inc); void continue_lists(); - void continue_listing(inode_list_t *lst); + bool continue_listing(inode_list_t *lst); + bool restart_listing(inode_list_t* lst); + void retry_start_pg_listing(inode_list_pg_t *pg); + int start_pg_listing(inode_list_pg_t *pg); void send_list(inode_list_osd_t *cur_list); void finish_list_pg(inode_list_pg_t *pg); bool check_finish_listing(inode_list_t *lst); diff --git a/src/client/cluster_client_list.cpp b/src/client/cluster_client_list.cpp index 214d66ce..2cdc6a92 100644 --- a/src/client/cluster_client_list.cpp +++ b/src/client/cluster_client_list.cpp @@ -2,9 +2,16 @@ // License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details) #include +#include "assert.h" #include "pg_states.h" #include "cluster_client.h" +#define LIST_PG_INIT 0 +#define LIST_PG_WAIT_ACTIVE 1 +#define LIST_PG_WAIT_CONNECT 2 +#define LIST_PG_SENT 3 +#define LIST_PG_DONE 4 + struct inode_list_t; struct inode_list_pg_t; @@ -13,21 +20,22 @@ struct inode_list_osd_t { inode_list_pg_t *pg = NULL; osd_num_t osd_num = 0; - bool sent = false; }; struct inode_list_pg_t { inode_list_t *lst = NULL; - int pos = 0; int errcode = 0; - pg_num_t pg_num; - osd_num_t cur_primary; - bool has_unstable = false; - int sent = 0; - int done = 0; + pg_num_t pg_num = 0; + osd_num_t cur_primary = 0; + int state = 0; + int inflight_ops = 0; + timespec wait_until; std::vector list_osds; + + bool has_unstable = false; std::set objects; + std::vector inactive_osds; }; struct inode_list_t @@ -35,17 +43,19 @@ struct inode_list_t cluster_client_t *cli = NULL; pool_id_t pool_id = 0; inode_t inode = 0; + int max_parallel_pgs = 16; + + int inflight_pgs = 0; + std::map inflight_per_osd; int done_pgs = 0; - int want = 0; int onstack = 0; - std::vector inactive_osds; - std::vector inactive_pgs; std::vector pgs; - std::function&& objects, pg_num_t pg_num, osd_num_t primary_osd, int errcode, int status)> callback; + pg_num_t real_pg_count = 0; + std::function&& objects, pg_num_t pg_num, std::vector && inactive_osds, int errcode, int status)> callback; }; -inode_list_t* cluster_client_t::list_inode_start(inode_t inode, - std::function&& objects, pg_num_t pg_num, osd_num_t primary_osd, int errcode, int status)> callback) +inode_list_t* cluster_client_t::list_inode_start(inode_t inode, int max_parallel_pgs, std::function&& objects, pg_num_t pg_num, std::vector && inactive_osds, int errcode, int status)> callback) { init_msgr(); pool_id_t pool_id = INODE_POOL(inode); @@ -62,180 +72,257 @@ inode_list_t* cluster_client_t::list_inode_start(inode_t inode, lst->pool_id = pool_id; lst->inode = inode; lst->callback = callback; - auto pool_cfg = st_cli.pool_config[pool_id]; - std::set inactive_osd_set; - for (auto & pg_item: pool_cfg.pg_config) - { - auto & pg = pg_item.second; - if (pg.pause || !pg.cur_primary || !(pg.cur_state & PG_ACTIVE)) - { - lst->inactive_pgs.push_back(pg_item.first); - if (log_level > 0) - { - fprintf(stderr, "PG %u is inactive, skipping\n", pg_item.first); - } - continue; - } - inode_list_pg_t *r = new inode_list_pg_t(); - r->lst = lst; - r->pg_num = pg_item.first; - r->cur_primary = pg.cur_primary; - if (pg.cur_state != PG_ACTIVE) - { - // Not clean - std::set all_peers; - for (osd_num_t pg_osd: pg.target_set) - { - if (pg_osd != 0) - { - all_peers.insert(pg_osd); - } - } - for (osd_num_t pg_osd: pg.all_peers) - { - if (pg_osd != 0) - { - all_peers.insert(pg_osd); - } - } - for (auto & hist_item: pg.target_history) - { - for (auto pg_osd: hist_item) - { - if (pg_osd != 0) - { - all_peers.insert(pg_osd); - } - } - } - for (osd_num_t peer_osd: all_peers) - { - if (st_cli.peer_states.find(peer_osd) != st_cli.peer_states.end()) - { - r->list_osds.push_back((inode_list_osd_t){ - .pg = r, - .osd_num = peer_osd, - .sent = false, - }); - } - else - { - inactive_osd_set.insert(peer_osd); - } - } - } - // FIXME: We should retry (or at least OPTIONALLY retry) connecting during delete, but only _for_some_time_ - // FIXME: Also we should handle the case when the OSD is disconnected during delete - else if (st_cli.peer_states.find(pg.cur_primary) != st_cli.peer_states.end()) - { - // Clean - r->list_osds.push_back((inode_list_osd_t){ - .pg = r, - .osd_num = pg.cur_primary, - .sent = false, - }); - } - else - { - inactive_osd_set.insert(pg.cur_primary); - } - lst->pgs.push_back(r); - } - std::sort(lst->pgs.begin(), lst->pgs.end(), [](inode_list_pg_t *a, inode_list_pg_t *b) - { - return a->cur_primary < b->cur_primary ? true : false; - }); - for (int i = 0; i < lst->pgs.size(); i++) - { - lst->pgs[i]->pos = i; - } - lst->inactive_osds.insert(lst->inactive_osds.end(), inactive_osd_set.begin(), inactive_osd_set.end()); + lst->max_parallel_pgs = max_parallel_pgs <= 0 ? 16 : max_parallel_pgs; lists.push_back(lst); + if (!continue_listing(lst)) + { + return NULL; + } return lst; } int cluster_client_t::list_pg_count(inode_list_t *lst) { - return lst->pgs.size(); + return lst->pgs.size() - lst->done_pgs; } -const std::vector & cluster_client_t::list_inode_get_inactive_osds(inode_list_t *lst) +void cluster_client_t::list_inode_next(inode_list_t *lst) { - return lst->inactive_osds; -} - -const std::vector & cluster_client_t::list_inode_get_inactive_pgs(inode_list_t *lst) -{ - return lst->inactive_pgs; -} - -void cluster_client_t::list_inode_next(inode_list_t *lst, int next_pgs) -{ - if (next_pgs >= 0) - { - lst->want += next_pgs; - } continue_listing(lst); } -void cluster_client_t::continue_listing(inode_list_t *lst) +bool cluster_client_t::continue_listing(inode_list_t *lst) { - if (lst->done_pgs >= lst->pgs.size()) - { - return; - } - if (lst->want <= 0) - { - return; - } if (lst->onstack > 0) { - return; + return true; } lst->onstack++; - for (int i = 0; i < lst->pgs.size(); i++) + if (restart_listing(lst)) { - if (!lst->pgs[i]) + for (int i = 0; i < lst->pgs.size() && lst->inflight_pgs < lst->max_parallel_pgs; i++) { + retry_start_pg_listing(lst->pgs[i]); } - else if (lst->pgs[i]->sent < lst->pgs[i]->list_osds.size()) + } + if (check_finish_listing(lst)) + { + // Do not change lst->onstack because it's already freed + return false; + } + lst->onstack--; + return true; +} + +bool cluster_client_t::restart_listing(inode_list_t* lst) +{ + auto pool_it = st_cli.pool_config.find(lst->pool_id); + // We want listing to be consistent. To achieve it we should: + // 1) retry listing of each PG if its state changes + // 2) abort listing if PG count changes during listing + // 3) ideally, only talk to the primary OSD - this will be done separately + // So first we add all PGs without checking their state + if (pool_it == st_cli.pool_config.end() || + lst->real_pg_count != pool_it->second.real_pg_count) + { + for (auto pg: lst->pgs) { - for (int j = 0; j < lst->pgs[i]->list_osds.size(); j++) + if (pg->inflight_ops > 0) { - send_list(&lst->pgs[i]->list_osds[j]); - if (lst->want <= 0) - { - lst->onstack--; - return; - } + // Wait until all in-progress listings complete or fail + return false; } } - else if (!lst->pgs[i]->list_osds.size()) + for (auto pg: lst->pgs) { - lst->pgs[i]->errcode = -EIO; - finish_list_pg(lst->pgs[i]); - if (check_finish_listing(lst)) + delete pg; + } + if (log_level > 0 && lst->real_pg_count) + { + fprintf(stderr, "PG count in pool %u changed during listing\n", lst->pool_id); + } + lst->pgs.clear(); + if (pool_it == st_cli.pool_config.end()) + { + // Unknown pool + lst->callback(lst, std::set(), 0, std::vector(), -EINVAL, INODE_LIST_DONE); + return false; + } + else if (lst->done_pgs) + { + // PG count changed during listing, it should fail + lst->callback(lst, std::set(), 0, std::vector(), -EAGAIN, INODE_LIST_DONE); + return false; + } + else + { + lst->real_pg_count = pool_it->second.real_pg_count; + for (pg_num_t pg_num = 1; pg_num <= lst->real_pg_count; pg_num++) { - // Do not change lst->onstack because it's already freed - return; + inode_list_pg_t *pg = new inode_list_pg_t(); + pg->lst = lst; + pg->pg_num = pg_num; + lst->pgs.push_back(pg); } } } - lst->onstack--; + return true; +} + +void cluster_client_t::retry_start_pg_listing(inode_list_pg_t *pg) +{ + if (pg->state == LIST_PG_SENT || pg->state == LIST_PG_DONE) + { + return; + } + int new_st = start_pg_listing(pg); + if (new_st == LIST_PG_SENT) + { + pg->state = LIST_PG_SENT; + return; + } + if (new_st == LIST_PG_WAIT_ACTIVE && pg->state != LIST_PG_WAIT_ACTIVE || + new_st == LIST_PG_WAIT_CONNECT && pg->state != LIST_PG_WAIT_CONNECT) + { + int sec = (new_st == LIST_PG_WAIT_ACTIVE ? wait_up_timeout : peer_connect_timeout); + if (sec) + { + pg->state = new_st; + clock_gettime(CLOCK_REALTIME, &pg->wait_until); + pg->wait_until.tv_sec += sec; + if (new_st == LIST_PG_WAIT_ACTIVE) + { + if (log_level > 1) + fprintf(stderr, "Waiting for PG %u/%u to become active for %d seconds\n", pg->lst->pool_id, pg->pg_num, wait_up_timeout); + } + else + { + if (log_level > 2) + fprintf(stderr, "Waiting for connection to PG %u/%u OSDs for %d seconds\n", pg->lst->pool_id, pg->pg_num, peer_connect_timeout); + } + if (!list_retry_time.tv_sec || list_retry_time.tv_sec > pg->wait_until.tv_sec || + list_retry_time.tv_sec == pg->wait_until.tv_sec && list_retry_time.tv_nsec > pg->wait_until.tv_nsec) + { + list_retry_time = pg->wait_until; + if (list_retry_timeout_id >= 0) + { + tfd->clear_timer(list_retry_timeout_id); + } + list_retry_timeout_id = tfd->set_timer(sec*1000, false, [this](int timer_id) + { + continue_lists(); + }); + } + return; + } + } + assert(pg->state == LIST_PG_WAIT_ACTIVE || pg->state == LIST_PG_WAIT_CONNECT); + // Check if the timeout expired + timespec tv; + clock_gettime(CLOCK_REALTIME, &tv); + if (tv.tv_sec > pg->wait_until.tv_sec || + tv.tv_sec == pg->wait_until.tv_sec && tv.tv_nsec >= pg->wait_until.tv_nsec) + { + fprintf(stderr, "Failed to wait for PG %u/%u to become %s, skipping listing\n", pg->lst->pool_id, pg->pg_num, + pg->state == LIST_PG_WAIT_ACTIVE ? "active" : "connected"); + pg->errcode = -EPIPE; + pg->list_osds.clear(); + pg->objects.clear(); + finish_list_pg(pg); + } +} + +int cluster_client_t::start_pg_listing(inode_list_pg_t *pg) +{ + auto & pool_cfg = st_cli.pool_config[pg->lst->pool_id]; + auto pg_it = pool_cfg.pg_config.find(pg->pg_num); + assert(pg->lst->real_pg_count == pool_cfg.real_pg_count); + if (pg_it == pool_cfg.pg_config.end() || + pg_it->second.pause || + !pg_it->second.cur_primary || + !(pg_it->second.cur_state & PG_ACTIVE)) + { + // PG is (temporarily?) unavailable + return LIST_PG_WAIT_ACTIVE; + } + pg->inactive_osds.clear(); + std::set all_peers; + if (pg_it->second.cur_state != PG_ACTIVE) + { + // Not clean + for (osd_num_t pg_osd: pg_it->second.target_set) + all_peers.insert(pg_osd); + for (osd_num_t pg_osd: pg_it->second.all_peers) + all_peers.insert(pg_osd); + for (auto & hist_item: pg_it->second.target_history) + for (auto pg_osd: hist_item) + all_peers.insert(pg_osd); + // Remove zero OSD number + all_peers.erase(0); + // Remove unconnectable peers except cur_primary + for (auto peer_it = all_peers.begin(); peer_it != all_peers.end(); ) + { + if (*peer_it != pg_it->second.cur_primary && + st_cli.peer_states[*peer_it].is_null()) + { + pg->inactive_osds.push_back(*peer_it); +printf("OSD is inactive: %lu\n", *peer_it); + all_peers.erase(peer_it++); + } + else + peer_it++; + } + } + else + { + // Clean + all_peers.insert(pg_it->second.cur_primary); + } + // Check that we're connected to all PG OSDs + bool conn = true; + for (osd_num_t peer_osd: all_peers) + { + if (msgr.osd_peer_fds.find(peer_osd) == msgr.osd_peer_fds.end()) + { + // Initiate connection + if (st_cli.peer_states[peer_osd].is_null()) + { + return LIST_PG_WAIT_ACTIVE; + } + msgr.connect_peer(peer_osd, st_cli.peer_states[peer_osd]); + conn = false; + } + } + if (!conn) + { + return LIST_PG_WAIT_CONNECT; + } + // Send all listings at once as the simplest way to guarantee that we connect + // to the exact same OSDs that are listed in PG state + pg->errcode = 0; + pg->list_osds.clear(); + pg->has_unstable = false; + pg->objects.clear(); + pg->cur_primary = pg_it->second.cur_primary; + for (osd_num_t peer_osd: all_peers) + { + pg->list_osds.push_back((inode_list_osd_t){ + .pg = pg, + .osd_num = peer_osd, + }); + } + for (auto & list_osd: pg->list_osds) + { + send_list(&list_osd); + } + return LIST_PG_SENT; } void cluster_client_t::send_list(inode_list_osd_t *cur_list) { - if (cur_list->sent) - { - return; - } - if (msgr.osd_peer_fds.find(cur_list->osd_num) == msgr.osd_peer_fds.end()) - { - // Initiate connection - msgr.connect_peer(cur_list->osd_num, st_cli.peer_states[cur_list->osd_num]); - return; - } + if (!cur_list->pg->inflight_ops) + cur_list->pg->lst->inflight_pgs++; + cur_list->pg->inflight_ops++; auto & pool_cfg = st_cli.pool_config[cur_list->pg->lst->pool_id]; osd_op_t *op = new osd_op_t(); op->op_type = OSD_OP_OUT; @@ -290,27 +377,24 @@ void cluster_client_t::send_list(inode_list_osd_t *cur_list) } } delete op; - auto lst = cur_list->pg->lst; - cur_list->pg->done++; + cur_list->pg->inflight_ops--; + if (!cur_list->pg->inflight_ops) + cur_list->pg->lst->inflight_pgs--; + // FIXME: Retry listing after ms on EPIPE finish_list_pg(cur_list->pg); - if (!check_finish_listing(lst)) - { - continue_listing(lst); - } + continue_listing(cur_list->pg->lst); }; msgr.outbox_push(op); - cur_list->sent = true; - cur_list->pg->sent++; - cur_list->pg->lst->want--; } void cluster_client_t::finish_list_pg(inode_list_pg_t *pg) { auto lst = pg->lst; - if (pg->done >= pg->list_osds.size()) + if (pg->inflight_ops == 0) { - int status = 0; lst->done_pgs++; + pg->state = LIST_PG_DONE; + int status = 0; if (lst->done_pgs >= lst->pgs.size()) { status |= INODE_LIST_DONE; @@ -319,13 +403,15 @@ void cluster_client_t::finish_list_pg(inode_list_pg_t *pg) { status |= INODE_LIST_HAS_UNSTABLE; } - lst->pgs[pg->pos] = NULL; - lst->callback(lst, std::move(pg->objects), pg->pg_num, pg->cur_primary, pg->errcode, status); - delete pg; + lst->callback(lst, std::move(pg->objects), pg->pg_num, std::move(pg->inactive_osds), pg->errcode, status); } - else +} + +void cluster_client_t::continue_lists() +{ + for (int i = lists.size()-1; i >= 0; i--) { - lst->want++; + continue_listing(lists[i]); } } @@ -333,7 +419,11 @@ bool cluster_client_t::check_finish_listing(inode_list_t *lst) { if (lst->done_pgs >= lst->pgs.size()) { - // All done + for (auto pg: lst->pgs) + { + delete pg; + } + lst->pgs.clear(); for (int i = 0; i < lists.size(); i++) { if (lists[i] == lst) @@ -347,11 +437,3 @@ bool cluster_client_t::check_finish_listing(inode_list_t *lst) } return false; } - -void cluster_client_t::continue_lists() -{ - for (auto lst: lists) - { - continue_listing(lst); - } -} diff --git a/src/cmd/cli.cpp b/src/cmd/cli.cpp index a13f4c58..c917e383 100644 --- a/src/cmd/cli.cpp +++ b/src/cmd/cli.cpp @@ -98,10 +98,11 @@ static const char* help_text = "\n" "vitastor-cli rm-data --pool --inode [--wait-list] [--min-offset ]\n" " Remove inode data without changing metadata.\n" - " --wait-list Retrieve full objects listings before starting to remove objects.\n" - " Requires more memory, but allows to show correct removal progress.\n" - " --min-offset Purge only data starting with specified offset.\n" - " --max-offset Purge only data before specified offset.\n" + " --wait-list Retrieve full objects listings before starting to remove objects.\n" + " Requires more memory, but allows to show correct removal progress.\n" + " --min-offset Purge only data starting with specified offset.\n" + " --max-offset Purge only data before specified offset.\n" + " --wait_up_timeout 16 Timeout for waiting until PGs are up in seconds.\n" "\n" "vitastor-cli merge-data [--target ]\n" " Merge layer data without changing metadata. Merge .. to .\n" diff --git a/src/cmd/cli.h b/src/cmd/cli.h index c392c9a5..cf022e85 100644 --- a/src/cmd/cli.h +++ b/src/cmd/cli.h @@ -30,7 +30,7 @@ struct cli_result_t class cli_tool_t { public: - uint64_t iodepth = 4, parallel_osds = 32; + uint64_t iodepth = 32, parallel_osds = 4; bool progress = false; bool list_first = false; bool json_output = false; diff --git a/src/cmd/cli_merge.cpp b/src/cmd/cli_merge.cpp index bcb6f93e..60f6d97c 100644 --- a/src/cmd/cli_merge.cpp +++ b/src/cmd/cli_merge.cpp @@ -389,8 +389,8 @@ struct snap_merger_t if (lower ? (sp.second < target_rank) : (sp.second > target_rank)) { lists_todo++; - inode_list_t* lst = parent->cli->list_inode_start(src, [this, src]( - inode_list_t *lst, std::set&& objects, pg_num_t pg_num, osd_num_t primary_osd, int errcode, int status) + inode_list_t* lst = parent->cli->list_inode_start(src, parent->parallel_osds, [this, src]( + inode_list_t *lst, std::set&& objects, pg_num_t pg_num, std::vector && inactive_osds, int errcode, int status) { if (errcode) { @@ -439,10 +439,10 @@ struct snap_merger_t } else { - parent->cli->list_inode_next(lst, 1); + parent->cli->list_inode_next(lst); } }); - parent->cli->list_inode_next(lst, parent->parallel_osds); + parent->cli->list_inode_next(lst); } } } diff --git a/src/cmd/cli_rm_data.cpp b/src/cmd/cli_rm_data.cpp index 8ab6aca6..66f8c2b0 100644 --- a/src/cmd/cli_rm_data.cpp +++ b/src/cmd/cli_rm_data.cpp @@ -31,11 +31,11 @@ struct rm_inode_t cli_tool_t *parent = NULL; inode_list_t *lister = NULL; std::vector lists; - std::vector inactive_osds; + std::set inactive_osds; std::set inactive_pgs; uint64_t total_count = 0, total_done = 0, total_prev_pct = 0; - uint64_t pgs_to_list = 0; bool lists_done = false; + int pgs_to_list = 0; int state = 0; int error_count = 0; @@ -43,47 +43,70 @@ struct rm_inode_t void start_delete() { - lister = parent->cli->list_inode_start(inode, [this](inode_list_t *lst, - std::set&& objects, pg_num_t pg_num, osd_num_t primary_osd, int errcode, int status) + lister = parent->cli->list_inode_start(inode, parent->parallel_osds, [this](inode_list_t *lst, + std::set&& objects, pg_num_t pg_num, std::vector && inactive_osds, int errcode, int status) { + osd_num_t rm_osd_num = 0; + auto pool_it = parent->cli->st_cli.pool_config.find(pool_id); + if (pool_it != parent->cli->st_cli.pool_config.end()) + { + auto pg_it = pool_it->second.pg_config.find(pg_num); + if (pg_it != pool_it->second.pg_config.end()) + rm_osd_num = pg_it->second.primary; + } + if (!rm_osd_num) + { + fprintf(stderr, "PG %u/%u is down, skipping\n", pool_id, pg_num); + errcode = -EPIPE; + } if (errcode) { inactive_pgs.insert(pg_num); } - rm_pg_t *rm = new rm_pg_t((rm_pg_t){ - .pg_num = pg_num, - .rm_osd_num = primary_osd, - .objects = objects, - .obj_count = objects.size(), - .obj_done = 0, - .synced = !objects.size() || parent->cli->get_immediate_commit(inode), - }); - if (min_offset == 0 && max_offset == 0) - { - total_count += objects.size(); - } else { - for (object_id oid: objects) + for (auto osd_num: inactive_osds) { - if (oid.stripe >= min_offset && (!max_offset || oid.stripe < max_offset)) + this->inactive_osds.insert(osd_num); + } + rm_pg_t *rm = new rm_pg_t((rm_pg_t){ + .pg_num = pg_num, + .rm_osd_num = rm_osd_num, + .objects = std::move(objects), + .obj_done = 0, + .synced = !objects.size() || parent->cli->get_immediate_commit(inode), + }); + if (min_offset == 0 && max_offset == 0) + { + total_count += objects.size(); + } + else + { + for (object_id oid: objects) { - total_count++; + if (oid.stripe >= min_offset && (!max_offset || oid.stripe < max_offset)) + { + total_count++; + } } } + rm->obj_pos = rm->objects.begin(); + lists.push_back(rm); } - rm->obj_pos = rm->objects.begin(); - lists.push_back(rm); if (parent->list_first && !(status & INODE_LIST_DONE)) { // The listing object is dead when DONE => don't call next() - parent->cli->list_inode_next(lister, 1); + parent->cli->list_inode_next(lister); } if (status & INODE_LIST_DONE) { lists_done = true; + pgs_to_list = 0; + } + else + { + pgs_to_list = parent->cli->list_pg_count(lister); } - pgs_to_list--; continue_delete(); }); if (!lister) @@ -96,18 +119,8 @@ struct rm_inode_t state = 100; return; } - inactive_osds = parent->cli->list_inode_get_inactive_osds(lister); - if (inactive_osds.size() && !parent->json_output) - { - fprintf(stderr, "Some data may remain after delete on OSDs which are currently down: "); - for (int i = 0; i < inactive_osds.size(); i++) - { - fprintf(stderr, i > 0 ? ", %ju" : "%ju", inactive_osds[i]); - } - fprintf(stderr, "\n"); - } pgs_to_list = parent->cli->list_pg_count(lister); - parent->cli->list_inode_next(lister, parent->parallel_osds); + parent->cli->list_inode_next(lister); } void send_ops(rm_pg_t *cur_list) @@ -209,7 +222,7 @@ struct rm_inode_t i--; if (!lists_done) { - parent->cli->list_inode_next(lister, 1); + parent->cli->list_inode_next(lister); } } else @@ -220,8 +233,8 @@ struct rm_inode_t if (parent->progress && total_count > 0 && total_done*1000/total_count != total_prev_pct) { fprintf(stderr, parent->color - ? "\rRemoved %ju/%ju objects, %ju more PGs to list..." - : "Removed %ju/%ju objects, %ju more PGs to list...\n", total_done, total_count, pgs_to_list); + ? "\rRemoved %ju/%ju objects, %u more PGs to list..." + : "Removed %ju/%ju objects, %u more PGs to list...\n", total_done, total_count, pgs_to_list); total_prev_pct = total_done*1000/total_count; } if (lists_done && !lists.size()) @@ -230,6 +243,16 @@ struct rm_inode_t { fprintf(stderr, "\n"); } + if (inactive_osds.size() && !parent->json_output) + { + fprintf(stderr, "Some data may remain after delete on OSDs which are currently down: "); + int i = 0; + for (auto osd_num: inactive_osds) + { + fprintf(stderr, (i++) ? ", %ju" : "%ju", osd_num); + } + fprintf(stderr, "\n"); + } if (inactive_pgs.size() && !parent->json_output) { fprintf(stderr, "Failed to list some PGs, deletion is not complete: PG "); @@ -240,33 +263,45 @@ struct rm_inode_t } fprintf(stderr, "\n"); } - bool is_error = (total_done < total_count || inactive_osds.size() > 0 || inactive_pgs.size() > 0 || error_count > 0); - if (parent->progress && is_error) - { - fprintf( - stderr, "Warning: Pool:%u,ID:%ju inode data may not have been fully removed.\n" - "Use `vitastor-cli rm-data --pool %u --inode %ju` if you encounter it in listings.\n", - pool_id, INODE_NO_POOL(inode), pool_id, INODE_NO_POOL(inode) - ); - } json11::Json::array inactive_pgs_json; for (auto pg_num: inactive_pgs) { inactive_pgs_json.push_back((uint64_t)pg_num); } - result = (cli_result_t){ - .err = is_error && !down_ok ? EIO : 0, - .text = is_error ? "Some blocks were not removed" : ( - "Done, inode "+std::to_string(INODE_NO_POOL(inode))+" from pool "+ - std::to_string(pool_id)+" removed"), - .data = json11::Json::object { - { "removed_objects", total_done }, - { "total_objects", total_count }, - { "inactive_osds", inactive_osds }, - { "inactive_pgs", inactive_pgs_json }, - }, + json11::Json data = json11::Json::object { + { "removed_objects", total_done }, + { "total_objects", total_count }, + { "inactive_osds", json11::Json::array(inactive_osds.begin(), inactive_osds.end()) }, + { "inactive_pgs", inactive_pgs_json }, }; state = 100; + if (total_done < total_count || inactive_pgs.size() > 0 || error_count > 0 || + inactive_osds.size() > 0 && !down_ok) + { + // Error + result = (cli_result_t){ + .err = EIO, + .text = "Some blocks were not removed", + .data = data, + }; + } + else + { + if (parent->progress && inactive_osds.size() > 0 && down_ok) + { + fprintf( + stderr, "Warning: --down-ok is set and some OSDs are down.\n" + "Pool:%u,ID:%ju inode data may not have been fully removed.\n" + "Use `vitastor-cli rm-data --pool %u --inode %ju` if you encounter it in listings.\n", + pool_id, INODE_NO_POOL(inode), pool_id, INODE_NO_POOL(inode) + ); + } + result = (cli_result_t){ + .text = "Done, inode "+std::to_string(INODE_NO_POOL(inode))+" from pool "+ + std::to_string(pool_id)+" removed", + .data = data, + }; + } } } diff --git a/tests/test_rm_degraded.sh b/tests/test_rm_degraded.sh index 0296bf09..6c24f767 100755 --- a/tests/test_rm_degraded.sh +++ b/tests/test_rm_degraded.sh @@ -22,4 +22,23 @@ if ! ( build/src/cmd/vitastor-cli --etcd_address $ETCD_URL ls | grep testimg | g format_error "Image should be marked as partially deleted" fi +start_osd 3 +sleep 5 + +# Now do the same but without del /vitastor/osd/state + +LD_PRELOAD="build/src/client/libfio_vitastor.so" \ + fio -thread -name=test -ioengine=build/src/client/libfio_vitastor.so -bs=4M -direct=1 -iodepth=1 \ + -end_fsync=1 -fsync=1 -rw=write -etcd=$ETCD_URL -image=testimg -size=128M -cluster_log_level=10 + +kill -9 $OSD3_PID + +if build/src/cmd/vitastor-cli --etcd_address $ETCD_URL rm testimg --log_level 10 ; then + format_error "Delete should not be successful with inactive OSDs" +fi + +if ! ( build/src/cmd/vitastor-cli --etcd_address $ETCD_URL ls | grep testimg | grep DEL ) ; then + format_error "Image should be marked as partially deleted" +fi + format_green OK