Adding a new progress tracking class in the cephfs offline tools dir that is consumed by multiple
utilities namely cephfs-data-scan, cephfs-journal-tool, cephfs-table-tool.
Multiple commands under each utility now have progress tracking included by default.
The progress tracking is not option based- i.e. no "--progress" option required.
This is because if we keep it option based and the user forgets to include the option while
firing the command, then the only recourse is to kill the execution and restart. This is not
suitable for state changing operations, hence better to include progress tracking by default.
Exact command list where progress tracking is included is specified in the tracker.
Testing updates also posted in tracker, initial testing with single and multi-threaded
loads on FS with 4M objects succeeded for cephfs-data-scan.
- Updated DataScan to use worker_n and worker_m for better clarity and functionality.
- Added set_progress_operation_name method to manage operation names during progress tracking.
- Enhanced ProgressTracker with new methods for detailed progress reporting and ETA calculations.
- Improved thread safety and performance in progress updates.
- Added terminal width detection and line padding for better display.
- Introduced write_console_line function to manage console output.
- Updated progress display logic to throttle updates based on time.
- Removed display_interval variable to streamline progress updates.
Fixes: https://tracker.ceph.com/issues/63191
Signed-off-by: Edwin Rodriguez <edwin.rodriguez1@ibm.com>
Signed-off-by: Edwin Rodriguez <edwin.rodriguez1@ibm.com>
list(APPEND BOOST_COMPONENTS locale)
endif()
+if(WITH_CEPHFS)
+ list(APPEND BOOST_COMPONENTS filesystem)
+endif()
+
set(Boost_USE_MULTITHREADED ON)
CMAKE_DEPENDENT_OPTION(WITH_BOOST_VALGRIND "Boost support for valgrind" OFF
return r;
}
+std::string
+to_pretty_timedelta(timespan duration)
+{
+ using namespace std::chrono;
+
+ auto duration_seconds = duration_cast<seconds>(duration).count();
+
+ if (duration < seconds{120}) {
+ return fmt::format("{}s", duration_seconds);
+ }
+ if (duration < minutes{120}) {
+ return fmt::format("{}m", duration_seconds / 60);
+ }
+ if (duration < hours{48}) {
+ return fmt::format("{}h", duration_seconds / 3600);
+ }
+ if (duration < hours{24 * 14}) {
+ return fmt::format("{}d", duration_seconds / (3600 * 24));
+ }
+ if (duration < hours{24 * 7 * 12}) {
+ return fmt::format("{}w", duration_seconds / (3600 * 24 * 7));
+ }
+ if (duration < hours{24 * 365 * 2}) {
+ return fmt::format("{}M", duration_seconds / (3600 * 24 * 30));
+ }
+ return fmt::format("{}y", duration_seconds / (3600 * 24 * 365));
+}
}
namespace std {
std::string exact_timespan_str(timespan t);
std::chrono::seconds parse_timespan(const std::string& s);
+/**
+ * Convert a duration to a human-readable string representation.
+ *
+ * Formats the duration using the largest appropriate time unit:
+ * - seconds (s) for durations < 2 minutes
+ * - minutes (m) for durations < 2 hours
+ * - hours (h) for durations < 2 days
+ * - days (d) for durations < 2 weeks
+ * - weeks (w) for durations < ~3 months
+ * - months (M) for durations < 2 years
+ * - years (y) for durations >= 2 years
+ *
+ * @param duration The time duration to format
+ * @return A formatted string (e.g., "90m", "3h", "5d", "2y")
+*/
+std::string to_pretty_timedelta(timespan duration);
+
// detects presence of Clock::to_timespec() and from_timespec()
template <typename Clock, typename = std::void_t<>>
struct converts_to_timespec : std::false_type {};
set(cephfs_journal_tool_srcs
cephfs-journal-tool.cc
JournalTool.cc
+ ProgressTracker.cc
JournalFilter.cc
JournalScanner.cc
EventOutput.cc
add_executable(cephfs-journal-tool ${cephfs_journal_tool_srcs})
target_link_libraries(cephfs-journal-tool
legacy-option-headers
- librados mds osdc global
- ${BLKID_LIBRARIES} ${CMAKE_DL_LIBS})
+ ceph-common librados mds osdc global uring::uring
+ ${BLKID_LIBRARIES} ${CMAKE_DL_LIBS} Boost::filesystem)
set(cephfs-meta-injection_srcs
cephfs-meta-injection.cc
add_executable(cephfs-meta-injection ${cephfs-meta-injection_srcs})
target_link_libraries(cephfs-meta-injection
legacy-option-headers
- librados mds osdc global
+ ceph-common librados mds osdc global uring::uring
${BLKID_LIBRARIES} ${CMAKE_DL_LIBS})
set(cephfs_table_tool_srcs
cephfs-table-tool.cc
TableTool.cc
+ ProgressTracker.cc
RoleSelector.cc
MDSUtility.cc)
add_executable(cephfs-table-tool ${cephfs_table_tool_srcs})
-target_link_libraries(cephfs-table-tool librados mds osdc global
- ${BLKID_LIBRARIES} ${CMAKE_DL_LIBS})
+target_link_libraries(cephfs-table-tool ceph-common librados mds osdc global uring::uring
+ ${BLKID_LIBRARIES} ${CMAKE_DL_LIBS} Boost::filesystem)
set(cephfs_data_scan_srcs
cephfs-data-scan.cc
DataScan.cc
+ ProgressTracker.cc
RoleSelector.cc
PgFiles.cc
MDSUtility.cc)
add_executable(cephfs-data-scan ${cephfs_data_scan_srcs})
-target_link_libraries(cephfs-data-scan librados cephfs mds osdc global
+target_link_libraries(cephfs-data-scan ceph-common librados cephfs mds osdc global uring::uring
cls_cephfs_client
- ${BLKID_LIBRARIES} ${CMAKE_DL_LIBS})
+ ${BLKID_LIBRARIES} ${CMAKE_DL_LIBS} Boost::filesystem)
install(TARGETS
cephfs-journal-tool
#include "DataScan.h"
-#include "include/compat.h"
+#include <fmt/format.h>
+
+#include <fstream>
+
#include "common/debug.h"
-#include "common/errno.h"
+
+#include "cls/cephfs/cls_cephfs_client.h"
#include "common/ceph_argparse.h"
-#include <fstream>
-#include "include/util.h"
+#include "common/errno.h"
#include "include/ceph_fs.h"
-
+#include "include/compat.h"
+#include "include/util.h"
#include "mds/CDentry.h"
#include "mds/CInode.h"
-#include "mds/CDentry.h"
#include "mds/InoTable.h"
-#include "mds/snap.h" // for struct sr_t
#include "mds/SnapServer.h"
-#include "cls/cephfs/cls_cephfs_client.h"
+#include "mds/snap.h" // for struct sr_t
#include "PgFiles.h"
-#include "include/compat.h"
#define dout_context g_ceph_context
#define dout_subsys ceph_subsys_mds
return true;
} else if (arg == std::string("--worker_n")) {
std::string err;
- n = strict_strtoll(val.c_str(), 10, &err);
+ worker_n = strict_strtoll(val.c_str(), 10, &err);
if (!err.empty()) {
std::cerr << "Invalid worker number '" << val << "'" << std::endl;
*r = -EINVAL;
return true;
} else if (arg == std::string("--worker_m")) {
std::string err;
- m = strict_strtoll(val.c_str(), 10, &err);
+ worker_m = strict_strtoll(val.c_str(), 10, &err);
if (!err.empty()) {
std::cerr << "Invalid worker count '" << val << "'" << std::endl;
*r = -EINVAL;
} else if (arg == std::string("--alternate-pool")) {
metadata_pool_name = val;
return true;
+ } else if (arg == std::string("--progress_update_interval")) {
+ std::string err;
+ progress_update_interval = std::chrono::seconds(strict_strtoll(val.c_str(), 10, &err));
+ if (!err.empty()) {
+ std::cerr << "Invalid progress update interval '" << val << "'" << std::endl;
+ *r = -EINVAL;
+ return false;
+ }
+ return true;
} else {
return false;
}
// Common RADOS init: open metadata pool
// =====================================
- librados::Rados rados;
int r = rados.init_with_context(g_ceph_context);
if (r < 0) {
derr << "RADOS unavailable" << dendl;
return r;
}
+
std::string const &command = args[0];
std::string data_pool_name;
std::set<std::string> extra_data_pool_names;
return 0;
}
+std::string
+DataScan::get_progress_operation_name(std::string_view op_name) const
+{
+ if (worker_m > 1) {
+ return fmt::format("{} {}/{}", op_name, worker_n, worker_m);
+ } else {
+ return std::string(op_name);
+ }
+}
int DataScan::scan_extents()
{
data_ios.push_back(&extra_data_io);
}
+ uint64_t total_objects = get_pool_objects(data_ios);
+ auto progress_tracker = std::make_unique<ProgressTracker>(get_progress_operation_name("scan_extents"));
+ progress_tracker->set_enable_progress_update(true);
+ progress_tracker->start(total_objects);
+
+ if (total_objects == 0) {
+ dout(4) << "No objects found in data pools" << dendl;
+ return 0;
+ }
+
for (auto ioctx : data_ios) {
- int r = forall_objects(*ioctx, false, [this, ioctx](
+ int r = forall_objects(*ioctx, false, [this, ioctx, &progress_tracker](
std::string const &oid,
uint64_t obj_name_ino,
uint64_t obj_name_offset) -> int
return r;
}
+ progress_tracker->increment();
+ progress_tracker->display_progress();
+
return r;
});
if (r < 0) {
return r;
}
}
-
return 0;
}
librados::ObjectCursor range_i;
librados::ObjectCursor range_end;
ioctx.object_list_slice(
- ioctx.object_list_begin(),
- ioctx.object_list_end(),
- n,
- m,
- &range_i,
- &range_end);
+ ioctx.object_list_begin(), ioctx.object_list_end(), worker_n, worker_m,
+ &range_i, &range_end);
bufferlist filter_bl;
return r;
}
+uint64_t
+DataScan::get_pool_objects(const std::vector<librados::IoCtx*>& data_ios)
+{
+ uint64_t total = 0;
+ std::list<std::string> pool_names;
+
+ for (auto ioctx : data_ios) {
+ if (!ioctx)
+ continue;
+ pool_names.push_back(ioctx->get_pool_name());
+ }
+
+ librados::stats_map stats;
+ int ret = rados.get_pool_stats(pool_names, stats);
+ if (ret < 0) {
+ dout(1) << "Failed to get pool stats: " << cpp_strerror(ret) << dendl;
+ return 0;
+ }
+
+ for (const auto& stat : stats) {
+ dout(20) << "Pool " << stat.first << " has " << stat.second.num_objects
+ << " objects" << dendl;
+ total += stat.second.num_objects;
+ }
+
+ if (worker_m > 1) {
+ // For multi-worker scenarios, estimate this worker's share
+ // Each worker processes roughly 1/m of the total objects
+ uint64_t estimated_worker_objects = total / worker_m;
+ dout(4) << "Worker " << worker_n << "/" << worker_m
+ << " estimated to process ~" << estimated_worker_objects
+ << " objects (out of " << total << " total)" << dendl;
+ return estimated_worker_objects;
+ }
+
+ dout(4) << "Single worker processing " << total << " objects" << dendl;
+ return total;
+}
+
+uint64_t
+DataScan::get_metadata_pool_objects(
+ librados::IoCtx& metadata_io,
+ bool worker_sliced)
+{
+ uint64_t total = 0;
+ std::list<std::string> pool_names;
+ pool_names.push_back(metadata_io.get_pool_name());
+
+ librados::stats_map stats;
+ int ret = rados.get_pool_stats(pool_names, stats);
+ if (ret < 0) {
+ dout(1) << "Failed to get metadata pool stats: " << cpp_strerror(ret)
+ << dendl;
+ return 0;
+ }
+
+ for (const auto& stat : stats) {
+ dout(20) << "Metadata pool " << stat.first << " has "
+ << stat.second.num_objects << " objects" << dendl;
+ total += stat.second.num_objects;
+ }
+
+ if (worker_sliced && worker_m > 1) {
+ // For operations that use forall_objects() with worker slicing
+ uint64_t estimated_worker_objects = total / worker_m;
+ dout(4) << "Worker " << worker_n << "/" << worker_m
+ << " estimated to process ~" << estimated_worker_objects
+ << " metadata objects (out of " << total << " total, worker-sliced)"
+ << dendl;
+ return estimated_worker_objects;
+ } else {
+ // For operations that process ALL metadata objects in each worker (like scan_links)
+ dout(4) << "Worker " << worker_n << "/" << worker_m << " processing all "
+ << total << " metadata objects (no worker slicing)" << dendl;
+ return total;
+ }
+}
+
int DataScan::scan_inodes()
{
bool roots_present;
return -EIO;
}
- return forall_objects(data_io, true, [this](
- std::string const &oid,
- uint64_t obj_name_ino,
- uint64_t obj_name_offset) -> int
- {
- int r = 0;
+ uint64_t total_objects = get_pool_objects({&data_io});
+ auto progress_tracker = std::make_unique<ProgressTracker>(get_progress_operation_name("scan_inodes"));
+ progress_tracker->set_enable_progress_update(true);
+ progress_tracker->start(total_objects);
+
+ if (total_objects == 0) {
+ dout(4) << "No objects found in data pool" << dendl;
+ return 0;
+ }
+
+ r = forall_objects(
+ data_io, true,
+ [this, &progress_tracker](
+ std::string const& oid, uint64_t obj_name_ino,
+ uint64_t obj_name_offset) -> int {
+ int r = 0;
dout(10) << "handling object "
<< std::hex << obj_name_ino << "." << obj_name_offset << std::dec
}
}
- return r;
- });
+ progress_tracker->increment();
+ progress_tracker->display_progress();
+
+ return r;
+ });
+ return r;
}
int DataScan::cleanup()
{
+ uint64_t total_objects = get_pool_objects({&data_io});
+ auto progress_tracker = std::make_unique<ProgressTracker>(get_progress_operation_name("scan_cleanup"));
+ progress_tracker->set_enable_progress_update(true);
+ progress_tracker->start(total_objects);
+
+ if (total_objects == 0) {
+ dout(4) << "No objects found in data pool" << dendl;
+ return 0;
+ }
+
// We are looking for only zeroth object
- //
- return forall_objects(data_io, true, [this](
- std::string const &oid,
- uint64_t obj_name_ino,
- uint64_t obj_name_offset) -> int
- {
- int r = 0;
- r = ClsCephFSClient::delete_inode_accumulate_result(data_io, oid);
- if (r < 0) {
- dout(4) << "Error deleting accumulated metadata from '"
- << oid << "': " << cpp_strerror(r) << dendl;
- }
- return r;
+ return forall_objects(
+ data_io, true,
+ [this, &progress_tracker](
+ std::string const& oid, uint64_t obj_name_ino,
+ uint64_t obj_name_offset) -> int {
+ int r = ClsCephFSClient::delete_inode_accumulate_result(data_io, oid);
+ if (r < 0) {
+ dout(4) << "Error deleting accumulated metadata from '" << oid
+ << "': " << cpp_strerror(r) << dendl;
+ }
+
+ progress_tracker->increment();
+ progress_tracker->display_progress();
+
+ return r;
});
}
return -EINVAL;
}
+ // Initialize progress tracking
+ // We'll estimate total objects in metadata pool for progress tracking
+ // scan_links processes ALL metadata objects in each worker (no slicing)
+ uint64_t total_objects = get_metadata_pool_objects(metadata_io, false);
+
+ auto progress_tracker = std::make_unique<ProgressTracker>(get_progress_operation_name("scan_links"));
+ progress_tracker->set_enable_progress_update(true);
+ // Since scan_links processes each object twice (SCAN_INOS + CHECK_LINK phases),
+ // we'll track total iterations as 2 * object_count
+ progress_tracker->start(total_objects * 2);
+
interval_set<uint64_t> used_inos;
map<inodeno_t, int> remote_links;
map<inodeno_t, vector<uint64_t>> referent_inodes; //referent inode list of primary inode
<< std::dec << "/" << dname << dendl;
return -EINVAL;
}
+ progress_tracker->increment();
+ progress_tracker->display_progress();
}
}
}
int DataScan::scan_frags()
{
+ uint64_t total_objects = get_metadata_pool_objects(metadata_io, true);
+ auto progress_tracker = std::make_unique<ProgressTracker>(get_progress_operation_name("scan_frags"));
+ progress_tracker->set_enable_progress_update(true);
+ progress_tracker->start(total_objects);
+ if (total_objects == 0) {
+ dout(4) << "No objects found in metadata pool" << dendl;
+ return 0;
+ }
+
bool roots_present;
int r = driver->check_roots(&roots_present);
if (r != 0) {
return -EIO;
}
- return forall_objects(metadata_io, true, [this](
+ return forall_objects(metadata_io, true, [this, &progress_tracker](
std::string const &oid,
uint64_t obj_name_ino,
uint64_t obj_name_offset) -> int
}
}
+ progress_tracker->increment();
+ progress_tracker->display_progress();
+
return r;
});
}
inode->uid = g_conf()->mds_root_ino_uid;
inode->gid = g_conf()->mds_root_ino_gid;
}
-
*/
-#include "MDSUtility.h"
#include "include/rados/librados.hpp"
+#include "MDSUtility.h"
+#include "ProgressTracker.h"
+
struct inode_backtrace_t;
class InodeStore;
class MDSTable;
class DataScan : public MDSUtility, public MetadataTool
{
- protected:
- RecoveryDriver *driver;
- fs_cluster_id_t fscid;
+private:
+ librados::Rados rados;
- std::string metadata_pool_name;
- std::vector<int64_t> data_pools;
+ uint64_t get_pool_objects(const std::vector<librados::IoCtx*>& data_ios);
+ uint64_t get_metadata_pool_objects(
+ librados::IoCtx& metadata_io,
+ bool worker_sliced = false);
- // IoCtx for data pool (where we scrape file backtraces from)
- librados::IoCtx data_io;
- // Remember the data pool ID for use in layouts
- int64_t data_pool_id;
- // IoCtxs for extra data pools
- std::vector<librados::IoCtx> extra_data_ios;
+protected:
+ RecoveryDriver* driver;
+ fs_cluster_id_t fscid;
- uint32_t n;
- uint32_t m;
+ std::string get_progress_operation_name(std::string_view op_name) const;
- /**
+ // IoCtx for data pool (where we scrape file backtraces from)
+ librados::IoCtx data_io;
+ // Remember the data pool ID for use in layouts
+ int64_t data_pool_id;
+ // IoCtxs for extra data pools
+ std::vector<librados::IoCtx> extra_data_ios;
+
+ uint32_t worker_n;
+ uint32_t worker_m;
+
+ std::string metadata_pool_name;
+ std::vector<int64_t> data_pools;
+
+ /**
* Scan data pool for backtraces, and inject inodes to metadata pool
*/
- int scan_inodes();
+ int scan_inodes();
- /**
+ /**
* Scan data pool for file sizes and mtimes
*/
- int scan_extents();
+ int scan_extents();
- /**
+ /**
* Scan metadata pool for 0th dirfrags to link orphaned
* directory inodes.
*/
- int scan_frags();
+ int scan_frags();
- /**
+ /**
* Cleanup xattrs from data pool
*/
- int cleanup();
+ int cleanup();
- /**
+ /**
* Check if an inode number is in the permitted ranges
*/
- bool valid_ino(inodeno_t ino) const;
+ bool valid_ino(inodeno_t ino) const;
- int scan_links();
+ int scan_links();
- // Accept pools which are not in the FSMap
- bool force_pool;
- // Respond to decode errors by overwriting
- bool force_corrupt;
- // Overwrite root objects even if they exist
- bool force_init;
- // Only scan inodes without this scrub tag
- std::string filter_tag;
+ // Accept pools which are not in the FSMap
+ bool force_pool;
+ // Respond to decode errors by overwriting
+ bool force_corrupt;
+ // Overwrite root objects even if they exist
+ bool force_init;
+ // Only scan inodes without this scrub tag
+ std::string filter_tag;
+ // Interval to update progress
+ ProgressTracker::duration progress_update_interval;
- /**
+ /**
* @param r set to error on valid key with invalid value
* @return true if argument consumed, else false
*/
- bool parse_kwarg(
- const std::vector<const char*> &args,
- std::vector<const char *>::const_iterator &i,
- int *r);
+ bool parse_kwarg(
+ const std::vector<const char*>& args,
+ std::vector<const char*>::const_iterator& i,
+ int* r);
- /**
+ /**
* @return true if argument consumed, else false
*/
- bool parse_arg(
- const std::vector<const char*> &arg,
- std::vector<const char *>::const_iterator &i);
+ bool parse_arg(
+ const std::vector<const char*>& arg,
+ std::vector<const char*>::const_iterator& i);
- int probe_filter(librados::IoCtx &ioctx);
+ int probe_filter(librados::IoCtx& ioctx);
- /**
+ /**
* Apply a function to all objects in an ioctx's pool, optionally
* restricted to only those objects with a 00000000 offset and
* no tag matching DataScan::scrub_tag.
*/
- int forall_objects(
- librados::IoCtx &ioctx,
- bool untagged_only,
- std::function<int(std::string, uint64_t, uint64_t)> handler);
-
- public:
- static void usage();
- int main(const std::vector<const char *> &args);
-
- DataScan()
- : driver(NULL), fscid(FS_CLUSTER_ID_NONE),
- data_pool_id(-1), n(0), m(1),
- force_pool(false), force_corrupt(false),
- force_init(false)
- {
- }
+ int forall_objects(
+ librados::IoCtx& ioctx,
+ bool untagged_only,
+ std::function<int(std::string, uint64_t, uint64_t)> handler);
+
+public:
+ static void usage();
+ int main(const std::vector<const char*>& args);
+
+ DataScan() :
+ driver(NULL),
+ fscid(FS_CLUSTER_ID_NONE),
+ data_pool_id(-1),
+ worker_n(0),
+ worker_m(1),
+ force_pool(false),
+ force_corrupt(false),
+ force_init(false)
+ {
+ }
~DataScan() override
{
* Iterate over log entries, attempting to scavenge from each one
*/
std::set<inodeno_t> consumed_inos;
+ uint64_t event_count = js.events.size();
+ progress_tracker->set_operation_name("Processing events");
+ progress_tracker->start(event_count);
+
+
for (JournalScanner::EventMap::iterator i = js.events.begin();
i != js.events.end(); ++i) {
auto& le = i->second.log_event;
JournalScanner::EventError(scav_r, cpp_strerror(r))));
}
}
+
+ progress_tracker->increment();
+ progress_tracker->display_progress();
}
+ progress_tracker->display_final_summary();
+
+
/**
* Update InoTable to reflect any inode numbers consumed during scavenge
*/
{
int r;
+ progress_tracker->set_operation_name("Scanning journal");
+ progress_tracker->start(0); // Unknown total initially
+
JournalFilter filter(type);
JournalScanner js(input, rank, type, filter);
r = js.scan();
return r;
}
+ progress_tracker->display_final_summary();
js.report(std::cout);
return 0;
int r = 0;
+ // Initialize progress tracking for dentry recovery
+ progress_tracker->set_operation_name("Recovering dentries");
+ progress_tracker->start(metablob.lump_order.size());
+
// Replay fullbits (dentry+inode)
for (const auto& frag : metablob.lump_order) {
EMetaBlob::dirlump const &lump = metablob.lump_map.find(frag)->second;
return r;
}
}
+
+ progress_tracker->increment();
+ progress_tracker->display_progress();
}
+ progress_tracker->display_final_summary();
+
+
/* Now that we've looked at the dirlumps, we finally pay attention to
* the roots (i.e. inodes without ancestry). This is necessary in order
* to pick up dirstat updates on ROOT_INO. dirstat updates are functionally
* Foundation. See file COPYING.
*/
-#include "MDSUtility.h"
-#include "RoleSelector.h"
#include <vector>
-#include "mds/mdstypes.h"
+#include "include/rados/librados.hpp"
#include "mds/LogEvent.h"
#include "mds/events/EMetaBlob.h"
-
-#include "include/rados/librados.hpp"
+#include "mds/mdstypes.h"
#include "JournalFilter.h"
+#include "MDSUtility.h"
+#include "ProgressTracker.h"
+#include "RoleSelector.h"
class JournalScanner;
class JournalTool : public MDSUtility
{
private:
+ std::unique_ptr<ProgressTracker> progress_tracker;
MDSRoleSelector role_selector;
// Bit hacky, use this `rank` member to control behaviour of the
// various main_ functions.
const std::string &command);
public:
static void usage();
+
JournalTool() :
- rank(0), other_pool(false) {}
+ rank(0), other_pool(false)
+ {
+ progress_tracker =
+ std::make_unique<ProgressTracker>("Journal processing");
+ }
int main(std::vector<const char*> &argv);
};
--- /dev/null
+#include "ProgressTracker.h"
+
+#include <fmt/format.h>
+#include <sys/ioctl.h>
+
+#include <string_view>
+
+#include <boost/asio.hpp>
+#include <boost/process.hpp>
+
+#include "common/ceph_context.h"
+#include "global/global_context.h"
+#include "include/cephfs/libcephfs.h"
+
+namespace bp = boost::process;
+namespace asio = boost::asio;
+
+namespace {
+
+/**
+ * Get the width of the terminal console
+ * @return Terminal width in columns, or 80 as default if not available
+ */
+int get_terminal_width() {
+ struct winsize w{0,0,0,0};
+ if (ioctl(STDOUT_FILENO, TIOCGWINSZ, &w) == 0 && w.ws_col > 0) {
+ return w.ws_col;
+ }
+ return 80; // Default fallback width
+}
+
+/**
+ * Output a line to the terminal, padding to full width if stdout is a TTY.
+ * This is useful for progress displays that need to overwrite previous output.
+ *
+ * @param line The text to output (without trailing newline)
+ * @param use_carriage_return If true, uses \r instead of \n for line ending
+ */
+void write_console_line(const std::string& line, bool use_carriage_return = true) {
+ static bool is_tty = isatty(fileno(stdout));
+
+ if (is_tty) {
+ int terminal_width = get_terminal_width();
+ std::string output = line;
+
+ // Truncate if line is longer than terminal width
+ if (static_cast<int>(output.length()) > terminal_width - 1) {
+ output = output.substr(0, terminal_width - 1);
+ }
+
+ // Pad with spaces to fill the entire line
+ int padding = terminal_width - static_cast<int>(output.length());
+ if (padding > 0) {
+ output.append(padding, ' ');
+ }
+
+ using namespace std::literals;
+ // Use carriage return to overwrite the line, or newline for permanent output
+ std::cout << (use_carriage_return ? "\r"sv : ""sv) << output
+ << (use_carriage_return ? ""sv : "\n"sv) << std::flush;
+ } else {
+ // Not a TTY, just output normally with newline
+ std::cout << line << std::endl;
+ }
+}
+
+} // anonymous namespace
+
+ProgressTracker::ProgressTracker(std::string_view operation_name) :
+ operation_name(operation_name)
+{}
+
+ProgressTracker::~ProgressTracker() {
+ display_final_summary();
+}
+
+void
+ProgressTracker::start(uint64_t total)
+{
+ processed_items.store(0);
+ total_items.store(total);
+ start_time = std::chrono::steady_clock::now();
+ started = true;
+ last_console_update = last_progress_update = clock::now();
+}
+
+void
+ProgressTracker::increment(uint64_t count)
+{
+ processed_items.fetch_add(count);
+}
+
+void
+ProgressTracker::set_processed(uint64_t count)
+{
+ processed_items.store(count);
+}
+
+void
+ProgressTracker::set_total(uint64_t total)
+{
+ total_items.store(total);
+}
+
+void
+ProgressTracker::display_progress() const
+{
+ if (!started) {
+ return;
+ }
+ const time_point now = clock::now();
+ if ((now - last_console_update) < console_refresh_interval) {
+ return;
+ }
+ display_progress_internal();
+ last_console_update = clock::now();
+}
+
+std::string
+ProgressTracker::get_ceph_progress_event() const
+{
+ return fmt::format("{}_{}", operation_name, getpid());
+}
+
+std::string
+ProgressTracker::get_progress_status(const float progress) const
+{
+ uint64_t current_processed = get_processed();
+ uint64_t current_total = get_total();
+ if (current_total > 0) {
+ std::chrono::seconds eta_seconds =
+ get_eta_seconds(current_processed, current_total);
+
+ std::string eta_str = (eta_seconds.count() > 0)
+ ? ceph::to_pretty_timedelta(eta_seconds)
+ : "calculating";
+
+ return fmt::format(
+ "{}/{} objects ({:.2f}%%), ETA: {}", current_processed,
+ current_total, progress, eta_str);
+ } else {
+ return fmt::format("{} objects", current_processed);
+ }
+}
+
+std::string
+ProgressTracker::get_completed_status() const
+{
+ auto now = std::chrono::steady_clock::now();
+ auto duration =
+ std::chrono::duration_cast<std::chrono::seconds>(now - start_time);
+
+ uint64_t final_processed = processed_items.load();
+ float avg_rate = duration.count() > 0
+ ? static_cast<float>(final_processed) /
+ static_cast<float>(duration.count())
+ : 0.0f;
+
+ return fmt::format(
+ "{} items in {}s (avg: {} items/sec)", final_processed, duration.count(),
+ avg_rate);
+}
+
+void
+ProgressTracker::update_ceph_progress_internal(
+ const std::string& message,
+ float progress,
+ bool force_update) const
+{
+ if (!enable_progress_update) {
+ return;
+ }
+ const auto ceph_path = bp::search_path("ceph");
+ if (ceph_path.empty()) {
+ std::cerr << "Warning: \"ceph\" not found on PATH."
+ << std::endl;
+ enable_progress_update = false;
+ return;
+ }
+
+ // Check how long since the last update
+ const time_point now = clock::now();
+ if (!force_update &&
+ (now - last_progress_update) < progress_refresh_interval) {
+ return;
+ }
+
+ // Set CEPH_CONF from the current runtime configuration if available
+ auto env = boost::this_process::environment();
+ try {
+ const std::string conf_path =
+ g_conf().get_conf_path(); // g_conf().get_val<std::string>("conf");
+ if (!conf_path.empty()) {
+ env["CEPH_CONF"] = conf_path;
+ }
+ } catch (...) {
+ std::cerr << "Warning: Failed to retrieve CEPH_CONF path in "
+ "update_ceph_progress_internal."
+ << std::endl;
+ // If the config value isn't available, leave CEPH_CONF unchanged
+ }
+
+ try {
+ // Create io_context for async operations
+ asio::io_context io_context;
+
+ // Async buffer for process output
+ std::future<std::string> output_future;
+ asio::streambuf output_buffer;
+
+ // Launch process with async pipe
+ bp::async_pipe merged_pipe(io_context);
+
+ bp::child ceph_cmd(
+ ceph_path, "mgr", "cli", "update_progress_event",
+ get_ceph_progress_event(), operation_name + ": " + message,
+ std::to_string(progress), "--add-to-ceph-s",
+ (bp::std_out & bp::std_err) > merged_pipe,
+ env,
+ io_context);
+
+ // Asynchronously read all output
+ std::string output;
+ std::function<void(const boost::system::error_code&, std::size_t)> read_handler;
+ read_handler = [&](const boost::system::error_code& ec, std::size_t bytes_transferred) {
+ if (!ec && bytes_transferred > 0) {
+ std::string chunk(
+ asio::buffers_begin(output_buffer.data()),
+ asio::buffers_begin(output_buffer.data()) + bytes_transferred);
+ output.append(chunk);
+ output_buffer.consume(bytes_transferred);
+
+ // Continue reading
+ asio::async_read(merged_pipe, output_buffer,
+ asio::transfer_at_least(1), read_handler);
+ } else if (ec == asio::error::eof) {
+ // End of stream - normal completion
+ } else if (ec) {
+ // Other error
+ std::cerr << "Error reading ceph command output: " << ec.message() << std::endl;
+ }
+ };
+
+ // Start async read
+ asio::async_read(merged_pipe, output_buffer,
+ asio::transfer_at_least(1), read_handler);
+
+ // Run io_context to completion
+ io_context.run();
+
+ // Wait for process to complete
+ ceph_cmd.wait();
+
+ if (auto ret = ceph_cmd.exit_code()) {
+ if (ret == ENOTSUP) {
+ static std::once_flag enotsup_msg;
+ std::call_once(enotsup_msg, [&output]() {
+ std::cerr << output << std::endl;
+ });
+ } else {
+ std::cout << output << std::endl;
+ enable_progress_update = false;
+ }
+ }
+ } catch (bp::process_error& ec) {
+ std::cerr << ec.what() << std::endl;
+ } catch (std::exception& ex) {
+ std::cerr << "Exception in update_ceph_progress_internal: " << ex.what() << std::endl;
+ }
+
+ last_progress_update = clock::now();
+}
+
+
+void
+ProgressTracker::display_progress_internal() const
+{
+ std::lock_guard<std::mutex> lock(display_mutex);
+
+ float progress = get_progress_percent();
+
+ std::string progress_status = get_progress_status(progress);
+ write_console_line(fmt::format("Processed {}", progress_status));
+ update_ceph_progress_internal(progress_status, progress);
+}
+
+void
+ProgressTracker::display_final_summary() const
+{
+ if (!started) {
+ return;
+ }
+ std::lock_guard<std::mutex> lock(display_mutex);
+ std::string completed_status = get_completed_status();
+ write_console_line(fmt::format("Completed {}! Processed {}", operation_name, completed_status), false);
+ update_ceph_progress_internal(completed_status, 100.0, true);
+}
+
+float
+ProgressTracker::get_progress_percent() const
+{
+ uint64_t current_total = total_items.load();
+ if (current_total == 0) {
+ return 0.0f;
+ }
+
+ uint64_t current_processed = processed_items.load();
+ return (static_cast<float>(current_processed) /
+ static_cast<float>(current_total)) *
+ 100.0f;
+}
+
+std::chrono::seconds
+ProgressTracker::get_eta_seconds(
+ uint64_t current_processed,
+ uint64_t current_total) const
+{
+ if (!started) {
+ return std::chrono::seconds{0};
+ }
+
+ auto now = std::chrono::steady_clock::now();
+ auto duration =
+ std::chrono::duration_cast<std::chrono::seconds>(now - start_time);
+
+ if (duration.count() == 0 || current_total == 0 || current_processed == 0) {
+ return std::chrono::seconds{0};
+ }
+
+ // If we've processed more than estimated, ETA is 0 (completing)
+ if (current_processed >= current_total) {
+ return std::chrono::seconds{0};
+ }
+
+ float objects_per_sec = static_cast<float>(current_processed) /
+ static_cast<float>(duration.count());
+ if (objects_per_sec <= 0) {
+ return std::chrono::seconds{0};
+ }
+
+ int64_t remaining = static_cast<int64_t>(current_total) -
+ static_cast<int64_t>(current_processed);
+ if (remaining <= 0) {
+ return std::chrono::seconds{0};
+ }
+
+ return std::chrono::seconds(
+ static_cast<int64_t>(remaining / objects_per_sec));
+}
--- /dev/null
+#ifndef PROGRESS_TRACKER_H
+#define PROGRESS_TRACKER_H
+
+#include <atomic>
+#include <chrono>
+#include <iostream>
+#include <mutex>
+#include <string>
+
+#include "tools/cephfs_mirror/Types.h"
+
+/**
+ * Reusable progress tracking utility for cephfs offline tools
+ * Provides thread-safe progress tracking with ETA calculation and consistent display formatting
+ */
+class ProgressTracker {
+public:
+ using clock = std::chrono::high_resolution_clock;
+ using duration = clock::duration;
+ using time_point = clock::time_point;
+
+ /**
+ * Constructor
+ * @param operation_name Name of the operation being tracked (e.g. "Processing objects")
+ */
+ explicit ProgressTracker(std::string_view operation_name = "Processing");
+
+ ~ProgressTracker();
+
+ /**
+ * Initialize progress tracking
+ * @param total_items Total number of items to process (0 if unknown)
+ */
+ void start(uint64_t total_items = 0);
+
+ /**
+ * Update progress by incrementing processed count
+ * @param count Number of items processed (default: 1)
+ */
+ void increment(uint64_t count = 1);
+
+ /**
+ * Set the current processed count directly
+ * @param count Current number of processed items
+ */
+ void set_processed(uint64_t count);
+
+ /**
+ * Set or update the total number of items
+ * @param total Total number of items to process
+ */
+ void set_total(uint64_t total);
+
+ /**
+ * Display current progress (call this periodically)
+ * Will only display if enough items have been processed since the last display
+ */
+ void display_progress() const;
+
+ /**
+ * Display final summary when operation is complete
+ */
+ void display_final_summary() const;
+
+ /**
+ * Get the current number of processed items
+ */
+ uint64_t
+ get_processed() const
+ {
+ return processed_items.load();
+ }
+
+ /**
+ * Get the total number of items
+ */
+ uint64_t
+ get_total() const
+ {
+ return total_items.load();
+ }
+
+ /**
+ * Get progress as a percentage (0.0 to 100.0)
+ */
+ float get_progress_percent() const;
+
+ /**
+ * Get estimated time remaining in seconds
+ */
+ std::chrono::seconds get_eta_seconds(
+ uint64_t current_processed,
+ uint64_t current_total) const;
+
+ /**
+ * Configuration methods
+ */
+ void
+ set_operation_name(const std::string& name)
+ {
+ operation_name = name;
+ }
+
+ /**
+ * Check if progress tracking has been started
+ */
+ bool
+ is_started() const
+ {
+ return started;
+ }
+
+ void
+ set_enable_progress_update(bool val) const
+ {
+ enable_progress_update = val;
+ }
+
+ void set_progress_update_interval(const duration& interval)
+ {
+ progress_refresh_interval = std::chrono::duration_cast<duration>(interval);
+ }
+
+private:
+
+ /**
+ * Display the current progress to stdout.
+ *
+ * Thread-safety:
+ * - Uses an internal mutex to serialize console output across threads.
+ * - Reads atomics for processed/total counts.
+ *
+ * Intended usage:
+ * - Called by display_progress() when sufficient progress has been made.
+ */
+ void display_progress_internal() const;
+
+ /**
+ * Build a unique progress event identifier for Ceph mgr progress.
+ *
+ * Format:
+ * - "<operation_name>_<pid>"
+ *
+ * @return string identifier suitable for Ceph "update_progress_event".
+ */
+ std::string get_ceph_progress_event() const;
+
+ /**
+ * Build a human-readable status line of the current progress.
+ *
+ * When the total is known (> 0):
+ * - "<processed>/<total> objects (<percent>%), ETA: <mm>m<ss>s"
+ * When the total is unknown:
+ * - "<processed> objects"
+ *
+ * @return formatted status description for the current progress.
+ */
+ std::string get_progress_status(float progress) const;
+
+ /**
+ * Build a final summary string with total duration and average rate.
+ *
+ * Format:
+ * - "<processed> items in <seconds>s (avg: <items/sec> items/sec)"
+ *
+ * @return formatted completion summary.
+ */
+ std::string get_completed_status() const;
+
+
+ /**
+ * Push the current progress status to Ceph mgr.
+ *
+ * Behavior:
+ * - No-ops if progress updates are disabled or the 'ceph' binary is not found.
+ * - Rate-limited by progress_refresh_interval.
+ * - Spawns a child process to call:
+ * ceph mgr cli update_progress_event <event> "<title>: <status>" <percent> --add-to-ceph-s
+ *
+ * Thread-safety:
+ * - Reads atomics for processed/total counts.
+ * - Uses internal timestamps for rate limiting.
+ *
+ * @param message Human-readable status to include in the event payload.
+ * @param progress Progress percentage (0-100).
+ * @param force_update If true, bypass the rate limiter and send the update immediately.
+ */
+ void update_ceph_progress_internal(
+ const std::string& message,
+ float progress,
+ bool force_update = false) const;
+
+ /**
+ * Number of items processed so far.
+ * Atomic to support updates from multiple threads.
+ */
+ std::atomic<uint64_t> processed_items{0};
+
+ /**
+ * Total number of items to process.
+ * 0 indicates an unknown total.
+ */
+ std::atomic<uint64_t> total_items{0};
+
+ /**
+ * Wall-clock time when tracking started.
+ * Used to compute elapsed time, throughput, and ETA.
+ */
+ std::chrono::steady_clock::time_point start_time;
+
+ /**
+ * Short description of the tracked operation (e.g., "scan_extents").
+ * Appears in console output and Ceph progress events.
+ */
+ std::string operation_name;
+
+ /**
+ * Indicates whether start() was called and tracking is active.
+ */
+ bool started{false};
+
+ /**
+ * Guards console output to keep status lines consistent across threads.
+ */
+ mutable std::mutex display_mutex; // For thread-safe console output
+
+ /**
+ * Timestamp of the last console update.
+ * Used to throttle progress updates.
+ */
+ mutable time_point last_console_update{time_point::min()};
+
+ /**
+ * Minimum time interval between console updates.
+ */
+ static constexpr duration console_refresh_interval =
+ std::chrono::duration_cast<duration>(std::chrono::seconds(1));
+
+ /**
+ * Timestamp of the last Ceph mgr progress update.
+ * Used to throttle external progress event updates.
+ */
+ mutable time_point last_progress_update{time_point::min()};
+
+ /**
+ * Enables sending progress updates to the Ceph mgr.
+ * Disabled automatically if prerequisites are not met.
+ */
+ mutable bool enable_progress_update{false};
+
+ /**
+ * Minimum time interval between Ceph mgr progress updates.
+ */
+ duration progress_refresh_interval =
+ std::chrono::duration_cast<duration>(std::chrono::seconds(5));
+};
+
+#endif // PROGRESS_TRACKER_H
ceph_assert(f != NULL);
int r = 0;
+ auto roles = role_selector.get_roles();
+ // Start progress tracking for multiple ranks (if more than 1)
+ if (roles.size() > 1) {
+ progress_tracker->start(roles.size());
+ }
f->open_object_section("ranks");
- for (auto role : role_selector.get_roles()) {
+ for (auto role : roles) {
std::ostringstream rank_str;
rank_str << role.rank;
f->open_object_section(rank_str.str().c_str());
f->dump_int("result", rank_r);
f->close_section();
-
-
+ // Update progress after processing each rank (only if multiple ranks)
+ if (roles.size() > 1) {
+ progress_tracker->increment();
+ progress_tracker->display_progress();
+ }
}
f->close_section();
+
+ // Display final summary at the end (only if multiple ranks)
+ if (roles.size() > 1) {
+ progress_tracker->display_final_summary();
+ }
+
+ // Display final summary at the end (only if multiple ranks)
+ if (roles.size() > 1) {
+ progress_tracker->display_final_summary();
+ }
return r;
}
*/
+#include "include/rados/librados.hpp"
+
#include "MDSUtility.h"
+#include "ProgressTracker.h"
#include "RoleSelector.h"
-#include "include/rados/librados.hpp"
-
/**
* Command line tool for debugging the backing store of
* MDSTable instances.
librados::Rados rados;
librados::IoCtx io;
+ std::unique_ptr<ProgressTracker> progress_tracker;
+
int apply_role_fn(std::function<int(mds_role_t, Formatter *)> fptr, Formatter *f);
public:
static void usage();
int main(std::vector<const char*> &argv);
+ TableTool()
+ {
+ progress_tracker = std::make_unique<ProgressTracker>("Table operations");
+ }
};
projects / ceph.git / commitdiff