diff options
Diffstat (limited to 'src/tools/cephfs_mirror')
25 files changed, 5373 insertions, 0 deletions
diff --git a/src/tools/cephfs_mirror/CMakeLists.txt b/src/tools/cephfs_mirror/CMakeLists.txt new file mode 100644 index 000000000..4b6dea7a1 --- /dev/null +++ b/src/tools/cephfs_mirror/CMakeLists.txt @@ -0,0 +1,30 @@ +set(cephfs_mirror_internal + ClusterWatcher.cc + Mirror.cc + FSMirror.cc + InstanceWatcher.cc + MirrorWatcher.cc + PeerReplayer.cc + ServiceDaemon.cc + Types.cc + Utils.cc + Watcher.cc + watcher/RewatchRequest.cc) + +add_executable(cephfs-mirror + main.cc) + +add_library(cephfs_mirror_internal STATIC + ${cephfs_mirror_internal}) + +target_link_libraries(cephfs-mirror + cephfs_mirror_internal + global + ceph-common + cls_cephfs_client + librados + mds + cephfs + ${ALLOC_LIBS}) + +install(TARGETS cephfs-mirror DESTINATION bin) diff --git a/src/tools/cephfs_mirror/ClusterWatcher.cc b/src/tools/cephfs_mirror/ClusterWatcher.cc new file mode 100644 index 000000000..b5f6f81d7 --- /dev/null +++ b/src/tools/cephfs_mirror/ClusterWatcher.cc @@ -0,0 +1,182 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include <mutex> +#include <vector> + +#include "common/ceph_context.h" +#include "common/debug.h" +#include "common/errno.h" +#include "mon/MonClient.h" + +#include "ClusterWatcher.h" +#include "ServiceDaemon.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_cephfs_mirror +#undef dout_prefix +#define dout_prefix *_dout << "cephfs::mirror::ClusterWatcher " << __func__ + +namespace cephfs { +namespace mirror { + +ClusterWatcher::ClusterWatcher(CephContext *cct, MonClient *monc, ServiceDaemon *service_daemon, + Listener &listener) + : Dispatcher(cct), + m_monc(monc), + m_service_daemon(service_daemon), + m_listener(listener) { +} + +ClusterWatcher::~ClusterWatcher() { +} + +bool ClusterWatcher::ms_can_fast_dispatch2(const cref_t<Message> &m) const { + return m->get_type() == CEPH_MSG_FS_MAP; +} + +void ClusterWatcher::ms_fast_dispatch2(const ref_t<Message> &m) { + bool handled = ms_dispatch2(m); + ceph_assert(handled); +} + +bool ClusterWatcher::ms_dispatch2(const ref_t<Message> &m) { + if (m->get_type() == CEPH_MSG_FS_MAP) { + if (m->get_connection()->get_peer_type() == CEPH_ENTITY_TYPE_MON) { + handle_fsmap(ref_cast<MFSMap>(m)); + } + return true; + } + + return false; +} + +int ClusterWatcher::init() { + dout(20) << dendl; + + bool sub = m_monc->sub_want("fsmap", 0, 0); + if (!sub) { + derr << ": failed subscribing to FSMap" << dendl; + return -1; + } + + m_monc->renew_subs(); + dout(10) << ": subscribed to FSMap" << dendl; + return 0; +} + +void ClusterWatcher::shutdown() { + dout(20) << dendl; + std::scoped_lock locker(m_lock); + m_stopping = true; + m_monc->sub_unwant("fsmap"); +} + +void ClusterWatcher::handle_fsmap(const cref_t<MFSMap> &m) { + dout(20) << dendl; + + auto fsmap = m->get_fsmap(); + auto filesystems = fsmap.get_filesystems(); + + std::vector<Filesystem> mirroring_enabled; + std::vector<Filesystem> mirroring_disabled; + std::map<Filesystem, Peers> peers_added; + std::map<Filesystem, Peers> peers_removed; + std::map<Filesystem, uint64_t> fs_metadata_pools; + { + std::scoped_lock locker(m_lock); + if (m_stopping) { + return; + } + + // deleted filesystems are considered mirroring disabled + for (auto it = m_filesystem_peers.begin(); it != m_filesystem_peers.end();) { + if (!fsmap.filesystem_exists(it->first.fscid)) { + mirroring_disabled.emplace_back(it->first); + it = m_filesystem_peers.erase(it); + continue; + } + ++it; + } + + for (auto &filesystem : filesystems) { + auto fs = Filesystem{filesystem->fscid, + std::string(filesystem->mds_map.get_fs_name())}; + auto pool_id = filesystem->mds_map.get_metadata_pool(); + auto &mirror_info = filesystem->mirror_info; + + if (!mirror_info.is_mirrored()) { + auto it = m_filesystem_peers.find(fs); + if (it != m_filesystem_peers.end()) { + mirroring_disabled.emplace_back(fs); + m_filesystem_peers.erase(it); + } + } else { + auto [fspeersit, enabled] = m_filesystem_peers.emplace(fs, Peers{}); + auto &peers = fspeersit->second; + + if (enabled) { + mirroring_enabled.emplace_back(fs); + fs_metadata_pools.emplace(fs, pool_id); + } + + // peers added + Peers added; + std::set_difference(mirror_info.peers.begin(), mirror_info.peers.end(), + peers.begin(), peers.end(), std::inserter(added, added.end())); + + // peers removed + Peers removed; + std::set_difference(peers.begin(), peers.end(), + mirror_info.peers.begin(), mirror_info.peers.end(), + std::inserter(removed, removed.end())); + + // update set + if (!added.empty()) { + peers_added.emplace(fs, added); + peers.insert(added.begin(), added.end()); + } + if (!removed.empty()) { + peers_removed.emplace(fs, removed); + for (auto &p : removed) { + peers.erase(p); + } + } + } + } + } + + dout(5) << ": mirroring enabled=" << mirroring_enabled << ", mirroring_disabled=" + << mirroring_disabled << dendl; + for (auto &fs : mirroring_enabled) { + m_service_daemon->add_filesystem(fs.fscid, fs.fs_name); + m_listener.handle_mirroring_enabled(FilesystemSpec(fs, fs_metadata_pools.at(fs))); + } + for (auto &fs : mirroring_disabled) { + m_service_daemon->remove_filesystem(fs.fscid); + m_listener.handle_mirroring_disabled(fs); + } + + dout(5) << ": peers added=" << peers_added << ", peers removed=" << peers_removed << dendl; + + for (auto &[fs, peers] : peers_added) { + for (auto &peer : peers) { + m_service_daemon->add_peer(fs.fscid, peer); + m_listener.handle_peers_added(fs, peer); + } + } + for (auto &[fs, peers] : peers_removed) { + for (auto &peer : peers) { + m_service_daemon->remove_peer(fs.fscid, peer); + m_listener.handle_peers_removed(fs, peer); + } + } + + std::scoped_lock locker(m_lock); + if (!m_stopping) { + m_monc->sub_got("fsmap", fsmap.get_epoch()); + } // else we have already done a sub_unwant() +} + +} // namespace mirror +} // namespace cephfs diff --git a/src/tools/cephfs_mirror/ClusterWatcher.h b/src/tools/cephfs_mirror/ClusterWatcher.h new file mode 100644 index 000000000..a418898f5 --- /dev/null +++ b/src/tools/cephfs_mirror/ClusterWatcher.h @@ -0,0 +1,77 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPHFS_MIRROR_CLUSTER_WATCHER_H +#define CEPHFS_MIRROR_CLUSTER_WATCHER_H + +#include <map> + +#include "common/ceph_mutex.h" +#include "common/async/context_pool.h" +#include "messages/MFSMap.h" +#include "msg/Dispatcher.h" +#include "Types.h" + +class MonClient; + +namespace cephfs { +namespace mirror { + +class ServiceDaemon; + +// watch peer changes for filesystems via FSMap updates + +class ClusterWatcher : public Dispatcher { +public: + struct Listener { + virtual ~Listener() { + } + + virtual void handle_mirroring_enabled(const FilesystemSpec &spec) = 0; + virtual void handle_mirroring_disabled(const Filesystem &filesystem) = 0; + + virtual void handle_peers_added(const Filesystem &filesystem, const Peer &peer) = 0; + virtual void handle_peers_removed(const Filesystem &filesystem, const Peer &peer) = 0; + }; + + ClusterWatcher(CephContext *cct, MonClient *monc, ServiceDaemon *service_daemon, + Listener &listener); + ~ClusterWatcher(); + + bool ms_can_fast_dispatch_any() const override { + return true; + } + bool ms_can_fast_dispatch2(const cref_t<Message> &m) const override; + void ms_fast_dispatch2(const ref_t<Message> &m) override; + bool ms_dispatch2(const ref_t<Message> &m) override; + + void ms_handle_connect(Connection *c) override { + } + bool ms_handle_reset(Connection *c) override { + return false; + } + void ms_handle_remote_reset(Connection *c) override { + } + bool ms_handle_refused(Connection *c) override { + return false; + } + + int init(); + void shutdown(); + +private: + ceph::mutex m_lock = ceph::make_mutex("cephfs::mirror::cluster_watcher"); + MonClient *m_monc; + ServiceDaemon *m_service_daemon; + Listener &m_listener; + + bool m_stopping = false; + std::map<Filesystem, Peers> m_filesystem_peers; + + void handle_fsmap(const cref_t<MFSMap> &m); +}; + +} // namespace mirror +} // namespace cephfs + +#endif // CEPHFS_MIRROR_CLUSTER_WATCHER_H diff --git a/src/tools/cephfs_mirror/FSMirror.cc b/src/tools/cephfs_mirror/FSMirror.cc new file mode 100644 index 000000000..76dcc11f6 --- /dev/null +++ b/src/tools/cephfs_mirror/FSMirror.cc @@ -0,0 +1,441 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "common/admin_socket.h" +#include "common/ceph_argparse.h" +#include "common/ceph_context.h" +#include "common/common_init.h" +#include "common/debug.h" +#include "common/errno.h" +#include "common/WorkQueue.h" +#include "include/stringify.h" +#include "msg/Messenger.h" +#include "FSMirror.h" +#include "PeerReplayer.h" +#include "aio_utils.h" +#include "ServiceDaemon.h" +#include "Utils.h" + +#include "common/Cond.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_cephfs_mirror +#undef dout_prefix +#define dout_prefix *_dout << "cephfs::mirror::FSMirror " << __func__ + +namespace cephfs { +namespace mirror { + +namespace { + +const std::string SERVICE_DAEMON_DIR_COUNT_KEY("directory_count"); +const std::string SERVICE_DAEMON_PEER_INIT_FAILED_KEY("peer_init_failed"); + +class MirrorAdminSocketCommand { +public: + virtual ~MirrorAdminSocketCommand() { + } + virtual int call(Formatter *f) = 0; +}; + +class StatusCommand : public MirrorAdminSocketCommand { +public: + explicit StatusCommand(FSMirror *fs_mirror) + : fs_mirror(fs_mirror) { + } + + int call(Formatter *f) override { + fs_mirror->mirror_status(f); + return 0; + } + +private: + FSMirror *fs_mirror; +}; + +} // anonymous namespace + +class MirrorAdminSocketHook : public AdminSocketHook { +public: + MirrorAdminSocketHook(CephContext *cct, const Filesystem &filesystem, FSMirror *fs_mirror) + : admin_socket(cct->get_admin_socket()) { + int r; + std::string cmd; + + // mirror status format is name@fscid + cmd = "fs mirror status " + stringify(filesystem.fs_name) + "@" + stringify(filesystem.fscid); + r = admin_socket->register_command( + cmd, this, "get filesystem mirror status"); + if (r == 0) { + commands[cmd] = new StatusCommand(fs_mirror); + } + } + + ~MirrorAdminSocketHook() override { + admin_socket->unregister_commands(this); + for (auto &[command, cmdptr] : commands) { + delete cmdptr; + } + } + + int call(std::string_view command, const cmdmap_t& cmdmap, + Formatter *f, std::ostream &errss, bufferlist &out) override { + auto p = commands.at(std::string(command)); + return p->call(f); + } + +private: + typedef std::map<std::string, MirrorAdminSocketCommand*, std::less<>> Commands; + + AdminSocket *admin_socket; + Commands commands; +}; + +FSMirror::FSMirror(CephContext *cct, const Filesystem &filesystem, uint64_t pool_id, + ServiceDaemon *service_daemon, std::vector<const char*> args, + ContextWQ *work_queue) + : m_cct(cct), + m_filesystem(filesystem), + m_pool_id(pool_id), + m_service_daemon(service_daemon), + m_args(args), + m_work_queue(work_queue), + m_snap_listener(this), + m_asok_hook(new MirrorAdminSocketHook(cct, filesystem, this)) { + m_service_daemon->add_or_update_fs_attribute(m_filesystem.fscid, SERVICE_DAEMON_DIR_COUNT_KEY, + (uint64_t)0); +} + +FSMirror::~FSMirror() { + dout(20) << dendl; + + { + std::scoped_lock locker(m_lock); + delete m_instance_watcher; + delete m_mirror_watcher; + } + // outside the lock so that in-progress commands can acquire + // lock and finish executing. + delete m_asok_hook; +} + +int FSMirror::init_replayer(PeerReplayer *peer_replayer) { + ceph_assert(ceph_mutex_is_locked(m_lock)); + return peer_replayer->init(); +} + +void FSMirror::shutdown_replayer(PeerReplayer *peer_replayer) { + peer_replayer->shutdown(); +} + +void FSMirror::cleanup() { + dout(20) << dendl; + ceph_unmount(m_mount); + ceph_release(m_mount); + m_ioctx.close(); + m_cluster.reset(); +} + +void FSMirror::reopen_logs() { + std::scoped_lock locker(m_lock); + + if (m_cluster) { + reinterpret_cast<CephContext *>(m_cluster->cct())->reopen_logs(); + } + for (auto &[peer, replayer] : m_peer_replayers) { + replayer->reopen_logs(); + } +} + +void FSMirror::init(Context *on_finish) { + dout(20) << dendl; + + std::scoped_lock locker(m_lock); + int r = connect(g_ceph_context->_conf->name.to_str(), + g_ceph_context->_conf->cluster, &m_cluster, "", "", m_args); + if (r < 0) { + m_init_failed = true; + on_finish->complete(r); + return; + } + + r = m_cluster->ioctx_create2(m_pool_id, m_ioctx); + if (r < 0) { + m_init_failed = true; + m_cluster.reset(); + derr << ": error accessing local pool (id=" << m_pool_id << "): " + << cpp_strerror(r) << dendl; + on_finish->complete(r); + return; + } + + r = mount(m_cluster, m_filesystem, true, &m_mount); + if (r < 0) { + m_init_failed = true; + m_ioctx.close(); + m_cluster.reset(); + on_finish->complete(r); + return; + } + + m_addrs = m_cluster->get_addrs(); + dout(10) << ": rados addrs=" << m_addrs << dendl; + + init_instance_watcher(on_finish); +} + +void FSMirror::shutdown(Context *on_finish) { + dout(20) << dendl; + + { + std::scoped_lock locker(m_lock); + m_stopping = true; + if (m_on_init_finish != nullptr) { + dout(10) << ": delaying shutdown -- init in progress" << dendl; + m_on_shutdown_finish = new LambdaContext([this, on_finish](int r) { + if (r < 0) { + on_finish->complete(0); + return; + } + m_on_shutdown_finish = on_finish; + shutdown_peer_replayers(); + }); + return; + } + + m_on_shutdown_finish = on_finish; + } + + shutdown_peer_replayers(); +} + +void FSMirror::shutdown_peer_replayers() { + dout(20) << dendl; + + for (auto &[peer, peer_replayer] : m_peer_replayers) { + dout(5) << ": shutting down replayer for peer=" << peer << dendl; + shutdown_replayer(peer_replayer.get()); + } + m_peer_replayers.clear(); + + shutdown_mirror_watcher(); +} + +void FSMirror::init_instance_watcher(Context *on_finish) { + dout(20) << dendl; + + m_on_init_finish = new LambdaContext([this, on_finish](int r) { + { + std::scoped_lock locker(m_lock); + if (r < 0) { + m_init_failed = true; + } + } + on_finish->complete(r); + if (m_on_shutdown_finish != nullptr) { + m_on_shutdown_finish->complete(r); + } + }); + + Context *ctx = new C_CallbackAdapter< + FSMirror, &FSMirror::handle_init_instance_watcher>(this); + m_instance_watcher = InstanceWatcher::create(m_ioctx, m_snap_listener, m_work_queue); + m_instance_watcher->init(ctx); +} + +void FSMirror::handle_init_instance_watcher(int r) { + dout(20) << ": r=" << r << dendl; + + Context *on_init_finish = nullptr; + { + std::scoped_lock locker(m_lock); + if (r < 0) { + std::swap(on_init_finish, m_on_init_finish); + } + } + + if (on_init_finish != nullptr) { + on_init_finish->complete(r); + return; + } + + init_mirror_watcher(); +} + +void FSMirror::init_mirror_watcher() { + dout(20) << dendl; + + std::scoped_lock locker(m_lock); + Context *ctx = new C_CallbackAdapter< + FSMirror, &FSMirror::handle_init_mirror_watcher>(this); + m_mirror_watcher = MirrorWatcher::create(m_ioctx, this, m_work_queue); + m_mirror_watcher->init(ctx); +} + +void FSMirror::handle_init_mirror_watcher(int r) { + dout(20) << ": r=" << r << dendl; + + Context *on_init_finish = nullptr; + { + std::scoped_lock locker(m_lock); + if (r == 0) { + std::swap(on_init_finish, m_on_init_finish); + } + } + + if (on_init_finish != nullptr) { + on_init_finish->complete(r); + return; + } + + m_retval = r; // save errcode for init context callback + shutdown_instance_watcher(); +} + +void FSMirror::shutdown_mirror_watcher() { + dout(20) << dendl; + + std::scoped_lock locker(m_lock); + Context *ctx = new C_CallbackAdapter< + FSMirror, &FSMirror::handle_shutdown_mirror_watcher>(this); + m_mirror_watcher->shutdown(ctx); +} + +void FSMirror::handle_shutdown_mirror_watcher(int r) { + dout(20) << ": r=" << r << dendl; + + shutdown_instance_watcher(); +} + +void FSMirror::shutdown_instance_watcher() { + dout(20) << dendl; + + std::scoped_lock locker(m_lock); + Context *ctx = new C_CallbackAdapter< + FSMirror, &FSMirror::handle_shutdown_instance_watcher>(this); + m_instance_watcher->shutdown(new C_AsyncCallback<ContextWQ>(m_work_queue, ctx)); +} + +void FSMirror::handle_shutdown_instance_watcher(int r) { + dout(20) << ": r=" << r << dendl; + + cleanup(); + + Context *on_init_finish = nullptr; + Context *on_shutdown_finish = nullptr; + + { + std::scoped_lock locker(m_lock); + std::swap(on_init_finish, m_on_init_finish); + std::swap(on_shutdown_finish, m_on_shutdown_finish); + } + + if (on_init_finish != nullptr) { + on_init_finish->complete(m_retval); + } + if (on_shutdown_finish != nullptr) { + on_shutdown_finish->complete(r); + } +} + +void FSMirror::handle_acquire_directory(string_view dir_path) { + dout(5) << ": dir_path=" << dir_path << dendl; + + { + std::scoped_lock locker(m_lock); + m_directories.emplace(dir_path); + m_service_daemon->add_or_update_fs_attribute(m_filesystem.fscid, SERVICE_DAEMON_DIR_COUNT_KEY, + m_directories.size()); + + for (auto &[peer, peer_replayer] : m_peer_replayers) { + dout(10) << ": peer=" << peer << dendl; + peer_replayer->add_directory(dir_path); + } + } +} + +void FSMirror::handle_release_directory(string_view dir_path) { + dout(5) << ": dir_path=" << dir_path << dendl; + + { + std::scoped_lock locker(m_lock); + auto it = m_directories.find(dir_path); + if (it != m_directories.end()) { + m_directories.erase(it); + m_service_daemon->add_or_update_fs_attribute(m_filesystem.fscid, SERVICE_DAEMON_DIR_COUNT_KEY, + m_directories.size()); + for (auto &[peer, peer_replayer] : m_peer_replayers) { + dout(10) << ": peer=" << peer << dendl; + peer_replayer->remove_directory(dir_path); + } + } + } +} + +void FSMirror::add_peer(const Peer &peer) { + dout(10) << ": peer=" << peer << dendl; + + std::scoped_lock locker(m_lock); + m_all_peers.emplace(peer); + if (m_peer_replayers.find(peer) != m_peer_replayers.end()) { + return; + } + + auto replayer = std::make_unique<PeerReplayer>( + m_cct, this, m_cluster, m_filesystem, peer, m_directories, m_mount, m_service_daemon); + int r = init_replayer(replayer.get()); + if (r < 0) { + m_service_daemon->add_or_update_peer_attribute(m_filesystem.fscid, peer, + SERVICE_DAEMON_PEER_INIT_FAILED_KEY, + true); + return; + } + m_peer_replayers.emplace(peer, std::move(replayer)); + ceph_assert(m_peer_replayers.size() == 1); // support only a single peer +} + +void FSMirror::remove_peer(const Peer &peer) { + dout(10) << ": peer=" << peer << dendl; + + std::unique_ptr<PeerReplayer> replayer; + { + std::scoped_lock locker(m_lock); + m_all_peers.erase(peer); + auto it = m_peer_replayers.find(peer); + if (it != m_peer_replayers.end()) { + replayer = std::move(it->second); + m_peer_replayers.erase(it); + } + } + + if (replayer) { + dout(5) << ": shutting down replayers for peer=" << peer << dendl; + shutdown_replayer(replayer.get()); + } +} + +void FSMirror::mirror_status(Formatter *f) { + std::scoped_lock locker(m_lock); + f->open_object_section("status"); + if (m_init_failed) { + f->dump_string("state", "failed"); + } else if (is_blocklisted(locker)) { + f->dump_string("state", "blocklisted"); + } else { + // dump rados addr for blocklist test + f->dump_string("rados_inst", m_addrs); + f->open_object_section("peers"); + for ([[maybe_unused]] auto &[peer, peer_replayer] : m_peer_replayers) { + peer.dump(f); + } + f->close_section(); // peers + f->open_object_section("snap_dirs"); + f->dump_int("dir_count", m_directories.size()); + f->close_section(); // snap_dirs + } + f->close_section(); // status +} + + +} // namespace mirror +} // namespace cephfs diff --git a/src/tools/cephfs_mirror/FSMirror.h b/src/tools/cephfs_mirror/FSMirror.h new file mode 100644 index 000000000..bae5a38e1 --- /dev/null +++ b/src/tools/cephfs_mirror/FSMirror.h @@ -0,0 +1,158 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPHFS_MIRROR_FS_MIRROR_H +#define CEPHFS_MIRROR_FS_MIRROR_H + +#include "common/Formatter.h" +#include "common/Thread.h" +#include "mds/FSMap.h" +#include "Types.h" +#include "InstanceWatcher.h" +#include "MirrorWatcher.h" + +class ContextWQ; + +namespace cephfs { +namespace mirror { + +class MirrorAdminSocketHook; +class PeerReplayer; +class ServiceDaemon; + +// handle mirroring for a filesystem to a set of peers + +class FSMirror { +public: + FSMirror(CephContext *cct, const Filesystem &filesystem, uint64_t pool_id, + ServiceDaemon *service_daemon, std::vector<const char*> args, + ContextWQ *work_queue); + ~FSMirror(); + + void init(Context *on_finish); + void shutdown(Context *on_finish); + + void add_peer(const Peer &peer); + void remove_peer(const Peer &peer); + + bool is_stopping() { + std::scoped_lock locker(m_lock); + return m_stopping; + } + + bool is_init_failed() { + std::scoped_lock locker(m_lock); + return m_init_failed; + } + + bool is_failed() { + std::scoped_lock locker(m_lock); + return m_init_failed || + m_instance_watcher->is_failed() || + m_mirror_watcher->is_failed(); + } + + bool is_blocklisted() { + std::scoped_lock locker(m_lock); + return is_blocklisted(locker); + } + + Peers get_peers() { + std::scoped_lock locker(m_lock); + return m_all_peers; + } + + std::string get_instance_addr() { + std::scoped_lock locker(m_lock); + return m_addrs; + } + + // admin socket helpers + void mirror_status(Formatter *f); + + void reopen_logs(); + +private: + bool is_blocklisted(const std::scoped_lock<ceph::mutex> &locker) const { + bool blocklisted = false; + if (m_instance_watcher) { + blocklisted = m_instance_watcher->is_blocklisted(); + } + if (m_mirror_watcher) { + blocklisted |= m_mirror_watcher->is_blocklisted(); + } + + return blocklisted; + } + + struct SnapListener : public InstanceWatcher::Listener { + FSMirror *fs_mirror; + + SnapListener(FSMirror *fs_mirror) + : fs_mirror(fs_mirror) { + } + + void acquire_directory(string_view dir_path) override { + fs_mirror->handle_acquire_directory(dir_path); + } + + void release_directory(string_view dir_path) override { + fs_mirror->handle_release_directory(dir_path); + } + }; + + CephContext *m_cct; + Filesystem m_filesystem; + uint64_t m_pool_id; + ServiceDaemon *m_service_daemon; + std::vector<const char *> m_args; + ContextWQ *m_work_queue; + + ceph::mutex m_lock = ceph::make_mutex("cephfs::mirror::fs_mirror"); + SnapListener m_snap_listener; + std::set<std::string, std::less<>> m_directories; + Peers m_all_peers; + std::map<Peer, std::unique_ptr<PeerReplayer>> m_peer_replayers; + + RadosRef m_cluster; + std::string m_addrs; + librados::IoCtx m_ioctx; + InstanceWatcher *m_instance_watcher = nullptr; + MirrorWatcher *m_mirror_watcher = nullptr; + + int m_retval = 0; + bool m_stopping = false; + bool m_init_failed = false; + Context *m_on_init_finish = nullptr; + Context *m_on_shutdown_finish = nullptr; + + MirrorAdminSocketHook *m_asok_hook = nullptr; + + MountRef m_mount; + + int init_replayer(PeerReplayer *peer_replayer); + void shutdown_replayer(PeerReplayer *peer_replayer); + void cleanup(); + + void init_instance_watcher(Context *on_finish); + void handle_init_instance_watcher(int r); + + void init_mirror_watcher(); + void handle_init_mirror_watcher(int r); + + void shutdown_peer_replayers(); + + void shutdown_mirror_watcher(); + void handle_shutdown_mirror_watcher(int r); + + void shutdown_instance_watcher(); + void handle_shutdown_instance_watcher(int r); + + void handle_acquire_directory(string_view dir_path); + void handle_release_directory(string_view dir_path); +}; + +} // namespace mirror +} // namespace cephfs + +#endif // CEPHFS_MIRROR_FS_MIRROR_H diff --git a/src/tools/cephfs_mirror/InstanceWatcher.cc b/src/tools/cephfs_mirror/InstanceWatcher.cc new file mode 100644 index 000000000..9c357da31 --- /dev/null +++ b/src/tools/cephfs_mirror/InstanceWatcher.cc @@ -0,0 +1,251 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "common/ceph_context.h" +#include "common/ceph_json.h" +#include "common/debug.h" +#include "common/errno.h" +#include "common/WorkQueue.h" +#include "cls/cephfs/cls_cephfs_client.h" +#include "include/stringify.h" +#include "aio_utils.h" +#include "InstanceWatcher.h" +#include "Types.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_cephfs_mirror +#undef dout_prefix +#define dout_prefix *_dout << "cephfs::mirror::InstanceWatcher " << __func__ + +namespace cephfs { +namespace mirror { + +namespace { + +std::string instance_oid(const std::string &instance_id) { + return CEPHFS_MIRROR_OBJECT + "." + instance_id; +} + +} // anonymous namespace + +InstanceWatcher::InstanceWatcher(librados::IoCtx &ioctx, + Listener &listener, ContextWQ *work_queue) + : Watcher(ioctx, instance_oid(stringify(ioctx.get_instance_id())), work_queue), + m_ioctx(ioctx), + m_listener(listener), + m_work_queue(work_queue), + m_lock(ceph::make_mutex("cephfs::mirror::instance_watcher")) { +} + +InstanceWatcher::~InstanceWatcher() { +} + +void InstanceWatcher::init(Context *on_finish) { + dout(20) << dendl; + + { + std::scoped_lock locker(m_lock); + ceph_assert(m_on_init_finish == nullptr); + m_on_init_finish = new LambdaContext([this, on_finish](int r) { + on_finish->complete(r); + if (m_on_shutdown_finish != nullptr) { + m_on_shutdown_finish->complete(0); + } + }); + } + + create_instance(); +} + +void InstanceWatcher::shutdown(Context *on_finish) { + dout(20) << dendl; + + { + std::scoped_lock locker(m_lock); + ceph_assert(m_on_shutdown_finish == nullptr); + if (m_on_init_finish != nullptr) { + dout(10) << ": delaying shutdown -- init in progress" << dendl; + m_on_shutdown_finish = new LambdaContext([this, on_finish](int r) { + m_on_shutdown_finish = nullptr; + shutdown(on_finish); + }); + return; + } + + m_on_shutdown_finish = on_finish; + } + + unregister_watcher(); +} + +void InstanceWatcher::handle_notify(uint64_t notify_id, uint64_t handle, + uint64_t notifier_id, bufferlist& bl) { + dout(20) << dendl; + + std::string dir_path; + std::string mode; + try { + JSONDecoder jd(bl); + JSONDecoder::decode_json("dir_path", dir_path, &jd.parser, true); + JSONDecoder::decode_json("mode", mode, &jd.parser, true); + } catch (const JSONDecoder::err &e) { + derr << ": failed to decode notify json: " << e.what() << dendl; + } + + dout(20) << ": notifier_id=" << notifier_id << ", dir_path=" << dir_path + << ", mode=" << mode << dendl; + + if (mode == "acquire") { + m_listener.acquire_directory(dir_path); + } else if (mode == "release") { + m_listener.release_directory(dir_path); + } else { + derr << ": unknown mode" << dendl; + } + + bufferlist outbl; + acknowledge_notify(notify_id, handle, outbl); +} + +void InstanceWatcher::handle_rewatch_complete(int r) { + dout(5) << ": r=" << r << dendl; + + if (r == -EBLOCKLISTED) { + dout(0) << ": client blocklisted" <<dendl; + std::scoped_lock locker(m_lock); + m_blocklisted = true; + } else if (r == -ENOENT) { + derr << ": mirroring object deleted" << dendl; + m_failed = true; + } else if (r < 0) { + derr << ": rewatch error: " << cpp_strerror(r) << dendl; + m_failed = true; + } +} + +void InstanceWatcher::create_instance() { + dout(20) << dendl; + + std::scoped_lock locker(m_lock); + librados::ObjectWriteOperation op; + op.create(false); + + librados::AioCompletion *aio_comp = + librados::Rados::aio_create_completion( + this, &rados_callback<InstanceWatcher, &InstanceWatcher::handle_create_instance>); + int r = m_ioctx.aio_operate(m_oid, aio_comp, &op); + ceph_assert(r == 0); + aio_comp->release(); +} + +void InstanceWatcher::handle_create_instance(int r) { + dout(20) << ": r=" << r << dendl; + + Context *on_init_finish = nullptr; + { + std::scoped_lock locker(m_lock); + if (r < 0) { + std::swap(on_init_finish, m_on_init_finish); + } + } + + if (on_init_finish != nullptr) { + on_init_finish->complete(r); + return; + } + + register_watcher(); +} + +void InstanceWatcher::register_watcher() { + dout(20) << dendl; + + std::scoped_lock locker(m_lock); + Context *on_finish = new C_CallbackAdapter< + InstanceWatcher, &InstanceWatcher::handle_register_watcher>(this); + register_watch(on_finish); +} + +void InstanceWatcher::handle_register_watcher(int r) { + dout(20) << ": r=" << r << dendl; + + Context *on_init_finish = nullptr; + { + std::scoped_lock locker(m_lock); + if (r == 0) { + std::swap(on_init_finish, m_on_init_finish); + } + } + + if (on_init_finish != nullptr) { + on_init_finish->complete(r); + return; + } + + remove_instance(); +} + +void InstanceWatcher::unregister_watcher() { + dout(20) << dendl; + + std::scoped_lock locker(m_lock); + Context *on_finish = new C_CallbackAdapter< + InstanceWatcher, &InstanceWatcher::handle_unregister_watcher>(this); + unregister_watch(new C_AsyncCallback<ContextWQ>(m_work_queue, on_finish)); +} + +void InstanceWatcher::handle_unregister_watcher(int r) { + dout(20) << ": r=" << r << dendl; + + Context *on_shutdown_finish = nullptr; + { + std::scoped_lock locker(m_lock); + if (r < 0) { + std::swap(on_shutdown_finish, m_on_shutdown_finish); + } + } + + if (on_shutdown_finish != nullptr) { + on_shutdown_finish->complete(r); + return; + } + + remove_instance(); +} + +void InstanceWatcher::remove_instance() { + dout(20) << dendl; + + std::scoped_lock locker(m_lock); + librados::ObjectWriteOperation op; + op.remove(); + + librados::AioCompletion *aio_comp = + librados::Rados::aio_create_completion( + this, &rados_callback<InstanceWatcher, &InstanceWatcher::handle_remove_instance>); + int r = m_ioctx.aio_operate(m_oid, aio_comp, &op); + ceph_assert(r == 0); + aio_comp->release(); +} + +void InstanceWatcher::handle_remove_instance(int r) { + dout(20) << ": r=" << r << dendl; + + Context *on_init_finish = nullptr; + Context *on_shutdown_finish = nullptr; + { + std::scoped_lock locker(m_lock); + std::swap(on_init_finish, m_on_init_finish); + std::swap(on_shutdown_finish, m_on_shutdown_finish); + } + + if (on_init_finish != nullptr) { + on_init_finish->complete(r); + } + if (on_shutdown_finish != nullptr) { + on_shutdown_finish->complete(r); + } +} + +} // namespace mirror +} // namespace cephfs diff --git a/src/tools/cephfs_mirror/InstanceWatcher.h b/src/tools/cephfs_mirror/InstanceWatcher.h new file mode 100644 index 000000000..06edf5da9 --- /dev/null +++ b/src/tools/cephfs_mirror/InstanceWatcher.h @@ -0,0 +1,85 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPHFS_MIRROR_INSTANCE_WATCHER_H +#define CEPHFS_MIRROR_INSTANCE_WATCHER_H + +#include <string_view> + +#include "common/ceph_mutex.h" +#include "include/Context.h" +#include "include/rados/librados.hpp" +#include "Watcher.h" + +class ContextWQ; + +namespace cephfs { +namespace mirror { + +// watch directory update notifications via per daemon rados +// object and invoke listener callback. + +class InstanceWatcher : public Watcher { +public: + struct Listener { + virtual ~Listener() { + } + + virtual void acquire_directory(string_view dir_path) = 0; + virtual void release_directory(string_view dir_path) = 0; + }; + + static InstanceWatcher *create(librados::IoCtx &ioctx, + Listener &listener, ContextWQ *work_queue) { + return new InstanceWatcher(ioctx, listener, work_queue); + } + + InstanceWatcher(librados::IoCtx &ioctx, Listener &listener, ContextWQ *work_queue); + ~InstanceWatcher(); + + void init(Context *on_finish); + void shutdown(Context *on_finish); + + void handle_notify(uint64_t notify_id, uint64_t handle, + uint64_t notifier_id, bufferlist& bl) override; + void handle_rewatch_complete(int r) override; + + bool is_blocklisted() { + std::scoped_lock locker(m_lock); + return m_blocklisted; + } + + bool is_failed() { + std::scoped_lock locker(m_lock); + return m_failed; + } + +private: + librados::IoCtx &m_ioctx; + Listener &m_listener; + ContextWQ *m_work_queue; + + ceph::mutex m_lock; + Context *m_on_init_finish = nullptr; + Context *m_on_shutdown_finish = nullptr; + + bool m_blocklisted = false; + bool m_failed = false; + + void create_instance(); + void handle_create_instance(int r); + + void register_watcher(); + void handle_register_watcher(int r); + + void remove_instance(); + void handle_remove_instance(int r); + + void unregister_watcher(); + void handle_unregister_watcher(int r); +}; + +} // namespace mirror +} // namespace cephfs + +#endif // CEPHFS_MIRROR_INSTANCE_WATCHER_H diff --git a/src/tools/cephfs_mirror/Mirror.cc b/src/tools/cephfs_mirror/Mirror.cc new file mode 100644 index 000000000..890805764 --- /dev/null +++ b/src/tools/cephfs_mirror/Mirror.cc @@ -0,0 +1,602 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "common/ceph_argparse.h" +#include "common/ceph_context.h" +#include "common/common_init.h" +#include "common/Cond.h" +#include "common/debug.h" +#include "common/errno.h" +#include "common/Timer.h" +#include "common/WorkQueue.h" +#include "include/types.h" +#include "mon/MonClient.h" +#include "msg/Messenger.h" +#include "aio_utils.h" +#include "Mirror.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_cephfs_mirror +#undef dout_prefix +#define dout_prefix *_dout << "cephfs::mirror::Mirror " << __func__ + +namespace cephfs { +namespace mirror { + +namespace { + +const std::string SERVICE_DAEMON_MIRROR_ENABLE_FAILED_KEY("mirroring_failed"); + +class SafeTimerSingleton : public CommonSafeTimer<ceph::mutex> { +public: + ceph::mutex timer_lock = ceph::make_mutex("cephfs::mirror::timer_lock"); + + explicit SafeTimerSingleton(CephContext *cct) + : SafeTimer(cct, timer_lock, true) { + init(); + } +}; + +class ThreadPoolSingleton : public ThreadPool { +public: + ContextWQ *work_queue = nullptr; + + explicit ThreadPoolSingleton(CephContext *cct) + : ThreadPool(cct, "Mirror::thread_pool", "tp_mirror", 1) { + work_queue = new ContextWQ("Mirror::work_queue", ceph::make_timespan(60), this); + + start(); + } +}; + +} // anonymous namespace + +struct Mirror::C_EnableMirroring : Context { + Mirror *mirror; + Filesystem filesystem; + uint64_t pool_id; + + C_EnableMirroring(Mirror *mirror, const Filesystem &filesystem, uint64_t pool_id) + : mirror(mirror), + filesystem(filesystem), + pool_id(pool_id) { + } + + void finish(int r) override { + enable_mirroring(); + } + + void enable_mirroring() { + Context *ctx = new C_CallbackAdapter<C_EnableMirroring, + &C_EnableMirroring::handle_enable_mirroring>(this); + mirror->enable_mirroring(filesystem, pool_id, ctx); + } + + void handle_enable_mirroring(int r) { + mirror->handle_enable_mirroring(filesystem, r); + delete this; + } + + // context needs to live post completion + void complete(int r) override { + finish(r); + } +}; + +struct Mirror::C_DisableMirroring : Context { + Mirror *mirror; + Filesystem filesystem; + + C_DisableMirroring(Mirror *mirror, const Filesystem &filesystem) + : mirror(mirror), + filesystem(filesystem) { + } + + void finish(int r) override { + disable_mirroring(); + } + + void disable_mirroring() { + Context *ctx = new C_CallbackAdapter<C_DisableMirroring, + &C_DisableMirroring::handle_disable_mirroring>(this); + mirror->disable_mirroring(filesystem, ctx); + } + + void handle_disable_mirroring(int r) { + mirror->handle_disable_mirroring(filesystem, r); + delete this; + } + + // context needs to live post completion + void complete(int r) override { + finish(r); + } +}; + +struct Mirror::C_PeerUpdate : Context { + Mirror *mirror; + Filesystem filesystem; + Peer peer; + bool remove = false; + + C_PeerUpdate(Mirror *mirror, const Filesystem &filesystem, + const Peer &peer) + : mirror(mirror), + filesystem(filesystem), + peer(peer) { + } + C_PeerUpdate(Mirror *mirror, const Filesystem &filesystem, + const Peer &peer, bool remove) + : mirror(mirror), + filesystem(filesystem), + peer(peer), + remove(remove) { + } + + void finish(int r) override { + if (remove) { + mirror->remove_peer(filesystem, peer); + } else { + mirror->add_peer(filesystem, peer); + } + } +}; + +struct Mirror::C_RestartMirroring : Context { + Mirror *mirror; + Filesystem filesystem; + uint64_t pool_id; + Peers peers; + + C_RestartMirroring(Mirror *mirror, const Filesystem &filesystem, + uint64_t pool_id, const Peers &peers) + : mirror(mirror), + filesystem(filesystem), + pool_id(pool_id), + peers(peers) { + } + + void finish(int r) override { + disable_mirroring(); + } + + void disable_mirroring() { + Context *ctx = new C_CallbackAdapter<C_RestartMirroring, + &C_RestartMirroring::handle_disable_mirroring>(this); + mirror->disable_mirroring(filesystem, ctx); + } + + void handle_disable_mirroring(int r) { + enable_mirroring(); + } + + void enable_mirroring() { + std::scoped_lock locker(mirror->m_lock); + Context *ctx = new C_CallbackAdapter<C_RestartMirroring, + &C_RestartMirroring::handle_enable_mirroring>(this); + mirror->enable_mirroring(filesystem, pool_id, ctx, true); + } + + void handle_enable_mirroring(int r) { + mirror->handle_enable_mirroring(filesystem, peers, r); + delete this; + } + + // context needs to live post completion + void complete(int r) override { + finish(r); + } +}; + +Mirror::Mirror(CephContext *cct, const std::vector<const char*> &args, + MonClient *monc, Messenger *msgr) + : m_cct(cct), + m_args(args), + m_monc(monc), + m_msgr(msgr), + m_listener(this), + m_last_blocklist_check(ceph_clock_now()), + m_last_failure_check(ceph_clock_now()), + m_local(new librados::Rados()) { + auto thread_pool = &(cct->lookup_or_create_singleton_object<ThreadPoolSingleton>( + "cephfs::mirror::thread_pool", false, cct)); + auto safe_timer = &(cct->lookup_or_create_singleton_object<SafeTimerSingleton>( + "cephfs::mirror::safe_timer", false, cct)); + m_thread_pool = thread_pool; + m_work_queue = thread_pool->work_queue; + m_timer = safe_timer; + m_timer_lock = &safe_timer->timer_lock; + std::scoped_lock timer_lock(*m_timer_lock); + schedule_mirror_update_task(); +} + +Mirror::~Mirror() { + dout(10) << dendl; + { + std::scoped_lock timer_lock(*m_timer_lock); + m_timer->shutdown(); + } + + m_work_queue->drain(); + delete m_work_queue; + { + std::scoped_lock locker(m_lock); + m_thread_pool->stop(); + } +} + +int Mirror::init_mon_client() { + dout(20) << dendl; + + m_monc->set_messenger(m_msgr); + m_msgr->add_dispatcher_head(m_monc); + m_monc->set_want_keys(CEPH_ENTITY_TYPE_MON); + + int r = m_monc->init(); + if (r < 0) { + derr << ": failed to init mon client: " << cpp_strerror(r) << dendl; + return r; + } + + r = m_monc->authenticate(std::chrono::duration<double>(m_cct->_conf.get_val<std::chrono::seconds>("client_mount_timeout")).count()); + if (r < 0) { + derr << ": failed to authenticate to monitor: " << cpp_strerror(r) << dendl; + return r; + } + + client_t me = m_monc->get_global_id(); + m_msgr->set_myname(entity_name_t::CLIENT(me.v)); + return 0; +} + +int Mirror::init(std::string &reason) { + dout(20) << dendl; + + std::scoped_lock locker(m_lock); + + int r = m_local->init_with_context(m_cct); + if (r < 0) { + derr << ": could not initialize rados handler" << dendl; + return r; + } + + r = m_local->connect(); + if (r < 0) { + derr << ": error connecting to local cluster" << dendl; + return r; + } + + m_service_daemon = std::make_unique<ServiceDaemon>(m_cct, m_local); + r = m_service_daemon->init(); + if (r < 0) { + derr << ": error registering service daemon: " << cpp_strerror(r) << dendl; + return r; + } + + r = init_mon_client(); + if (r < 0) { + return r; + } + + return 0; +} + +void Mirror::shutdown() { + dout(20) << dendl; + m_stopping = true; + m_cluster_watcher->shutdown(); + m_cond.notify_all(); +} + +void Mirror::reopen_logs() { + for (auto &[filesystem, mirror_action] : m_mirror_actions) { + mirror_action.fs_mirror->reopen_logs(); + } + g_ceph_context->reopen_logs(); +} + +void Mirror::handle_signal(int signum) { + dout(10) << ": signal=" << signum << dendl; + + std::scoped_lock locker(m_lock); + switch (signum) { + case SIGHUP: + reopen_logs(); + break; + case SIGINT: + case SIGTERM: + shutdown(); + break; + default: + ceph_abort_msgf("unexpected signal %d", signum); + } +} + +void Mirror::handle_enable_mirroring(const Filesystem &filesystem, + const Peers &peers, int r) { + dout(20) << ": filesystem=" << filesystem << ", peers=" << peers + << ", r=" << r << dendl; + + std::scoped_lock locker(m_lock); + auto &mirror_action = m_mirror_actions.at(filesystem); + ceph_assert(mirror_action.action_in_progress); + + mirror_action.action_in_progress = false; + m_cond.notify_all(); + if (r < 0) { + derr << ": failed to initialize FSMirror for filesystem=" << filesystem + << ": " << cpp_strerror(r) << dendl; + m_service_daemon->add_or_update_fs_attribute(filesystem.fscid, + SERVICE_DAEMON_MIRROR_ENABLE_FAILED_KEY, + true); + return; + } + + for (auto &peer : peers) { + mirror_action.fs_mirror->add_peer(peer); + } + + dout(10) << ": Initialized FSMirror for filesystem=" << filesystem << dendl; +} + +void Mirror::handle_enable_mirroring(const Filesystem &filesystem, int r) { + dout(20) << ": filesystem=" << filesystem << ", r=" << r << dendl; + + std::scoped_lock locker(m_lock); + auto &mirror_action = m_mirror_actions.at(filesystem); + ceph_assert(mirror_action.action_in_progress); + + mirror_action.action_in_progress = false; + m_cond.notify_all(); + if (r < 0) { + derr << ": failed to initialize FSMirror for filesystem=" << filesystem + << ": " << cpp_strerror(r) << dendl; + m_service_daemon->add_or_update_fs_attribute(filesystem.fscid, + SERVICE_DAEMON_MIRROR_ENABLE_FAILED_KEY, + true); + return; + } + + dout(10) << ": Initialized FSMirror for filesystem=" << filesystem << dendl; +} + +void Mirror::enable_mirroring(const Filesystem &filesystem, uint64_t local_pool_id, + Context *on_finish, bool is_restart) { + ceph_assert(ceph_mutex_is_locked(m_lock)); + + auto &mirror_action = m_mirror_actions.at(filesystem); + if (is_restart) { + mirror_action.fs_mirror.reset(); + } else { + ceph_assert(!mirror_action.action_in_progress); + } + + ceph_assert(!mirror_action.fs_mirror); + + dout(10) << ": starting FSMirror: filesystem=" << filesystem << dendl; + + mirror_action.action_in_progress = true; + mirror_action.fs_mirror = std::make_unique<FSMirror>(m_cct, filesystem, local_pool_id, + m_service_daemon.get(), m_args, m_work_queue); + mirror_action.fs_mirror->init(new C_AsyncCallback<ContextWQ>(m_work_queue, on_finish)); +} + +void Mirror::mirroring_enabled(const Filesystem &filesystem, uint64_t local_pool_id) { + dout(10) << ": filesystem=" << filesystem << ", pool_id=" << local_pool_id << dendl; + + std::scoped_lock locker(m_lock); + if (m_stopping) { + return; + } + + auto p = m_mirror_actions.emplace(filesystem, MirrorAction(local_pool_id)); + auto &mirror_action = p.first->second; + mirror_action.action_ctxs.push_back(new C_EnableMirroring(this, filesystem, local_pool_id)); +} + +void Mirror::handle_disable_mirroring(const Filesystem &filesystem, int r) { + dout(10) << ": filesystem=" << filesystem << ", r=" << r << dendl; + + std::scoped_lock locker(m_lock); + auto &mirror_action = m_mirror_actions.at(filesystem); + + if (!mirror_action.fs_mirror->is_init_failed()) { + ceph_assert(mirror_action.action_in_progress); + mirror_action.action_in_progress = false; + m_cond.notify_all(); + } + + if (!m_stopping) { + mirror_action.fs_mirror.reset(); + if (mirror_action.action_ctxs.empty()) { + dout(10) << ": no pending actions for filesystem=" << filesystem << dendl; + m_mirror_actions.erase(filesystem); + } + } +} + +void Mirror::disable_mirroring(const Filesystem &filesystem, Context *on_finish) { + ceph_assert(ceph_mutex_is_locked(m_lock)); + + auto &mirror_action = m_mirror_actions.at(filesystem); + ceph_assert(mirror_action.fs_mirror); + ceph_assert(!mirror_action.action_in_progress); + + if (mirror_action.fs_mirror->is_init_failed()) { + dout(10) << ": init failed for filesystem=" << filesystem << dendl; + m_work_queue->queue(on_finish, -EINVAL); + return; + } + + mirror_action.action_in_progress = true; + mirror_action.fs_mirror->shutdown(new C_AsyncCallback<ContextWQ>(m_work_queue, on_finish)); +} + +void Mirror::mirroring_disabled(const Filesystem &filesystem) { + dout(10) << ": filesystem=" << filesystem << dendl; + + std::scoped_lock locker(m_lock); + if (m_stopping) { + dout(5) << "shutting down" << dendl; + return; + } + + auto &mirror_action = m_mirror_actions.at(filesystem); + mirror_action.action_ctxs.push_back(new C_DisableMirroring(this, filesystem)); +} + +void Mirror::add_peer(const Filesystem &filesystem, const Peer &peer) { + ceph_assert(ceph_mutex_is_locked(m_lock)); + + auto &mirror_action = m_mirror_actions.at(filesystem); + ceph_assert(mirror_action.fs_mirror); + ceph_assert(!mirror_action.action_in_progress); + + mirror_action.fs_mirror->add_peer(peer); +} + +void Mirror::peer_added(const Filesystem &filesystem, const Peer &peer) { + dout(20) << ": filesystem=" << filesystem << ", peer=" << peer << dendl; + + std::scoped_lock locker(m_lock); + if (m_stopping) { + dout(5) << "shutting down" << dendl; + return; + } + + auto &mirror_action = m_mirror_actions.at(filesystem); + mirror_action.action_ctxs.push_back(new C_PeerUpdate(this, filesystem, peer)); +} + +void Mirror::remove_peer(const Filesystem &filesystem, const Peer &peer) { + ceph_assert(ceph_mutex_is_locked(m_lock)); + + auto &mirror_action = m_mirror_actions.at(filesystem); + ceph_assert(mirror_action.fs_mirror); + ceph_assert(!mirror_action.action_in_progress); + + mirror_action.fs_mirror->remove_peer(peer); +} + +void Mirror::peer_removed(const Filesystem &filesystem, const Peer &peer) { + dout(20) << ": filesystem=" << filesystem << ", peer=" << peer << dendl; + + std::scoped_lock locker(m_lock); + if (m_stopping) { + dout(5) << "shutting down" << dendl; + return; + } + + auto &mirror_action = m_mirror_actions.at(filesystem); + mirror_action.action_ctxs.push_back(new C_PeerUpdate(this, filesystem, peer, true)); +} + +void Mirror::update_fs_mirrors() { + dout(20) << dendl; + + auto now = ceph_clock_now(); + double blocklist_interval = g_ceph_context->_conf.get_val<std::chrono::seconds> + ("cephfs_mirror_restart_mirror_on_blocklist_interval").count(); + bool check_blocklist = blocklist_interval > 0 && ((now - m_last_blocklist_check) >= blocklist_interval); + + double failed_interval = g_ceph_context->_conf.get_val<std::chrono::seconds> + ("cephfs_mirror_restart_mirror_on_failure_interval").count(); + bool check_failure = failed_interval > 0 && ((now - m_last_failure_check) >= failed_interval); + + { + std::scoped_lock locker(m_lock); + for (auto &[filesystem, mirror_action] : m_mirror_actions) { + auto failed = mirror_action.fs_mirror && mirror_action.fs_mirror->is_failed(); + auto blocklisted = mirror_action.fs_mirror && mirror_action.fs_mirror->is_blocklisted(); + + if (check_failure && !mirror_action.action_in_progress && failed) { + // about to restart failed mirror instance -- nothing + // should interfere + dout(5) << ": filesystem=" << filesystem << " failed mirroring -- restarting" << dendl; + auto peers = mirror_action.fs_mirror->get_peers(); + auto ctx = new C_RestartMirroring(this, filesystem, mirror_action.pool_id, peers); + ctx->complete(0); + } else if (check_blocklist && !mirror_action.action_in_progress && blocklisted) { + // about to restart blocklisted mirror instance -- nothing + // should interfere + dout(5) << ": filesystem=" << filesystem << " is blocklisted -- restarting" << dendl; + auto peers = mirror_action.fs_mirror->get_peers(); + auto ctx = new C_RestartMirroring(this, filesystem, mirror_action.pool_id, peers); + ctx->complete(0); + } + if (!failed && !blocklisted && !mirror_action.action_ctxs.empty() + && !mirror_action.action_in_progress) { + auto ctx = std::move(mirror_action.action_ctxs.front()); + mirror_action.action_ctxs.pop_front(); + ctx->complete(0); + } + } + + if (check_blocklist) { + m_last_blocklist_check = now; + } + if (check_failure) { + m_last_failure_check = now; + } + } + + schedule_mirror_update_task(); +} + +void Mirror::schedule_mirror_update_task() { + ceph_assert(m_timer_task == nullptr); + ceph_assert(ceph_mutex_is_locked(*m_timer_lock)); + + m_timer_task = new LambdaContext([this](int _) { + m_timer_task = nullptr; + update_fs_mirrors(); + }); + double after = g_ceph_context->_conf.get_val<std::chrono::seconds> + ("cephfs_mirror_action_update_interval").count(); + dout(20) << ": scheduling fs mirror update (" << m_timer_task << ") after " + << after << " seconds" << dendl; + m_timer->add_event_after(after, m_timer_task); +} + +void Mirror::run() { + dout(20) << dendl; + + std::unique_lock locker(m_lock); + m_cluster_watcher.reset(new ClusterWatcher(m_cct, m_monc, m_service_daemon.get(), m_listener)); + m_msgr->add_dispatcher_tail(m_cluster_watcher.get()); + + m_cluster_watcher->init(); + m_cond.wait(locker, [this]{return m_stopping;}); + + locker.unlock(); + { + std::scoped_lock timer_lock(*m_timer_lock); + if (m_timer_task != nullptr) { + dout(10) << ": canceling timer task=" << m_timer_task << dendl; + m_timer->cancel_event(m_timer_task); + m_timer_task = nullptr; + } + } + locker.lock(); + + for (auto &[filesystem, mirror_action] : m_mirror_actions) { + dout(10) << ": trying to shutdown filesystem=" << filesystem << dendl; + // wait for in-progress action and shutdown + m_cond.wait(locker, [&mirror_action=mirror_action] + {return !mirror_action.action_in_progress;}); + if (mirror_action.fs_mirror && + !mirror_action.fs_mirror->is_stopping() && + !mirror_action.fs_mirror->is_init_failed()) { + C_SaferCond cond; + mirror_action.fs_mirror->shutdown(new C_AsyncCallback<ContextWQ>(m_work_queue, &cond)); + int r = cond.wait(); + dout(10) << ": shutdown filesystem=" << filesystem << ", r=" << r << dendl; + } + + mirror_action.fs_mirror.reset(); + } +} + +} // namespace mirror +} // namespace cephfs + diff --git a/src/tools/cephfs_mirror/Mirror.h b/src/tools/cephfs_mirror/Mirror.h new file mode 100644 index 000000000..f0ffdd516 --- /dev/null +++ b/src/tools/cephfs_mirror/Mirror.h @@ -0,0 +1,140 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPHFS_MIRROR_H +#define CEPHFS_MIRROR_H + +#include <map> +#include <set> +#include <vector> + +#include "common/ceph_mutex.h" +#include "common/WorkQueue.h" +#include "mds/FSMap.h" +#include "ClusterWatcher.h" +#include "FSMirror.h" +#include "ServiceDaemon.h" +#include "Types.h" + +class Messenger; +class MonClient; +class ContextWQ; + +namespace cephfs { +namespace mirror { + +// this wraps up ClusterWatcher and FSMirrors to implement mirroring +// for ceph filesystems. + +class Mirror { +public: + Mirror(CephContext *cct, const std::vector<const char*> &args, + MonClient *monc, Messenger *msgr); + ~Mirror(); + + int init(std::string &reason); + void shutdown(); + void run(); + + void handle_signal(int signum); + +private: + static constexpr std::string_view MIRRORING_MODULE = "mirroring"; + + struct C_EnableMirroring; + struct C_DisableMirroring; + struct C_PeerUpdate; + struct C_RestartMirroring; + + struct ClusterListener : ClusterWatcher::Listener { + Mirror *mirror; + + ClusterListener(Mirror *mirror) + : mirror(mirror) { + } + + void handle_mirroring_enabled(const FilesystemSpec &spec) override { + mirror->mirroring_enabled(spec.filesystem, spec.pool_id); + } + + void handle_mirroring_disabled(const Filesystem &filesystem) override { + mirror->mirroring_disabled(filesystem); + } + + void handle_peers_added(const Filesystem &filesystem, const Peer &peer) override { + mirror->peer_added(filesystem, peer); + } + + void handle_peers_removed(const Filesystem &filesystem, const Peer &peer) override { + mirror->peer_removed(filesystem, peer); + } + }; + + struct MirrorAction { + MirrorAction(uint64_t pool_id) : + pool_id(pool_id) { + } + + uint64_t pool_id; // for restarting blocklisted mirror instance + bool action_in_progress = false; + std::list<Context *> action_ctxs; + std::unique_ptr<FSMirror> fs_mirror; + }; + + ceph::mutex m_lock = ceph::make_mutex("cephfs::mirror::Mirror"); + ceph::condition_variable m_cond; + + CephContext *m_cct; + std::vector<const char *> m_args; + MonClient *m_monc; + Messenger *m_msgr; + ClusterListener m_listener; + + ThreadPool *m_thread_pool = nullptr; + ContextWQ *m_work_queue = nullptr; + SafeTimer *m_timer = nullptr; + ceph::mutex *m_timer_lock = nullptr; + Context *m_timer_task = nullptr; + + bool m_stopping = false; + std::unique_ptr<ClusterWatcher> m_cluster_watcher; + std::map<Filesystem, MirrorAction> m_mirror_actions; + + utime_t m_last_blocklist_check; + utime_t m_last_failure_check; + + RadosRef m_local; + std::unique_ptr<ServiceDaemon> m_service_daemon; + + int init_mon_client(); + + // called via listener + void mirroring_enabled(const Filesystem &filesystem, uint64_t local_pool_id); + void mirroring_disabled(const Filesystem &filesystem); + void peer_added(const Filesystem &filesystem, const Peer &peer); + void peer_removed(const Filesystem &filesystem, const Peer &peer); + + // mirror enable callback + void enable_mirroring(const Filesystem &filesystem, uint64_t local_pool_id, + Context *on_finish, bool is_restart=false); + void handle_enable_mirroring(const Filesystem &filesystem, int r); + void handle_enable_mirroring(const Filesystem &filesystem, const Peers &peers, int r); + + // mirror disable callback + void disable_mirroring(const Filesystem &filesystem, Context *on_finish); + void handle_disable_mirroring(const Filesystem &filesystem, int r); + + // peer update callback + void add_peer(const Filesystem &filesystem, const Peer &peer); + void remove_peer(const Filesystem &filesystem, const Peer &peer); + + void schedule_mirror_update_task(); + void update_fs_mirrors(); + + void reopen_logs(); +}; + +} // namespace mirror +} // namespace cephfs + +#endif // CEPHFS_MIRROR_H diff --git a/src/tools/cephfs_mirror/MirrorWatcher.cc b/src/tools/cephfs_mirror/MirrorWatcher.cc new file mode 100644 index 000000000..26b88d077 --- /dev/null +++ b/src/tools/cephfs_mirror/MirrorWatcher.cc @@ -0,0 +1,148 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "common/ceph_context.h" +#include "common/ceph_json.h" +#include "common/debug.h" +#include "common/errno.h" +#include "common/WorkQueue.h" +#include "include/stringify.h" +#include "msg/Messenger.h" +#include "aio_utils.h" +#include "MirrorWatcher.h" +#include "FSMirror.h" +#include "Types.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_cephfs_mirror +#undef dout_prefix +#define dout_prefix *_dout << "cephfs::mirror::MirrorWatcher " << __func__ + +namespace cephfs { +namespace mirror { + +MirrorWatcher::MirrorWatcher(librados::IoCtx &ioctx, FSMirror *fs_mirror, + ContextWQ *work_queue) + : Watcher(ioctx, CEPHFS_MIRROR_OBJECT, work_queue), + m_ioctx(ioctx), + m_fs_mirror(fs_mirror), + m_work_queue(work_queue), + m_lock(ceph::make_mutex("cephfs::mirror::mirror_watcher")), + m_instance_id(stringify(m_ioctx.get_instance_id())) { +} + +MirrorWatcher::~MirrorWatcher() { +} + +void MirrorWatcher::init(Context *on_finish) { + dout(20) << dendl; + + { + std::scoped_lock locker(m_lock); + ceph_assert(m_on_init_finish == nullptr); + m_on_init_finish = new LambdaContext([this, on_finish](int r) { + on_finish->complete(r); + if (m_on_shutdown_finish != nullptr) { + m_on_shutdown_finish->complete(0); + } + }); + } + + register_watcher(); +} + +void MirrorWatcher::shutdown(Context *on_finish) { + dout(20) << dendl; + + { + std::scoped_lock locker(m_lock); + ceph_assert(m_on_shutdown_finish == nullptr); + if (m_on_init_finish != nullptr) { + dout(10) << ": delaying shutdown -- init in progress" << dendl; + m_on_shutdown_finish = new LambdaContext([this, on_finish](int r) { + m_on_shutdown_finish = nullptr; + shutdown(on_finish); + }); + return; + } + + m_on_shutdown_finish = on_finish; + } + + unregister_watcher(); +} + +void MirrorWatcher::handle_notify(uint64_t notify_id, uint64_t handle, + uint64_t notifier_id, bufferlist& bl) { + dout(20) << dendl; + + JSONFormatter f; + f.open_object_section("info"); + encode_json("addr", m_fs_mirror->get_instance_addr(), &f); + f.close_section(); + + bufferlist outbl; + f.flush(outbl); + acknowledge_notify(notify_id, handle, outbl); +} + +void MirrorWatcher::handle_rewatch_complete(int r) { + dout(5) << ": r=" << r << dendl; + + if (r == -EBLOCKLISTED) { + dout(0) << ": client blocklisted" <<dendl; + std::scoped_lock locker(m_lock); + m_blocklisted = true; + } else if (r == -ENOENT) { + derr << ": mirroring object deleted" << dendl; + m_failed = true; + } else if (r < 0) { + derr << ": rewatch error: " << cpp_strerror(r) << dendl; + m_failed = true; + } +} + +void MirrorWatcher::register_watcher() { + dout(20) << dendl; + + std::scoped_lock locker(m_lock); + Context *on_finish = new C_CallbackAdapter< + MirrorWatcher, &MirrorWatcher::handle_register_watcher>(this); + register_watch(on_finish); +} + +void MirrorWatcher::handle_register_watcher(int r) { + dout(20) << ": r=" << r << dendl; + + Context *on_init_finish = nullptr; + { + std::scoped_lock locker(m_lock); + std::swap(on_init_finish, m_on_init_finish); + } + + on_init_finish->complete(r); +} + +void MirrorWatcher::unregister_watcher() { + dout(20) << dendl; + + std::scoped_lock locker(m_lock); + Context *on_finish = new C_CallbackAdapter< + MirrorWatcher, &MirrorWatcher::handle_unregister_watcher>(this); + unregister_watch(new C_AsyncCallback<ContextWQ>(m_work_queue, on_finish)); +} + +void MirrorWatcher::handle_unregister_watcher(int r) { + dout(20) << ": r=" << r << dendl; + + Context *on_shutdown_finish = nullptr; + { + std::scoped_lock locker(m_lock); + std::swap(on_shutdown_finish, m_on_shutdown_finish); + } + + on_shutdown_finish->complete(r); +} + +} // namespace mirror +} // namespace cephfs diff --git a/src/tools/cephfs_mirror/MirrorWatcher.h b/src/tools/cephfs_mirror/MirrorWatcher.h new file mode 100644 index 000000000..c4d4f4522 --- /dev/null +++ b/src/tools/cephfs_mirror/MirrorWatcher.h @@ -0,0 +1,79 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPHFS_MIRROR_MIRROR_WATCHER_H +#define CEPHFS_MIRROR_MIRROR_WATCHER_H + +#include <string_view> + +#include "common/ceph_mutex.h" +#include "include/Context.h" +#include "include/rados/librados.hpp" +#include "Watcher.h" + +class ContextWQ; +class Messenger; + +namespace cephfs { +namespace mirror { + +class FSMirror; + +// watch for notifications via cephfs_mirror object (in metadata +// pool). this is used sending keepalived with keepalive payload +// being the rados instance address (used by the manager module +// to blocklist when needed). + +class MirrorWatcher : public Watcher { +public: + static MirrorWatcher *create(librados::IoCtx &ioctx, FSMirror *fs_mirror, + ContextWQ *work_queue) { + return new MirrorWatcher(ioctx, fs_mirror, work_queue); + } + + MirrorWatcher(librados::IoCtx &ioctx, FSMirror *fs_mirror, + ContextWQ *work_queue); + ~MirrorWatcher(); + + void init(Context *on_finish); + void shutdown(Context *on_finish); + + void handle_notify(uint64_t notify_id, uint64_t handle, + uint64_t notifier_id, bufferlist& bl) override; + void handle_rewatch_complete(int r) override; + + bool is_blocklisted() { + std::scoped_lock locker(m_lock); + return m_blocklisted; + } + + bool is_failed() { + std::scoped_lock locker(m_lock); + return m_failed; + } + +private: + librados::IoCtx &m_ioctx; + FSMirror *m_fs_mirror; + ContextWQ *m_work_queue; + + ceph::mutex m_lock; + std::string m_instance_id; + + Context *m_on_init_finish = nullptr; + Context *m_on_shutdown_finish = nullptr; + + bool m_blocklisted = false; + bool m_failed = false; + + void register_watcher(); + void handle_register_watcher(int r); + + void unregister_watcher(); + void handle_unregister_watcher(int r); +}; + +} // namespace mirror +} // namespace cephfs + +#endif // CEPHFS_MIRROR_MIRROR_WATCHER_H diff --git a/src/tools/cephfs_mirror/PeerReplayer.cc b/src/tools/cephfs_mirror/PeerReplayer.cc new file mode 100644 index 000000000..aaf97b868 --- /dev/null +++ b/src/tools/cephfs_mirror/PeerReplayer.cc @@ -0,0 +1,1552 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include <stack> +#include <fcntl.h> +#include <algorithm> +#include <sys/time.h> +#include <sys/file.h> +#include <boost/scope_exit.hpp> + +#include "common/admin_socket.h" +#include "common/ceph_context.h" +#include "common/debug.h" +#include "common/errno.h" +#include "FSMirror.h" +#include "PeerReplayer.h" +#include "Utils.h" + +#include "json_spirit/json_spirit.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_cephfs_mirror +#undef dout_prefix +#define dout_prefix *_dout << "cephfs::mirror::PeerReplayer(" \ + << m_peer.uuid << ") " << __func__ + +namespace cephfs { +namespace mirror { + +namespace { + +const std::string PEER_CONFIG_KEY_PREFIX = "cephfs/mirror/peer"; + +std::string snapshot_dir_path(CephContext *cct, const std::string &path) { + return path + "/" + cct->_conf->client_snapdir; +} + +std::string snapshot_path(const std::string &snap_dir, const std::string &snap_name) { + return snap_dir + "/" + snap_name; +} + +std::string snapshot_path(CephContext *cct, const std::string &path, const std::string &snap_name) { + return path + "/" + cct->_conf->client_snapdir + "/" + snap_name; +} + +std::string entry_path(const std::string &dir, const std::string &name) { + return dir + "/" + name; +} + +std::map<std::string, std::string> decode_snap_metadata(snap_metadata *snap_metadata, + size_t nr_snap_metadata) { + std::map<std::string, std::string> metadata; + for (size_t i = 0; i < nr_snap_metadata; ++i) { + metadata.emplace(snap_metadata[i].key, snap_metadata[i].value); + } + + return metadata; +} + +std::string peer_config_key(const std::string &fs_name, const std::string &uuid) { + return PEER_CONFIG_KEY_PREFIX + "/" + fs_name + "/" + uuid; +} + +class PeerAdminSocketCommand { +public: + virtual ~PeerAdminSocketCommand() { + } + virtual int call(Formatter *f) = 0; +}; + +class StatusCommand : public PeerAdminSocketCommand { +public: + explicit StatusCommand(PeerReplayer *peer_replayer) + : peer_replayer(peer_replayer) { + } + + int call(Formatter *f) override { + peer_replayer->peer_status(f); + return 0; + } + +private: + PeerReplayer *peer_replayer; +}; + +// helper to open a directory relative to a file descriptor +int opendirat(MountRef mnt, int dirfd, const std::string &relpath, int flags, + ceph_dir_result **dirp) { + int r = ceph_openat(mnt, dirfd, relpath.c_str(), flags, 0); + if (r < 0) { + return r; + } + + int fd = r; + r = ceph_fdopendir(mnt, fd, dirp); + ceph_close(mnt, fd); + return r; +} + +} // anonymous namespace + +class PeerReplayerAdminSocketHook : public AdminSocketHook { +public: + PeerReplayerAdminSocketHook(CephContext *cct, const Filesystem &filesystem, + const Peer &peer, PeerReplayer *peer_replayer) + : admin_socket(cct->get_admin_socket()) { + int r; + std::string cmd; + + // mirror peer status format is name@id uuid + cmd = "fs mirror peer status " + + stringify(filesystem.fs_name) + "@" + stringify(filesystem.fscid) + + " " + + stringify(peer.uuid); + r = admin_socket->register_command( + cmd, this, "get peer mirror status"); + if (r == 0) { + commands[cmd] = new StatusCommand(peer_replayer); + } + } + + ~PeerReplayerAdminSocketHook() override { + admin_socket->unregister_commands(this); + for (auto &[command, cmdptr] : commands) { + delete cmdptr; + } + } + + int call(std::string_view command, const cmdmap_t& cmdmap, + Formatter *f, std::ostream &errss, bufferlist &out) override { + auto p = commands.at(std::string(command)); + return p->call(f); + } + +private: + typedef std::map<std::string, PeerAdminSocketCommand*, std::less<>> Commands; + + AdminSocket *admin_socket; + Commands commands; +}; + +PeerReplayer::PeerReplayer(CephContext *cct, FSMirror *fs_mirror, + RadosRef local_cluster, const Filesystem &filesystem, + const Peer &peer, const std::set<std::string, std::less<>> &directories, + MountRef mount, ServiceDaemon *service_daemon) + : m_cct(cct), + m_fs_mirror(fs_mirror), + m_local_cluster(local_cluster), + m_filesystem(filesystem), + m_peer(peer), + m_directories(directories.begin(), directories.end()), + m_local_mount(mount), + m_service_daemon(service_daemon), + m_asok_hook(new PeerReplayerAdminSocketHook(cct, filesystem, peer, this)), + m_lock(ceph::make_mutex("cephfs::mirror::PeerReplayer::" + stringify(peer.uuid))) { + // reset sync stats sent via service daemon + m_service_daemon->add_or_update_peer_attribute(m_filesystem.fscid, m_peer, + SERVICE_DAEMON_FAILED_DIR_COUNT_KEY, (uint64_t)0); + m_service_daemon->add_or_update_peer_attribute(m_filesystem.fscid, m_peer, + SERVICE_DAEMON_RECOVERED_DIR_COUNT_KEY, (uint64_t)0); +} + +PeerReplayer::~PeerReplayer() { + delete m_asok_hook; +} + +int PeerReplayer::init() { + dout(20) << ": initial dir list=[" << m_directories << "]" << dendl; + for (auto &dir_root : m_directories) { + m_snap_sync_stats.emplace(dir_root, SnapSyncStat()); + } + + auto &remote_client = m_peer.remote.client_name; + auto &remote_cluster = m_peer.remote.cluster_name; + auto remote_filesystem = Filesystem{0, m_peer.remote.fs_name}; + + std::string key = peer_config_key(m_filesystem.fs_name, m_peer.uuid); + std::string cmd = + "{" + "\"prefix\": \"config-key get\", " + "\"key\": \"" + key + "\"" + "}"; + + bufferlist in_bl; + bufferlist out_bl; + + int r = m_local_cluster->mon_command(cmd, in_bl, &out_bl, nullptr); + dout(5) << ": mon command r=" << r << dendl; + if (r < 0 && r != -ENOENT) { + return r; + } + + std::string mon_host; + std::string cephx_key; + if (!r) { + json_spirit::mValue root; + if (!json_spirit::read(out_bl.to_str(), root)) { + derr << ": invalid config-key JSON" << dendl; + return -EBADMSG; + } + try { + auto &root_obj = root.get_obj(); + mon_host = root_obj.at("mon_host").get_str(); + cephx_key = root_obj.at("key").get_str(); + dout(0) << ": remote monitor host=" << mon_host << dendl; + } catch (std::runtime_error&) { + derr << ": unexpected JSON received" << dendl; + return -EBADMSG; + } + } + + r = connect(remote_client, remote_cluster, &m_remote_cluster, mon_host, cephx_key); + if (r < 0) { + derr << ": error connecting to remote cluster: " << cpp_strerror(r) + << dendl; + return r; + } + + r = mount(m_remote_cluster, remote_filesystem, false, &m_remote_mount); + if (r < 0) { + m_remote_cluster.reset(); + derr << ": error mounting remote filesystem=" << remote_filesystem << dendl; + return r; + } + + std::scoped_lock locker(m_lock); + auto nr_replayers = g_ceph_context->_conf.get_val<uint64_t>( + "cephfs_mirror_max_concurrent_directory_syncs"); + dout(20) << ": spawning " << nr_replayers << " snapshot replayer(s)" << dendl; + + while (nr_replayers-- > 0) { + std::unique_ptr<SnapshotReplayerThread> replayer( + new SnapshotReplayerThread(this)); + std::string name("replayer-" + stringify(nr_replayers)); + replayer->create(name.c_str()); + m_replayers.push_back(std::move(replayer)); + } + + return 0; +} + +void PeerReplayer::shutdown() { + dout(20) << dendl; + + { + std::scoped_lock locker(m_lock); + ceph_assert(!m_stopping); + m_stopping = true; + m_cond.notify_all(); + } + + for (auto &replayer : m_replayers) { + replayer->join(); + } + m_replayers.clear(); + ceph_unmount(m_remote_mount); + ceph_release(m_remote_mount); + m_remote_mount = nullptr; + m_remote_cluster.reset(); +} + +void PeerReplayer::add_directory(string_view dir_root) { + dout(20) << ": dir_root=" << dir_root << dendl; + + std::scoped_lock locker(m_lock); + m_directories.emplace_back(dir_root); + m_snap_sync_stats.emplace(dir_root, SnapSyncStat()); + m_cond.notify_all(); +} + +void PeerReplayer::remove_directory(string_view dir_root) { + dout(20) << ": dir_root=" << dir_root << dendl; + auto _dir_root = std::string(dir_root); + + std::scoped_lock locker(m_lock); + auto it = std::find(m_directories.begin(), m_directories.end(), _dir_root); + if (it != m_directories.end()) { + m_directories.erase(it); + } + + auto it1 = m_registered.find(_dir_root); + if (it1 == m_registered.end()) { + m_snap_sync_stats.erase(_dir_root); + } else { + it1->second.canceled = true; + } + m_cond.notify_all(); +} + +boost::optional<std::string> PeerReplayer::pick_directory() { + dout(20) << dendl; + + auto now = clock::now(); + auto retry_timo = g_ceph_context->_conf.get_val<uint64_t>( + "cephfs_mirror_retry_failed_directories_interval"); + + boost::optional<std::string> candidate; + for (auto &dir_root : m_directories) { + auto &sync_stat = m_snap_sync_stats.at(dir_root); + if (sync_stat.failed) { + std::chrono::duration<double> d = now - *sync_stat.last_failed; + if (d.count() < retry_timo) { + continue; + } + } + if (!m_registered.count(dir_root)) { + candidate = dir_root; + break; + } + } + + std::rotate(m_directories.begin(), m_directories.begin() + 1, m_directories.end()); + return candidate; +} + +int PeerReplayer::register_directory(const std::string &dir_root, + SnapshotReplayerThread *replayer) { + dout(20) << ": dir_root=" << dir_root << dendl; + ceph_assert(m_registered.find(dir_root) == m_registered.end()); + + DirRegistry registry; + int r = try_lock_directory(dir_root, replayer, ®istry); + if (r < 0) { + return r; + } + + dout(5) << ": dir_root=" << dir_root << " registered with replayer=" + << replayer << dendl; + m_registered.emplace(dir_root, std::move(registry)); + return 0; +} + +void PeerReplayer::unregister_directory(const std::string &dir_root) { + dout(20) << ": dir_root=" << dir_root << dendl; + + auto it = m_registered.find(dir_root); + ceph_assert(it != m_registered.end()); + + unlock_directory(it->first, it->second); + m_registered.erase(it); + if (std::find(m_directories.begin(), m_directories.end(), dir_root) == m_directories.end()) { + m_snap_sync_stats.erase(dir_root); + } +} + +int PeerReplayer::try_lock_directory(const std::string &dir_root, + SnapshotReplayerThread *replayer, DirRegistry *registry) { + dout(20) << ": dir_root=" << dir_root << dendl; + + int r = ceph_open(m_remote_mount, dir_root.c_str(), O_RDONLY | O_DIRECTORY, 0); + if (r < 0 && r != -ENOENT) { + derr << ": failed to open remote dir_root=" << dir_root << ": " << cpp_strerror(r) + << dendl; + return r; + } + + if (r == -ENOENT) { + // we snap under dir_root, so mode does not matter much + r = ceph_mkdirs(m_remote_mount, dir_root.c_str(), 0755); + if (r < 0) { + derr << ": failed to create remote directory=" << dir_root << ": " << cpp_strerror(r) + << dendl; + return r; + } + + r = ceph_open(m_remote_mount, dir_root.c_str(), O_RDONLY | O_DIRECTORY, 0); + if (r < 0) { + derr << ": failed to open remote dir_root=" << dir_root << ": " << cpp_strerror(r) + << dendl; + return r; + } + } + + int fd = r; + r = ceph_flock(m_remote_mount, fd, LOCK_EX | LOCK_NB, (uint64_t)replayer->get_thread_id()); + if (r != 0) { + if (r == -EWOULDBLOCK) { + dout(5) << ": dir_root=" << dir_root << " is locked by cephfs-mirror, " + << "will retry again" << dendl; + } else { + derr << ": failed to lock dir_root=" << dir_root << ": " << cpp_strerror(r) + << dendl; + } + + if (ceph_close(m_remote_mount, fd) < 0) { + derr << ": failed to close (cleanup) remote dir_root=" << dir_root << ": " + << cpp_strerror(r) << dendl; + } + return r; + } + + dout(10) << ": dir_root=" << dir_root << " locked" << dendl; + + registry->fd = fd; + registry->replayer = replayer; + return 0; +} + +void PeerReplayer::unlock_directory(const std::string &dir_root, const DirRegistry ®istry) { + dout(20) << ": dir_root=" << dir_root << dendl; + + int r = ceph_flock(m_remote_mount, registry.fd, LOCK_UN, + (uint64_t)registry.replayer->get_thread_id()); + if (r < 0) { + derr << ": failed to unlock remote dir_root=" << dir_root << ": " << cpp_strerror(r) + << dendl; + return; + } + + r = ceph_close(m_remote_mount, registry.fd); + if (r < 0) { + derr << ": failed to close remote dir_root=" << dir_root << ": " << cpp_strerror(r) + << dendl; + } + + dout(10) << ": dir_root=" << dir_root << " unlocked" << dendl; +} + +int PeerReplayer::build_snap_map(const std::string &dir_root, + std::map<uint64_t, std::string> *snap_map, bool is_remote) { + auto snap_dir = snapshot_dir_path(m_cct, dir_root); + dout(20) << ": dir_root=" << dir_root << ", snap_dir=" << snap_dir + << ", is_remote=" << is_remote << dendl; + + auto lr_str = is_remote ? "remote" : "local"; + auto mnt = is_remote ? m_remote_mount : m_local_mount; + + ceph_dir_result *dirp = nullptr; + int r = ceph_opendir(mnt, snap_dir.c_str(), &dirp); + if (r < 0) { + if (is_remote && r == -ENOENT) { + return 0; + } + derr << ": failed to open " << lr_str << " snap directory=" << snap_dir + << ": " << cpp_strerror(r) << dendl; + return r; + } + + std::set<std::string> snaps; + auto entry = ceph_readdir(mnt, dirp); + while (entry != NULL) { + auto d_name = std::string(entry->d_name); + dout(20) << ": entry=" << d_name << dendl; + if (d_name != "." && d_name != ".." && d_name.rfind("_", 0) != 0) { + snaps.emplace(d_name); + } + + entry = ceph_readdir(mnt, dirp); + } + + int rv = 0; + for (auto &snap : snaps) { + snap_info info; + auto snap_path = snapshot_path(snap_dir, snap); + r = ceph_get_snap_info(mnt, snap_path.c_str(), &info); + if (r < 0) { + derr << ": failed to fetch " << lr_str << " snap info for snap_path=" << snap_path + << ": " << cpp_strerror(r) << dendl; + rv = r; + break; + } + + uint64_t snap_id; + if (is_remote) { + if (!info.nr_snap_metadata) { + derr << ": snap_path=" << snap_path << " has invalid metadata in remote snapshot" + << dendl; + rv = -EINVAL; + } else { + auto metadata = decode_snap_metadata(info.snap_metadata, info.nr_snap_metadata); + dout(20) << ": snap_path=" << snap_path << ", metadata=" << metadata << dendl; + auto it = metadata.find(PRIMARY_SNAP_ID_KEY); + if (it == metadata.end()) { + derr << ": snap_path=" << snap_path << " has missing \"" << PRIMARY_SNAP_ID_KEY + << "\" in metadata" << dendl; + rv = -EINVAL; + } else { + snap_id = std::stoull(it->second); + } + ceph_free_snap_info_buffer(&info); + } + } else { + snap_id = info.id; + } + + if (rv != 0) { + break; + } + snap_map->emplace(snap_id, snap); + } + + r = ceph_closedir(mnt, dirp); + if (r < 0) { + derr << ": failed to close " << lr_str << " snap directory=" << snap_dir + << ": " << cpp_strerror(r) << dendl; + } + + dout(10) << ": " << lr_str << " snap_map=" << *snap_map << dendl; + return rv; +} + +int PeerReplayer::propagate_snap_deletes(const std::string &dir_root, + const std::set<std::string> &snaps) { + dout(5) << ": dir_root=" << dir_root << ", deleted snapshots=" << snaps << dendl; + + for (auto &snap : snaps) { + dout(20) << ": deleting dir_root=" << dir_root << ", snapshot=" << snap + << dendl; + int r = ceph_rmsnap(m_remote_mount, dir_root.c_str(), snap.c_str()); + if (r < 0) { + derr << ": failed to delete remote snap dir_root=" << dir_root + << ", snapshot=" << snaps << ": " << cpp_strerror(r) << dendl; + return r; + } + inc_deleted_snap(dir_root); + } + + return 0; +} + +int PeerReplayer::propagate_snap_renames( + const std::string &dir_root, + const std::set<std::pair<std::string,std::string>> &snaps) { + dout(10) << ": dir_root=" << dir_root << ", renamed snapshots=" << snaps << dendl; + + for (auto &snapp : snaps) { + auto from = snapshot_path(m_cct, dir_root, snapp.first); + auto to = snapshot_path(m_cct, dir_root, snapp.second); + dout(20) << ": renaming dir_root=" << dir_root << ", snapshot from=" + << from << ", to=" << to << dendl; + int r = ceph_rename(m_remote_mount, from.c_str(), to.c_str()); + if (r < 0) { + derr << ": failed to rename remote snap dir_root=" << dir_root + << ", snapshot from =" << from << ", to=" << to << ": " + << cpp_strerror(r) << dendl; + return r; + } + inc_renamed_snap(dir_root); + } + + return 0; +} + +int PeerReplayer::remote_mkdir(const std::string &epath, const struct ceph_statx &stx, + const FHandles &fh) { + dout(10) << ": remote epath=" << epath << dendl; + + int r = ceph_mkdirat(m_remote_mount, fh.r_fd_dir_root, epath.c_str(), stx.stx_mode & ~S_IFDIR); + if (r < 0 && r != -EEXIST) { + derr << ": failed to create remote directory=" << epath << ": " << cpp_strerror(r) + << dendl; + return r; + } + + r = ceph_chownat(m_remote_mount, fh.r_fd_dir_root, epath.c_str(), stx.stx_uid, stx.stx_gid, + AT_SYMLINK_NOFOLLOW); + if (r < 0) { + derr << ": failed to chown remote directory=" << epath << ": " << cpp_strerror(r) + << dendl; + return r; + } + + r = ceph_chmodat(m_remote_mount, fh.r_fd_dir_root, epath.c_str(), stx.stx_mode & ~S_IFMT, + AT_SYMLINK_NOFOLLOW); + if (r < 0) { + derr << ": failed to chmod remote directory=" << epath << ": " << cpp_strerror(r) + << dendl; + return r; + } + + struct timespec times[] = {{stx.stx_atime.tv_sec, stx.stx_atime.tv_nsec}, + {stx.stx_mtime.tv_sec, stx.stx_mtime.tv_nsec}}; + r = ceph_utimensat(m_remote_mount, fh.r_fd_dir_root, epath.c_str(), times, AT_SYMLINK_NOFOLLOW); + if (r < 0) { + derr << ": failed to change [am]time on remote directory=" << epath << ": " + << cpp_strerror(r) << dendl; + return r; + } + + return 0; +} + +#define NR_IOVECS 8 // # iovecs +#define IOVEC_SIZE (8 * 1024 * 1024) // buffer size for each iovec +int PeerReplayer::copy_to_remote(const std::string &dir_root, const std::string &epath, + const struct ceph_statx &stx, const FHandles &fh) { + dout(10) << ": dir_root=" << dir_root << ", epath=" << epath << dendl; + int l_fd; + int r_fd; + void *ptr; + struct iovec iov[NR_IOVECS]; + + int r = ceph_openat(m_local_mount, fh.c_fd, epath.c_str(), O_RDONLY | O_NOFOLLOW, 0); + if (r < 0) { + derr << ": failed to open local file path=" << epath << ": " + << cpp_strerror(r) << dendl; + return r; + } + + l_fd = r; + r = ceph_openat(m_remote_mount, fh.r_fd_dir_root, epath.c_str(), + O_CREAT | O_TRUNC | O_WRONLY | O_NOFOLLOW, stx.stx_mode); + if (r < 0) { + derr << ": failed to create remote file path=" << epath << ": " + << cpp_strerror(r) << dendl; + goto close_local_fd; + } + + r_fd = r; + ptr = malloc(NR_IOVECS * IOVEC_SIZE); + if (!ptr) { + r = -ENOMEM; + derr << ": failed to allocate memory" << dendl; + goto close_remote_fd; + } + + while (true) { + if (should_backoff(dir_root, &r)) { + dout(0) << ": backing off r=" << r << dendl; + break; + } + + for (int i = 0; i < NR_IOVECS; ++i) { + iov[i].iov_base = (char*)ptr + IOVEC_SIZE*i; + iov[i].iov_len = IOVEC_SIZE; + } + + r = ceph_preadv(m_local_mount, l_fd, iov, NR_IOVECS, -1); + if (r < 0) { + derr << ": failed to read local file path=" << epath << ": " + << cpp_strerror(r) << dendl; + break; + } + if (r == 0) { + break; + } + + int iovs = (int)(r / IOVEC_SIZE); + int t = r % IOVEC_SIZE; + if (t) { + iov[iovs].iov_len = t; + ++iovs; + } + + r = ceph_pwritev(m_remote_mount, r_fd, iov, iovs, -1); + if (r < 0) { + derr << ": failed to write remote file path=" << epath << ": " + << cpp_strerror(r) << dendl; + break; + } + } + + if (r == 0) { + r = ceph_fsync(m_remote_mount, r_fd, 0); + if (r < 0) { + derr << ": failed to sync data for file path=" << epath << ": " + << cpp_strerror(r) << dendl; + } + } + + free(ptr); + +close_remote_fd: + if (ceph_close(m_remote_mount, r_fd) < 0) { + derr << ": failed to close remote fd path=" << epath << ": " << cpp_strerror(r) + << dendl; + return -EINVAL; + } + +close_local_fd: + if (ceph_close(m_local_mount, l_fd) < 0) { + derr << ": failed to close local fd path=" << epath << ": " << cpp_strerror(r) + << dendl; + return -EINVAL; + } + + return r == 0 ? 0 : r; +} + +int PeerReplayer::remote_file_op(const std::string &dir_root, const std::string &epath, + const struct ceph_statx &stx, const FHandles &fh, + bool need_data_sync, bool need_attr_sync) { + dout(10) << ": dir_root=" << dir_root << ", epath=" << epath << ", need_data_sync=" << need_data_sync + << ", need_attr_sync=" << need_attr_sync << dendl; + + int r; + if (need_data_sync) { + if (S_ISREG(stx.stx_mode)) { + r = copy_to_remote(dir_root, epath, stx, fh); + if (r < 0) { + derr << ": failed to copy path=" << epath << ": " << cpp_strerror(r) << dendl; + return r; + } + } else if (S_ISLNK(stx.stx_mode)) { + // free the remote link before relinking + r = ceph_unlinkat(m_remote_mount, fh.r_fd_dir_root, epath.c_str(), 0); + if (r < 0 && r != -ENOENT) { + derr << ": failed to remove remote symlink=" << epath << dendl; + return r; + } + char *target = (char *)alloca(stx.stx_size+1); + r = ceph_readlinkat(m_local_mount, fh.c_fd, epath.c_str(), target, stx.stx_size); + if (r < 0) { + derr << ": failed to readlink local path=" << epath << ": " << cpp_strerror(r) + << dendl; + return r; + } + + target[stx.stx_size] = '\0'; + r = ceph_symlinkat(m_remote_mount, target, fh.r_fd_dir_root, epath.c_str()); + if (r < 0 && r != EEXIST) { + derr << ": failed to symlink remote path=" << epath << " to target=" << target + << ": " << cpp_strerror(r) << dendl; + return r; + } + } else { + dout(5) << ": skipping entry=" << epath << ": unsupported mode=" << stx.stx_mode + << dendl; + return 0; + } + } + + if (need_attr_sync) { + r = ceph_chownat(m_remote_mount, fh.r_fd_dir_root, epath.c_str(), stx.stx_uid, stx.stx_gid, + AT_SYMLINK_NOFOLLOW); + if (r < 0) { + derr << ": failed to chown remote directory=" << epath << ": " << cpp_strerror(r) + << dendl; + return r; + } + + r = ceph_chmodat(m_remote_mount, fh.r_fd_dir_root, epath.c_str(), stx.stx_mode & ~S_IFMT, + AT_SYMLINK_NOFOLLOW); + if (r < 0) { + derr << ": failed to chmod remote directory=" << epath << ": " << cpp_strerror(r) + << dendl; + return r; + } + + struct timespec times[] = {{stx.stx_atime.tv_sec, stx.stx_atime.tv_nsec}, + {stx.stx_mtime.tv_sec, stx.stx_mtime.tv_nsec}}; + r = ceph_utimensat(m_remote_mount, fh.r_fd_dir_root, epath.c_str(), times, AT_SYMLINK_NOFOLLOW); + if (r < 0) { + derr << ": failed to change [am]time on remote directory=" << epath << ": " + << cpp_strerror(r) << dendl; + return r; + } + } + + return 0; +} + +int PeerReplayer::cleanup_remote_dir(const std::string &dir_root, + const std::string &epath, const FHandles &fh) { + dout(20) << ": dir_root=" << dir_root << ", epath=" << epath + << dendl; + + struct ceph_statx tstx; + int r = ceph_statxat(m_remote_mount, fh.r_fd_dir_root, epath.c_str(), &tstx, + CEPH_STATX_MODE | CEPH_STATX_UID | CEPH_STATX_GID | + CEPH_STATX_SIZE | CEPH_STATX_ATIME | CEPH_STATX_MTIME, + AT_STATX_DONT_SYNC | AT_SYMLINK_NOFOLLOW); + if (r < 0) { + derr << ": failed to stat remote directory=" << epath << ": " + << cpp_strerror(r) << dendl; + return r; + } + + ceph_dir_result *tdirp; + r = opendirat(m_remote_mount, fh.r_fd_dir_root, epath, AT_SYMLINK_NOFOLLOW, + &tdirp); + if (r < 0) { + derr << ": failed to open remote directory=" << epath << ": " + << cpp_strerror(r) << dendl; + return r; + } + + std::stack<SyncEntry> rm_stack; + rm_stack.emplace(SyncEntry(epath, tdirp, tstx)); + while (!rm_stack.empty()) { + if (should_backoff(dir_root, &r)) { + dout(0) << ": backing off r=" << r << dendl; + break; + } + + dout(20) << ": " << rm_stack.size() << " entries in stack" << dendl; + std::string e_name; + auto &entry = rm_stack.top(); + dout(20) << ": top of stack path=" << entry.epath << dendl; + if (entry.is_directory()) { + struct ceph_statx stx; + struct dirent de; + while (true) { + r = ceph_readdirplus_r(m_remote_mount, entry.dirp, &de, &stx, + CEPH_STATX_MODE, AT_STATX_DONT_SYNC | AT_SYMLINK_NOFOLLOW, NULL); + if (r < 0) { + derr << ": failed to read remote directory=" << entry.epath << dendl; + break; + } + if (r == 0) { + break; + } + + auto d_name = std::string(de.d_name); + if (d_name != "." && d_name != "..") { + e_name = d_name; + break; + } + } + + if (r == 0) { + r = ceph_unlinkat(m_remote_mount, fh.r_fd_dir_root, entry.epath.c_str(), AT_REMOVEDIR); + if (r < 0) { + derr << ": failed to remove remote directory=" << entry.epath << ": " + << cpp_strerror(r) << dendl; + break; + } + + dout(10) << ": done for remote directory=" << entry.epath << dendl; + if (ceph_closedir(m_remote_mount, entry.dirp) < 0) { + derr << ": failed to close remote directory=" << entry.epath << dendl; + } + rm_stack.pop(); + continue; + } + if (r < 0) { + break; + } + + auto epath = entry_path(entry.epath, e_name); + if (S_ISDIR(stx.stx_mode)) { + ceph_dir_result *dirp; + r = opendirat(m_remote_mount, fh.r_fd_dir_root, epath, AT_SYMLINK_NOFOLLOW, + &dirp); + if (r < 0) { + derr << ": failed to open remote directory=" << epath << ": " + << cpp_strerror(r) << dendl; + break; + } + rm_stack.emplace(SyncEntry(epath, dirp, stx)); + } else { + rm_stack.emplace(SyncEntry(epath, stx)); + } + } else { + r = ceph_unlinkat(m_remote_mount, fh.r_fd_dir_root, entry.epath.c_str(), 0); + if (r < 0) { + derr << ": failed to remove remote directory=" << entry.epath << ": " + << cpp_strerror(r) << dendl; + break; + } + dout(10) << ": done for remote file=" << entry.epath << dendl; + rm_stack.pop(); + } + } + + while (!rm_stack.empty()) { + auto &entry = rm_stack.top(); + if (entry.is_directory()) { + dout(20) << ": closing remote directory=" << entry.epath << dendl; + if (ceph_closedir(m_remote_mount, entry.dirp) < 0) { + derr << ": failed to close remote directory=" << entry.epath << dendl; + } + } + + rm_stack.pop(); + } + + return r; +} + +int PeerReplayer::should_sync_entry(const std::string &epath, const struct ceph_statx &cstx, + const FHandles &fh, bool *need_data_sync, bool *need_attr_sync) { + dout(10) << ": epath=" << epath << dendl; + + *need_data_sync = false; + *need_attr_sync = false; + struct ceph_statx pstx; + int r = ceph_statxat(fh.p_mnt, fh.p_fd, epath.c_str(), &pstx, + CEPH_STATX_MODE | CEPH_STATX_UID | CEPH_STATX_GID | + CEPH_STATX_SIZE | CEPH_STATX_CTIME | CEPH_STATX_MTIME, + AT_STATX_DONT_SYNC | AT_SYMLINK_NOFOLLOW); + if (r < 0 && r != -ENOENT && r != -ENOTDIR) { + derr << ": failed to stat prev entry= " << epath << ": " << cpp_strerror(r) + << dendl; + return r; + } + + if (r < 0) { + // inode does not exist in prev snapshot or file type has changed + // (file was S_IFREG earlier, S_IFDIR now). + dout(5) << ": entry=" << epath << ", r=" << r << dendl; + *need_data_sync = true; + *need_attr_sync = true; + return 0; + } + + dout(10) << ": local cur statx: mode=" << cstx.stx_mode << ", uid=" << cstx.stx_uid + << ", gid=" << cstx.stx_gid << ", size=" << cstx.stx_size << ", ctime=" + << cstx.stx_ctime << ", mtime=" << cstx.stx_mtime << dendl; + dout(10) << ": local prev statx: mode=" << pstx.stx_mode << ", uid=" << pstx.stx_uid + << ", gid=" << pstx.stx_gid << ", size=" << pstx.stx_size << ", ctime=" + << pstx.stx_ctime << ", mtime=" << pstx.stx_mtime << dendl; + if ((cstx.stx_mode & S_IFMT) != (pstx.stx_mode & S_IFMT)) { + dout(5) << ": entry=" << epath << " has mode mismatch" << dendl; + *need_data_sync = true; + *need_attr_sync = true; + } else { + *need_data_sync = (cstx.stx_size != pstx.stx_size) || (cstx.stx_mtime != pstx.stx_mtime); + *need_attr_sync = (cstx.stx_ctime != pstx.stx_ctime); + } + + return 0; +} + +int PeerReplayer::propagate_deleted_entries(const std::string &dir_root, + const std::string &epath, const FHandles &fh) { + dout(10) << ": dir_root=" << dir_root << ", epath=" << epath << dendl; + + ceph_dir_result *dirp; + int r = opendirat(fh.p_mnt, fh.p_fd, epath, AT_SYMLINK_NOFOLLOW, &dirp); + if (r < 0) { + if (r == -ELOOP) { + dout(5) << ": epath=" << epath << " is a symbolic link -- mode sync" + << " done when traversing parent" << dendl; + return 0; + } + if (r == -ENOTDIR) { + dout(5) << ": epath=" << epath << " is not a directory -- mode sync" + << " done when traversing parent" << dendl; + return 0; + } + if (r == -ENOENT) { + dout(5) << ": epath=" << epath << " missing in previous-snap/remote dir-root" + << dendl; + } + return r; + } + + struct dirent *dire = (struct dirent *)alloca(512 * sizeof(struct dirent)); + while (true) { + if (should_backoff(dir_root, &r)) { + dout(0) << ": backing off r=" << r << dendl; + break; + } + + int len = ceph_getdents(fh.p_mnt, dirp, (char *)dire, 512); + if (len < 0) { + derr << ": failed to read directory entries: " << cpp_strerror(len) << dendl; + r = len; + // flip errno to signal that we got an err (possible the + // snapshot getting deleted in midst). + if (r == -ENOENT) { + r = -EINVAL; + } + break; + } + if (len == 0) { + dout(10) << ": reached EOD" << dendl; + break; + } + int nr = len / sizeof(struct dirent); + for (int i = 0; i < nr; ++i) { + if (should_backoff(dir_root, &r)) { + dout(0) << ": backing off r=" << r << dendl; + break; + } + std::string d_name = std::string(dire[i].d_name); + if (d_name == "." || d_name == "..") { + continue; + } + + struct ceph_statx pstx; + auto dpath = entry_path(epath, d_name); + r = ceph_statxat(fh.p_mnt, fh.p_fd, dpath.c_str(), &pstx, + CEPH_STATX_MODE, AT_STATX_DONT_SYNC | AT_SYMLINK_NOFOLLOW); + if (r < 0) { + derr << ": failed to stat (prev) directory=" << dpath << ": " + << cpp_strerror(r) << dendl; + // flip errno to signal that we got an err (possible the + // snapshot getting deleted in midst). + if (r == -ENOENT) { + r = -EINVAL; + } + return r; + } + + struct ceph_statx cstx; + r = ceph_statxat(m_local_mount, fh.c_fd, dpath.c_str(), &cstx, + CEPH_STATX_MODE, AT_STATX_DONT_SYNC | AT_SYMLINK_NOFOLLOW); + if (r < 0 && r != -ENOENT) { + derr << ": failed to stat local (cur) directory=" << dpath << ": " + << cpp_strerror(r) << dendl; + return r; + } + + bool purge_remote = true; + if (r == 0) { + // directory entry present in both snapshots -- check inode + // type + if ((pstx.stx_mode & S_IFMT) == (cstx.stx_mode & S_IFMT)) { + dout(5) << ": mode matches for entry=" << d_name << dendl; + purge_remote = false; + } else { + dout(5) << ": mode mismatch for entry=" << d_name << dendl; + } + } else { + dout(5) << ": entry=" << d_name << " missing in current snapshot" << dendl; + } + + if (purge_remote) { + dout(5) << ": purging remote entry=" << dpath << dendl; + if (S_ISDIR(pstx.stx_mode)) { + r = cleanup_remote_dir(dir_root, dpath, fh); + } else { + r = ceph_unlinkat(m_remote_mount, fh.r_fd_dir_root, dpath.c_str(), 0); + } + + if (r < 0 && r != -ENOENT) { + derr << ": failed to cleanup remote entry=" << d_name << ": " + << cpp_strerror(r) << dendl; + return r; + } + } + } + } + + ceph_closedir(fh.p_mnt, dirp); + return r; +} + +int PeerReplayer::open_dir(MountRef mnt, const std::string &dir_path, + boost::optional<uint64_t> snap_id) { + dout(20) << ": dir_path=" << dir_path << dendl; + if (snap_id) { + dout(20) << ": expected snapshot id=" << *snap_id << dendl; + } + + int fd = ceph_open(mnt, dir_path.c_str(), O_DIRECTORY | O_RDONLY, 0); + if (fd < 0) { + derr << ": cannot open dir_path=" << dir_path << ": " << cpp_strerror(fd) + << dendl; + return fd; + } + + if (!snap_id) { + return fd; + } + + snap_info info; + int r = ceph_get_snap_info(mnt, dir_path.c_str(), &info); + if (r < 0) { + derr << ": failed to fetch snap_info for path=" << dir_path + << ": " << cpp_strerror(r) << dendl; + ceph_close(mnt, fd); + return r; + } + + if (info.id != *snap_id) { + dout(5) << ": got mismatching snapshot id for path=" << dir_path << " (" << info.id + << " vs " << *snap_id << ") -- possible recreate" << dendl; + ceph_close(mnt, fd); + return -EINVAL; + } + + return fd; +} + +int PeerReplayer::pre_sync_check_and_open_handles( + const std::string &dir_root, + const Snapshot ¤t, boost::optional<Snapshot> prev, + FHandles *fh) { + dout(20) << ": dir_root=" << dir_root << ", current=" << current << dendl; + if (prev) { + dout(20) << ": prev=" << prev << dendl; + } + + auto cur_snap_path = snapshot_path(m_cct, dir_root, current.first); + auto fd = open_dir(m_local_mount, cur_snap_path, current.second); + if (fd < 0) { + return fd; + } + + // current snapshot file descriptor + fh->c_fd = fd; + + MountRef mnt; + if (prev) { + mnt = m_local_mount; + auto prev_snap_path = snapshot_path(m_cct, dir_root, (*prev).first); + fd = open_dir(mnt, prev_snap_path, (*prev).second); + } else { + mnt = m_remote_mount; + fd = open_dir(mnt, dir_root, boost::none); + } + + if (fd < 0) { + if (!prev || fd != -ENOENT) { + ceph_close(m_local_mount, fh->c_fd); + return fd; + } + + // ENOENT of previous snap + dout(5) << ": previous snapshot=" << *prev << " missing" << dendl; + mnt = m_remote_mount; + fd = open_dir(mnt, dir_root, boost::none); + if (fd < 0) { + ceph_close(m_local_mount, fh->c_fd); + return fd; + } + } + + // "previous" snapshot or dir_root file descriptor + fh->p_fd = fd; + fh->p_mnt = mnt; + + { + std::scoped_lock locker(m_lock); + auto it = m_registered.find(dir_root); + ceph_assert(it != m_registered.end()); + fh->r_fd_dir_root = it->second.fd; + } + + dout(5) << ": using " << ((fh->p_mnt == m_local_mount) ? "local (previous) snapshot" : "remote dir_root") + << " for incremental transfer" << dendl; + return 0; +} + +void PeerReplayer::post_sync_close_handles(const FHandles &fh) { + dout(20) << dendl; + + // @FHandles.r_fd_dir_root is closed in @unregister_directory since + // its used to acquire an exclusive lock on remote dir_root. + ceph_close(m_local_mount, fh.c_fd); + ceph_close(fh.p_mnt, fh.p_fd); +} + +int PeerReplayer::do_synchronize(const std::string &dir_root, const Snapshot ¤t, + boost::optional<Snapshot> prev) { + dout(20) << ": dir_root=" << dir_root << ", current=" << current << dendl; + if (prev) { + dout(20) << ": incremental sync check from prev=" << prev << dendl; + } + + FHandles fh; + int r = pre_sync_check_and_open_handles(dir_root, current, prev, &fh); + if (r < 0) { + dout(5) << ": cannot proceeed with sync: " << cpp_strerror(r) << dendl; + return r; + } + + BOOST_SCOPE_EXIT_ALL( (this)(&fh) ) { + post_sync_close_handles(fh); + }; + + // record that we are going to "dirty" the data under this + // directory root + auto snap_id_str{stringify(current.second)}; + r = ceph_fsetxattr(m_remote_mount, fh.r_fd_dir_root, "ceph.mirror.dirty_snap_id", + snap_id_str.c_str(), snap_id_str.size(), 0); + if (r < 0) { + derr << ": error setting \"ceph.mirror.dirty_snap_id\" on dir_root=" << dir_root + << ": " << cpp_strerror(r) << dendl; + return r; + } + + struct ceph_statx tstx; + r = ceph_fstatx(m_local_mount, fh.c_fd, &tstx, + CEPH_STATX_MODE | CEPH_STATX_UID | CEPH_STATX_GID | + CEPH_STATX_SIZE | CEPH_STATX_ATIME | CEPH_STATX_MTIME, + AT_STATX_DONT_SYNC | AT_SYMLINK_NOFOLLOW); + if (r < 0) { + derr << ": failed to stat snap=" << current.first << ": " << cpp_strerror(r) + << dendl; + return r; + } + + ceph_dir_result *tdirp; + r = ceph_fdopendir(m_local_mount, fh.c_fd, &tdirp); + if (r < 0) { + derr << ": failed to open local snap=" << current.first << ": " << cpp_strerror(r) + << dendl; + return r; + } + + std::stack<SyncEntry> sync_stack; + sync_stack.emplace(SyncEntry(".", tdirp, tstx)); + while (!sync_stack.empty()) { + if (should_backoff(dir_root, &r)) { + dout(0) << ": backing off r=" << r << dendl; + break; + } + + dout(20) << ": " << sync_stack.size() << " entries in stack" << dendl; + std::string e_name; + auto &entry = sync_stack.top(); + dout(20) << ": top of stack path=" << entry.epath << dendl; + if (entry.is_directory()) { + // entry is a directory -- propagate deletes for missing entries + // (and changed inode types) to the remote filesystem. + if (!entry.needs_remote_sync()) { + r = propagate_deleted_entries(dir_root, entry.epath, fh); + if (r < 0 && r != -ENOENT) { + derr << ": failed to propagate missing dirs: " << cpp_strerror(r) << dendl; + break; + } + entry.set_remote_synced(); + } + + struct ceph_statx stx; + struct dirent de; + while (true) { + r = ceph_readdirplus_r(m_local_mount, entry.dirp, &de, &stx, + CEPH_STATX_MODE | CEPH_STATX_UID | CEPH_STATX_GID | + CEPH_STATX_SIZE | CEPH_STATX_ATIME | CEPH_STATX_MTIME, + AT_STATX_DONT_SYNC | AT_SYMLINK_NOFOLLOW, NULL); + if (r < 0) { + derr << ": failed to local read directory=" << entry.epath << dendl; + break; + } + if (r == 0) { + break; + } + + auto d_name = std::string(de.d_name); + if (d_name != "." && d_name != "..") { + e_name = d_name; + break; + } + } + + if (r == 0) { + dout(10) << ": done for directory=" << entry.epath << dendl; + if (ceph_closedir(m_local_mount, entry.dirp) < 0) { + derr << ": failed to close local directory=" << entry.epath << dendl; + } + sync_stack.pop(); + continue; + } + if (r < 0) { + break; + } + + auto epath = entry_path(entry.epath, e_name); + if (S_ISDIR(stx.stx_mode)) { + r = remote_mkdir(epath, stx, fh); + if (r < 0) { + break; + } + ceph_dir_result *dirp; + r = opendirat(m_local_mount, fh.c_fd, epath, AT_SYMLINK_NOFOLLOW, &dirp); + if (r < 0) { + derr << ": failed to open local directory=" << epath << ": " + << cpp_strerror(r) << dendl; + break; + } + sync_stack.emplace(SyncEntry(epath, dirp, stx)); + } else { + sync_stack.emplace(SyncEntry(epath, stx)); + } + } else { + bool need_data_sync = true; + bool need_attr_sync = true; + r = should_sync_entry(entry.epath, entry.stx, fh, + &need_data_sync, &need_attr_sync); + if (r < 0) { + break; + } + + dout(5) << ": entry=" << entry.epath << ", data_sync=" << need_data_sync + << ", attr_sync=" << need_attr_sync << dendl; + if (need_data_sync || need_attr_sync) { + r = remote_file_op(dir_root, entry.epath, entry.stx, fh, need_data_sync, + need_attr_sync); + if (r < 0) { + break; + } + } + dout(10) << ": done for epath=" << entry.epath << dendl; + sync_stack.pop(); + } + } + + while (!sync_stack.empty()) { + auto &entry = sync_stack.top(); + if (entry.is_directory()) { + dout(20) << ": closing local directory=" << entry.epath << dendl; + if (ceph_closedir(m_local_mount, entry.dirp) < 0) { + derr << ": failed to close local directory=" << entry.epath << dendl; + } + } + + sync_stack.pop(); + } + + return r; +} + +int PeerReplayer::synchronize(const std::string &dir_root, const Snapshot ¤t, + boost::optional<Snapshot> prev) { + dout(20) << ": dir_root=" << dir_root << ", current=" << current << dendl; + if (prev) { + dout(20) << ": prev=" << prev << dendl; + } + + int r = ceph_getxattr(m_remote_mount, dir_root.c_str(), "ceph.mirror.dirty_snap_id", nullptr, 0); + if (r < 0 && r != -ENODATA) { + derr << ": failed to fetch primary_snap_id length from dir_root=" << dir_root + << ": " << cpp_strerror(r) << dendl; + return r; + } + + // no xattr, can't determine which snap the data belongs to! + if (r < 0) { + dout(5) << ": missing \"ceph.mirror.dirty_snap_id\" xattr on remote -- using" + << " incremental sync with remote scan" << dendl; + r = do_synchronize(dir_root, current, boost::none); + } else { + size_t xlen = r; + char *val = (char *)alloca(xlen+1); + r = ceph_getxattr(m_remote_mount, dir_root.c_str(), "ceph.mirror.dirty_snap_id", (void*)val, xlen); + if (r < 0) { + derr << ": failed to fetch \"dirty_snap_id\" for dir_root: " << dir_root + << ": " << cpp_strerror(r) << dendl; + return r; + } + + val[xlen] = '\0'; + uint64_t dirty_snap_id = atoll(val); + + dout(20) << ": dirty_snap_id: " << dirty_snap_id << " vs (" << current.second + << "," << (prev ? stringify((*prev).second) : "~") << ")" << dendl; + if (prev && (dirty_snap_id == (*prev).second || dirty_snap_id == current.second)) { + dout(5) << ": match -- using incremental sync with local scan" << dendl; + r = do_synchronize(dir_root, current, prev); + } else { + dout(5) << ": mismatch -- using incremental sync with remote scan" << dendl; + r = do_synchronize(dir_root, current, boost::none); + } + } + + // snap sync failed -- bail out! + if (r < 0) { + return r; + } + + auto cur_snap_id_str{stringify(current.second)}; + snap_metadata snap_meta[] = {{PRIMARY_SNAP_ID_KEY.c_str(), cur_snap_id_str.c_str()}}; + r = ceph_mksnap(m_remote_mount, dir_root.c_str(), current.first.c_str(), 0755, + snap_meta, sizeof(snap_meta)/sizeof(snap_metadata)); + if (r < 0) { + derr << ": failed to snap remote directory dir_root=" << dir_root + << ": " << cpp_strerror(r) << dendl; + } + + return r; +} + +int PeerReplayer::do_sync_snaps(const std::string &dir_root) { + dout(20) << ": dir_root=" << dir_root << dendl; + + std::map<uint64_t, std::string> local_snap_map; + std::map<uint64_t, std::string> remote_snap_map; + + int r = build_snap_map(dir_root, &local_snap_map); + if (r < 0) { + derr << ": failed to build local snap map" << dendl; + return r; + } + + r = build_snap_map(dir_root, &remote_snap_map, true); + if (r < 0) { + derr << ": failed to build remote snap map" << dendl; + return r; + } + + // infer deleted and renamed snapshots from local and remote + // snap maps + std::set<std::string> snaps_deleted; + std::set<std::pair<std::string,std::string>> snaps_renamed; + for (auto &[primary_snap_id, snap_name] : remote_snap_map) { + auto it = local_snap_map.find(primary_snap_id); + if (it == local_snap_map.end()) { + snaps_deleted.emplace(snap_name); + } else if (it->second != snap_name) { + snaps_renamed.emplace(std::make_pair(snap_name, it->second)); + } + } + + r = propagate_snap_deletes(dir_root, snaps_deleted); + if (r < 0) { + derr << ": failed to propgate deleted snapshots" << dendl; + return r; + } + + r = propagate_snap_renames(dir_root, snaps_renamed); + if (r < 0) { + derr << ": failed to propgate renamed snapshots" << dendl; + return r; + } + + // start mirroring snapshots from the last snap-id synchronized + uint64_t last_snap_id = 0; + std::string last_snap_name; + if (!remote_snap_map.empty()) { + auto last = remote_snap_map.rbegin(); + last_snap_id = last->first; + last_snap_name = last->second; + set_last_synced_snap(dir_root, last_snap_id, last_snap_name); + } + + dout(5) << ": last snap-id transferred=" << last_snap_id << dendl; + auto it = local_snap_map.upper_bound(last_snap_id); + if (it == local_snap_map.end()) { + dout(20) << ": nothing to synchronize" << dendl; + return 0; + } + + auto snaps_per_cycle = g_ceph_context->_conf.get_val<uint64_t>( + "cephfs_mirror_max_snapshot_sync_per_cycle"); + + dout(10) << ": synchronizing from snap-id=" << it->first << dendl; + for (; it != local_snap_map.end(); ++it) { + set_current_syncing_snap(dir_root, it->first, it->second); + auto start = clock::now(); + boost::optional<Snapshot> prev = boost::none; + if (last_snap_id != 0) { + prev = std::make_pair(last_snap_name, last_snap_id); + } + r = synchronize(dir_root, std::make_pair(it->second, it->first), prev); + if (r < 0) { + derr << ": failed to synchronize dir_root=" << dir_root + << ", snapshot=" << it->second << dendl; + clear_current_syncing_snap(dir_root); + return r; + } + std::chrono::duration<double> duration = clock::now() - start; + set_last_synced_stat(dir_root, it->first, it->second, duration.count()); + if (--snaps_per_cycle == 0) { + break; + } + + last_snap_name = it->second; + last_snap_id = it->first; + } + + return 0; +} + +void PeerReplayer::sync_snaps(const std::string &dir_root, + std::unique_lock<ceph::mutex> &locker) { + dout(20) << ": dir_root=" << dir_root << dendl; + locker.unlock(); + int r = do_sync_snaps(dir_root); + if (r < 0) { + derr << ": failed to sync snapshots for dir_root=" << dir_root << dendl; + } + locker.lock(); + if (r < 0) { + _inc_failed_count(dir_root); + } else { + _reset_failed_count(dir_root); + } +} + +void PeerReplayer::run(SnapshotReplayerThread *replayer) { + dout(10) << ": snapshot replayer=" << replayer << dendl; + + time last_directory_scan = clock::zero(); + auto scan_interval = g_ceph_context->_conf.get_val<uint64_t>( + "cephfs_mirror_directory_scan_interval"); + + std::unique_lock locker(m_lock); + while (true) { + // do not check if client is blocklisted under lock + m_cond.wait_for(locker, 1s, [this]{return is_stopping();}); + if (is_stopping()) { + dout(5) << ": exiting" << dendl; + break; + } + + locker.unlock(); + + if (m_fs_mirror->is_blocklisted()) { + dout(5) << ": exiting as client is blocklisted" << dendl; + break; + } + + locker.lock(); + + auto now = clock::now(); + std::chrono::duration<double> timo = now - last_directory_scan; + if (timo.count() >= scan_interval && m_directories.size()) { + dout(20) << ": trying to pick from " << m_directories.size() << " directories" << dendl; + auto dir_root = pick_directory(); + if (dir_root) { + dout(5) << ": picked dir_root=" << *dir_root << dendl; + int r = register_directory(*dir_root, replayer); + if (r == 0) { + sync_snaps(*dir_root, locker); + unregister_directory(*dir_root); + } + } + + last_directory_scan = now; + } + } +} + +void PeerReplayer::peer_status(Formatter *f) { + std::scoped_lock locker(m_lock); + f->open_object_section("stats"); + for (auto &[dir_root, sync_stat] : m_snap_sync_stats) { + f->open_object_section(dir_root); + if (sync_stat.failed) { + f->dump_string("state", "failed"); + } else if (!sync_stat.current_syncing_snap) { + f->dump_string("state", "idle"); + } else { + f->dump_string("state", "syncing"); + f->open_object_section("current_sycning_snap"); + f->dump_unsigned("id", (*sync_stat.current_syncing_snap).first); + f->dump_string("name", (*sync_stat.current_syncing_snap).second); + f->close_section(); + } + if (sync_stat.last_synced_snap) { + f->open_object_section("last_synced_snap"); + f->dump_unsigned("id", (*sync_stat.last_synced_snap).first); + f->dump_string("name", (*sync_stat.last_synced_snap).second); + if (sync_stat.last_sync_duration) { + f->dump_float("sync_duration", *sync_stat.last_sync_duration); + f->dump_stream("sync_time_stamp") << sync_stat.last_synced; + } + f->close_section(); + } + f->dump_unsigned("snaps_synced", sync_stat.synced_snap_count); + f->dump_unsigned("snaps_deleted", sync_stat.deleted_snap_count); + f->dump_unsigned("snaps_renamed", sync_stat.renamed_snap_count); + f->close_section(); // dir_root + } + f->close_section(); // stats +} + +void PeerReplayer::reopen_logs() { + std::scoped_lock locker(m_lock); + + if (m_remote_cluster) { + reinterpret_cast<CephContext *>(m_remote_cluster->cct())->reopen_logs(); + } +} + +} // namespace mirror +} // namespace cephfs diff --git a/src/tools/cephfs_mirror/PeerReplayer.h b/src/tools/cephfs_mirror/PeerReplayer.h new file mode 100644 index 000000000..886c95329 --- /dev/null +++ b/src/tools/cephfs_mirror/PeerReplayer.h @@ -0,0 +1,319 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPHFS_MIRROR_PEER_REPLAYER_H +#define CEPHFS_MIRROR_PEER_REPLAYER_H + +#include "common/Formatter.h" +#include "common/Thread.h" +#include "mds/FSMap.h" +#include "ServiceDaemon.h" +#include "Types.h" + +namespace cephfs { +namespace mirror { + +class FSMirror; +class PeerReplayerAdminSocketHook; + +class PeerReplayer { +public: + PeerReplayer(CephContext *cct, FSMirror *fs_mirror, + RadosRef local_cluster, const Filesystem &filesystem, + const Peer &peer, const std::set<std::string, std::less<>> &directories, + MountRef mount, ServiceDaemon *service_daemon); + ~PeerReplayer(); + + // initialize replayer for a peer + int init(); + + // shutdown replayer for a peer + void shutdown(); + + // add a directory to mirror queue + void add_directory(string_view dir_root); + + // remove a directory from queue + void remove_directory(string_view dir_root); + + // admin socket helpers + void peer_status(Formatter *f); + + // reopen logs + void reopen_logs(); + +private: + inline static const std::string PRIMARY_SNAP_ID_KEY = "primary_snap_id"; + + inline static const std::string SERVICE_DAEMON_FAILED_DIR_COUNT_KEY = "failure_count"; + inline static const std::string SERVICE_DAEMON_RECOVERED_DIR_COUNT_KEY = "recovery_count"; + + using Snapshot = std::pair<std::string, uint64_t>; + + // file descriptor "triplet" for synchronizing a snapshot + // w/ an added MountRef for accessing "previous" snapshot. + struct FHandles { + // open file descriptor on the snap directory for snapshot + // currently being synchronized. Always use this fd with + // @m_local_mount. + int c_fd; + + // open file descriptor on the "previous" snapshot or on + // dir_root on remote filesystem (based on if the snapshot + // can be used for incremental transfer). Always use this + // fd with p_mnt which either points to @m_local_mount ( + // for local incremental comparison) or @m_remote_mount ( + // for remote incremental comparison). + int p_fd; + MountRef p_mnt; + + // open file descriptor on dir_root on remote filesystem. + // Always use this fd with @m_remote_mount. + int r_fd_dir_root; + }; + + bool is_stopping() { + return m_stopping; + } + + struct Replayer; + class SnapshotReplayerThread : public Thread { + public: + SnapshotReplayerThread(PeerReplayer *peer_replayer) + : m_peer_replayer(peer_replayer) { + } + + void *entry() override { + m_peer_replayer->run(this); + return 0; + } + + private: + PeerReplayer *m_peer_replayer; + }; + + struct DirRegistry { + int fd; + bool canceled = false; + SnapshotReplayerThread *replayer; + }; + + struct SyncEntry { + std::string epath; + ceph_dir_result *dirp; // valid for directories + struct ceph_statx stx; + // set by incremental sync _after_ ensuring missing entries + // in the currently synced snapshot have been propagated to + // the remote filesystem. + bool remote_synced = false; + + SyncEntry(std::string_view path, + const struct ceph_statx &stx) + : epath(path), + stx(stx) { + } + SyncEntry(std::string_view path, + ceph_dir_result *dirp, + const struct ceph_statx &stx) + : epath(path), + dirp(dirp), + stx(stx) { + } + + bool is_directory() const { + return S_ISDIR(stx.stx_mode); + } + + bool needs_remote_sync() const { + return remote_synced; + } + void set_remote_synced() { + remote_synced = true; + } + }; + + using clock = ceph::coarse_mono_clock; + using time = ceph::coarse_mono_time; + + // stats sent to service daemon + struct ServiceDaemonStats { + uint64_t failed_dir_count = 0; + uint64_t recovered_dir_count = 0; + }; + + struct SnapSyncStat { + uint64_t nr_failures = 0; // number of consecutive failures + boost::optional<time> last_failed; // lat failed timestamp + bool failed = false; // hit upper cap for consecutive failures + boost::optional<std::pair<uint64_t, std::string>> last_synced_snap; + boost::optional<std::pair<uint64_t, std::string>> current_syncing_snap; + uint64_t synced_snap_count = 0; + uint64_t deleted_snap_count = 0; + uint64_t renamed_snap_count = 0; + time last_synced = clock::zero(); + boost::optional<double> last_sync_duration; + }; + + void _inc_failed_count(const std::string &dir_root) { + auto max_failures = g_ceph_context->_conf.get_val<uint64_t>( + "cephfs_mirror_max_consecutive_failures_per_directory"); + auto &sync_stat = m_snap_sync_stats.at(dir_root); + sync_stat.last_failed = clock::now(); + if (++sync_stat.nr_failures >= max_failures && !sync_stat.failed) { + sync_stat.failed = true; + ++m_service_daemon_stats.failed_dir_count; + m_service_daemon->add_or_update_peer_attribute(m_filesystem.fscid, m_peer, + SERVICE_DAEMON_FAILED_DIR_COUNT_KEY, + m_service_daemon_stats.failed_dir_count); + } + } + void _reset_failed_count(const std::string &dir_root) { + auto &sync_stat = m_snap_sync_stats.at(dir_root); + if (sync_stat.failed) { + ++m_service_daemon_stats.recovered_dir_count; + m_service_daemon->add_or_update_peer_attribute(m_filesystem.fscid, m_peer, + SERVICE_DAEMON_RECOVERED_DIR_COUNT_KEY, + m_service_daemon_stats.recovered_dir_count); + } + sync_stat.nr_failures = 0; + sync_stat.failed = false; + sync_stat.last_failed = boost::none; + } + + void _set_last_synced_snap(const std::string &dir_root, uint64_t snap_id, + const std::string &snap_name) { + auto &sync_stat = m_snap_sync_stats.at(dir_root); + sync_stat.last_synced_snap = std::make_pair(snap_id, snap_name); + sync_stat.current_syncing_snap = boost::none; + } + void set_last_synced_snap(const std::string &dir_root, uint64_t snap_id, + const std::string &snap_name) { + std::scoped_lock locker(m_lock); + _set_last_synced_snap(dir_root, snap_id, snap_name); + } + void set_current_syncing_snap(const std::string &dir_root, uint64_t snap_id, + const std::string &snap_name) { + std::scoped_lock locker(m_lock); + auto &sync_stat = m_snap_sync_stats.at(dir_root); + sync_stat.current_syncing_snap = std::make_pair(snap_id, snap_name); + } + void clear_current_syncing_snap(const std::string &dir_root) { + std::scoped_lock locker(m_lock); + auto &sync_stat = m_snap_sync_stats.at(dir_root); + sync_stat.current_syncing_snap = boost::none; + } + void inc_deleted_snap(const std::string &dir_root) { + std::scoped_lock locker(m_lock); + auto &sync_stat = m_snap_sync_stats.at(dir_root); + ++sync_stat.deleted_snap_count; + } + void inc_renamed_snap(const std::string &dir_root) { + std::scoped_lock locker(m_lock); + auto &sync_stat = m_snap_sync_stats.at(dir_root); + ++sync_stat.renamed_snap_count; + } + void set_last_synced_stat(const std::string &dir_root, uint64_t snap_id, + const std::string &snap_name, double duration) { + std::scoped_lock locker(m_lock); + _set_last_synced_snap(dir_root, snap_id, snap_name); + auto &sync_stat = m_snap_sync_stats.at(dir_root); + sync_stat.last_synced = clock::now(); + sync_stat.last_sync_duration = duration; + ++sync_stat.synced_snap_count; + } + + bool should_backoff(const std::string &dir_root, int *retval) { + if (m_fs_mirror->is_blocklisted()) { + *retval = -EBLOCKLISTED; + return true; + } + + std::scoped_lock locker(m_lock); + if (is_stopping()) { + // ceph defines EBLOCKLISTED to ESHUTDOWN (108). so use + // EINPROGRESS to identify shutdown. + *retval = -EINPROGRESS; + return true; + } + auto &dr = m_registered.at(dir_root); + if (dr.canceled) { + *retval = -ECANCELED; + return true; + } + + *retval = 0; + return false; + } + + typedef std::vector<std::unique_ptr<SnapshotReplayerThread>> SnapshotReplayers; + + CephContext *m_cct; + FSMirror *m_fs_mirror; + RadosRef m_local_cluster; + Filesystem m_filesystem; + Peer m_peer; + // probably need to be encapsulated when supporting cancelations + std::map<std::string, DirRegistry> m_registered; + std::vector<std::string> m_directories; + std::map<std::string, SnapSyncStat> m_snap_sync_stats; + MountRef m_local_mount; + ServiceDaemon *m_service_daemon; + PeerReplayerAdminSocketHook *m_asok_hook = nullptr; + + ceph::mutex m_lock; + ceph::condition_variable m_cond; + RadosRef m_remote_cluster; + MountRef m_remote_mount; + bool m_stopping = false; + SnapshotReplayers m_replayers; + + ServiceDaemonStats m_service_daemon_stats; + + void run(SnapshotReplayerThread *replayer); + + boost::optional<std::string> pick_directory(); + int register_directory(const std::string &dir_root, SnapshotReplayerThread *replayer); + void unregister_directory(const std::string &dir_root); + int try_lock_directory(const std::string &dir_root, SnapshotReplayerThread *replayer, + DirRegistry *registry); + void unlock_directory(const std::string &dir_root, const DirRegistry ®istry); + void sync_snaps(const std::string &dir_root, std::unique_lock<ceph::mutex> &locker); + + + int build_snap_map(const std::string &dir_root, std::map<uint64_t, std::string> *snap_map, + bool is_remote=false); + + int propagate_snap_deletes(const std::string &dir_root, const std::set<std::string> &snaps); + int propagate_snap_renames(const std::string &dir_root, + const std::set<std::pair<std::string,std::string>> &snaps); + int propagate_deleted_entries(const std::string &dir_root, const std::string &epath, + const FHandles &fh); + int cleanup_remote_dir(const std::string &dir_root, const std::string &epath, + const FHandles &fh); + + int should_sync_entry(const std::string &epath, const struct ceph_statx &cstx, + const FHandles &fh, bool *need_data_sync, bool *need_attr_sync); + + int open_dir(MountRef mnt, const std::string &dir_path, boost::optional<uint64_t> snap_id); + int pre_sync_check_and_open_handles(const std::string &dir_root, const Snapshot ¤t, + boost::optional<Snapshot> prev, FHandles *fh); + void post_sync_close_handles(const FHandles &fh); + + int do_synchronize(const std::string &dir_root, const Snapshot ¤t, + boost::optional<Snapshot> prev); + + int synchronize(const std::string &dir_root, const Snapshot ¤t, + boost::optional<Snapshot> prev); + int do_sync_snaps(const std::string &dir_root); + + int remote_mkdir(const std::string &epath, const struct ceph_statx &stx, const FHandles &fh); + int remote_file_op(const std::string &dir_root, const std::string &epath, const struct ceph_statx &stx, + const FHandles &fh, bool need_data_sync, bool need_attr_sync); + int copy_to_remote(const std::string &dir_root, const std::string &epath, const struct ceph_statx &stx, + const FHandles &fh); +}; + +} // namespace mirror +} // namespace cephfs + +#endif // CEPHFS_MIRROR_PEER_REPLAYER_H diff --git a/src/tools/cephfs_mirror/ServiceDaemon.cc b/src/tools/cephfs_mirror/ServiceDaemon.cc new file mode 100644 index 000000000..f66dd46bf --- /dev/null +++ b/src/tools/cephfs_mirror/ServiceDaemon.cc @@ -0,0 +1,225 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "common/debug.h" +#include "common/errno.h" +#include "common/Timer.h" +#include "include/stringify.h" +#include "ServiceDaemon.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_cephfs_mirror +#undef dout_prefix +#define dout_prefix *_dout << "cephfs::mirror::ServiceDaemon: " << this << " " \ + << __func__ + +namespace cephfs { +namespace mirror { + +namespace { + +struct AttributeDumpVisitor : public boost::static_visitor<void> { + ceph::Formatter *f; + std::string name; + + AttributeDumpVisitor(ceph::Formatter *f, std::string_view name) + : f(f), name(name) { + } + + void operator()(bool val) const { + f->dump_bool(name.c_str(), val); + } + void operator()(uint64_t val) const { + f->dump_unsigned(name.c_str(), val); + } + void operator()(const std::string &val) const { + f->dump_string(name.c_str(), val); + } +}; + +} // anonymous namespace + +ServiceDaemon::ServiceDaemon(CephContext *cct, RadosRef rados) + : m_cct(cct), + m_rados(rados), + m_timer(new SafeTimer(cct, m_timer_lock, true)) { + m_timer->init(); +} + +ServiceDaemon::~ServiceDaemon() { + dout(10) << dendl; + { + std::scoped_lock timer_lock(m_timer_lock); + if (m_timer_ctx != nullptr) { + dout(5) << ": canceling timer task=" << m_timer_ctx << dendl; + m_timer->cancel_event(m_timer_ctx); + } + m_timer->shutdown(); + } + + delete m_timer; +} + +int ServiceDaemon::init() { + dout(20) << dendl; + + std::string id = m_cct->_conf->name.get_id(); + if (id.find(CEPHFS_MIRROR_AUTH_ID_PREFIX) == 0) { + id = id.substr(CEPHFS_MIRROR_AUTH_ID_PREFIX.size()); + } + std::string instance_id = stringify(m_rados->get_instance_id()); + + std::map<std::string, std::string> service_metadata = {{"id", id}, + {"instance_id", instance_id}}; + int r = m_rados->service_daemon_register("cephfs-mirror", instance_id, + service_metadata); + if (r < 0) { + return r; + } + return 0; +} + +void ServiceDaemon::add_filesystem(fs_cluster_id_t fscid, std::string_view fs_name) { + dout(10) << ": fscid=" << fscid << ", fs_name=" << fs_name << dendl; + + { + std::scoped_lock locker(m_lock); + m_filesystems.emplace(fscid, Filesystem(fs_name)); + } + schedule_update_status(); +} + +void ServiceDaemon::remove_filesystem(fs_cluster_id_t fscid) { + dout(10) << ": fscid=" << fscid << dendl; + + { + std::scoped_lock locker(m_lock); + m_filesystems.erase(fscid); + } + schedule_update_status(); +} + +void ServiceDaemon::add_peer(fs_cluster_id_t fscid, const Peer &peer) { + dout(10) << ": peer=" << peer << dendl; + + { + std::scoped_lock locker(m_lock); + auto fs_it = m_filesystems.find(fscid); + if (fs_it == m_filesystems.end()) { + return; + } + fs_it->second.peer_attributes.emplace(peer, Attributes{}); + } + schedule_update_status(); +} + +void ServiceDaemon::remove_peer(fs_cluster_id_t fscid, const Peer &peer) { + dout(10) << ": peer=" << peer << dendl; + + { + std::scoped_lock locker(m_lock); + auto fs_it = m_filesystems.find(fscid); + if (fs_it == m_filesystems.end()) { + return; + } + fs_it->second.peer_attributes.erase(peer); + } + schedule_update_status(); +} + +void ServiceDaemon::add_or_update_fs_attribute(fs_cluster_id_t fscid, std::string_view key, + AttributeValue value) { + dout(10) << ": fscid=" << fscid << dendl; + + { + std::scoped_lock locker(m_lock); + auto fs_it = m_filesystems.find(fscid); + if (fs_it == m_filesystems.end()) { + return; + } + + fs_it->second.fs_attributes[std::string(key)] = value; + } + schedule_update_status(); +} + +void ServiceDaemon::add_or_update_peer_attribute(fs_cluster_id_t fscid, const Peer &peer, + std::string_view key, AttributeValue value) { + dout(10) << ": fscid=" << fscid << dendl; + + { + std::scoped_lock locker(m_lock); + auto fs_it = m_filesystems.find(fscid); + if (fs_it == m_filesystems.end()) { + return; + } + + auto peer_it = fs_it->second.peer_attributes.find(peer); + if (peer_it == fs_it->second.peer_attributes.end()) { + return; + } + + peer_it->second[std::string(key)] = value; + } + schedule_update_status(); +} + +void ServiceDaemon::schedule_update_status() { + dout(10) << dendl; + + std::scoped_lock timer_lock(m_timer_lock); + if (m_timer_ctx != nullptr) { + return; + } + + m_timer_ctx = new LambdaContext([this] { + m_timer_ctx = nullptr; + update_status(); + }); + m_timer->add_event_after(1, m_timer_ctx); +} + +void ServiceDaemon::update_status() { + dout(20) << ": " << m_filesystems.size() << " filesystem(s)" << dendl; + + ceph::JSONFormatter f; + { + std::scoped_lock locker(m_lock); + f.open_object_section("filesystems"); + for (auto &[fscid, filesystem] : m_filesystems) { + f.open_object_section(stringify(fscid).c_str()); + f.dump_string("name", filesystem.fs_name); + for (auto &[attr_name, attr_value] : filesystem.fs_attributes) { + AttributeDumpVisitor visitor(&f, attr_name); + boost::apply_visitor(visitor, attr_value); + } + f.open_object_section("peers"); + for (auto &[peer, attributes] : filesystem.peer_attributes) { + f.open_object_section(peer.uuid); + f.dump_object("remote", peer.remote); + f.open_object_section("stats"); + for (auto &[attr_name, attr_value] : attributes) { + AttributeDumpVisitor visitor(&f, attr_name); + boost::apply_visitor(visitor, attr_value); + } + f.close_section(); // stats + f.close_section(); // peer.uuid + } + f.close_section(); // peers + f.close_section(); // fscid + } + f.close_section(); // filesystems + } + + std::stringstream ss; + f.flush(ss); + + int r = m_rados->service_daemon_update_status({{"status_json", ss.str()}}); + if (r < 0) { + derr << ": failed to update service daemon status: " << cpp_strerror(r) + << dendl; + } +} + +} // namespace mirror +} // namespace cephfs diff --git a/src/tools/cephfs_mirror/ServiceDaemon.h b/src/tools/cephfs_mirror/ServiceDaemon.h new file mode 100644 index 000000000..83eee286d --- /dev/null +++ b/src/tools/cephfs_mirror/ServiceDaemon.h @@ -0,0 +1,62 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPHFS_MIRROR_SERVICE_DAEMON_H +#define CEPHFS_MIRROR_SERVICE_DAEMON_H + +#include "common/ceph_mutex.h" +#include "common/Timer.h" +#include "mds/FSMap.h" +#include "Types.h" + +namespace cephfs { +namespace mirror { + +class ServiceDaemon { +public: + ServiceDaemon(CephContext *cct, RadosRef rados); + ~ServiceDaemon(); + + int init(); + + void add_filesystem(fs_cluster_id_t fscid, std::string_view fs_name); + void remove_filesystem(fs_cluster_id_t fscid); + + void add_peer(fs_cluster_id_t fscid, const Peer &peer); + void remove_peer(fs_cluster_id_t fscid, const Peer &peer); + + void add_or_update_fs_attribute(fs_cluster_id_t fscid, std::string_view key, + AttributeValue value); + void add_or_update_peer_attribute(fs_cluster_id_t fscid, const Peer &peer, + std::string_view key, AttributeValue value); + +private: + struct Filesystem { + std::string fs_name; + Attributes fs_attributes; + std::map<Peer, Attributes> peer_attributes; + + Filesystem(std::string_view fs_name) + : fs_name(fs_name) { + } + }; + + const std::string CEPHFS_MIRROR_AUTH_ID_PREFIX = "cephfs-mirror."; + + CephContext *m_cct; + RadosRef m_rados; + SafeTimer *m_timer; + ceph::mutex m_timer_lock = ceph::make_mutex("cephfs::mirror::ServiceDaemon"); + + ceph::mutex m_lock = ceph::make_mutex("cephfs::mirror::service_daemon"); + Context *m_timer_ctx = nullptr; + std::map<fs_cluster_id_t, Filesystem> m_filesystems; + + void schedule_update_status(); + void update_status(); +}; + +} // namespace mirror +} // namespace cephfs + +#endif // CEPHFS_MIRROR_SERVICE_DAEMON_H diff --git a/src/tools/cephfs_mirror/Types.cc b/src/tools/cephfs_mirror/Types.cc new file mode 100644 index 000000000..0049f9d79 --- /dev/null +++ b/src/tools/cephfs_mirror/Types.cc @@ -0,0 +1,21 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "Types.h" + +namespace cephfs { +namespace mirror { + +std::ostream& operator<<(std::ostream& out, const Filesystem &filesystem) { + out << "{fscid=" << filesystem.fscid << ", fs_name=" << filesystem.fs_name << "}"; + return out; +} + +std::ostream& operator<<(std::ostream& out, const FilesystemSpec &spec) { + out << "{filesystem=" << spec.filesystem << ", pool_id=" << spec.pool_id << "}"; + return out; +} + +} // namespace mirror +} // namespace cephfs + diff --git a/src/tools/cephfs_mirror/Types.h b/src/tools/cephfs_mirror/Types.h new file mode 100644 index 000000000..016a8dc86 --- /dev/null +++ b/src/tools/cephfs_mirror/Types.h @@ -0,0 +1,87 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPHFS_MIRROR_TYPES_H +#define CEPHFS_MIRROR_TYPES_H + +#include <set> +#include <iostream> +#include <string_view> + +#include "include/rados/librados.hpp" +#include "include/cephfs/libcephfs.h" +#include "mds/mdstypes.h" + +namespace cephfs { +namespace mirror { + +static const std::string CEPHFS_MIRROR_OBJECT("cephfs_mirror"); + +typedef boost::variant<bool, uint64_t, std::string> AttributeValue; +typedef std::map<std::string, AttributeValue> Attributes; + +// distinct filesystem identifier +struct Filesystem { + fs_cluster_id_t fscid; + std::string fs_name; + + bool operator==(const Filesystem &rhs) const { + return (fscid == rhs.fscid && + fs_name == rhs.fs_name); + } + + bool operator!=(const Filesystem &rhs) const { + return !(*this == rhs); + } + + bool operator<(const Filesystem &rhs) const { + if (fscid != rhs.fscid) { + return fscid < rhs.fscid; + } + + return fs_name < rhs.fs_name; + } +}; + +// specification of a filesystem -- pool id the metadata pool id. +struct FilesystemSpec { + FilesystemSpec() = default; + FilesystemSpec(const Filesystem &filesystem, uint64_t pool_id) + : filesystem(filesystem), + pool_id(pool_id) { + } + FilesystemSpec(fs_cluster_id_t fscid, std::string_view fs_name, uint64_t pool_id) + : filesystem(Filesystem{fscid, std::string(fs_name)}), + pool_id(pool_id) { + } + + Filesystem filesystem; + uint64_t pool_id; + + bool operator==(const FilesystemSpec &rhs) const { + return (filesystem == rhs.filesystem && + pool_id == rhs.pool_id); + } + + bool operator<(const FilesystemSpec &rhs) const { + if (filesystem != rhs.filesystem) { + return filesystem < rhs.filesystem; + } + + return pool_id < rhs.pool_id; + } +}; + +std::ostream& operator<<(std::ostream& out, const Filesystem &filesystem); +std::ostream& operator<<(std::ostream& out, const FilesystemSpec &spec); + +typedef std::shared_ptr<librados::Rados> RadosRef; +typedef std::shared_ptr<librados::IoCtx> IoCtxRef; + +// not a shared_ptr since the type is incomplete +typedef ceph_mount_info *MountRef; + +} // namespace mirror +} // namespace cephfs + +#endif // CEPHFS_MIRROR_TYPES_H diff --git a/src/tools/cephfs_mirror/Utils.cc b/src/tools/cephfs_mirror/Utils.cc new file mode 100644 index 000000000..1a8b8e0ac --- /dev/null +++ b/src/tools/cephfs_mirror/Utils.cc @@ -0,0 +1,166 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "common/ceph_argparse.h" +#include "common/ceph_context.h" +#include "common/common_init.h" +#include "common/debug.h" +#include "common/errno.h" + +#include "Utils.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_cephfs_mirror +#undef dout_prefix +#define dout_prefix *_dout << "cephfs::mirror::Utils " << __func__ + +namespace cephfs { +namespace mirror { + +int connect(std::string_view client_name, std::string_view cluster_name, + RadosRef *cluster, std::string_view mon_host, std::string_view cephx_key, + std::vector<const char *> args) { + dout(20) << ": connecting to cluster=" << cluster_name << ", client=" << client_name + << ", mon_host=" << mon_host << dendl; + + CephInitParameters iparams(CEPH_ENTITY_TYPE_CLIENT); + if (client_name.empty() || !iparams.name.from_str(client_name)) { + derr << ": error initializing cluster handle for " << cluster_name << dendl; + return -EINVAL; + } + + CephContext *cct = common_preinit(iparams, CODE_ENVIRONMENT_LIBRARY, + CINIT_FLAG_UNPRIVILEGED_DAEMON_DEFAULTS); + if (mon_host.empty()) { + cct->_conf->cluster = cluster_name; + } + + int r = cct->_conf.parse_config_files(nullptr, nullptr, 0); + if (r < 0 && r != -ENOENT) { + derr << ": could not read ceph conf: " << ": " << cpp_strerror(r) << dendl; + return r; + } + + cct->_conf.parse_env(cct->get_module_type()); + + if (!args.empty()) { + r = cct->_conf.parse_argv(args); + if (r < 0) { + derr << ": could not parse command line args: " << cpp_strerror(r) << dendl; + cct->put(); + return r; + } + } + cct->_conf.parse_env(cct->get_module_type()); + + if (!mon_host.empty()) { + r = cct->_conf.set_val("mon_host", std::string(mon_host)); + if (r < 0) { + derr << "failed to set mon_host config: " << cpp_strerror(r) << dendl; + cct->put(); + return r; + } + } + if (!cephx_key.empty()) { + r = cct->_conf.set_val("key", std::string(cephx_key)); + if (r < 0) { + derr << "failed to set key config: " << cpp_strerror(r) << dendl; + cct->put(); + return r; + } + } + + dout(10) << ": using mon addr=" << cct->_conf.get_val<std::string>("mon_host") << dendl; + + cluster->reset(new librados::Rados()); + + r = (*cluster)->init_with_context(cct); + ceph_assert(r == 0); + cct->put(); + + r = (*cluster)->connect(); + if (r < 0) { + derr << ": error connecting to " << cluster_name << ": " << cpp_strerror(r) + << dendl; + return r; + } + + dout(10) << ": connected to cluster=" << cluster_name << " using client=" + << client_name << dendl; + + return 0; +} + +int mount(RadosRef cluster, const Filesystem &filesystem, bool cross_check_fscid, + MountRef *mount) { + dout(20) << ": filesystem=" << filesystem << dendl; + + ceph_mount_info *cmi; + int r = ceph_create_with_context(&cmi, reinterpret_cast<CephContext*>(cluster->cct())); + if (r < 0) { + derr << ": mount error: " << cpp_strerror(r) << dendl; + return r; + } + + r = ceph_conf_set(cmi, "client_mount_uid", "0"); + if (r < 0) { + derr << ": mount error: " << cpp_strerror(r) << dendl; + return r; + } + + r = ceph_conf_set(cmi, "client_mount_gid", "0"); + if (r < 0) { + derr << ": mount error: " << cpp_strerror(r) << dendl; + return r; + } + + // mount timeout applies for local and remote mounts. + auto mount_timeout = g_ceph_context->_conf.get_val<std::chrono::seconds> + ("cephfs_mirror_mount_timeout").count(); + r = ceph_set_mount_timeout(cmi, mount_timeout); + if (r < 0) { + derr << ": mount error: " << cpp_strerror(r) << dendl; + return r; + } + + r = ceph_init(cmi); + if (r < 0) { + derr << ": mount error: " << cpp_strerror(r) << dendl; + return r; + } + + r = ceph_select_filesystem(cmi, filesystem.fs_name.c_str()); + if (r < 0) { + derr << ": mount error: " << cpp_strerror(r) << dendl; + return r; + } + + r = ceph_mount(cmi, NULL); + if (r < 0) { + derr << ": mount error: " << cpp_strerror(r) << dendl; + return r; + } + + auto fs_id = ceph_get_fs_cid(cmi); + if (cross_check_fscid && fs_id != filesystem.fscid) { + // this can happen in the most remotest possibility when a + // filesystem is deleted and recreated with the same name. + // since all this is driven asynchronously, we were able to + // mount the recreated filesystem. so bubble up the error. + // cleanup will eventually happen since a mirror disable event + // would have been queued. + derr << ": filesystem-id mismatch " << fs_id << " vs " << filesystem.fscid + << dendl; + // ignore errors, we are shutting down anyway. + ceph_unmount(cmi); + return -EINVAL; + } + + dout(10) << ": mounted filesystem=" << filesystem << dendl; + + *mount = cmi; + return 0; +} + +} // namespace mirror +} // namespace cephfs diff --git a/src/tools/cephfs_mirror/Utils.h b/src/tools/cephfs_mirror/Utils.h new file mode 100644 index 000000000..76b0c0726 --- /dev/null +++ b/src/tools/cephfs_mirror/Utils.h @@ -0,0 +1,22 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPHFS_MIRROR_UTILS_H +#define CEPHFS_MIRROR_UTILS_H + +#include "Types.h" + +namespace cephfs { +namespace mirror { + +int connect(std::string_view client_name, std::string_view cluster_name, + RadosRef *cluster, std::string_view mon_host={}, std::string_view cephx_key={}, + std::vector<const char *> args={}); + +int mount(RadosRef cluster, const Filesystem &filesystem, bool cross_check_fscid, + MountRef *mount); + +} // namespace mirror +} // namespace cephfs + +#endif // CEPHFS_MIRROR_UTILS_H diff --git a/src/tools/cephfs_mirror/Watcher.cc b/src/tools/cephfs_mirror/Watcher.cc new file mode 100644 index 000000000..1445fce5f --- /dev/null +++ b/src/tools/cephfs_mirror/Watcher.cc @@ -0,0 +1,285 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "common/ceph_context.h" +#include "common/debug.h" +#include "common/errno.h" +#include "common/WorkQueue.h" +#include "include/stringify.h" +#include "aio_utils.h" +#include "watcher/RewatchRequest.h" +#include "Watcher.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_cephfs_mirror +#undef dout_prefix +#define dout_prefix *_dout << "cephfs::mirror::Watcher " << __func__ + +using cephfs::mirror::watcher::RewatchRequest; + +namespace cephfs { +namespace mirror { + +namespace { + +struct C_UnwatchAndFlush : public Context { + librados::Rados rados; + Context *on_finish; + bool flushing = false; + int ret_val = 0; + + C_UnwatchAndFlush(librados::IoCtx &ioctx, Context *on_finish) + : rados(ioctx), on_finish(on_finish) { + } + + void complete(int r) override { + if (ret_val == 0 && r < 0) { + ret_val = r; + } + + if (!flushing) { + flushing = true; + + librados::AioCompletion *aio_comp = + librados::Rados::aio_create_completion( + this, &rados_callback<Context, &Context::complete>); + r = rados.aio_watch_flush(aio_comp); + + ceph_assert(r == 0); + aio_comp->release(); + return; + } + + // ensure our reference to the RadosClient is released prior + // to completing the callback to avoid racing an explicit + // librados shutdown + Context *ctx = on_finish; + r = ret_val; + delete this; + + ctx->complete(r); + } + + void finish(int r) override { + } +}; + +} // anonymous namespace + +Watcher::Watcher(librados::IoCtx &ioctx, std::string_view oid, ContextWQ *work_queue) + : m_oid(oid), + m_ioctx(ioctx), + m_work_queue(work_queue), + m_lock(ceph::make_shared_mutex("cephfs::mirror::snap_watcher")), + m_state(STATE_IDLE), + m_watch_ctx(*this) { +} + +Watcher::~Watcher() { +} + +void Watcher::register_watch(Context *on_finish) { + dout(20) << dendl; + + std::scoped_lock locker(m_lock); + m_state = STATE_REGISTERING; + + on_finish = new C_RegisterWatch(this, on_finish); + librados::AioCompletion *aio_comp = + librados::Rados::aio_create_completion(on_finish, &rados_callback<Context, &Context::complete>); + int r = m_ioctx.aio_watch(m_oid, aio_comp, &m_watch_handle, &m_watch_ctx); + ceph_assert(r == 0); + aio_comp->release(); +} + +void Watcher::handle_register_watch(int r, Context *on_finish) { + dout(20) << ": r=" << r << dendl; + + bool watch_error = false; + Context *unregister_watch_ctx = nullptr; + { + std::scoped_lock locker(m_lock); + ceph_assert(m_state == STATE_REGISTERING); + + m_state = STATE_IDLE; + if (r < 0) { + derr << ": failed to register watch: " << cpp_strerror(r) << dendl; + m_watch_handle = 0; + } + + if (m_unregister_watch_ctx != nullptr) { + std::swap(unregister_watch_ctx, m_unregister_watch_ctx); + } else if (r == 0 && m_watch_error) { + derr << ": re-registering after watch error" << dendl; + m_state = STATE_REGISTERING; + watch_error = true; + } else { + m_watch_blocklisted = (r == -EBLOCKLISTED); + } + } + + on_finish->complete(r); + if (unregister_watch_ctx != nullptr) { + unregister_watch_ctx->complete(0); + } else if (watch_error) { + rewatch(); + } +} + +void Watcher::unregister_watch(Context *on_finish) { + dout(20) << dendl; + + { + std::scoped_lock locker(m_lock); + if (m_state != STATE_IDLE) { + dout(10) << ": delaying unregister -- watch register in progress" << dendl; + ceph_assert(m_unregister_watch_ctx == nullptr); + m_unregister_watch_ctx = new LambdaContext([this, on_finish](int r) { + unregister_watch(on_finish); + }); + return; + } else if (is_registered()) { + // watch is registered -- unwatch + librados::AioCompletion *aio_comp = + librados::Rados::aio_create_completion(new C_UnwatchAndFlush(m_ioctx, on_finish), + &rados_callback<Context, &Context::complete>); + int r = m_ioctx.aio_unwatch(m_watch_handle, aio_comp); + ceph_assert(r == 0); + aio_comp->release(); + m_watch_handle = 0; + m_watch_blocklisted = false; + return; + } + } + + on_finish->complete(0); +} + +void Watcher::handle_error(uint64_t handle, int err) { + derr << ": handle=" << handle << ": " << cpp_strerror(err) << dendl; + + std::scoped_lock locker(m_lock); + m_watch_error = true; + + if (is_registered()) { + m_state = STATE_REWATCHING; + if (err == -EBLOCKLISTED) { + m_watch_blocklisted = true; + } + m_work_queue->queue(new LambdaContext([this] { + rewatch(); + }), 0); + } +} + +void Watcher::rewatch() { + dout(20) << dendl; + + Context *unregister_watch_ctx = nullptr; + { + std::unique_lock locker(m_lock); + ceph_assert(m_state == STATE_REWATCHING); + + if (m_unregister_watch_ctx != nullptr) { + m_state = STATE_IDLE; + std::swap(unregister_watch_ctx, m_unregister_watch_ctx); + } else { + m_watch_error = false; + Context *ctx = new C_CallbackAdapter<Watcher, &Watcher::handle_rewatch>(this); + auto req = RewatchRequest::create(m_ioctx, m_oid, m_lock, + &m_watch_ctx, &m_watch_handle, ctx); + req->send(); + return; + } + } + + unregister_watch_ctx->complete(0); +} + +void Watcher::handle_rewatch(int r) { + dout(20) << ": r=" << r << dendl; + + bool watch_error = false; + Context *unregister_watch_ctx = nullptr; + { + std::scoped_lock locker(m_lock); + ceph_assert(m_state == STATE_REWATCHING); + + m_watch_blocklisted = false; + if (m_unregister_watch_ctx != nullptr) { + dout(10) << ": skipping rewatch -- unregistering" << dendl; + m_state = STATE_IDLE; + std::swap(unregister_watch_ctx, m_unregister_watch_ctx); + } else if (r == -EBLOCKLISTED) { + m_watch_blocklisted = true; + derr << ": client blocklisted" << dendl; + } else if (r == -ENOENT) { + dout(5) << ": object " << m_oid << " does not exist" << dendl; + } else if (r < 0) { + derr << ": failed to rewatch: " << cpp_strerror(r) << dendl; + watch_error = true; + } else if (m_watch_error) { + derr << ": re-registering watch after error" << dendl; + watch_error = true; + } + } + + if (unregister_watch_ctx != nullptr) { + unregister_watch_ctx->complete(0); + return; + } else if (watch_error) { + rewatch(); + return; + } + + Context *ctx = new C_CallbackAdapter<Watcher, &Watcher::handle_rewatch_callback>(this); + m_work_queue->queue(ctx, r); +} + +void Watcher::handle_rewatch_callback(int r) { + dout(10) << ": r=" << r << dendl; + handle_rewatch_complete(r); + + bool watch_error = false; + Context *unregister_watch_ctx = nullptr; + { + std::scoped_lock locker(m_lock); + ceph_assert(m_state == STATE_REWATCHING); + + if (m_unregister_watch_ctx != nullptr) { + m_state = STATE_IDLE; + std::swap(unregister_watch_ctx, m_unregister_watch_ctx); + } else if (r == -EBLOCKLISTED || r == -ENOENT) { + m_state = STATE_IDLE; + } else if (r < 0 || m_watch_error) { + watch_error = true; + } else { + m_state = STATE_IDLE; + } + } + + if (unregister_watch_ctx != nullptr) { + unregister_watch_ctx->complete(0); + } else if (watch_error) { + rewatch(); + } +} + +void Watcher::acknowledge_notify(uint64_t notify_id, uint64_t handle, bufferlist &bl) { + m_ioctx.notify_ack(m_oid, notify_id, handle, bl); +} + +void Watcher::WatchCtx::handle_notify(uint64_t notify_id, uint64_t handle, + uint64_t notifier_id, bufferlist& bl) { + dout(20) << ": notify_id=" << notify_id << ", handle=" << handle + << ", notifier_id=" << notifier_id << dendl; + watcher.handle_notify(notify_id, handle, notifier_id, bl); +} + +void Watcher::WatchCtx::handle_error(uint64_t handle, int err) { + dout(20) << dendl; + watcher.handle_error(handle, err); +} + +} // namespace mirror +} // namespace cephfs diff --git a/src/tools/cephfs_mirror/Watcher.h b/src/tools/cephfs_mirror/Watcher.h new file mode 100644 index 000000000..9e7c54eeb --- /dev/null +++ b/src/tools/cephfs_mirror/Watcher.h @@ -0,0 +1,102 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPHFS_MIRROR_WATCHER_H +#define CEPHFS_MIRROR_WATCHER_H + +#include <string_view> + +#include "common/ceph_mutex.h" +#include "include/Context.h" +#include "include/rados/librados.hpp" + +class ContextWQ; + +namespace cephfs { +namespace mirror { + +// generic watcher class -- establish watch on a given rados object +// and invoke handle_notify() when notified. On notify error, try +// to re-establish the watch. Errors during rewatch are notified via +// handle_rewatch_complete(). + +class Watcher { +public: + Watcher(librados::IoCtx &ioctx, std::string_view oid, ContextWQ *work_queue); + virtual ~Watcher(); + + void register_watch(Context *on_finish); + void unregister_watch(Context *on_finish); + +protected: + std::string m_oid; + + void acknowledge_notify(uint64_t notify_if, uint64_t handle, bufferlist &bl); + + bool is_registered() const { + return m_state == STATE_IDLE && m_watch_handle != 0; + } + bool is_unregistered() const { + return m_state == STATE_IDLE && m_watch_handle == 0; + } + + virtual void handle_rewatch_complete(int r) { } + +private: + enum State { + STATE_IDLE, + STATE_REGISTERING, + STATE_REWATCHING + }; + + struct WatchCtx : public librados::WatchCtx2 { + Watcher &watcher; + + WatchCtx(Watcher &parent) : watcher(parent) {} + + void handle_notify(uint64_t notify_id, + uint64_t handle, + uint64_t notifier_id, + bufferlist& bl) override; + void handle_error(uint64_t handle, int err) override; + }; + + struct C_RegisterWatch : public Context { + Watcher *watcher; + Context *on_finish; + + C_RegisterWatch(Watcher *watcher, Context *on_finish) + : watcher(watcher), + on_finish(on_finish) { + } + + void finish(int r) override { + watcher->handle_register_watch(r, on_finish); + } + }; + + librados::IoCtx &m_ioctx; + ContextWQ *m_work_queue; + + mutable ceph::shared_mutex m_lock; + State m_state; + bool m_watch_error = false; + bool m_watch_blocklisted = false; + uint64_t m_watch_handle; + WatchCtx m_watch_ctx; + Context *m_unregister_watch_ctx = nullptr; + + virtual void handle_notify(uint64_t notify_id, uint64_t handle, + uint64_t notifier_id, bufferlist& bl) = 0; + void handle_error(uint64_t handle, int err); + + void rewatch(); + void handle_rewatch(int r); + void handle_rewatch_callback(int r); + void handle_register_watch(int r, Context *on_finish); +}; + +} // namespace mirror +} // namespace cephfs + +#endif // CEPHFS_MIRROR_WATCHER_H diff --git a/src/tools/cephfs_mirror/aio_utils.h b/src/tools/cephfs_mirror/aio_utils.h new file mode 100644 index 000000000..43f356381 --- /dev/null +++ b/src/tools/cephfs_mirror/aio_utils.h @@ -0,0 +1,53 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPHFS_MIRROR_AIO_UTILS_H +#define CEPHFS_MIRROR_AIO_UTILS_H + +#include "include/rados/librados.hpp" + +namespace cephfs { +namespace mirror { + +template <typename T, void(T::*MF)(int)> +void rados_callback(rados_completion_t c, void *arg) { + T *obj = reinterpret_cast<T*>(arg); + int r = rados_aio_get_return_value(c); + (obj->*MF)(r); +} + +template <typename T, void (T::*MF)(int)> +class C_CallbackAdapter : public Context { + T *obj; +public: + C_CallbackAdapter(T *obj) + : obj(obj) { + } + +protected: + void finish(int r) override { + (obj->*MF)(r); + } +}; + +template <typename WQ> +struct C_AsyncCallback : public Context { + WQ *op_work_queue; + Context *on_finish; + + C_AsyncCallback(WQ *op_work_queue, Context *on_finish) + : op_work_queue(op_work_queue), on_finish(on_finish) { + } + ~C_AsyncCallback() override { + delete on_finish; + } + void finish(int r) override { + op_work_queue->queue(on_finish, r); + on_finish = nullptr; + } +}; + +} // namespace mirror +} // namespace cephfs + +#endif // CEPHFS_MIRROR_AIO_UTILS_H diff --git a/src/tools/cephfs_mirror/main.cc b/src/tools/cephfs_mirror/main.cc new file mode 100644 index 000000000..efaa89c35 --- /dev/null +++ b/src/tools/cephfs_mirror/main.cc @@ -0,0 +1,124 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "common/ceph_argparse.h" +#include "common/config.h" +#include "common/debug.h" +#include "common/errno.h" +#include "common/async/context_pool.h" +#include "common/Preforker.h" +#include "global/global_init.h" +#include "global/signal_handler.h" +#include "mon/MonClient.h" +#include "msg/Messenger.h" +#include "Mirror.h" + +#include <vector> + +void usage() { + std::cout << "usage: cephfs-mirror [options...]" << std::endl; + std::cout << "options:\n"; + std::cout << " --mon-host monaddress[:port] connect to specified monitor\n"; + std::cout << " --keyring=<path> path to keyring for local cluster\n"; + std::cout << " --log-file=<logfile> file to log debug output\n"; + std::cout << " --debug-cephfs-mirror=<log-level>/<memory-level> set cephfs-mirror debug level\n"; + generic_server_usage(); +} + +cephfs::mirror::Mirror *mirror = nullptr; + +static void handle_signal(int signum) { + if (mirror) { + mirror->handle_signal(signum); + } +} + +int main(int argc, const char **argv) { + std::vector<const char*> args; + argv_to_vec(argc, argv, args); + if (args.empty()) { + cerr << argv[0] << ": -h or --help for usage" << std::endl; + ::exit(1); + } + + if (ceph_argparse_need_usage(args)) { + usage(); + ::exit(0); + } + + auto cct = global_init(nullptr, args, CEPH_ENTITY_TYPE_CLIENT, + CODE_ENVIRONMENT_DAEMON, + CINIT_FLAG_UNPRIVILEGED_DAEMON_DEFAULTS); + + Preforker forker; + if (global_init_prefork(g_ceph_context) >= 0) { + std::string err; + int r = forker.prefork(err); + if (r < 0) { + cerr << err << std::endl; + return r; + } + if (forker.is_parent()) { + g_ceph_context->_log->start(); + if (forker.parent_wait(err) != 0) { + return -ENXIO; + } + return 0; + } + global_init_postfork_start(g_ceph_context); + } + + common_init_finish(g_ceph_context); + + bool daemonize = g_conf().get_val<bool>("daemonize"); + if (daemonize) { + global_init_postfork_finish(g_ceph_context); + forker.daemonize(); + } + + init_async_signal_handler(); + register_async_signal_handler(SIGHUP, handle_signal); + register_async_signal_handler_oneshot(SIGINT, handle_signal); + register_async_signal_handler_oneshot(SIGTERM, handle_signal); + + std::vector<const char*> cmd_args; + argv_to_vec(argc, argv, cmd_args); + + Messenger *msgr = Messenger::create_client_messenger(g_ceph_context, "client"); + msgr->set_default_policy(Messenger::Policy::lossy_client(0)); + + std::string reason; + ceph::async::io_context_pool ctxpool(1); + MonClient monc(MonClient(g_ceph_context, ctxpool)); + int r = monc.build_initial_monmap(); + if (r < 0) { + cerr << "failed to generate initial monmap" << std::endl; + goto cleanup_messenger; + } + + msgr->start(); + + mirror = new cephfs::mirror::Mirror(g_ceph_context, cmd_args, &monc, msgr); + r = mirror->init(reason); + if (r < 0) { + std::cerr << "failed to initialize cephfs-mirror: " << reason << std::endl; + goto cleanup; + } + + mirror->run(); + delete mirror; + +cleanup: + monc.shutdown(); +cleanup_messenger: + msgr->shutdown(); + msgr->wait(); + delete msgr; + + unregister_async_signal_handler(SIGHUP, handle_signal); + unregister_async_signal_handler(SIGINT, handle_signal); + unregister_async_signal_handler(SIGTERM, handle_signal); + shutdown_async_signal_handler(); + + return forker.signal_exit(r); +} diff --git a/src/tools/cephfs_mirror/watcher/RewatchRequest.cc b/src/tools/cephfs_mirror/watcher/RewatchRequest.cc new file mode 100644 index 000000000..3070e6f8b --- /dev/null +++ b/src/tools/cephfs_mirror/watcher/RewatchRequest.cc @@ -0,0 +1,102 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "common/ceph_mutex.h" +#include "common/debug.h" +#include "common/errno.h" +#include "include/Context.h" +#include "tools/cephfs_mirror/aio_utils.h" +#include "RewatchRequest.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_cephfs_mirror +#undef dout_prefix +#define dout_prefix *_dout << "cephfs::mirror::watcher:RewatchRequest " << __func__ + +namespace cephfs { +namespace mirror { +namespace watcher { + +RewatchRequest::RewatchRequest(librados::IoCtx &ioctx, const std::string &oid, + ceph::shared_mutex &watch_lock, + librados::WatchCtx2 *watch_ctx, + uint64_t *watch_handle, Context *on_finish) + : m_ioctx(ioctx), m_oid(oid), m_lock(watch_lock), + m_watch_ctx(watch_ctx), m_watch_handle(watch_handle), + m_on_finish(on_finish) { +} + +void RewatchRequest::send() { + unwatch(); +} + +void RewatchRequest::unwatch() { + ceph_assert(ceph_mutex_is_wlocked(m_lock)); + if (*m_watch_handle == 0) { + rewatch(); + return; + } + + dout(10) << dendl; + + uint64_t watch_handle = 0; + std::swap(*m_watch_handle, watch_handle); + + librados::AioCompletion *aio_comp = + librados::Rados::aio_create_completion( + this, &rados_callback<RewatchRequest, &RewatchRequest::handle_unwatch>); + int r = m_ioctx.aio_unwatch(watch_handle, aio_comp); + ceph_assert(r == 0); + aio_comp->release(); +} + +void RewatchRequest::handle_unwatch(int r) { + dout(20) << ": r=" << r << dendl; + + if (r == -EBLOCKLISTED) { + derr << ": client blocklisted" << dendl; + finish(r); + return; + } else if (r < 0) { + derr << ": failed to unwatch: " << cpp_strerror(r) << dendl; + } + + rewatch(); +} + +void RewatchRequest::rewatch() { + dout(20) << dendl; + + librados::AioCompletion *aio_comp = + librados::Rados::aio_create_completion( + this, &rados_callback<RewatchRequest, &RewatchRequest::handle_rewatch>); + int r = m_ioctx.aio_watch(m_oid, aio_comp, &m_rewatch_handle, m_watch_ctx); + ceph_assert(r == 0); + aio_comp->release(); +} + +void RewatchRequest::handle_rewatch(int r) { + dout(20) << ": r=" << r << dendl; + + if (r < 0) { + derr << ": failed to watch object: " << cpp_strerror(r) << dendl; + m_rewatch_handle = 0; + } + + { + std::unique_lock locker(m_lock); + *m_watch_handle = m_rewatch_handle; + } + + finish(r); +} + +void RewatchRequest::finish(int r) { + dout(20) << ": r=" << r << dendl; + m_on_finish->complete(r); + delete this; +} + +} // namespace watcher +} // namespace mirror +} // namespace cephfs diff --git a/src/tools/cephfs_mirror/watcher/RewatchRequest.h b/src/tools/cephfs_mirror/watcher/RewatchRequest.h new file mode 100644 index 000000000..453fcb219 --- /dev/null +++ b/src/tools/cephfs_mirror/watcher/RewatchRequest.h @@ -0,0 +1,60 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPHFS_MIRROR_WATCHER_REWATCH_REQUEST_H +#define CEPHFS_MIRROR_WATCHER_REWATCH_REQUEST_H + +#include "common/ceph_mutex.h" +#include "include/int_types.h" +#include "include/rados/librados.hpp" + +struct Context; + +namespace cephfs { +namespace mirror { +namespace watcher { + +// Rewatch an existing watch -- the watch can be in an operatioal +// or error state. + +class RewatchRequest { +public: + + static RewatchRequest *create(librados::IoCtx &ioctx, const std::string &oid, + ceph::shared_mutex &watch_lock, + librados::WatchCtx2 *watch_ctx, + uint64_t *watch_handle, Context *on_finish) { + return new RewatchRequest(ioctx, oid, watch_lock, watch_ctx, watch_handle, + on_finish); + } + + RewatchRequest(librados::IoCtx &ioctx, const std::string &oid, + ceph::shared_mutex &watch_lock, librados::WatchCtx2 *watch_ctx, + uint64_t *watch_handle, Context *on_finish); + + void send(); + +private: + librados::IoCtx& m_ioctx; + std::string m_oid; + ceph::shared_mutex &m_lock; + librados::WatchCtx2 *m_watch_ctx; + uint64_t *m_watch_handle; + Context *m_on_finish; + + uint64_t m_rewatch_handle = 0; + + void unwatch(); + void handle_unwatch(int r); + + void rewatch(); + void handle_rewatch(int r); + + void finish(int r); +}; + +} // namespace watcher +} // namespace mirror +} // namespace cephfs + +#endif // CEPHFS_MIRROR_WATCHER_REWATCH_REQUEST_H |