--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_RBD_MIRROR_IMAGE_REPLAYER_H
+#define CEPH_RBD_MIRROR_IMAGE_REPLAYER_H
+
+#include "common/AsyncOpTracker.h"
+#include "common/Mutex.h"
+#include "common/WorkQueue.h"
+#include "include/rados/librados.hpp"
+#include "cls/journal/cls_journal_types.h"
+#include "cls/rbd/cls_rbd_types.h"
+#include "journal/JournalMetadataListener.h"
+#include "journal/ReplayEntry.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/journal/Types.h"
+#include "librbd/journal/TypeTraits.h"
+#include "ProgressContext.h"
+#include "types.h"
+#include "tools/rbd_mirror/image_replayer/Types.h"
+
+#include <boost/noncopyable.hpp>
+#include <boost/optional.hpp>
+
+#include <set>
+#include <map>
+#include <atomic>
+#include <string>
+#include <vector>
+
+class AdminSocketHook;
+
+namespace journal {
+
+class Journaler;
+class ReplayHandler;
+
+}
+
+namespace librbd {
+
+class ImageCtx;
+namespace journal { template <typename> class Replay; }
+
+}
+
+namespace rbd {
+namespace mirror {
+
+template <typename> struct ImageDeleter;
+template <typename> struct InstanceWatcher;
+template <typename> struct Threads;
+
+namespace image_replayer { template <typename> class BootstrapRequest; }
+namespace image_replayer { template <typename> class EventPreprocessor; }
+namespace image_replayer { template <typename> class ReplayStatusFormatter; }
+
+/**
+ * Replays changes from a remote cluster for a single image.
+ */
+template <typename ImageCtxT = librbd::ImageCtx>
+class ImageReplayer {
+public:
+ static ImageReplayer *create(
+ Threads<ImageCtxT> *threads, ImageDeleter<ImageCtxT>* image_deleter,
+ InstanceWatcher<ImageCtxT> *instance_watcher,
+ RadosRef local, const std::string &local_mirror_uuid, int64_t local_pool_id,
+ const std::string &global_image_id) {
+ return new ImageReplayer(threads, image_deleter, instance_watcher,
+ local, local_mirror_uuid, local_pool_id,
+ global_image_id);
+ }
+ void destroy() {
+ delete this;
+ }
+
+ ImageReplayer(Threads<ImageCtxT> *threads,
+ ImageDeleter<ImageCtxT>* image_deleter,
+ InstanceWatcher<ImageCtxT> *instance_watcher,
+ RadosRef local, const std::string &local_mirror_uuid,
+ int64_t local_pool_id, const std::string &global_image_id);
+ virtual ~ImageReplayer();
+ ImageReplayer(const ImageReplayer&) = delete;
+ ImageReplayer& operator=(const ImageReplayer&) = delete;
+
+ bool is_stopped() { Mutex::Locker l(m_lock); return is_stopped_(); }
+ bool is_running() { Mutex::Locker l(m_lock); return is_running_(); }
+ bool is_replaying() { Mutex::Locker l(m_lock); return is_replaying_(); }
+
+ std::string get_name() { Mutex::Locker l(m_lock); return m_name; };
+ void set_state_description(int r, const std::string &desc);
+
+ // TODO temporary until policy handles release of image replayers
+ inline bool is_finished() const {
+ Mutex::Locker locker(m_lock);
+ return m_finished;
+ }
+ inline void set_finished(bool finished) {
+ Mutex::Locker locker(m_lock);
+ m_finished = finished;
+ }
+
+ inline bool is_blacklisted() const {
+ Mutex::Locker locker(m_lock);
+ return (m_last_r == -EBLACKLISTED);
+ }
+
+ image_replayer::HealthState get_health_state() const;
+
+ void add_peer(const std::string &peer_uuid, librados::IoCtx &remote_io_ctx);
+
+ inline int64_t get_local_pool_id() const {
+ return m_local_pool_id;
+ }
+ inline const std::string& get_global_image_id() const {
+ return m_global_image_id;
+ }
+
+ void start(Context *on_finish = nullptr, bool manual = false);
+ void stop(Context *on_finish = nullptr, bool manual = false,
+ int r = 0, const std::string& desc = "");
+ void restart(Context *on_finish = nullptr);
+ void flush(Context *on_finish = nullptr);
+
+ void resync_image(Context *on_finish=nullptr);
+
+ void print_status(Formatter *f, stringstream *ss);
+
+ virtual void handle_replay_ready();
+ virtual void handle_replay_complete(int r, const std::string &error_desc);
+
+protected:
+ /**
+ * @verbatim
+ * (error)
+ * <uninitialized> <------------------------------------ FAIL
+ * | ^
+ * v *
+ * <starting> *
+ * | *
+ * v *
+ * WAIT_FOR_DELETION *
+ * | *
+ * v (error) *
+ * PREPARE_LOCAL_IMAGE * * * * * * * * * * * * * * * * * *
+ * | *
+ * v (error) *
+ * PREPARE_REMOTE_IMAGE * * * * * * * * * * * * * * * * * *
+ * | *
+ * v (error) *
+ * BOOTSTRAP_IMAGE * * * * * * * * * * * * * * * * * * * *
+ * | *
+ * v (error) *
+ * INIT_REMOTE_JOURNALER * * * * * * * * * * * * * * * * *
+ * | *
+ * v (error) *
+ * START_REPLAY * * * * * * * * * * * * * * * * * * * * * *
+ * |
+ * | /--------------------------------------------\
+ * | | |
+ * v v (asok flush) |
+ * REPLAYING -------------> LOCAL_REPLAY_FLUSH |
+ * | \ | |
+ * | | v |
+ * | | FLUSH_COMMIT_POSITION |
+ * | | | |
+ * | | \--------------------/|
+ * | | |
+ * | | (entries available) |
+ * | \-----------> REPLAY_READY |
+ * | | |
+ * | | (skip if not |
+ * | v needed) (error)
+ * | REPLAY_FLUSH * * * * * * * * *
+ * | | | *
+ * | | (skip if not | *
+ * | v needed) (error) *
+ * | GET_REMOTE_TAG * * * * * * * *
+ * | | | *
+ * | | (skip if not | *
+ * | v needed) (error) *
+ * | ALLOCATE_LOCAL_TAG * * * * * *
+ * | | | *
+ * | v (error) *
+ * | PREPROCESS_ENTRY * * * * * * *
+ * | | | *
+ * | v (error) *
+ * | PROCESS_ENTRY * * * * * * * * *
+ * | | | *
+ * | \---------------------/ *
+ * v *
+ * REPLAY_COMPLETE < * * * * * * * * * * * * * * * * * * *
+ * |
+ * v
+ * JOURNAL_REPLAY_SHUT_DOWN
+ * |
+ * v
+ * LOCAL_IMAGE_CLOSE
+ * |
+ * v
+ * <stopped>
+ *
+ * @endverbatim
+ */
+
+ virtual void on_start_fail(int r, const std::string &desc = "");
+ virtual bool on_start_interrupted();
+
+ virtual void on_stop_journal_replay(int r = 0, const std::string &desc = "");
+
+ virtual void on_flush_local_replay_flush_start(Context *on_flush);
+ virtual void on_flush_local_replay_flush_finish(Context *on_flush, int r);
+ virtual void on_flush_flush_commit_position_start(Context *on_flush);
+ virtual void on_flush_flush_commit_position_finish(Context *on_flush, int r);
+
+ bool on_replay_interrupted();
+
+private:
+ typedef typename librbd::journal::TypeTraits<ImageCtxT>::ReplayEntry ReplayEntry;
+
+ enum State {
+ STATE_UNKNOWN,
+ STATE_STARTING,
+ STATE_REPLAYING,
+ STATE_REPLAY_FLUSHING,
+ STATE_STOPPING,
+ STATE_STOPPED,
+ };
+
+ struct RemoteImage {
+ std::string mirror_uuid;
+ std::string image_id;
+ librados::IoCtx io_ctx;
+
+ RemoteImage() {
+ }
+ RemoteImage(const Peer& peer) : io_ctx(peer.io_ctx) {
+ }
+ };
+
+ typedef typename librbd::journal::TypeTraits<ImageCtxT>::Journaler Journaler;
+ typedef boost::optional<State> OptionalState;
+ typedef boost::optional<cls::rbd::MirrorImageStatusState>
+ OptionalMirrorImageStatusState;
+
+ struct JournalListener : public librbd::journal::Listener {
+ ImageReplayer *img_replayer;
+
+ JournalListener(ImageReplayer *img_replayer)
+ : img_replayer(img_replayer) {
+ }
+
+ void handle_close() override {
+ img_replayer->on_stop_journal_replay();
+ }
+
+ void handle_promoted() override {
+ img_replayer->on_stop_journal_replay(0, "force promoted");
+ }
+
+ void handle_resync() override {
+ img_replayer->resync_image();
+ }
+ };
+
+ class BootstrapProgressContext : public ProgressContext {
+ public:
+ BootstrapProgressContext(ImageReplayer<ImageCtxT> *replayer) :
+ replayer(replayer) {
+ }
+
+ void update_progress(const std::string &description,
+ bool flush = true) override;
+ private:
+ ImageReplayer<ImageCtxT> *replayer;
+ };
+
+ Threads<ImageCtxT> *m_threads;
+ ImageDeleter<ImageCtxT>* m_image_deleter;
+ InstanceWatcher<ImageCtxT> *m_instance_watcher;
+
+ Peers m_peers;
+ RemoteImage m_remote_image;
+
+ RadosRef m_local;
+ std::string m_local_mirror_uuid;
+ int64_t m_local_pool_id;
+ std::string m_local_image_id;
+ std::string m_global_image_id;
+ std::string m_name;
+
+ mutable Mutex m_lock;
+ State m_state = STATE_STOPPED;
+ std::string m_state_desc;
+
+ OptionalMirrorImageStatusState m_mirror_image_status_state = boost::none;
+ int m_last_r = 0;
+
+ BootstrapProgressContext m_progress_cxt;
+
+ bool m_finished = false;
+ bool m_delete_requested = false;
+ bool m_resync_requested = false;
+
+ image_replayer::EventPreprocessor<ImageCtxT> *m_event_preprocessor = nullptr;
+ image_replayer::ReplayStatusFormatter<ImageCtxT> *m_replay_status_formatter =
+ nullptr;
+ librados::IoCtx m_local_ioctx;
+ ImageCtxT *m_local_image_ctx = nullptr;
+ std::string m_local_image_tag_owner;
+
+ decltype(ImageCtxT::journal) m_local_journal = nullptr;
+ librbd::journal::Replay<ImageCtxT> *m_local_replay = nullptr;
+ Journaler* m_remote_journaler = nullptr;
+ ::journal::ReplayHandler *m_replay_handler = nullptr;
+ librbd::journal::Listener *m_journal_listener;
+
+ Context *m_on_start_finish = nullptr;
+ Context *m_on_stop_finish = nullptr;
+ Context *m_update_status_task = nullptr;
+ int m_update_status_interval = 0;
+ librados::AioCompletion *m_update_status_comp = nullptr;
+ bool m_stop_requested = false;
+ bool m_manual_stop = false;
+
+ AdminSocketHook *m_asok_hook = nullptr;
+
+ image_replayer::BootstrapRequest<ImageCtxT> *m_bootstrap_request = nullptr;
+
+ uint32_t m_in_flight_status_updates = 0;
+ bool m_update_status_requested = false;
+ Context *m_on_update_status_finish = nullptr;
+
+ librbd::journal::MirrorPeerClientMeta m_client_meta;
+
+ ReplayEntry m_replay_entry;
+ bool m_replay_tag_valid = false;
+ uint64_t m_replay_tag_tid = 0;
+ cls::journal::Tag m_replay_tag;
+ librbd::journal::TagData m_replay_tag_data;
+ librbd::journal::EventEntry m_event_entry;
+ AsyncOpTracker m_event_replay_tracker;
+ Context *m_delayed_preprocess_task = nullptr;
+
+ struct RemoteJournalerListener : public ::journal::JournalMetadataListener {
+ ImageReplayer *replayer;
+
+ RemoteJournalerListener(ImageReplayer *replayer) : replayer(replayer) { }
+
+ void handle_update(::journal::JournalMetadata *) override;
+ } m_remote_listener;
+
+ struct C_ReplayCommitted : public Context {
+ ImageReplayer *replayer;
+ ReplayEntry replay_entry;
+
+ C_ReplayCommitted(ImageReplayer *replayer,
+ ReplayEntry &&replay_entry)
+ : replayer(replayer), replay_entry(std::move(replay_entry)) {
+ }
+ void finish(int r) override {
+ replayer->handle_process_entry_safe(replay_entry, r);
+ }
+ };
+
+ static std::string to_string(const State state);
+
+ bool is_stopped_() const {
+ return m_state == STATE_STOPPED;
+ }
+ bool is_running_() const {
+ return !is_stopped_() && m_state != STATE_STOPPING && !m_stop_requested;
+ }
+ bool is_replaying_() const {
+ return (m_state == STATE_REPLAYING ||
+ m_state == STATE_REPLAY_FLUSHING);
+ }
+
+ bool update_mirror_image_status(bool force, const OptionalState &state);
+ bool start_mirror_image_status_update(bool force, bool restarting);
+ void finish_mirror_image_status_update();
+ void queue_mirror_image_status_update(const OptionalState &state);
+ void send_mirror_status_update(const OptionalState &state);
+ void handle_mirror_status_update(int r);
+ void reschedule_update_status_task(int new_interval = 0);
+
+ void shut_down(int r);
+ void handle_shut_down(int r);
+ void handle_remote_journal_metadata_updated();
+
+ void wait_for_deletion();
+ void handle_wait_for_deletion(int r);
+
+ void prepare_local_image();
+ void handle_prepare_local_image(int r);
+
+ void prepare_remote_image();
+ void handle_prepare_remote_image(int r);
+
+ void bootstrap();
+ void handle_bootstrap(int r);
+
+ void init_remote_journaler();
+ void handle_init_remote_journaler(int r);
+
+ void start_replay();
+ void handle_start_replay(int r);
+
+ void replay_flush();
+ void handle_replay_flush(int r);
+
+ void get_remote_tag();
+ void handle_get_remote_tag(int r);
+
+ void allocate_local_tag();
+ void handle_allocate_local_tag(int r);
+
+ void preprocess_entry();
+ void handle_preprocess_entry_ready(int r);
+ void handle_preprocess_entry_safe(int r);
+
+ void process_entry();
+ void handle_process_entry_ready(int r);
+ void handle_process_entry_safe(const ReplayEntry& replay_entry, int r);
+
+ void register_admin_socket_hook();
+ void unregister_admin_socket_hook();
+
+ void on_name_changed();
+};
+
+} // namespace mirror
+} // namespace rbd
+
+extern template class rbd::mirror::ImageReplayer<librbd::ImageCtx>;
+
+#endif // CEPH_RBD_MIRROR_IMAGE_REPLAYER_H
Code Review / stor4nfv.git / blobdiff