diff options
Diffstat (limited to '')
30 files changed, 6277 insertions, 0 deletions
diff --git a/src/librbd/migration/FileStream.cc b/src/librbd/migration/FileStream.cc new file mode 100644 index 000000000..63cd722dd --- /dev/null +++ b/src/librbd/migration/FileStream.cc @@ -0,0 +1,232 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef _LARGEFILE64_SOURCE +#define _LARGEFILE64_SOURCE +#endif // _LARGEFILE64_SOURCE + +#include "librbd/migration/FileStream.h" +#include "common/dout.h" +#include "common/errno.h" +#include "librbd/AsioEngine.h" +#include "librbd/ImageCtx.h" +#include "librbd/asio/Utils.h" +#include <boost/asio/buffer.hpp> +#include <boost/asio/post.hpp> +#include <boost/asio/read.hpp> +#include <fcntl.h> +#include <unistd.h> + +namespace librbd { +namespace migration { + +namespace { + +const std::string FILE_PATH {"file_path"}; + +} // anonymous namespace + +#ifdef BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR + +#define dout_subsys ceph_subsys_rbd +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::FileStream::ReadRequest " \ + << this << " " << __func__ << ": " + +template <typename I> +struct FileStream<I>::ReadRequest { + FileStream* file_stream; + io::Extents byte_extents; + bufferlist* data; + Context* on_finish; + size_t index = 0; + + ReadRequest(FileStream* file_stream, io::Extents&& byte_extents, + bufferlist* data, Context* on_finish) + : file_stream(file_stream), byte_extents(std::move(byte_extents)), + data(data), on_finish(on_finish) { + auto cct = file_stream->m_cct; + ldout(cct, 20) << dendl; + } + + void send() { + data->clear(); + read(); + } + + void read() { + auto cct = file_stream->m_cct; + if (index >= byte_extents.size()) { + finish(0); + return; + } + + auto& byte_extent = byte_extents[index++]; + ldout(cct, 20) << "byte_extent=" << byte_extent << dendl; + + auto ptr = buffer::ptr_node::create(buffer::create_small_page_aligned( + byte_extent.second)); + auto buffer = boost::asio::mutable_buffer( + ptr->c_str(), byte_extent.second); + data->push_back(std::move(ptr)); + + int r; + auto offset = lseek64(file_stream->m_file_no, byte_extent.first, SEEK_SET); + if (offset == -1) { + r = -errno; + lderr(cct) << "failed to seek file stream: " << cpp_strerror(r) << dendl; + finish(r); + return; + } + + boost::system::error_code ec; + size_t bytes_read = boost::asio::read( + *file_stream->m_stream_descriptor, std::move(buffer), ec); + r = -ec.value(); + if (r < 0 && r != -ENOENT) { + lderr(cct) << "failed to read from file stream: " << cpp_strerror(r) + << dendl; + finish(r); + return; + } else if (bytes_read < byte_extent.second) { + lderr(cct) << "failed to read " << byte_extent.second << " bytes from " + << "file stream" << dendl; + finish(-ERANGE); + return; + } + + // re-queue the remainder of the read requests + boost::asio::post(file_stream->m_strand, [this]() { read(); }); + } + + void finish(int r) { + auto cct = file_stream->m_cct; + ldout(cct, 20) << "r=" << r << dendl; + + if (r < 0) { + data->clear(); + } + + on_finish->complete(r); + delete this; + } +}; + +#endif // BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR + +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::FileStream: " << this \ + << " " << __func__ << ": " + +template <typename I> +FileStream<I>::FileStream(I* image_ctx, const json_spirit::mObject& json_object) + : m_cct(image_ctx->cct), m_asio_engine(image_ctx->asio_engine), + m_json_object(json_object), + m_strand(boost::asio::make_strand(*m_asio_engine)) { +} + +template <typename I> +FileStream<I>::~FileStream() { + if (m_file_no != -1) { + ::close(m_file_no); + } +} + +#ifdef BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR + +template <typename I> +void FileStream<I>::open(Context* on_finish) { + auto& file_path_value = m_json_object[FILE_PATH]; + if (file_path_value.type() != json_spirit::str_type) { + lderr(m_cct) << "failed to locate '" << FILE_PATH << "' key" << dendl; + on_finish->complete(-EINVAL); + return; + } + + auto& file_path = file_path_value.get_str(); + ldout(m_cct, 10) << "file_path=" << file_path << dendl; + + m_file_no = ::open(file_path.c_str(), O_RDONLY); + if (m_file_no < 0) { + int r = -errno; + lderr(m_cct) << "failed to open file stream '" << file_path << "': " + << cpp_strerror(r) << dendl; + on_finish->complete(r); + return; + } + + m_stream_descriptor = std::make_optional< + boost::asio::posix::stream_descriptor>(m_strand, m_file_no); + on_finish->complete(0); +} + +template <typename I> +void FileStream<I>::close(Context* on_finish) { + ldout(m_cct, 10) << dendl; + + m_stream_descriptor.reset(); + on_finish->complete(0); +} + +template <typename I> +void FileStream<I>::get_size(uint64_t* size, Context* on_finish) { + ldout(m_cct, 10) << dendl; + + // execute IO operations in a single strand to prevent seek races + boost::asio::post( + m_strand, [this, size, on_finish]() { + auto offset = lseek64(m_file_no, 0, SEEK_END); + if (offset == -1) { + int r = -errno; + lderr(m_cct) << "failed to seek to file end: " << cpp_strerror(r) + << dendl; + on_finish->complete(r); + return; + } + + ldout(m_cct, 10) << "size=" << offset << dendl; + *size = offset; + on_finish->complete(0); + }); +} + +template <typename I> +void FileStream<I>::read(io::Extents&& byte_extents, bufferlist* data, + Context* on_finish) { + ldout(m_cct, 20) << byte_extents << dendl; + + auto ctx = new ReadRequest(this, std::move(byte_extents), data, on_finish); + + // execute IO operations in a single strand to prevent seek races + boost::asio::post(m_strand, [ctx]() { ctx->send(); }); +} + +#else // BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR + +template <typename I> +void FileStream<I>::open(Context* on_finish) { + on_finish->complete(-EIO); +} + +template <typename I> +void FileStream<I>::close(Context* on_finish) { + on_finish->complete(-EIO); +} + +template <typename I> +void FileStream<I>::get_size(uint64_t* size, Context* on_finish) { + on_finish->complete(-EIO); +} + +template <typename I> +void FileStream<I>::read(io::Extents&& byte_extents, bufferlist* data, + Context* on_finish) { + on_finish->complete(-EIO); +} + +#endif // BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR + +} // namespace migration +} // namespace librbd + +template class librbd::migration::FileStream<librbd::ImageCtx>; diff --git a/src/librbd/migration/FileStream.h b/src/librbd/migration/FileStream.h new file mode 100644 index 000000000..32face71e --- /dev/null +++ b/src/librbd/migration/FileStream.h @@ -0,0 +1,68 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_MIGRATION_FILE_STREAM_H +#define CEPH_LIBRBD_MIGRATION_FILE_STREAM_H + +#include "include/int_types.h" +#include "librbd/migration/StreamInterface.h" +#include <boost/asio/io_context.hpp> +#include <boost/asio/strand.hpp> +#include <boost/asio/posix/basic_stream_descriptor.hpp> +#include <json_spirit/json_spirit.h> +#include <memory> +#include <string> + +struct Context; + +namespace librbd { + +struct AsioEngine; +struct ImageCtx; + +namespace migration { + +template <typename ImageCtxT> +class FileStream : public StreamInterface { +public: + static FileStream* create(ImageCtxT* image_ctx, + const json_spirit::mObject& json_object) { + return new FileStream(image_ctx, json_object); + } + + FileStream(ImageCtxT* image_ctx, const json_spirit::mObject& json_object); + ~FileStream() override; + + FileStream(const FileStream&) = delete; + FileStream& operator=(const FileStream&) = delete; + + void open(Context* on_finish) override; + void close(Context* on_finish) override; + + void get_size(uint64_t* size, Context* on_finish) override; + + void read(io::Extents&& byte_extents, bufferlist* data, + Context* on_finish) override; + +private: + CephContext* m_cct; + std::shared_ptr<AsioEngine> m_asio_engine; + json_spirit::mObject m_json_object; + + boost::asio::strand<boost::asio::io_context::executor_type> m_strand; +#ifdef BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR + std::optional<boost::asio::posix::stream_descriptor> m_stream_descriptor; + + struct ReadRequest; + +#endif // BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR + + int m_file_no = -1; +}; + +} // namespace migration +} // namespace librbd + +extern template class librbd::migration::FileStream<librbd::ImageCtx>; + +#endif // CEPH_LIBRBD_MIGRATION_FILE_STREAM_H diff --git a/src/librbd/migration/FormatInterface.h b/src/librbd/migration/FormatInterface.h new file mode 100644 index 000000000..d13521d11 --- /dev/null +++ b/src/librbd/migration/FormatInterface.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 CEPH_LIBRBD_MIGRATION_FORMAT_INTERFACE_H +#define CEPH_LIBRBD_MIGRATION_FORMAT_INTERFACE_H + +#include "include/buffer_fwd.h" +#include "include/int_types.h" +#include "common/zipkin_trace.h" +#include "librbd/Types.h" +#include "librbd/io/Types.h" +#include <map> + +struct Context; + +namespace librbd { + +namespace io { +struct AioCompletion; +struct ReadResult; +} // namespace io + +namespace migration { + +struct FormatInterface { + typedef std::map<uint64_t, SnapInfo> SnapInfos; + + virtual ~FormatInterface() { + } + + virtual void open(Context* on_finish) = 0; + virtual void close(Context* on_finish) = 0; + + virtual void get_snapshots(SnapInfos* snap_infos, Context* on_finish) = 0; + virtual void get_image_size(uint64_t snap_id, uint64_t* size, + Context* on_finish) = 0; + + virtual bool read(io::AioCompletion* aio_comp, uint64_t snap_id, + io::Extents&& image_extents, io::ReadResult&& read_result, + int op_flags, int read_flags, + const ZTracer::Trace &parent_trace) = 0; + + virtual void list_snaps(io::Extents&& image_extents, io::SnapIds&& snap_ids, + int list_snaps_flags, + io::SnapshotDelta* snapshot_delta, + const ZTracer::Trace &parent_trace, + Context* on_finish) = 0; +}; + +} // namespace migration +} // namespace librbd + +#endif // CEPH_LIBRBD_MIGRATION_FORMAT_INTERFACE_H diff --git a/src/librbd/migration/HttpClient.cc b/src/librbd/migration/HttpClient.cc new file mode 100644 index 000000000..90d5723ed --- /dev/null +++ b/src/librbd/migration/HttpClient.cc @@ -0,0 +1,947 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "librbd/migration/HttpClient.h" +#include "common/dout.h" +#include "common/errno.h" +#include "librbd/AsioEngine.h" +#include "librbd/ImageCtx.h" +#include "librbd/Utils.h" +#include "librbd/asio/Utils.h" +#include "librbd/io/AioCompletion.h" +#include "librbd/io/ReadResult.h" +#include "librbd/migration/Utils.h" +#include <boost/asio/buffer.hpp> +#include <boost/asio/post.hpp> +#include <boost/asio/ip/tcp.hpp> +#include <boost/asio/read.hpp> +#include <boost/asio/ssl.hpp> +#include <boost/beast/core.hpp> +#include <boost/beast/http/read.hpp> +#include <boost/lexical_cast.hpp> +#include <deque> + +namespace librbd { +namespace migration { + +#define dout_subsys ceph_subsys_rbd +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::HttpClient::" \ + << "HttpSession " << this << " " << __func__ \ + << ": " + +/** + * boost::beast utilizes non-inheriting template classes for handling plain vs + * encrypted TCP streams. Utilize a base-class for handling the majority of the + * logic for handling connecting, disconnecting, reseting, and sending requests. + */ + +template <typename I> +template <typename D> +class HttpClient<I>::HttpSession : public HttpSessionInterface { +public: + void init(Context* on_finish) override { + ceph_assert(m_http_client->m_strand.running_in_this_thread()); + + auto cct = m_http_client->m_cct; + ldout(cct, 15) << dendl; + + ceph_assert(m_state == STATE_UNINITIALIZED); + m_state = STATE_CONNECTING; + + resolve_host(on_finish); + } + + void shut_down(Context* on_finish) override { + ceph_assert(m_http_client->m_strand.running_in_this_thread()); + + auto cct = m_http_client->m_cct; + ldout(cct, 15) << dendl; + + ceph_assert(on_finish != nullptr); + ceph_assert(m_on_shutdown == nullptr); + m_on_shutdown = on_finish; + + auto current_state = m_state; + if (current_state == STATE_UNINITIALIZED) { + // never initialized or resolve/connect failed + on_finish->complete(0); + return; + } + + m_state = STATE_SHUTTING_DOWN; + if (current_state != STATE_READY) { + // delay shutdown until current state transition completes + return; + } + + disconnect(new LambdaContext([this](int r) { handle_shut_down(r); })); + } + + void issue(std::shared_ptr<Work>&& work) override { + ceph_assert(m_http_client->m_strand.running_in_this_thread()); + + auto cct = m_http_client->m_cct; + ldout(cct, 20) << "work=" << work.get() << dendl; + + if (is_shutdown()) { + lderr(cct) << "cannot issue HTTP request, client is shutdown" + << dendl; + work->complete(-ESHUTDOWN, {}); + return; + } + + bool first_issue = m_issue_queue.empty(); + m_issue_queue.emplace_back(work); + if (m_state == STATE_READY && first_issue) { + ldout(cct, 20) << "sending http request: work=" << work.get() << dendl; + finalize_issue(std::move(work)); + } else if (m_state == STATE_UNINITIALIZED) { + ldout(cct, 20) << "resetting HTTP session: work=" << work.get() << dendl; + m_state = STATE_RESET_CONNECTING; + resolve_host(nullptr); + } else { + ldout(cct, 20) << "queueing HTTP request: work=" << work.get() << dendl; + } + } + + void finalize_issue(std::shared_ptr<Work>&& work) { + auto cct = m_http_client->m_cct; + ldout(cct, 20) << "work=" << work.get() << dendl; + + ++m_in_flight_requests; + (*work)(derived().stream()); + } + + void handle_issue(boost::system::error_code ec, + std::shared_ptr<Work>&& work) override { + ceph_assert(m_http_client->m_strand.running_in_this_thread()); + + auto cct = m_http_client->m_cct; + ldout(cct, 20) << "work=" << work.get() << ", r=" << -ec.value() << dendl; + + ceph_assert(m_in_flight_requests > 0); + --m_in_flight_requests; + if (maybe_finalize_reset()) { + // previous request is attempting reset to this request will be resent + return; + } + + ceph_assert(!m_issue_queue.empty()); + m_issue_queue.pop_front(); + + if (is_shutdown()) { + lderr(cct) << "client shutdown during in-flight request" << dendl; + work->complete(-ESHUTDOWN, {}); + + maybe_finalize_shutdown(); + return; + } + + if (ec) { + if (ec == boost::asio::error::bad_descriptor || + ec == boost::asio::error::broken_pipe || + ec == boost::asio::error::connection_reset || + ec == boost::asio::error::operation_aborted || + ec == boost::asio::ssl::error::stream_truncated || + ec == boost::beast::http::error::end_of_stream || + ec == boost::beast::http::error::partial_message) { + ldout(cct, 5) << "remote peer stream closed, retrying request" << dendl; + m_issue_queue.push_front(work); + } else if (ec == boost::beast::error::timeout) { + lderr(cct) << "timed-out while issuing request" << dendl; + work->complete(-ETIMEDOUT, {}); + } else { + lderr(cct) << "failed to issue request: " << ec.message() << dendl; + work->complete(-ec.value(), {}); + } + + // attempt to recover the connection + reset(); + return; + } + + bool first_receive = m_receive_queue.empty(); + m_receive_queue.push_back(work); + if (first_receive) { + receive(std::move(work)); + } + + // TODO disable pipelining for non-idempotent requests + + // pipeline the next request into the stream + if (!m_issue_queue.empty()) { + work = m_issue_queue.front(); + ldout(cct, 20) << "sending http request: work=" << work.get() << dendl; + finalize_issue(std::move(work)); + } + } + +protected: + HttpClient* m_http_client; + + HttpSession(HttpClient* http_client) + : m_http_client(http_client), m_resolver(http_client->m_strand) { + } + + virtual void connect(boost::asio::ip::tcp::resolver::results_type results, + Context* on_finish) = 0; + virtual void disconnect(Context* on_finish) = 0; + + void close_socket() { + auto cct = m_http_client->m_cct; + ldout(cct, 15) << dendl; + + boost::system::error_code ec; + boost::beast::get_lowest_layer(derived().stream()).socket().close(ec); + } + +private: + enum State { + STATE_UNINITIALIZED, + STATE_CONNECTING, + STATE_READY, + STATE_RESET_PENDING, + STATE_RESET_DISCONNECTING, + STATE_RESET_CONNECTING, + STATE_SHUTTING_DOWN, + STATE_SHUTDOWN, + }; + + State m_state = STATE_UNINITIALIZED; + boost::asio::ip::tcp::resolver m_resolver; + + Context* m_on_shutdown = nullptr; + + uint64_t m_in_flight_requests = 0; + std::deque<std::shared_ptr<Work>> m_issue_queue; + std::deque<std::shared_ptr<Work>> m_receive_queue; + + boost::beast::flat_buffer m_buffer; + std::optional<boost::beast::http::parser<false, EmptyBody>> m_header_parser; + std::optional<boost::beast::http::parser<false, StringBody>> m_parser; + + D& derived() { + return static_cast<D&>(*this); + } + + void resolve_host(Context* on_finish) { + auto cct = m_http_client->m_cct; + ldout(cct, 15) << dendl; + + shutdown_socket(); + m_resolver.async_resolve( + m_http_client->m_url_spec.host, m_http_client->m_url_spec.port, + [this, on_finish](boost::system::error_code ec, auto results) { + handle_resolve_host(ec, results, on_finish); }); + } + + void handle_resolve_host( + boost::system::error_code ec, + boost::asio::ip::tcp::resolver::results_type results, + Context* on_finish) { + auto cct = m_http_client->m_cct; + int r = -ec.value(); + ldout(cct, 15) << "r=" << r << dendl; + + if (ec) { + if (ec == boost::asio::error::host_not_found) { + r = -ENOENT; + } else if (ec == boost::asio::error::host_not_found_try_again) { + // TODO: add retry throttle + r = -EAGAIN; + } + + lderr(cct) << "failed to resolve host '" + << m_http_client->m_url_spec.host << "': " + << cpp_strerror(r) << dendl; + advance_state(STATE_UNINITIALIZED, r, on_finish); + return; + } + + connect(results, new LambdaContext([this, on_finish](int r) { + handle_connect(r, on_finish); })); + } + + void handle_connect(int r, Context* on_finish) { + auto cct = m_http_client->m_cct; + ldout(cct, 15) << "r=" << r << dendl; + + if (r < 0) { + lderr(cct) << "failed to connect to host '" + << m_http_client->m_url_spec.host << "': " + << cpp_strerror(r) << dendl; + advance_state(STATE_UNINITIALIZED, r, on_finish); + return; + } + + advance_state(STATE_READY, 0, on_finish); + } + + void handle_shut_down(int r) { + auto cct = m_http_client->m_cct; + ldout(cct, 15) << "r=" << r << dendl; + + if (r < 0) { + lderr(cct) << "failed to disconnect stream: '" << cpp_strerror(r) + << dendl; + } + + // cancel all in-flight send/receives (if any) + shutdown_socket(); + + maybe_finalize_shutdown(); + } + + void maybe_finalize_shutdown() { + if (m_in_flight_requests > 0) { + return; + } + + // cancel any queued IOs + fail_queued_work(-ESHUTDOWN); + + advance_state(STATE_SHUTDOWN, 0, nullptr); + } + + bool is_shutdown() const { + ceph_assert(m_http_client->m_strand.running_in_this_thread()); + return (m_state == STATE_SHUTTING_DOWN || m_state == STATE_SHUTDOWN); + } + + void reset() { + ceph_assert(m_http_client->m_strand.running_in_this_thread()); + ceph_assert(m_state == STATE_READY); + + auto cct = m_http_client->m_cct; + ldout(cct, 15) << dendl; + + m_state = STATE_RESET_PENDING; + maybe_finalize_reset(); + } + + bool maybe_finalize_reset() { + if (m_state != STATE_RESET_PENDING) { + return false; + } + + if (m_in_flight_requests > 0) { + return true; + } + + ceph_assert(m_http_client->m_strand.running_in_this_thread()); + auto cct = m_http_client->m_cct; + ldout(cct, 15) << dendl; + + m_buffer.clear(); + + // move in-flight request back to the front of the issue queue + m_issue_queue.insert(m_issue_queue.begin(), + m_receive_queue.begin(), m_receive_queue.end()); + m_receive_queue.clear(); + + m_state = STATE_RESET_DISCONNECTING; + disconnect(new LambdaContext([this](int r) { handle_reset(r); })); + return true; + } + + void handle_reset(int r) { + auto cct = m_http_client->m_cct; + ldout(cct, 15) << "r=" << r << dendl; + + if (r < 0) { + lderr(cct) << "failed to disconnect stream: '" << cpp_strerror(r) + << dendl; + } + + advance_state(STATE_RESET_CONNECTING, r, nullptr); + } + + int shutdown_socket() { + if (!boost::beast::get_lowest_layer( + derived().stream()).socket().is_open()) { + return 0; + } + + auto cct = m_http_client->m_cct; + ldout(cct, 15) << dendl; + + boost::system::error_code ec; + boost::beast::get_lowest_layer(derived().stream()).socket().shutdown( + boost::asio::ip::tcp::socket::shutdown_both, ec); + + if (ec && ec != boost::beast::errc::not_connected) { + lderr(cct) << "failed to shutdown socket: " << ec.message() << dendl; + return -ec.value(); + } + + close_socket(); + return 0; + } + + void receive(std::shared_ptr<Work>&& work) { + auto cct = m_http_client->m_cct; + ldout(cct, 15) << "work=" << work.get() << dendl; + + ceph_assert(!m_receive_queue.empty()); + ++m_in_flight_requests; + + // receive the response for this request + m_parser.emplace(); + if (work->header_only()) { + // HEAD requests don't trasfer data but the parser still cares about max + // content-length + m_header_parser.emplace(); + m_header_parser->body_limit(std::numeric_limits<uint64_t>::max()); + + boost::beast::http::async_read_header( + derived().stream(), m_buffer, *m_header_parser, + [this, work=std::move(work)] + (boost::beast::error_code ec, std::size_t) mutable { + handle_receive(ec, std::move(work)); + }); + } else { + m_parser->body_limit(1 << 25); // max RBD object size + boost::beast::http::async_read( + derived().stream(), m_buffer, *m_parser, + [this, work=std::move(work)] + (boost::beast::error_code ec, std::size_t) mutable { + handle_receive(ec, std::move(work)); + }); + } + } + + void handle_receive(boost::system::error_code ec, + std::shared_ptr<Work>&& work) { + auto cct = m_http_client->m_cct; + ldout(cct, 15) << "work=" << work.get() << ", r=" << -ec.value() << dendl; + + ceph_assert(m_in_flight_requests > 0); + --m_in_flight_requests; + if (maybe_finalize_reset()) { + // previous request is attempting reset to this request will be resent + return; + } + + ceph_assert(!m_receive_queue.empty()); + m_receive_queue.pop_front(); + + if (is_shutdown()) { + lderr(cct) << "client shutdown with in-flight request" << dendl; + work->complete(-ESHUTDOWN, {}); + + maybe_finalize_shutdown(); + return; + } + + if (ec) { + if (ec == boost::asio::error::bad_descriptor || + ec == boost::asio::error::broken_pipe || + ec == boost::asio::error::connection_reset || + ec == boost::asio::error::operation_aborted || + ec == boost::asio::ssl::error::stream_truncated || + ec == boost::beast::http::error::end_of_stream || + ec == boost::beast::http::error::partial_message) { + ldout(cct, 5) << "remote peer stream closed, retrying request" << dendl; + m_receive_queue.push_front(work); + } else if (ec == boost::beast::error::timeout) { + lderr(cct) << "timed-out while issuing request" << dendl; + work->complete(-ETIMEDOUT, {}); + } else { + lderr(cct) << "failed to issue request: " << ec.message() << dendl; + work->complete(-ec.value(), {}); + } + + reset(); + return; + } + + Response response; + if (work->header_only()) { + m_parser.emplace(std::move(*m_header_parser)); + } + response = m_parser->release(); + + // basic response code handling in a common location + int r = 0; + auto result = response.result(); + if (result == boost::beast::http::status::not_found) { + lderr(cct) << "requested resource does not exist" << dendl; + r = -ENOENT; + } else if (result == boost::beast::http::status::forbidden) { + lderr(cct) << "permission denied attempting to access resource" << dendl; + r = -EACCES; + } else if (boost::beast::http::to_status_class(result) != + boost::beast::http::status_class::successful) { + lderr(cct) << "failed to retrieve size: HTTP " << result << dendl; + r = -EIO; + } + + bool need_eof = response.need_eof(); + if (r < 0) { + work->complete(r, {}); + } else { + work->complete(0, std::move(response)); + } + + if (need_eof) { + ldout(cct, 20) << "reset required for non-pipelined response: " + << "work=" << work.get() << dendl; + reset(); + } else if (!m_receive_queue.empty()) { + auto work = m_receive_queue.front(); + receive(std::move(work)); + } + } + + void advance_state(State next_state, int r, Context* on_finish) { + auto cct = m_http_client->m_cct; + auto current_state = m_state; + ldout(cct, 15) << "current_state=" << current_state << ", " + << "next_state=" << next_state << ", " + << "r=" << r << dendl; + + m_state = next_state; + if (current_state == STATE_CONNECTING) { + if (next_state == STATE_UNINITIALIZED) { + shutdown_socket(); + on_finish->complete(r); + return; + } else if (next_state == STATE_READY) { + on_finish->complete(r); + return; + } + } else if (current_state == STATE_SHUTTING_DOWN) { + if (next_state == STATE_READY) { + // shut down requested while connecting/resetting + disconnect(new LambdaContext([this](int r) { handle_shut_down(r); })); + return; + } else if (next_state == STATE_UNINITIALIZED || + next_state == STATE_SHUTDOWN || + next_state == STATE_RESET_CONNECTING) { + ceph_assert(m_on_shutdown != nullptr); + m_on_shutdown->complete(r); + return; + } + } else if (current_state == STATE_RESET_DISCONNECTING) { + // disconnected from peer -- ignore errors and reconnect + ceph_assert(next_state == STATE_RESET_CONNECTING); + ceph_assert(on_finish == nullptr); + shutdown_socket(); + resolve_host(nullptr); + return; + } else if (current_state == STATE_RESET_CONNECTING) { + ceph_assert(on_finish == nullptr); + if (next_state == STATE_READY) { + // restart queued IO + if (!m_issue_queue.empty()) { + auto& work = m_issue_queue.front(); + finalize_issue(std::move(work)); + } + return; + } else if (next_state == STATE_UNINITIALIZED) { + shutdown_socket(); + + // fail all queued IO + fail_queued_work(r); + return; + } + } + + lderr(cct) << "unexpected state transition: " + << "current_state=" << current_state << ", " + << "next_state=" << next_state << dendl; + ceph_assert(false); + } + + void complete_work(std::shared_ptr<Work> work, int r, Response&& response) { + auto cct = m_http_client->m_cct; + ldout(cct, 20) << "work=" << work.get() << ", r=" << r << dendl; + + work->complete(r, std::move(response)); + } + + void fail_queued_work(int r) { + auto cct = m_http_client->m_cct; + ldout(cct, 10) << "r=" << r << dendl; + + for (auto& work : m_issue_queue) { + complete_work(work, r, {}); + } + m_issue_queue.clear(); + ceph_assert(m_receive_queue.empty()); + } +}; + +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::HttpClient::" \ + << "PlainHttpSession " << this << " " << __func__ \ + << ": " + +template <typename I> +class HttpClient<I>::PlainHttpSession : public HttpSession<PlainHttpSession> { +public: + PlainHttpSession(HttpClient* http_client) + : HttpSession<PlainHttpSession>(http_client), + m_stream(http_client->m_strand) { + } + ~PlainHttpSession() override { + this->close_socket(); + } + + inline boost::beast::tcp_stream& + stream() { + return m_stream; + } + +protected: + void connect(boost::asio::ip::tcp::resolver::results_type results, + Context* on_finish) override { + auto http_client = this->m_http_client; + auto cct = http_client->m_cct; + ldout(cct, 15) << dendl; + + m_stream.async_connect( + results, + [on_finish](boost::system::error_code ec, const auto& endpoint) { + on_finish->complete(-ec.value()); + }); + } + + void disconnect(Context* on_finish) override { + on_finish->complete(0); + } + +private: + boost::beast::tcp_stream m_stream; + +}; + +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::HttpClient::" \ + << "SslHttpSession " << this << " " << __func__ \ + << ": " + +template <typename I> +class HttpClient<I>::SslHttpSession : public HttpSession<SslHttpSession> { +public: + SslHttpSession(HttpClient* http_client) + : HttpSession<SslHttpSession>(http_client), + m_stream(http_client->m_strand, http_client->m_ssl_context) { + } + ~SslHttpSession() override { + this->close_socket(); + } + + inline boost::beast::ssl_stream<boost::beast::tcp_stream>& + stream() { + return m_stream; + } + +protected: + void connect(boost::asio::ip::tcp::resolver::results_type results, + Context* on_finish) override { + auto http_client = this->m_http_client; + auto cct = http_client->m_cct; + ldout(cct, 15) << dendl; + + boost::beast::get_lowest_layer(m_stream).async_connect( + results, + [this, on_finish](boost::system::error_code ec, const auto& endpoint) { + handle_connect(-ec.value(), on_finish); + }); + } + + void disconnect(Context* on_finish) override { + auto http_client = this->m_http_client; + auto cct = http_client->m_cct; + ldout(cct, 15) << dendl; + + if (!m_ssl_enabled) { + on_finish->complete(0); + return; + } + + m_stream.async_shutdown( + asio::util::get_callback_adapter([this, on_finish](int r) { + shutdown(r, on_finish); })); + } + +private: + boost::beast::ssl_stream<boost::beast::tcp_stream> m_stream; + bool m_ssl_enabled = false; + + void handle_connect(int r, Context* on_finish) { + auto http_client = this->m_http_client; + auto cct = http_client->m_cct; + ldout(cct, 15) << dendl; + + if (r < 0) { + lderr(cct) << "failed to connect to host '" + << http_client->m_url_spec.host << "': " + << cpp_strerror(r) << dendl; + on_finish->complete(r); + return; + } + + handshake(on_finish); + } + + void handshake(Context* on_finish) { + auto http_client = this->m_http_client; + auto cct = http_client->m_cct; + ldout(cct, 15) << dendl; + + auto& host = http_client->m_url_spec.host; + m_stream.set_verify_mode( + boost::asio::ssl::verify_peer | + boost::asio::ssl::verify_fail_if_no_peer_cert); + m_stream.set_verify_callback( + [host, next=boost::asio::ssl::host_name_verification(host), + ignore_self_signed=http_client->m_ignore_self_signed_cert] + (bool preverified, boost::asio::ssl::verify_context& ctx) { + if (!preverified && ignore_self_signed) { + auto ec = X509_STORE_CTX_get_error(ctx.native_handle()); + switch (ec) { + case X509_V_ERR_DEPTH_ZERO_SELF_SIGNED_CERT: + case X509_V_ERR_SELF_SIGNED_CERT_IN_CHAIN: + // ignore self-signed cert issues + preverified = true; + break; + default: + break; + } + } + return next(preverified, ctx); + }); + + // Set SNI Hostname (many hosts need this to handshake successfully) + if(!SSL_set_tlsext_host_name(m_stream.native_handle(), + http_client->m_url_spec.host.c_str())) { + int r = -::ERR_get_error(); + lderr(cct) << "failed to initialize SNI hostname: " << cpp_strerror(r) + << dendl; + on_finish->complete(r); + return; + } + + // Perform the SSL/TLS handshake + m_stream.async_handshake( + boost::asio::ssl::stream_base::client, + asio::util::get_callback_adapter( + [this, on_finish](int r) { handle_handshake(r, on_finish); })); + } + + void handle_handshake(int r, Context* on_finish) { + auto http_client = this->m_http_client; + auto cct = http_client->m_cct; + ldout(cct, 15) << "r=" << r << dendl; + + if (r < 0) { + lderr(cct) << "failed to complete handshake: " << cpp_strerror(r) + << dendl; + disconnect(new LambdaContext([r, on_finish](int) { + on_finish->complete(r); })); + return; + } + + m_ssl_enabled = true; + on_finish->complete(0); + } + + void shutdown(int r, Context* on_finish) { + auto http_client = this->m_http_client; + auto cct = http_client->m_cct; + ldout(cct, 15) << "r=" << r << dendl; + + on_finish->complete(r); + } +}; + +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::HttpClient: " << this \ + << " " << __func__ << ": " + +template <typename I> +HttpClient<I>::HttpClient(I* image_ctx, const std::string& url) + : m_cct(image_ctx->cct), m_image_ctx(image_ctx), + m_asio_engine(image_ctx->asio_engine), m_url(url), + m_strand(boost::asio::make_strand(*m_asio_engine)), + m_ssl_context(boost::asio::ssl::context::sslv23_client) { + m_ssl_context.set_default_verify_paths(); +} + +template <typename I> +void HttpClient<I>::open(Context* on_finish) { + ldout(m_cct, 10) << "url=" << m_url << dendl; + + int r = util::parse_url(m_cct, m_url, &m_url_spec); + if (r < 0) { + lderr(m_cct) << "failed to parse url '" << m_url << "': " << cpp_strerror(r) + << dendl; + on_finish->complete(-EINVAL); + return; + } + + boost::asio::post(m_strand, [this, on_finish]() mutable { + create_http_session(on_finish); }); +} + +template <typename I> +void HttpClient<I>::close(Context* on_finish) { + boost::asio::post(m_strand, [this, on_finish]() mutable { + shut_down_http_session(on_finish); }); +} + +template <typename I> +void HttpClient<I>::get_size(uint64_t* size, Context* on_finish) { + ldout(m_cct, 10) << dendl; + + Request req; + req.method(boost::beast::http::verb::head); + + issue( + std::move(req), [this, size, on_finish](int r, Response&& response) { + handle_get_size(r, std::move(response), size, on_finish); + }); +} + +template <typename I> +void HttpClient<I>::handle_get_size(int r, Response&& response, uint64_t* size, + Context* on_finish) { + ldout(m_cct, 10) << "r=" << r << dendl; + + if (r < 0) { + lderr(m_cct) << "failed to retrieve size: " << cpp_strerror(r) << dendl; + on_finish->complete(r); + return; + } else if (!response.has_content_length()) { + lderr(m_cct) << "failed to retrieve size: missing content-length" << dendl; + on_finish->complete(-EINVAL); + return; + } + + auto content_length = response[boost::beast::http::field::content_length]; + try { + *size = boost::lexical_cast<uint64_t>(content_length); + } catch (boost::bad_lexical_cast&) { + lderr(m_cct) << "invalid content-length in response" << dendl; + on_finish->complete(-EBADMSG); + return; + } + + on_finish->complete(0); +} + +template <typename I> +void HttpClient<I>::read(io::Extents&& byte_extents, bufferlist* data, + Context* on_finish) { + ldout(m_cct, 20) << dendl; + + auto aio_comp = io::AioCompletion::create_and_start( + on_finish, librbd::util::get_image_ctx(m_image_ctx), io::AIO_TYPE_READ); + aio_comp->set_request_count(byte_extents.size()); + + // utilize ReadResult to assemble multiple byte extents into a single bl + // since boost::beast doesn't support multipart responses out-of-the-box + io::ReadResult read_result{data}; + aio_comp->read_result = std::move(read_result); + aio_comp->read_result.set_image_extents(byte_extents); + + // issue a range get request for each extent + uint64_t buffer_offset = 0; + for (auto [byte_offset, byte_length] : byte_extents) { + auto ctx = new io::ReadResult::C_ImageReadRequest( + aio_comp, buffer_offset, {{byte_offset, byte_length}}); + buffer_offset += byte_length; + + Request req; + req.method(boost::beast::http::verb::get); + + std::stringstream range; + ceph_assert(byte_length > 0); + range << "bytes=" << byte_offset << "-" << (byte_offset + byte_length - 1); + req.set(boost::beast::http::field::range, range.str()); + + issue( + std::move(req), + [this, byte_offset=byte_offset, byte_length=byte_length, ctx] + (int r, Response&& response) { + handle_read(r, std::move(response), byte_offset, byte_length, &ctx->bl, + ctx); + }); + } +} + +template <typename I> +void HttpClient<I>::handle_read(int r, Response&& response, + uint64_t byte_offset, uint64_t byte_length, + bufferlist* data, Context* on_finish) { + ldout(m_cct, 20) << "bytes=" << byte_offset << "~" << byte_length << ", " + << "r=" << r << dendl; + + if (r < 0) { + lderr(m_cct) << "failed to read requested byte range: " + << cpp_strerror(r) << dendl; + on_finish->complete(r); + return; + } else if (response.result() != boost::beast::http::status::partial_content) { + lderr(m_cct) << "failed to retrieve requested byte range: HTTP " + << response.result() << dendl; + on_finish->complete(-EIO); + return; + } else if (byte_length != response.body().size()) { + lderr(m_cct) << "unexpected short range read: " + << "wanted=" << byte_length << ", " + << "received=" << response.body().size() << dendl; + on_finish->complete(-EINVAL); + return; + } + + data->clear(); + data->append(response.body()); + on_finish->complete(data->length()); +} + +template <typename I> +void HttpClient<I>::issue(std::shared_ptr<Work>&& work) { + boost::asio::post(m_strand, [this, work=std::move(work)]() mutable { + m_http_session->issue(std::move(work)); }); +} + +template <typename I> +void HttpClient<I>::create_http_session(Context* on_finish) { + ldout(m_cct, 15) << dendl; + + ceph_assert(m_http_session == nullptr); + switch (m_url_spec.scheme) { + case URL_SCHEME_HTTP: + m_http_session = std::make_unique<PlainHttpSession>(this); + break; + case URL_SCHEME_HTTPS: + m_http_session = std::make_unique<SslHttpSession>(this); + break; + default: + ceph_assert(false); + break; + } + + m_http_session->init(on_finish); +} + +template <typename I> +void HttpClient<I>::shut_down_http_session(Context* on_finish) { + ldout(m_cct, 15) << dendl; + + if (m_http_session == nullptr) { + on_finish->complete(0); + return; + } + + m_http_session->shut_down(on_finish); +} + +} // namespace migration +} // namespace librbd + +template class librbd::migration::HttpClient<librbd::ImageCtx>; diff --git a/src/librbd/migration/HttpClient.h b/src/librbd/migration/HttpClient.h new file mode 100644 index 000000000..3997e6159 --- /dev/null +++ b/src/librbd/migration/HttpClient.h @@ -0,0 +1,205 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_MIGRATION_HTTP_CLIENT_H +#define CEPH_LIBRBD_MIGRATION_HTTP_CLIENT_H + +#include "include/common_fwd.h" +#include "include/int_types.h" +#include "librbd/io/Types.h" +#include "librbd/migration/HttpProcessorInterface.h" +#include "librbd/migration/Types.h" +#include <boost/asio/io_context.hpp> +#include <boost/asio/strand.hpp> +#include <boost/asio/ip/tcp.hpp> +#include <boost/asio/ssl/context.hpp> +#include <boost/beast/version.hpp> +#include <boost/beast/core/tcp_stream.hpp> +#include <boost/beast/http/empty_body.hpp> +#include <boost/beast/http/message.hpp> +#include <boost/beast/http/string_body.hpp> +#include <boost/beast/http/write.hpp> +#include <boost/beast/ssl/ssl_stream.hpp> +#include <functional> +#include <memory> +#include <string> +#include <utility> + +struct Context; + +namespace librbd { + +struct AsioEngine; +struct ImageCtx; + +namespace migration { + +template <typename ImageCtxT> +class HttpClient { +public: + using EmptyBody = boost::beast::http::empty_body; + using StringBody = boost::beast::http::string_body; + using Request = boost::beast::http::request<EmptyBody>; + using Response = boost::beast::http::response<StringBody>; + + using RequestPreprocessor = std::function<void(Request&)>; + + static HttpClient* create(ImageCtxT* image_ctx, const std::string& url) { + return new HttpClient(image_ctx, url); + } + + HttpClient(ImageCtxT* image_ctx, const std::string& url); + HttpClient(const HttpClient&) = delete; + HttpClient& operator=(const HttpClient&) = delete; + + void open(Context* on_finish); + void close(Context* on_finish); + + void get_size(uint64_t* size, Context* on_finish); + + void read(io::Extents&& byte_extents, bufferlist* data, + Context* on_finish); + + void set_ignore_self_signed_cert(bool ignore) { + m_ignore_self_signed_cert = ignore; + } + + void set_http_processor(HttpProcessorInterface* http_processor) { + m_http_processor = http_processor; + } + + template <class Body, typename Completion> + void issue(boost::beast::http::request<Body>&& request, + Completion&& completion) { + struct WorkImpl : Work { + HttpClient* http_client; + boost::beast::http::request<Body> request; + Completion completion; + + WorkImpl(HttpClient* http_client, + boost::beast::http::request<Body>&& request, + Completion&& completion) + : http_client(http_client), request(std::move(request)), + completion(std::move(completion)) { + } + WorkImpl(const WorkImpl&) = delete; + WorkImpl& operator=(const WorkImpl&) = delete; + + bool need_eof() const override { + return request.need_eof(); + } + + bool header_only() const override { + return (request.method() == boost::beast::http::verb::head); + } + + void complete(int r, Response&& response) override { + completion(r, std::move(response)); + } + + void operator()(boost::beast::tcp_stream& stream) override { + preprocess_request(); + + boost::beast::http::async_write( + stream, request, + [http_session=http_client->m_http_session.get(), + work=this->shared_from_this()] + (boost::beast::error_code ec, std::size_t) mutable { + http_session->handle_issue(ec, std::move(work)); + }); + } + + void operator()( + boost::beast::ssl_stream<boost::beast::tcp_stream>& stream) override { + preprocess_request(); + + boost::beast::http::async_write( + stream, request, + [http_session=http_client->m_http_session.get(), + work=this->shared_from_this()] + (boost::beast::error_code ec, std::size_t) mutable { + http_session->handle_issue(ec, std::move(work)); + }); + } + + void preprocess_request() { + if (http_client->m_http_processor) { + http_client->m_http_processor->process_request(request); + } + } + }; + + initialize_default_fields(request); + issue(std::make_shared<WorkImpl>(this, std::move(request), + std::move(completion))); + } + +private: + struct Work; + struct HttpSessionInterface { + virtual ~HttpSessionInterface() {} + + virtual void init(Context* on_finish) = 0; + virtual void shut_down(Context* on_finish) = 0; + + virtual void issue(std::shared_ptr<Work>&& work) = 0; + virtual void handle_issue(boost::system::error_code ec, + std::shared_ptr<Work>&& work) = 0; + }; + + struct Work : public std::enable_shared_from_this<Work> { + virtual ~Work() {} + virtual bool need_eof() const = 0; + virtual bool header_only() const = 0; + virtual void complete(int r, Response&&) = 0; + virtual void operator()(boost::beast::tcp_stream& stream) = 0; + virtual void operator()( + boost::beast::ssl_stream<boost::beast::tcp_stream>& stream) = 0; + }; + + template <typename D> struct HttpSession; + struct PlainHttpSession; + struct SslHttpSession; + + CephContext* m_cct; + ImageCtxT* m_image_ctx; + std::shared_ptr<AsioEngine> m_asio_engine; + std::string m_url; + + UrlSpec m_url_spec; + + bool m_ignore_self_signed_cert = false; + + HttpProcessorInterface* m_http_processor = nullptr; + + boost::asio::strand<boost::asio::io_context::executor_type> m_strand; + + boost::asio::ssl::context m_ssl_context; + std::unique_ptr<HttpSessionInterface> m_http_session; + + template <typename Fields> + void initialize_default_fields(Fields& fields) const { + fields.target(m_url_spec.path); + fields.set(boost::beast::http::field::host, m_url_spec.host); + fields.set(boost::beast::http::field::user_agent, + BOOST_BEAST_VERSION_STRING); + } + + void handle_get_size(int r, Response&& response, uint64_t* size, + Context* on_finish); + + void handle_read(int r, Response&& response, uint64_t byte_offset, + uint64_t byte_length, bufferlist* data, Context* on_finish); + + void issue(std::shared_ptr<Work>&& work); + + void create_http_session(Context* on_finish); + void shut_down_http_session(Context* on_finish); +}; + +} // namespace migration +} // namespace librbd + +extern template class librbd::migration::HttpClient<librbd::ImageCtx>; + +#endif // CEPH_LIBRBD_MIGRATION_HTTP_CLIENT_H diff --git a/src/librbd/migration/HttpProcessorInterface.h b/src/librbd/migration/HttpProcessorInterface.h new file mode 100644 index 000000000..3d9af88bd --- /dev/null +++ b/src/librbd/migration/HttpProcessorInterface.h @@ -0,0 +1,27 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_MIGRATION_HTTP_PROCESSOR_INTERFACE_H +#define CEPH_LIBRBD_MIGRATION_HTTP_PROCESSOR_INTERFACE_H + +#include <boost/beast/http/empty_body.hpp> +#include <boost/beast/http/message.hpp> + +namespace librbd { +namespace migration { + +struct HttpProcessorInterface { + using EmptyBody = boost::beast::http::empty_body; + using EmptyRequest = boost::beast::http::request<EmptyBody>; + + virtual ~HttpProcessorInterface() { + } + + virtual void process_request(EmptyRequest& request) = 0; + +}; + +} // namespace migration +} // namespace librbd + +#endif // CEPH_LIBRBD_MIGRATION_HTTP_PROCESSOR_INTERFACE_H diff --git a/src/librbd/migration/HttpStream.cc b/src/librbd/migration/HttpStream.cc new file mode 100644 index 000000000..fa3cc0032 --- /dev/null +++ b/src/librbd/migration/HttpStream.cc @@ -0,0 +1,83 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "librbd/migration/HttpStream.h" +#include "common/dout.h" +#include "common/errno.h" +#include "librbd/AsioEngine.h" +#include "librbd/ImageCtx.h" +#include "librbd/asio/Utils.h" +#include "librbd/migration/HttpClient.h" +#include <boost/beast/http.hpp> + +namespace librbd { +namespace migration { + +namespace { + +const std::string URL_KEY {"url"}; + +} // anonymous namespace + +#define dout_subsys ceph_subsys_rbd +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::HttpStream: " << this \ + << " " << __func__ << ": " + +template <typename I> +HttpStream<I>::HttpStream(I* image_ctx, const json_spirit::mObject& json_object) + : m_image_ctx(image_ctx), m_cct(image_ctx->cct), + m_asio_engine(image_ctx->asio_engine), m_json_object(json_object) { +} + +template <typename I> +HttpStream<I>::~HttpStream() { +} + +template <typename I> +void HttpStream<I>::open(Context* on_finish) { + auto& url_value = m_json_object[URL_KEY]; + if (url_value.type() != json_spirit::str_type) { + lderr(m_cct) << "failed to locate '" << URL_KEY << "' key" << dendl; + on_finish->complete(-EINVAL); + return; + } + + m_url = url_value.get_str(); + ldout(m_cct, 10) << "url=" << m_url << dendl; + + m_http_client.reset(HttpClient<I>::create(m_image_ctx, m_url)); + m_http_client->open(on_finish); +} + +template <typename I> +void HttpStream<I>::close(Context* on_finish) { + ldout(m_cct, 10) << dendl; + + if (!m_http_client) { + on_finish->complete(0); + return; + } + + m_http_client->close(on_finish); +} + +template <typename I> +void HttpStream<I>::get_size(uint64_t* size, Context* on_finish) { + ldout(m_cct, 10) << dendl; + + m_http_client->get_size(size, on_finish); +} + +template <typename I> +void HttpStream<I>::read(io::Extents&& byte_extents, bufferlist* data, + Context* on_finish) { + ldout(m_cct, 20) << "byte_extents=" << byte_extents << dendl; + + m_http_client->read(std::move(byte_extents), data, on_finish); +} + +} // namespace migration +} // namespace librbd + +template class librbd::migration::HttpStream<librbd::ImageCtx>; diff --git a/src/librbd/migration/HttpStream.h b/src/librbd/migration/HttpStream.h new file mode 100644 index 000000000..01a583714 --- /dev/null +++ b/src/librbd/migration/HttpStream.h @@ -0,0 +1,68 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_MIGRATION_HTTP_STREAM_H +#define CEPH_LIBRBD_MIGRATION_HTTP_STREAM_H + +#include "include/int_types.h" +#include "librbd/migration/StreamInterface.h" +#include <boost/beast/http/message.hpp> +#include <boost/beast/http/string_body.hpp> +#include <json_spirit/json_spirit.h> +#include <memory> +#include <string> + +struct Context; + +namespace librbd { + +struct AsioEngine; +struct ImageCtx; + +namespace migration { + +template <typename> class HttpClient; + +template <typename ImageCtxT> +class HttpStream : public StreamInterface { +public: + static HttpStream* create(ImageCtxT* image_ctx, + const json_spirit::mObject& json_object) { + return new HttpStream(image_ctx, json_object); + } + + HttpStream(ImageCtxT* image_ctx, const json_spirit::mObject& json_object); + ~HttpStream() override; + + HttpStream(const HttpStream&) = delete; + HttpStream& operator=(const HttpStream&) = delete; + + void open(Context* on_finish) override; + void close(Context* on_finish) override; + + void get_size(uint64_t* size, Context* on_finish) override; + + void read(io::Extents&& byte_extents, bufferlist* data, + Context* on_finish) override; + +private: + using HttpResponse = boost::beast::http::response< + boost::beast::http::string_body>; + + ImageCtxT* m_image_ctx; + CephContext* m_cct; + std::shared_ptr<AsioEngine> m_asio_engine; + json_spirit::mObject m_json_object; + + std::string m_url; + + std::unique_ptr<HttpClient<ImageCtxT>> m_http_client; + +}; + +} // namespace migration +} // namespace librbd + +extern template class librbd::migration::HttpStream<librbd::ImageCtx>; + +#endif // CEPH_LIBRBD_MIGRATION_HTTP_STREAM_H diff --git a/src/librbd/migration/ImageDispatch.cc b/src/librbd/migration/ImageDispatch.cc new file mode 100644 index 000000000..1ae143d78 --- /dev/null +++ b/src/librbd/migration/ImageDispatch.cc @@ -0,0 +1,156 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "librbd/migration/ImageDispatch.h" +#include "include/neorados/RADOS.hpp" +#include "common/dout.h" +#include "librbd/ImageCtx.h" +#include "librbd/io/AioCompletion.h" +#include "librbd/migration/FormatInterface.h" + +#define dout_subsys ceph_subsys_rbd +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::ImageDispatch: " << this \ + << " " << __func__ << ": " + +namespace librbd { +namespace migration { + +template <typename I> +ImageDispatch<I>::ImageDispatch(I* image_ctx, + std::unique_ptr<FormatInterface> format) + : m_image_ctx(image_ctx), m_format(std::move(format)) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << "ictx=" << image_ctx << dendl; +} + +template <typename I> +void ImageDispatch<I>::shut_down(Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + on_finish->complete(0); +} + +template <typename I> +bool ImageDispatch<I>::read( + io::AioCompletion* aio_comp, io::Extents &&image_extents, + io::ReadResult &&read_result, IOContext io_context, int op_flags, + int read_flags, const ZTracer::Trace &parent_trace, uint64_t tid, + std::atomic<uint32_t>* image_dispatch_flags, + io::DispatchResult* dispatch_result, Context** on_finish, + Context* on_dispatched) { + auto cct = m_image_ctx->cct; + ldout(cct, 20) << dendl; + + *dispatch_result = io::DISPATCH_RESULT_COMPLETE; + return m_format->read(aio_comp, io_context->read_snap().value_or(CEPH_NOSNAP), + std::move(image_extents), std::move(read_result), + op_flags, read_flags, parent_trace); +} + +template <typename I> +bool ImageDispatch<I>::write( + io::AioCompletion* aio_comp, io::Extents &&image_extents, bufferlist &&bl, + int op_flags, const ZTracer::Trace &parent_trace, + uint64_t tid, std::atomic<uint32_t>* image_dispatch_flags, + io::DispatchResult* dispatch_result, Context** on_finish, + Context* on_dispatched) { + auto cct = m_image_ctx->cct; + lderr(cct) << dendl; + + fail_io(-EROFS, aio_comp, dispatch_result); + return true; +} + +template <typename I> +bool ImageDispatch<I>::discard( + io::AioCompletion* aio_comp, io::Extents &&image_extents, + uint32_t discard_granularity_bytes, const ZTracer::Trace &parent_trace, + uint64_t tid, std::atomic<uint32_t>* image_dispatch_flags, + io::DispatchResult* dispatch_result, Context** on_finish, + Context* on_dispatched) { + auto cct = m_image_ctx->cct; + lderr(cct) << dendl; + + fail_io(-EROFS, aio_comp, dispatch_result); + return true; +} + +template <typename I> +bool ImageDispatch<I>::write_same( + io::AioCompletion* aio_comp, io::Extents &&image_extents, bufferlist &&bl, + int op_flags, const ZTracer::Trace &parent_trace, + uint64_t tid, std::atomic<uint32_t>* image_dispatch_flags, + io::DispatchResult* dispatch_result, Context** on_finish, + Context* on_dispatched) { + auto cct = m_image_ctx->cct; + lderr(cct) << dendl; + + fail_io(-EROFS, aio_comp, dispatch_result); + return true; +} + +template <typename I> +bool ImageDispatch<I>::compare_and_write( + io::AioCompletion* aio_comp, io::Extents &&image_extents, + bufferlist &&cmp_bl, bufferlist &&bl, uint64_t *mismatch_offset, + int op_flags, const ZTracer::Trace &parent_trace, + uint64_t tid, std::atomic<uint32_t>* image_dispatch_flags, + io::DispatchResult* dispatch_result, Context** on_finish, + Context* on_dispatched) { + auto cct = m_image_ctx->cct; + lderr(cct) << dendl; + + fail_io(-EROFS, aio_comp, dispatch_result); + return true; +} + +template <typename I> +bool ImageDispatch<I>::flush( + io::AioCompletion* aio_comp, io::FlushSource flush_source, + const ZTracer::Trace &parent_trace, uint64_t tid, + std::atomic<uint32_t>* image_dispatch_flags, + io::DispatchResult* dispatch_result, Context** on_finish, + Context* on_dispatched) { + auto cct = m_image_ctx->cct; + ldout(cct, 20) << dendl; + + *dispatch_result = io::DISPATCH_RESULT_COMPLETE; + aio_comp->set_request_count(0); + return true; +} + +template <typename I> +bool ImageDispatch<I>::list_snaps( + io::AioCompletion* aio_comp, io::Extents&& image_extents, + io::SnapIds&& snap_ids, int list_snaps_flags, + io::SnapshotDelta* snapshot_delta, const ZTracer::Trace &parent_trace, + uint64_t tid, std::atomic<uint32_t>* image_dispatch_flags, + io::DispatchResult* dispatch_result, Context** on_finish, + Context* on_dispatched) { + auto cct = m_image_ctx->cct; + ldout(cct, 20) << dendl; + + *dispatch_result = io::DISPATCH_RESULT_COMPLETE; + + aio_comp->set_request_count(1); + auto ctx = new io::C_AioRequest(aio_comp); + + m_format->list_snaps(std::move(image_extents), std::move(snap_ids), + list_snaps_flags, snapshot_delta, parent_trace, + ctx); + return true; +} + +template <typename I> +void ImageDispatch<I>::fail_io(int r, io::AioCompletion* aio_comp, + io::DispatchResult* dispatch_result) { + *dispatch_result = io::DISPATCH_RESULT_COMPLETE; + aio_comp->fail(r); +} + +} // namespace migration +} // namespace librbd + +template class librbd::migration::ImageDispatch<librbd::ImageCtx>; diff --git a/src/librbd/migration/ImageDispatch.h b/src/librbd/migration/ImageDispatch.h new file mode 100644 index 000000000..cd96141c0 --- /dev/null +++ b/src/librbd/migration/ImageDispatch.h @@ -0,0 +1,101 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_MIGRATION_IMAGE_DISPATCH_H +#define CEPH_LIBRBD_MIGRATION_IMAGE_DISPATCH_H + +#include "librbd/io/ImageDispatchInterface.h" +#include <memory> + +struct Context; + +namespace librbd { + +struct ImageCtx; + +namespace migration { + +struct FormatInterface; + +template <typename ImageCtxT> +class ImageDispatch : public io::ImageDispatchInterface { +public: + static ImageDispatch* create(ImageCtxT* image_ctx, + std::unique_ptr<FormatInterface> source) { + return new ImageDispatch(image_ctx, std::move(source)); + } + + ImageDispatch(ImageCtxT* image_ctx, std::unique_ptr<FormatInterface> source); + + void shut_down(Context* on_finish) override; + + io::ImageDispatchLayer get_dispatch_layer() const override { + return io::IMAGE_DISPATCH_LAYER_MIGRATION; + } + + bool read( + io::AioCompletion* aio_comp, io::Extents &&image_extents, + io::ReadResult &&read_result, IOContext io_context, int op_flags, + int read_flags, const ZTracer::Trace &parent_trace, uint64_t tid, + std::atomic<uint32_t>* image_dispatch_flags, + io::DispatchResult* dispatch_result, Context** on_finish, + Context* on_dispatched) override; + bool write( + io::AioCompletion* aio_comp, io::Extents &&image_extents, bufferlist &&bl, + int op_flags, const ZTracer::Trace &parent_trace, + uint64_t tid, std::atomic<uint32_t>* image_dispatch_flags, + io::DispatchResult* dispatch_result, Context** on_finish, + Context* on_dispatched) override; + bool discard( + io::AioCompletion* aio_comp, io::Extents &&image_extents, + uint32_t discard_granularity_bytes, const ZTracer::Trace &parent_trace, + uint64_t tid, std::atomic<uint32_t>* image_dispatch_flags, + io::DispatchResult* dispatch_result, Context** on_finish, + Context* on_dispatched) override; + bool write_same( + io::AioCompletion* aio_comp, io::Extents &&image_extents, bufferlist &&bl, + int op_flags, const ZTracer::Trace &parent_trace, + uint64_t tid, std::atomic<uint32_t>* image_dispatch_flags, + io::DispatchResult* dispatch_result, Context** on_finish, + Context* on_dispatched) override; + bool compare_and_write( + io::AioCompletion* aio_comp, io::Extents &&image_extents, + bufferlist &&cmp_bl, bufferlist &&bl, uint64_t *mismatch_offset, + int op_flags, const ZTracer::Trace &parent_trace, + uint64_t tid, std::atomic<uint32_t>* image_dispatch_flags, + io::DispatchResult* dispatch_result, Context** on_finish, + Context* on_dispatched) override; + bool flush( + io::AioCompletion* aio_comp, io::FlushSource flush_source, + const ZTracer::Trace &parent_trace, uint64_t tid, + std::atomic<uint32_t>* image_dispatch_flags, + io::DispatchResult* dispatch_result, Context** on_finish, + Context* on_dispatched) override; + + bool list_snaps( + io::AioCompletion* aio_comp, io::Extents&& image_extents, + io::SnapIds&& snap_ids, int list_snaps_flags, + io::SnapshotDelta* snapshot_delta, const ZTracer::Trace &parent_trace, + uint64_t tid, std::atomic<uint32_t>* image_dispatch_flags, + io::DispatchResult* dispatch_result, Context** on_finish, + Context* on_dispatched) override; + + bool invalidate_cache(Context* on_finish) override { + return false; + } + +private: + ImageCtxT* m_image_ctx; + std::unique_ptr<FormatInterface> m_format; + + void fail_io(int r, io::AioCompletion* aio_comp, + io::DispatchResult* dispatch_result); + +}; + +} // namespace migration +} // namespace librbd + +extern template class librbd::migration::ImageDispatch<librbd::ImageCtx>; + +#endif // CEPH_LIBRBD_MIGRATION_IMAGE_DISPATCH_H diff --git a/src/librbd/migration/NativeFormat.cc b/src/librbd/migration/NativeFormat.cc new file mode 100644 index 000000000..51248b95e --- /dev/null +++ b/src/librbd/migration/NativeFormat.cc @@ -0,0 +1,309 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "librbd/migration/NativeFormat.h" +#include "include/neorados/RADOS.hpp" +#include "common/dout.h" +#include "common/errno.h" +#include "librbd/ImageCtx.h" +#include "librbd/ImageState.h" +#include "librbd/Utils.h" +#include "librbd/asio/ContextWQ.h" +#include "librbd/io/ImageDispatchSpec.h" +#include "json_spirit/json_spirit.h" +#include "boost/lexical_cast.hpp" +#include <sstream> + +#define dout_subsys ceph_subsys_rbd +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::NativeFormat: " << this \ + << " " << __func__ << ": " + +namespace librbd { +namespace migration { + +namespace { + +const std::string TYPE_KEY{"type"}; +const std::string POOL_ID_KEY{"pool_id"}; +const std::string POOL_NAME_KEY{"pool_name"}; +const std::string POOL_NAMESPACE_KEY{"pool_namespace"}; +const std::string IMAGE_NAME_KEY{"image_name"}; +const std::string IMAGE_ID_KEY{"image_id"}; +const std::string SNAP_NAME_KEY{"snap_name"}; +const std::string SNAP_ID_KEY{"snap_id"}; + +} // anonymous namespace + +template <typename I> +std::string NativeFormat<I>::build_source_spec( + int64_t pool_id, const std::string& pool_namespace, + const std::string& image_name, const std::string& image_id) { + json_spirit::mObject source_spec; + source_spec[TYPE_KEY] = "native"; + source_spec[POOL_ID_KEY] = pool_id; + source_spec[POOL_NAMESPACE_KEY] = pool_namespace; + source_spec[IMAGE_NAME_KEY] = image_name; + if (!image_id.empty()) { + source_spec[IMAGE_ID_KEY] = image_id; + } + return json_spirit::write(source_spec); +} + +template <typename I> +NativeFormat<I>::NativeFormat( + I* image_ctx, const json_spirit::mObject& json_object, bool import_only) + : m_image_ctx(image_ctx), m_json_object(json_object), + m_import_only(import_only) { +} + +template <typename I> +void NativeFormat<I>::open(Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + auto& pool_name_val = m_json_object[POOL_NAME_KEY]; + if (pool_name_val.type() == json_spirit::str_type) { + librados::Rados rados(m_image_ctx->md_ctx); + librados::IoCtx io_ctx; + int r = rados.ioctx_create(pool_name_val.get_str().c_str(), io_ctx); + if (r < 0 ) { + lderr(cct) << "invalid pool name" << dendl; + on_finish->complete(r); + return; + } + + m_pool_id = io_ctx.get_id(); + } else if (pool_name_val.type() != json_spirit::null_type) { + lderr(cct) << "invalid pool name" << dendl; + on_finish->complete(-EINVAL); + return; + } + + auto& pool_id_val = m_json_object[POOL_ID_KEY]; + if (m_pool_id != -1 && pool_id_val.type() != json_spirit::null_type) { + lderr(cct) << "cannot specify both pool name and pool id" << dendl; + on_finish->complete(-EINVAL); + return; + } else if (pool_id_val.type() == json_spirit::int_type) { + m_pool_id = pool_id_val.get_int64(); + } else if (pool_id_val.type() == json_spirit::str_type) { + try { + m_pool_id = boost::lexical_cast<int64_t>(pool_id_val.get_str()); + } catch (boost::bad_lexical_cast &) { + } + } + + if (m_pool_id == -1) { + lderr(cct) << "missing or invalid pool id" << dendl; + on_finish->complete(-EINVAL); + return; + } + + auto& pool_namespace_val = m_json_object[POOL_NAMESPACE_KEY]; + if (pool_namespace_val.type() == json_spirit::str_type) { + m_pool_namespace = pool_namespace_val.get_str(); + } else if (pool_namespace_val.type() != json_spirit::null_type) { + lderr(cct) << "invalid pool namespace" << dendl; + on_finish->complete(-EINVAL); + return; + } + + auto& image_name_val = m_json_object[IMAGE_NAME_KEY]; + if (image_name_val.type() != json_spirit::str_type) { + lderr(cct) << "missing or invalid image name" << dendl; + on_finish->complete(-EINVAL); + return; + } + m_image_name = image_name_val.get_str(); + + auto& image_id_val = m_json_object[IMAGE_ID_KEY]; + if (image_id_val.type() == json_spirit::str_type) { + m_image_id = image_id_val.get_str(); + } else if (image_id_val.type() != json_spirit::null_type) { + lderr(cct) << "invalid image id" << dendl; + on_finish->complete(-EINVAL); + return; + } + + auto& snap_name_val = m_json_object[SNAP_NAME_KEY]; + if (snap_name_val.type() == json_spirit::str_type) { + m_snap_name = snap_name_val.get_str(); + } else if (snap_name_val.type() != json_spirit::null_type) { + lderr(cct) << "invalid snap name" << dendl; + on_finish->complete(-EINVAL); + return; + } + + auto& snap_id_val = m_json_object[SNAP_ID_KEY]; + if (!m_snap_name.empty() && snap_id_val.type() != json_spirit::null_type) { + lderr(cct) << "cannot specify both snap name and snap id" << dendl; + on_finish->complete(-EINVAL); + return; + } else if (snap_id_val.type() == json_spirit::str_type) { + try { + m_snap_id = boost::lexical_cast<uint64_t>(snap_id_val.get_str()); + } catch (boost::bad_lexical_cast &) { + } + } else if (snap_id_val.type() == json_spirit::int_type) { + m_snap_id = snap_id_val.get_uint64(); + } + + if (snap_id_val.type() != json_spirit::null_type && + m_snap_id == CEPH_NOSNAP) { + lderr(cct) << "invalid snap id" << dendl; + on_finish->complete(-EINVAL); + return; + } + + // snapshot is required for import to keep source read-only + if (m_import_only && m_snap_name.empty() && m_snap_id == CEPH_NOSNAP) { + lderr(cct) << "snapshot required for import" << dendl; + on_finish->complete(-EINVAL); + return; + } + + // TODO add support for external clusters + librados::IoCtx io_ctx; + int r = util::create_ioctx(m_image_ctx->md_ctx, "source image", + m_pool_id, m_pool_namespace, &io_ctx); + if (r < 0) { + on_finish->complete(r); + return; + } + + m_image_ctx->md_ctx.dup(io_ctx); + m_image_ctx->data_ctx.dup(io_ctx); + m_image_ctx->name = m_image_name; + + uint64_t flags = 0; + if (m_image_id.empty() && !m_import_only) { + flags |= OPEN_FLAG_OLD_FORMAT; + } else { + m_image_ctx->id = m_image_id; + } + + if (m_image_ctx->child != nullptr) { + // set rados flags for reading the parent image + if (m_image_ctx->child->config.template get_val<bool>("rbd_balance_parent_reads")) { + m_image_ctx->set_read_flag(librados::OPERATION_BALANCE_READS); + } else if (m_image_ctx->child->config.template get_val<bool>("rbd_localize_parent_reads")) { + m_image_ctx->set_read_flag(librados::OPERATION_LOCALIZE_READS); + } + } + + // open the source RBD image + on_finish = new LambdaContext([this, on_finish](int r) { + handle_open(r, on_finish); }); + m_image_ctx->state->open(flags, on_finish); +} + +template <typename I> +void NativeFormat<I>::handle_open(int r, Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << "r=" << r << dendl; + + if (r < 0) { + lderr(cct) << "failed to open image: " << cpp_strerror(r) << dendl; + on_finish->complete(r); + return; + } + + if (m_snap_id == CEPH_NOSNAP && m_snap_name.empty()) { + on_finish->complete(0); + return; + } + + if (!m_snap_name.empty()) { + std::shared_lock image_locker{m_image_ctx->image_lock}; + m_snap_id = m_image_ctx->get_snap_id(cls::rbd::UserSnapshotNamespace{}, + m_snap_name); + } + + if (m_snap_id == CEPH_NOSNAP) { + lderr(cct) << "failed to locate snapshot " << m_snap_name << dendl; + on_finish = new LambdaContext([on_finish](int) { + on_finish->complete(-ENOENT); }); + m_image_ctx->state->close(on_finish); + return; + } + + on_finish = new LambdaContext([this, on_finish](int r) { + handle_snap_set(r, on_finish); }); + m_image_ctx->state->snap_set(m_snap_id, on_finish); +} + +template <typename I> +void NativeFormat<I>::handle_snap_set(int r, Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << "r=" << r << dendl; + + if (r < 0) { + lderr(cct) << "failed to set snapshot " << m_snap_id << ": " + << cpp_strerror(r) << dendl; + on_finish = new LambdaContext([r, on_finish](int) { + on_finish->complete(r); }); + m_image_ctx->state->close(on_finish); + return; + } + + on_finish->complete(0); +} + +template <typename I> +void NativeFormat<I>::close(Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + // the native librbd::image::CloseRequest handles all cleanup + on_finish->complete(0); +} + +template <typename I> +void NativeFormat<I>::get_snapshots(SnapInfos* snap_infos, Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + m_image_ctx->image_lock.lock_shared(); + *snap_infos = m_image_ctx->snap_info; + m_image_ctx->image_lock.unlock_shared(); + + on_finish->complete(0); +} + +template <typename I> +void NativeFormat<I>::get_image_size(uint64_t snap_id, uint64_t* size, + Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + m_image_ctx->image_lock.lock_shared(); + *size = m_image_ctx->get_image_size(snap_id); + m_image_ctx->image_lock.unlock_shared(); + + + on_finish->complete(0); +} + +template <typename I> +void NativeFormat<I>::list_snaps(io::Extents&& image_extents, + io::SnapIds&& snap_ids, int list_snaps_flags, + io::SnapshotDelta* snapshot_delta, + const ZTracer::Trace &parent_trace, + Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 20) << "image_extents=" << image_extents << dendl; + + auto aio_comp = io::AioCompletion::create_and_start( + on_finish, util::get_image_ctx(m_image_ctx), io::AIO_TYPE_GENERIC); + auto req = io::ImageDispatchSpec::create_list_snaps( + *m_image_ctx, io::IMAGE_DISPATCH_LAYER_MIGRATION, aio_comp, + std::move(image_extents), io::ImageArea::DATA, std::move(snap_ids), + list_snaps_flags, snapshot_delta, {}); + req->send(); +} + +} // namespace migration +} // namespace librbd + +template class librbd::migration::NativeFormat<librbd::ImageCtx>; diff --git a/src/librbd/migration/NativeFormat.h b/src/librbd/migration/NativeFormat.h new file mode 100644 index 000000000..e58c04121 --- /dev/null +++ b/src/librbd/migration/NativeFormat.h @@ -0,0 +1,82 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_MIGRATION_NATIVE_FORMAT_H +#define CEPH_LIBRBD_MIGRATION_NATIVE_FORMAT_H + +#include "include/int_types.h" +#include "librbd/Types.h" +#include "librbd/migration/FormatInterface.h" +#include "json_spirit/json_spirit.h" +#include <memory> + +struct Context; + +namespace librbd { + +struct AsioEngine; +struct ImageCtx; + +namespace migration { + +template <typename ImageCtxT> +class NativeFormat : public FormatInterface { +public: + static std::string build_source_spec(int64_t pool_id, + const std::string& pool_namespace, + const std::string& image_name, + const std::string& image_id); + + static NativeFormat* create(ImageCtxT* image_ctx, + const json_spirit::mObject& json_object, + bool import_only) { + return new NativeFormat(image_ctx, json_object, import_only); + } + + NativeFormat(ImageCtxT* image_ctx, const json_spirit::mObject& json_object, + bool import_only); + NativeFormat(const NativeFormat&) = delete; + NativeFormat& operator=(const NativeFormat&) = delete; + + void open(Context* on_finish) override; + void close(Context* on_finish) override; + + void get_snapshots(SnapInfos* snap_infos, Context* on_finish) override; + void get_image_size(uint64_t snap_id, uint64_t* size, + Context* on_finish) override; + + bool read(io::AioCompletion* aio_comp, uint64_t snap_id, + io::Extents&& image_extents, io::ReadResult&& read_result, + int op_flags, int read_flags, + const ZTracer::Trace &parent_trace) override { + return false; + } + + void list_snaps(io::Extents&& image_extents, io::SnapIds&& snap_ids, + int list_snaps_flags, io::SnapshotDelta* snapshot_delta, + const ZTracer::Trace &parent_trace, + Context* on_finish) override; + +private: + ImageCtxT* m_image_ctx; + json_spirit::mObject m_json_object; + bool m_import_only; + + int64_t m_pool_id = -1; + std::string m_pool_namespace; + std::string m_image_name; + std::string m_image_id; + std::string m_snap_name; + uint64_t m_snap_id = CEPH_NOSNAP; + + void handle_open(int r, Context* on_finish); + void handle_snap_set(int r, Context* on_finish); + +}; + +} // namespace migration +} // namespace librbd + +extern template class librbd::migration::NativeFormat<librbd::ImageCtx>; + +#endif // CEPH_LIBRBD_MIGRATION_NATIVE_FORMAT_H diff --git a/src/librbd/migration/OpenSourceImageRequest.cc b/src/librbd/migration/OpenSourceImageRequest.cc new file mode 100644 index 000000000..8abdedf33 --- /dev/null +++ b/src/librbd/migration/OpenSourceImageRequest.cc @@ -0,0 +1,249 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "librbd/migration/OpenSourceImageRequest.h" +#include "common/dout.h" +#include "common/errno.h" +#include "librbd/ImageCtx.h" +#include "librbd/ImageState.h" +#include "librbd/Utils.h" +#include "librbd/io/ImageDispatcher.h" +#include "librbd/migration/ImageDispatch.h" +#include "librbd/migration/NativeFormat.h" +#include "librbd/migration/SourceSpecBuilder.h" + +#define dout_subsys ceph_subsys_rbd +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::OpenSourceImageRequest: " \ + << this << " " << __func__ << ": " + +namespace librbd { +namespace migration { + +template <typename I> +OpenSourceImageRequest<I>::OpenSourceImageRequest( + librados::IoCtx& io_ctx, I* dst_image_ctx, uint64_t src_snap_id, + const MigrationInfo &migration_info, I** src_image_ctx, Context* on_finish) + : m_cct(reinterpret_cast<CephContext*>(io_ctx.cct())), m_io_ctx(io_ctx), + m_dst_image_ctx(dst_image_ctx), m_src_snap_id(src_snap_id), + m_migration_info(migration_info), m_src_image_ctx(src_image_ctx), + m_on_finish(on_finish) { + ldout(m_cct, 10) << dendl; +} + +template <typename I> +void OpenSourceImageRequest<I>::send() { + open_source(); +} + +template <typename I> +void OpenSourceImageRequest<I>::open_source() { + ldout(m_cct, 10) << dendl; + + // note that all source image ctx properties are placeholders + *m_src_image_ctx = I::create("", "", CEPH_NOSNAP, m_io_ctx, true); + auto src_image_ctx = *m_src_image_ctx; + src_image_ctx->child = m_dst_image_ctx; + + // use default layout values (can be overridden by source layers later) + src_image_ctx->order = 22; + src_image_ctx->layout = file_layout_t(); + src_image_ctx->layout.stripe_count = 1; + src_image_ctx->layout.stripe_unit = 1ULL << src_image_ctx->order; + src_image_ctx->layout.object_size = 1Ull << src_image_ctx->order; + src_image_ctx->layout.pool_id = -1; + + bool import_only = true; + auto source_spec = m_migration_info.source_spec; + if (source_spec.empty()) { + // implies legacy migration from RBD image in same cluster + source_spec = NativeFormat<I>::build_source_spec( + m_migration_info.pool_id, m_migration_info.pool_namespace, + m_migration_info.image_name, m_migration_info.image_id); + import_only = false; + } + + ldout(m_cct, 15) << "source_spec=" << source_spec << ", " + << "source_snap_id=" << m_src_snap_id << ", " + << "import_only=" << import_only << dendl; + + SourceSpecBuilder<I> source_spec_builder{src_image_ctx}; + json_spirit::mObject source_spec_object; + int r = source_spec_builder.parse_source_spec(source_spec, + &source_spec_object); + if (r < 0) { + lderr(m_cct) << "failed to parse migration source-spec:" << cpp_strerror(r) + << dendl; + (*m_src_image_ctx)->state->close(); + finish(r); + return; + } + + r = source_spec_builder.build_format(source_spec_object, import_only, + &m_format); + if (r < 0) { + lderr(m_cct) << "failed to build migration format handler: " + << cpp_strerror(r) << dendl; + (*m_src_image_ctx)->state->close(); + finish(r); + return; + } + + auto ctx = util::create_context_callback< + OpenSourceImageRequest<I>, + &OpenSourceImageRequest<I>::handle_open_source>(this); + m_format->open(ctx); +} + +template <typename I> +void OpenSourceImageRequest<I>::handle_open_source(int r) { + ldout(m_cct, 10) << "r=" << r << dendl; + + if (r < 0) { + lderr(m_cct) << "failed to open migration source: " << cpp_strerror(r) + << dendl; + finish(r); + return; + } + + get_image_size(); +} + +template <typename I> +void OpenSourceImageRequest<I>::get_image_size() { + ldout(m_cct, 10) << dendl; + + auto ctx = util::create_context_callback< + OpenSourceImageRequest<I>, + &OpenSourceImageRequest<I>::handle_get_image_size>(this); + m_format->get_image_size(CEPH_NOSNAP, &m_image_size, ctx); +} + +template <typename I> +void OpenSourceImageRequest<I>::handle_get_image_size(int r) { + ldout(m_cct, 10) << "r=" << r << ", " + << "image_size=" << m_image_size << dendl; + + if (r < 0) { + lderr(m_cct) << "failed to retrieve image size: " << cpp_strerror(r) + << dendl; + close_image(r); + return; + } + + auto src_image_ctx = *m_src_image_ctx; + src_image_ctx->image_lock.lock(); + src_image_ctx->size = m_image_size; + src_image_ctx->image_lock.unlock(); + + get_snapshots(); +} + +template <typename I> +void OpenSourceImageRequest<I>::get_snapshots() { + ldout(m_cct, 10) << dendl; + + auto ctx = util::create_context_callback< + OpenSourceImageRequest<I>, + &OpenSourceImageRequest<I>::handle_get_snapshots>(this); + m_format->get_snapshots(&m_snap_infos, ctx); +} + +template <typename I> +void OpenSourceImageRequest<I>::handle_get_snapshots(int r) { + ldout(m_cct, 10) << "r=" << r << dendl; + + if (r < 0) { + lderr(m_cct) << "failed to retrieve snapshots: " << cpp_strerror(r) + << dendl; + close_image(r); + return; + } + + // copy snapshot metadata to image ctx + auto src_image_ctx = *m_src_image_ctx; + src_image_ctx->image_lock.lock(); + + src_image_ctx->snaps.clear(); + src_image_ctx->snap_info.clear(); + src_image_ctx->snap_ids.clear(); + + ::SnapContext snapc; + for (auto it = m_snap_infos.rbegin(); it != m_snap_infos.rend(); ++it) { + auto& [snap_id, snap_info] = *it; + snapc.snaps.push_back(snap_id); + + ldout(m_cct, 10) << "adding snap: ns=" << snap_info.snap_namespace << ", " + << "name=" << snap_info.name << ", " + << "id=" << snap_id << dendl; + src_image_ctx->add_snap( + snap_info.snap_namespace, snap_info.name, snap_id, + snap_info.size, snap_info.parent, snap_info.protection_status, + snap_info.flags, snap_info.timestamp); + } + if (!snapc.snaps.empty()) { + snapc.seq = snapc.snaps[0]; + } + src_image_ctx->snapc = snapc; + + ldout(m_cct, 15) << "read snap id: " << m_src_snap_id << ", " + << "write snapc={" + << "seq=" << snapc.seq << ", " + << "snaps=" << snapc.snaps << "}" << dendl; + + // ensure data_ctx and data_io_context are pointing to correct snapshot + if (m_src_snap_id != CEPH_NOSNAP) { + int r = src_image_ctx->snap_set(m_src_snap_id); + if (r < 0) { + src_image_ctx->image_lock.unlock(); + + lderr(m_cct) << "error setting source image snap id: " + << cpp_strerror(r) << dendl; + finish(r); + return; + } + } + + src_image_ctx->image_lock.unlock(); + + finish(0); +} + +template <typename I> +void OpenSourceImageRequest<I>::close_image(int r) { + ldout(m_cct, 10) << "r=" << r << dendl; + + auto ctx = new LambdaContext([this, r](int) { + finish(r); + }); + (*m_src_image_ctx)->state->close(ctx); +} + +template <typename I> +void OpenSourceImageRequest<I>::register_image_dispatch() { + ldout(m_cct, 10) << dendl; + + // intercept any IO requests to the source image + auto io_image_dispatch = ImageDispatch<I>::create( + *m_src_image_ctx, std::move(m_format)); + (*m_src_image_ctx)->io_image_dispatcher->register_dispatch(io_image_dispatch); +} + +template <typename I> +void OpenSourceImageRequest<I>::finish(int r) { + ldout(m_cct, 10) << "r=" << r << dendl; + + if (r < 0) { + *m_src_image_ctx = nullptr; + } else { + register_image_dispatch(); + } + + m_on_finish->complete(r); + delete this; +} + +} // namespace migration +} // namespace librbd + +template class librbd::migration::OpenSourceImageRequest<librbd::ImageCtx>; diff --git a/src/librbd/migration/OpenSourceImageRequest.h b/src/librbd/migration/OpenSourceImageRequest.h new file mode 100644 index 000000000..f0dab3ad9 --- /dev/null +++ b/src/librbd/migration/OpenSourceImageRequest.h @@ -0,0 +1,103 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_MIGRATION_OPEN_SOURCE_IMAGE_REQUEST_H +#define CEPH_LIBRBD_MIGRATION_OPEN_SOURCE_IMAGE_REQUEST_H + +#include "include/rados/librados_fwd.hpp" +#include "librbd/Types.h" +#include <map> +#include <memory> + +struct Context; + +namespace librbd { + +struct ImageCtx; + +namespace migration { + +struct FormatInterface; + +template <typename ImageCtxT> +class OpenSourceImageRequest { +public: + static OpenSourceImageRequest* create(librados::IoCtx& io_ctx, + ImageCtxT* destination_image_ctx, + uint64_t src_snap_id, + const MigrationInfo &migration_info, + ImageCtxT** source_image_ctx, + Context* on_finish) { + return new OpenSourceImageRequest(io_ctx, destination_image_ctx, + src_snap_id, migration_info, + source_image_ctx, on_finish); + } + + OpenSourceImageRequest(librados::IoCtx& io_ctx, + ImageCtxT* destination_image_ctx, + uint64_t src_snap_id, + const MigrationInfo &migration_info, + ImageCtxT** source_image_ctx, + Context* on_finish); + + void send(); + +private: + /** + * @verbatim + * + * <start> + * | + * v + * OPEN_SOURCE + * | + * v + * GET_IMAGE_SIZE * * * * * * * + * | * + * v v + * GET_SNAPSHOTS * * * * > CLOSE_IMAGE + * | | + * v | + * <finish> <------------------/ + * + * @endverbatim + */ + + typedef std::map<uint64_t, SnapInfo> SnapInfos; + + CephContext* m_cct; + librados::IoCtx& m_io_ctx; + ImageCtxT* m_dst_image_ctx; + uint64_t m_src_snap_id; + MigrationInfo m_migration_info; + ImageCtxT** m_src_image_ctx; + Context* m_on_finish; + + std::unique_ptr<FormatInterface> m_format; + + uint64_t m_image_size = 0; + SnapInfos m_snap_infos; + + void open_source(); + void handle_open_source(int r); + + void get_image_size(); + void handle_get_image_size(int r); + + void get_snapshots(); + void handle_get_snapshots(int r); + + void close_image(int r); + + void register_image_dispatch(); + + void finish(int r); + +}; + +} // namespace migration +} // namespace librbd + +extern template class librbd::migration::OpenSourceImageRequest<librbd::ImageCtx>; + +#endif // CEPH_LIBRBD_MIGRATION_OPEN_SOURCE_IMAGE_REQUEST_H diff --git a/src/librbd/migration/QCOW.h b/src/librbd/migration/QCOW.h new file mode 100644 index 000000000..23401e515 --- /dev/null +++ b/src/librbd/migration/QCOW.h @@ -0,0 +1,466 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +/* Based on QEMU block/qcow.cc and block/qcow2.h, which has this license: */ + +/* + * Block driver for the QCOW version 2 format + * + * Copyright (c) 2004-2006 Fabrice Bellard + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL + * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +#ifndef CEPH_LIBRBD_MIGRATION_QCOW2_H +#define CEPH_LIBRBD_MIGRATION_QCOW2_H + +#include "include/ceph_assert.h" +#include "include/int_types.h" +#include "librbd/migration/QCOW.h" + +#define QCOW_MAGIC (('Q' << 24) | ('F' << 16) | ('I' << 8) | 0xfb) + +#define QCOW_CRYPT_NONE 0 +#define QCOW_CRYPT_AES 1 +#define QCOW_CRYPT_LUKS 2 + +#define QCOW_MAX_CRYPT_CLUSTERS 32 +#define QCOW_MAX_SNAPSHOTS 65536 + +/* Field widths in qcow2 mean normal cluster offsets cannot reach + * 64PB; depending on cluster size, compressed clusters can have a + * smaller limit (64PB for up to 16k clusters, then ramps down to + * 512TB for 2M clusters). */ +#define QCOW_MAX_CLUSTER_OFFSET ((1ULL << 56) - 1) + +/* 8 MB refcount table is enough for 2 PB images at 64k cluster size + * (128 GB for 512 byte clusters, 2 EB for 2 MB clusters) */ +#define QCOW_MAX_REFTABLE_SIZE (1ULL << 23) + +/* 32 MB L1 table is enough for 2 PB images at 64k cluster size + * (128 GB for 512 byte clusters, 2 EB for 2 MB clusters) */ +#define QCOW_MAX_L1_SIZE (1ULL << 25) + +/* Allow for an average of 1k per snapshot table entry, should be plenty of + * space for snapshot names and IDs */ +#define QCOW_MAX_SNAPSHOTS_SIZE (1024 * QCOW_MAX_SNAPSHOTS) + +/* Maximum amount of extra data per snapshot table entry to accept */ +#define QCOW_MAX_SNAPSHOT_EXTRA_DATA 1024 + +/* Bitmap header extension constraints */ +#define QCOW2_MAX_BITMAPS 65535 +#define QCOW2_MAX_BITMAP_DIRECTORY_SIZE (1024 * QCOW2_MAX_BITMAPS) + +/* Maximum of parallel sub-request per guest request */ +#define QCOW2_MAX_WORKERS 8 + +/* indicate that the refcount of the referenced cluster is exactly one. */ +#define QCOW_OFLAG_COPIED (1ULL << 63) +/* indicate that the cluster is compressed (they never have the copied flag) */ +#define QCOW_OFLAG_COMPRESSED (1ULL << 62) +/* The cluster reads as all zeros */ +#define QCOW_OFLAG_ZERO (1ULL << 0) + +#define QCOW_EXTL2_SUBCLUSTERS_PER_CLUSTER 32 + +/* The subcluster X [0..31] is allocated */ +#define QCOW_OFLAG_SUB_ALLOC(X) (1ULL << (X)) +/* The subcluster X [0..31] reads as zeroes */ +#define QCOW_OFLAG_SUB_ZERO(X) (QCOW_OFLAG_SUB_ALLOC(X) << 32) +/* Subclusters [X, Y) (0 <= X <= Y <= 32) are allocated */ +#define QCOW_OFLAG_SUB_ALLOC_RANGE(X, Y) \ + (QCOW_OFLAG_SUB_ALLOC(Y) - QCOW_OFLAG_SUB_ALLOC(X)) +/* Subclusters [X, Y) (0 <= X <= Y <= 32) read as zeroes */ +#define QCOW_OFLAG_SUB_ZERO_RANGE(X, Y) \ + (QCOW_OFLAG_SUB_ALLOC_RANGE(X, Y) << 32) +/* L2 entry bitmap with all allocation bits set */ +#define QCOW_L2_BITMAP_ALL_ALLOC (QCOW_OFLAG_SUB_ALLOC_RANGE(0, 32)) +/* L2 entry bitmap with all "read as zeroes" bits set */ +#define QCOW_L2_BITMAP_ALL_ZEROES (QCOW_OFLAG_SUB_ZERO_RANGE(0, 32)) + +/* Size of normal and extended L2 entries */ +#define QCOW_L2E_SIZE_NORMAL (sizeof(uint64_t)) +#define QCOW_L2E_SIZE_EXTENDED (sizeof(uint64_t) * 2) + +/* Size of L1 table entries */ +#define QCOW_L1E_SIZE (sizeof(uint64_t)) + +/* Size of reftable entries */ +#define QCOW_REFTABLE_ENTRY_SIZE (sizeof(uint64_t)) + +#define QCOW_MIN_CLUSTER_BITS 9 +#define QCOW_MAX_CLUSTER_BITS 21 + +/* Defined in the qcow2 spec (compressed cluster descriptor) */ +#define QCOW2_COMPRESSED_SECTOR_SIZE 512U +#define QCOW2_COMPRESSED_SECTOR_MASK (~(QCOW2_COMPRESSED_SECTOR_SIZE - 1ULL)) + +#define QCOW_L2_CACHE_SIZE 16 + +/* Must be at least 2 to cover COW */ +#define QCOW_MIN_L2_CACHE_SIZE 2 /* cache entries */ + +/* Must be at least 4 to cover all cases of refcount table growth */ +#define QCOW_MIN_REFCOUNT_CACHE_SIZE 4 /* clusters */ + +#define QCOW_DEFAULT_L2_CACHE_MAX_SIZE (1ULL << 25) +#define QCOW_DEFAULT_CACHE_CLEAN_INTERVAL 600 /* seconds */ + +#define QCOW_DEFAULT_CLUSTER_SIZE 65536 + +#define QCOW2_OPT_DATA_FILE "data-file" +#define QCOW2_OPT_LAZY_REFCOUNTS "lazy-refcounts" +#define QCOW2_OPT_DISCARD_REQUEST "pass-discard-request" +#define QCOW2_OPT_DISCARD_SNAPSHOT "pass-discard-snapshot" +#define QCOW2_OPT_DISCARD_OTHER "pass-discard-other" +#define QCOW2_OPT_OVERLAP "overlap-check" +#define QCOW2_OPT_OVERLAP_TEMPLATE "overlap-check.template" +#define QCOW2_OPT_OVERLAP_MAIN_HEADER "overlap-check.main-header" +#define QCOW2_OPT_OVERLAP_ACTIVE_L1 "overlap-check.active-l1" +#define QCOW2_OPT_OVERLAP_ACTIVE_L2 "overlap-check.active-l2" +#define QCOW2_OPT_OVERLAP_REFCOUNT_TABLE "overlap-check.refcount-table" +#define QCOW2_OPT_OVERLAP_REFCOUNT_BLOCK "overlap-check.refcount-block" +#define QCOW2_OPT_OVERLAP_SNAPSHOT_TABLE "overlap-check.snapshot-table" +#define QCOW2_OPT_OVERLAP_INACTIVE_L1 "overlap-check.inactive-l1" +#define QCOW2_OPT_OVERLAP_INACTIVE_L2 "overlap-check.inactive-l2" +#define QCOW2_OPT_OVERLAP_BITMAP_DIRECTORY "overlap-check.bitmap-directory" +#define QCOW2_OPT_CACHE_SIZE "cache-size" +#define QCOW2_OPT_L2_CACHE_SIZE "l2-cache-size" +#define QCOW2_OPT_L2_CACHE_ENTRY_SIZE "l2-cache-entry-size" +#define QCOW2_OPT_REFCOUNT_CACHE_SIZE "refcount-cache-size" +#define QCOW2_OPT_CACHE_CLEAN_INTERVAL "cache-clean-interval" + +typedef struct QCowHeaderProbe { + uint32_t magic; + uint32_t version; +} __attribute__((__packed__)) QCowHeaderProbe; + +typedef struct QCowHeaderV1 +{ + uint32_t magic; + uint32_t version; + uint64_t backing_file_offset; + uint32_t backing_file_size; + uint32_t mtime; + uint64_t size; /* in bytes */ + uint8_t cluster_bits; + uint8_t l2_bits; + uint16_t padding; + uint32_t crypt_method; + uint64_t l1_table_offset; +} __attribute__((__packed__)) QCowHeaderV1; + +typedef struct QCowHeader { + uint32_t magic; + uint32_t version; + uint64_t backing_file_offset; + uint32_t backing_file_size; + uint32_t cluster_bits; + uint64_t size; /* in bytes */ + uint32_t crypt_method; + uint32_t l1_size; /* XXX: save number of clusters instead ? */ + uint64_t l1_table_offset; + uint64_t refcount_table_offset; + uint32_t refcount_table_clusters; + uint32_t nb_snapshots; + uint64_t snapshots_offset; + + /* The following fields are only valid for version >= 3 */ + uint64_t incompatible_features; + uint64_t compatible_features; + uint64_t autoclear_features; + + uint32_t refcount_order; + uint32_t header_length; + + /* Additional fields */ + uint8_t compression_type; + + /* header must be a multiple of 8 */ + uint8_t padding[7]; +} __attribute__((__packed__)) QCowHeader; + +typedef struct QCowSnapshotHeader { + /* header is 8 byte aligned */ + uint64_t l1_table_offset; + + uint32_t l1_size; + uint16_t id_str_size; + uint16_t name_size; + + uint32_t date_sec; + uint32_t date_nsec; + + uint64_t vm_clock_nsec; + + uint32_t vm_state_size; + uint32_t extra_data_size; /* for extension */ + /* extra data follows */ + /* id_str follows */ + /* name follows */ +} __attribute__((__packed__)) QCowSnapshotHeader; + +typedef struct QCowSnapshotExtraData { + uint64_t vm_state_size_large; + uint64_t disk_size; + uint64_t icount; +} __attribute__((__packed__)) QCowSnapshotExtraData; + + +typedef struct QCowSnapshot { + uint64_t l1_table_offset; + uint32_t l1_size; + char *id_str; + char *name; + uint64_t disk_size; + uint64_t vm_state_size; + uint32_t date_sec; + uint32_t date_nsec; + uint64_t vm_clock_nsec; + /* icount value for the moment when snapshot was taken */ + uint64_t icount; + /* Size of all extra data, including QCowSnapshotExtraData if available */ + uint32_t extra_data_size; + /* Data beyond QCowSnapshotExtraData, if any */ + void *unknown_extra_data; +} QCowSnapshot; + +typedef struct Qcow2CryptoHeaderExtension { + uint64_t offset; + uint64_t length; +} __attribute__((__packed__)) Qcow2CryptoHeaderExtension; + +typedef struct Qcow2UnknownHeaderExtension { + uint32_t magic; + uint32_t len; + uint8_t data[]; +} Qcow2UnknownHeaderExtension; + +enum { + QCOW2_FEAT_TYPE_INCOMPATIBLE = 0, + QCOW2_FEAT_TYPE_COMPATIBLE = 1, + QCOW2_FEAT_TYPE_AUTOCLEAR = 2, +}; + +/* Incompatible feature bits */ +enum { + QCOW2_INCOMPAT_DIRTY_BITNR = 0, + QCOW2_INCOMPAT_CORRUPT_BITNR = 1, + QCOW2_INCOMPAT_DATA_FILE_BITNR = 2, + QCOW2_INCOMPAT_COMPRESSION_BITNR = 3, + QCOW2_INCOMPAT_EXTL2_BITNR = 4, + QCOW2_INCOMPAT_DIRTY = 1 << QCOW2_INCOMPAT_DIRTY_BITNR, + QCOW2_INCOMPAT_CORRUPT = 1 << QCOW2_INCOMPAT_CORRUPT_BITNR, + QCOW2_INCOMPAT_DATA_FILE = 1 << QCOW2_INCOMPAT_DATA_FILE_BITNR, + QCOW2_INCOMPAT_COMPRESSION = 1 << QCOW2_INCOMPAT_COMPRESSION_BITNR, + QCOW2_INCOMPAT_EXTL2 = 1 << QCOW2_INCOMPAT_EXTL2_BITNR, + + QCOW2_INCOMPAT_MASK = QCOW2_INCOMPAT_DIRTY + | QCOW2_INCOMPAT_CORRUPT + | QCOW2_INCOMPAT_DATA_FILE + | QCOW2_INCOMPAT_COMPRESSION + | QCOW2_INCOMPAT_EXTL2, +}; + +/* Compatible feature bits */ +enum { + QCOW2_COMPAT_LAZY_REFCOUNTS_BITNR = 0, + QCOW2_COMPAT_LAZY_REFCOUNTS = 1 << QCOW2_COMPAT_LAZY_REFCOUNTS_BITNR, + + QCOW2_COMPAT_FEAT_MASK = QCOW2_COMPAT_LAZY_REFCOUNTS, +}; + +/* Autoclear feature bits */ +enum { + QCOW2_AUTOCLEAR_BITMAPS_BITNR = 0, + QCOW2_AUTOCLEAR_DATA_FILE_RAW_BITNR = 1, + QCOW2_AUTOCLEAR_BITMAPS = 1 << QCOW2_AUTOCLEAR_BITMAPS_BITNR, + QCOW2_AUTOCLEAR_DATA_FILE_RAW = 1 << QCOW2_AUTOCLEAR_DATA_FILE_RAW_BITNR, + + QCOW2_AUTOCLEAR_MASK = QCOW2_AUTOCLEAR_BITMAPS + | QCOW2_AUTOCLEAR_DATA_FILE_RAW, +}; + +enum qcow2_discard_type { + QCOW2_DISCARD_NEVER = 0, + QCOW2_DISCARD_ALWAYS, + QCOW2_DISCARD_REQUEST, + QCOW2_DISCARD_SNAPSHOT, + QCOW2_DISCARD_OTHER, + QCOW2_DISCARD_MAX +}; + +typedef struct Qcow2Feature { + uint8_t type; + uint8_t bit; + char name[46]; +} __attribute__((__packed__)) Qcow2Feature; + +typedef struct Qcow2DiscardRegion { + uint64_t offset; + uint64_t bytes; +} Qcow2DiscardRegion; + +typedef uint64_t Qcow2GetRefcountFunc(const void *refcount_array, + uint64_t index); +typedef void Qcow2SetRefcountFunc(void *refcount_array, + uint64_t index, uint64_t value); + +typedef struct Qcow2BitmapHeaderExt { + uint32_t nb_bitmaps; + uint32_t reserved32; + uint64_t bitmap_directory_size; + uint64_t bitmap_directory_offset; +} __attribute__((__packed__)) Qcow2BitmapHeaderExt; + +#define QCOW_RC_CACHE_SIZE QCOW_L2_CACHE_SIZE; + +typedef struct Qcow2COWRegion { + /** + * Offset of the COW region in bytes from the start of the first cluster + * touched by the request. + */ + unsigned offset; + + /** Number of bytes to copy */ + unsigned nb_bytes; +} Qcow2COWRegion; + +/** + * Describes an in-flight (part of a) write request that writes to clusters + * that are not referenced in their L2 table yet. + */ +typedef struct QCowL2Meta +{ + /** Guest offset of the first newly allocated cluster */ + uint64_t offset; + + /** Host offset of the first newly allocated cluster */ + uint64_t alloc_offset; + + /** Number of newly allocated clusters */ + int nb_clusters; + + /** Do not free the old clusters */ + bool keep_old_clusters; + + /** + * The COW Region between the start of the first allocated cluster and the + * area the guest actually writes to. + */ + Qcow2COWRegion cow_start; + + /** + * The COW Region between the area the guest actually writes to and the + * end of the last allocated cluster. + */ + Qcow2COWRegion cow_end; + + /* + * Indicates that COW regions are already handled and do not require + * any more processing. + */ + bool skip_cow; + + /** + * Indicates that this is not a normal write request but a preallocation. + * If the image has extended L2 entries this means that no new individual + * subclusters will be marked as allocated in the L2 bitmap (but any + * existing contents of that bitmap will be kept). + */ + bool prealloc; + + /** Pointer to next L2Meta of the same write request */ + struct QCowL2Meta *next; +} QCowL2Meta; + +typedef enum QCow2ClusterType { + QCOW2_CLUSTER_UNALLOCATED, + QCOW2_CLUSTER_ZERO_PLAIN, + QCOW2_CLUSTER_ZERO_ALLOC, + QCOW2_CLUSTER_NORMAL, + QCOW2_CLUSTER_COMPRESSED, +} QCow2ClusterType; + +typedef enum QCow2MetadataOverlap { + QCOW2_OL_MAIN_HEADER_BITNR = 0, + QCOW2_OL_ACTIVE_L1_BITNR = 1, + QCOW2_OL_ACTIVE_L2_BITNR = 2, + QCOW2_OL_REFCOUNT_TABLE_BITNR = 3, + QCOW2_OL_REFCOUNT_BLOCK_BITNR = 4, + QCOW2_OL_SNAPSHOT_TABLE_BITNR = 5, + QCOW2_OL_INACTIVE_L1_BITNR = 6, + QCOW2_OL_INACTIVE_L2_BITNR = 7, + QCOW2_OL_BITMAP_DIRECTORY_BITNR = 8, + + QCOW2_OL_MAX_BITNR = 9, + + QCOW2_OL_NONE = 0, + QCOW2_OL_MAIN_HEADER = (1 << QCOW2_OL_MAIN_HEADER_BITNR), + QCOW2_OL_ACTIVE_L1 = (1 << QCOW2_OL_ACTIVE_L1_BITNR), + QCOW2_OL_ACTIVE_L2 = (1 << QCOW2_OL_ACTIVE_L2_BITNR), + QCOW2_OL_REFCOUNT_TABLE = (1 << QCOW2_OL_REFCOUNT_TABLE_BITNR), + QCOW2_OL_REFCOUNT_BLOCK = (1 << QCOW2_OL_REFCOUNT_BLOCK_BITNR), + QCOW2_OL_SNAPSHOT_TABLE = (1 << QCOW2_OL_SNAPSHOT_TABLE_BITNR), + QCOW2_OL_INACTIVE_L1 = (1 << QCOW2_OL_INACTIVE_L1_BITNR), + /* NOTE: Checking overlaps with inactive L2 tables will result in bdrv + * reads. */ + QCOW2_OL_INACTIVE_L2 = (1 << QCOW2_OL_INACTIVE_L2_BITNR), + QCOW2_OL_BITMAP_DIRECTORY = (1 << QCOW2_OL_BITMAP_DIRECTORY_BITNR), +} QCow2MetadataOverlap; + +/* Perform all overlap checks which can be done in constant time */ +#define QCOW2_OL_CONSTANT \ + (QCOW2_OL_MAIN_HEADER | QCOW2_OL_ACTIVE_L1 | QCOW2_OL_REFCOUNT_TABLE | \ + QCOW2_OL_SNAPSHOT_TABLE | QCOW2_OL_BITMAP_DIRECTORY) + +/* Perform all overlap checks which don't require disk access */ +#define QCOW2_OL_CACHED \ + (QCOW2_OL_CONSTANT | QCOW2_OL_ACTIVE_L2 | QCOW2_OL_REFCOUNT_BLOCK | \ + QCOW2_OL_INACTIVE_L1) + +/* Perform all overlap checks */ +#define QCOW2_OL_ALL \ + (QCOW2_OL_CACHED | QCOW2_OL_INACTIVE_L2) + +#define QCOW_L1E_OFFSET_MASK 0x00fffffffffffe00ULL +#define QCOW_L2E_OFFSET_MASK 0x00fffffffffffe00ULL +#define QCOW_L2E_COMPRESSED_OFFSET_SIZE_MASK 0x3fffffffffffffffULL + +#define REFT_OFFSET_MASK 0xfffffffffffffe00ULL + +#define INV_OFFSET (-1ULL) + +static inline uint64_t l2meta_cow_start(QCowL2Meta *m) +{ + return m->offset + m->cow_start.offset; +} + +static inline uint64_t l2meta_cow_end(QCowL2Meta *m) +{ + return m->offset + m->cow_end.offset + m->cow_end.nb_bytes; +} + +static inline uint64_t refcount_diff(uint64_t r1, uint64_t r2) +{ + return r1 > r2 ? r1 - r2 : r2 - r1; +} + +#endif // CEPH_LIBRBD_MIGRATION_QCOW2_H diff --git a/src/librbd/migration/QCOWFormat.cc b/src/librbd/migration/QCOWFormat.cc new file mode 100644 index 000000000..cedf9aa20 --- /dev/null +++ b/src/librbd/migration/QCOWFormat.cc @@ -0,0 +1,1545 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "librbd/migration/QCOWFormat.h" +#include "common/Clock.h" +#include "common/dout.h" +#include "common/errno.h" +#include "include/intarith.h" +#include "librbd/AsioEngine.h" +#include "librbd/ImageCtx.h" +#include "librbd/ImageState.h" +#include "librbd/Utils.h" +#include "librbd/io/AioCompletion.h" +#include "librbd/io/ReadResult.h" +#include "librbd/migration/SnapshotInterface.h" +#include "librbd/migration/SourceSpecBuilder.h" +#include "librbd/migration/StreamInterface.h" +#include "librbd/migration/Utils.h" +#include <boost/asio/dispatch.hpp> +#include <boost/asio/post.hpp> +#include <boost/endian/conversion.hpp> +#include <deque> +#include <tuple> +#include <unordered_map> +#include <vector> + +#define dout_subsys ceph_subsys_rbd + +namespace librbd { +namespace migration { + +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::QCOWFormat: " \ + << __func__ << ": " + +using boost::endian::big_to_native; + +namespace qcow_format { + +struct ClusterExtent { + uint64_t cluster_offset; + uint64_t cluster_length; + uint64_t intra_cluster_offset; + uint64_t image_offset; + uint64_t buffer_offset; + + ClusterExtent(uint64_t cluster_offset, uint64_t cluster_length, + uint64_t intra_cluster_offset, uint64_t image_offset, + uint64_t buffer_offset) + : cluster_offset(cluster_offset), cluster_length(cluster_length), + intra_cluster_offset(intra_cluster_offset), image_offset(image_offset), + buffer_offset(buffer_offset) { + } +}; + +typedef std::vector<ClusterExtent> ClusterExtents; + +void LookupTable::init() { + if (cluster_offsets == nullptr) { + cluster_offsets = reinterpret_cast<uint64_t*>(bl.c_str()); + } +} + +void LookupTable::decode() { + init(); + + // L2 tables are selectively byte-swapped on demand if only requesting a + // single cluster offset + if (decoded) { + return; + } + + // translate the lookup table (big-endian -> CPU endianess) + for (auto idx = 0UL; idx < size; ++idx) { + cluster_offsets[idx] = big_to_native(cluster_offsets[idx]); + } + + decoded = true; +} + +void populate_cluster_extents(CephContext* cct, uint64_t cluster_size, + const io::Extents& image_extents, + ClusterExtents* cluster_extents) { + uint64_t buffer_offset = 0; + for (auto [image_offset, image_length] : image_extents) { + while (image_length > 0) { + auto intra_cluster_offset = image_offset & (cluster_size - 1); + auto intra_cluster_length = cluster_size - intra_cluster_offset; + auto cluster_length = std::min(image_length, intra_cluster_length); + + ldout(cct, 20) << "image_offset=" << image_offset << ", " + << "image_length=" << image_length << ", " + << "cluster_length=" << cluster_length << dendl; + + + cluster_extents->emplace_back(0, cluster_length, intra_cluster_offset, + image_offset, buffer_offset); + + image_offset += cluster_length; + image_length -= cluster_length; + buffer_offset += cluster_length; + } + } +} + +} // namespace qcow_format + +using namespace qcow_format; + +template <typename I> +struct QCOWFormat<I>::Cluster { + const uint64_t cluster_offset; + bufferlist cluster_data_bl; + + Cluster(uint64_t cluster_offset) : cluster_offset(cluster_offset) { + } +}; + +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::QCOWFormat::ClusterCache: " \ + << this << " " << __func__ << ": " + +template <typename I> +class QCOWFormat<I>::ClusterCache { +public: + ClusterCache(QCOWFormat* qcow_format) + : qcow_format(qcow_format), + m_strand(*qcow_format->m_image_ctx->asio_engine) { + } + + void get_cluster(uint64_t cluster_offset, uint64_t cluster_length, + uint64_t intra_cluster_offset, bufferlist* bl, + Context* on_finish) { + auto cct = qcow_format->m_image_ctx->cct; + ldout(cct, 20) << "cluster_offset=" << cluster_offset << dendl; + + // cache state machine runs in a single strand thread + boost::asio::dispatch( + m_strand, + [this, cluster_offset, cluster_length, intra_cluster_offset, bl, + on_finish]() { + execute_get_cluster(cluster_offset, cluster_length, + intra_cluster_offset, bl, on_finish); + }); + } + +private: + typedef std::tuple<uint64_t, uint64_t, bufferlist*, Context*> Completion; + typedef std::list<Completion> Completions; + + QCOWFormat* qcow_format; + boost::asio::io_context::strand m_strand; + + std::shared_ptr<Cluster> cluster; + std::unordered_map<uint64_t, Completions> cluster_completions; + + void execute_get_cluster(uint64_t cluster_offset, uint64_t cluster_length, + uint64_t intra_cluster_offset, bufferlist* bl, + Context* on_finish) { + auto cct = qcow_format->m_image_ctx->cct; + ldout(cct, 20) << "cluster_offset=" << cluster_offset << dendl; + + if (cluster && cluster->cluster_offset == cluster_offset) { + // most-recent cluster matches + bl->substr_of(cluster->cluster_data_bl, intra_cluster_offset, + cluster_length); + boost::asio::post(*qcow_format->m_image_ctx->asio_engine, + [on_finish]() { on_finish->complete(0); }); + return; + } + + // record callback for cluster + bool new_request = (cluster_completions.count(cluster_offset) == 0); + cluster_completions[cluster_offset].emplace_back( + intra_cluster_offset, cluster_length, bl, on_finish); + if (new_request) { + // start the new read request + read_cluster(std::make_shared<Cluster>(cluster_offset)); + } + } + + void read_cluster(std::shared_ptr<Cluster> cluster) { + auto cct = qcow_format->m_image_ctx->cct; + + uint64_t stream_offset = cluster->cluster_offset; + uint64_t stream_length = qcow_format->m_cluster_size; + if ((cluster->cluster_offset & QCOW_OFLAG_COMPRESSED) != 0) { + // compressed clusters encode the compressed length in the lower bits + stream_offset = cluster->cluster_offset & + qcow_format->m_cluster_offset_mask; + stream_length = (cluster->cluster_offset >> + (63 - qcow_format->m_cluster_bits)) & + (qcow_format->m_cluster_size - 1); + } + + ldout(cct, 20) << "cluster_offset=" << cluster->cluster_offset << ", " + << "stream_offset=" << stream_offset << ", " + << "stream_length=" << stream_length << dendl; + + // read the cluster into the cache entry + auto ctx = new LambdaContext([this, cluster](int r) { + boost::asio::post(m_strand, [this, cluster, r]() { + handle_read_cluster(r, cluster); }); }); + qcow_format->m_stream->read({{stream_offset, stream_length}}, + &cluster->cluster_data_bl, ctx); + } + + void handle_read_cluster(int r, std::shared_ptr<Cluster> cluster) { + auto cct = qcow_format->m_image_ctx->cct; + ldout(cct, 20) << "r=" << r << ", " + << "cluster_offset=" << cluster->cluster_offset << dendl; + + auto completions = std::move(cluster_completions[cluster->cluster_offset]); + cluster_completions.erase(cluster->cluster_offset); + + if (r < 0) { + lderr(cct) << "failed to read cluster offset " << cluster->cluster_offset + << ": " << cpp_strerror(r) << dendl; + } else { + if ((cluster->cluster_offset & QCOW_OFLAG_COMPRESSED) != 0) { + bufferlist compressed_bl{std::move(cluster->cluster_data_bl)}; + cluster->cluster_data_bl.clear(); + + // TODO + lderr(cct) << "support for compressed clusters is not available" + << dendl; + r = -EINVAL; + } else { + // cache the MRU cluster in case of sequential IO + this->cluster = cluster; + } + } + + // complete the IO back to caller + boost::asio::post(*qcow_format->m_image_ctx->asio_engine, + [r, cluster, completions=std::move(completions)]() { + for (auto completion : completions) { + if (r >= 0) { + std::get<2>(completion)->substr_of( + cluster->cluster_data_bl, + std::get<0>(completion), + std::get<1>(completion)); + } + std::get<3>(completion)->complete(r); + } + }); + } +}; + +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::QCOWFormat::L2TableCache: " \ + << this << " " << __func__ << ": " + +template <typename I> +class QCOWFormat<I>::L2TableCache { +public: + L2TableCache(QCOWFormat* qcow_format) + : qcow_format(qcow_format), + m_strand(*qcow_format->m_image_ctx->asio_engine), + l2_cache_entries(QCOW_L2_CACHE_SIZE) { + } + + void get_l2_table(const LookupTable* l1_table, uint64_t l2_table_offset, + std::shared_ptr<const LookupTable>* l2_table, + Context* on_finish) { + auto cct = qcow_format->m_image_ctx->cct; + ldout(cct, 20) << "l2_table_offset=" << l2_table_offset << dendl; + + // cache state machine runs in a single strand thread + Request request{l1_table, l2_table_offset, l2_table, on_finish}; + boost::asio::dispatch( + m_strand, [this, request=std::move(request)]() { + requests.push_back(std::move(request)); + }); + dispatch_request(); + } + + void get_cluster_offset(const LookupTable* l1_table, + uint64_t image_offset, uint64_t* cluster_offset, + Context* on_finish) { + auto cct = qcow_format->m_image_ctx->cct; + uint32_t l1_table_index = image_offset >> qcow_format->m_l1_shift; + uint64_t l2_table_offset = l1_table->cluster_offsets[std::min<uint32_t>( + l1_table_index, l1_table->size - 1)] & + qcow_format->m_cluster_mask; + uint32_t l2_table_index = (image_offset >> qcow_format->m_cluster_bits) & + (qcow_format->m_l2_size - 1); + ldout(cct, 20) << "image_offset=" << image_offset << ", " + << "l1_table_index=" << l1_table_index << ", " + << "l2_table_offset=" << l2_table_offset << ", " + << "l2_table_index=" << l2_table_index << dendl; + + if (l1_table_index >= l1_table->size) { + lderr(cct) << "L1 index " << l1_table_index << " out-of-bounds" << dendl; + on_finish->complete(-ERANGE); + return; + } else if (l2_table_offset == 0) { + // L2 table has not been allocated for specified offset + ldout(cct, 20) << "image_offset=" << image_offset << ", " + << "cluster_offset=DNE" << dendl; + *cluster_offset = 0; + on_finish->complete(-ENOENT); + return; + } + + // cache state machine runs in a single strand thread + Request request{l1_table, l2_table_offset, l2_table_index, cluster_offset, + on_finish}; + boost::asio::dispatch( + m_strand, [this, request=std::move(request)]() { + requests.push_back(std::move(request)); + }); + dispatch_request(); + } + +private: + QCOWFormat* qcow_format; + + boost::asio::io_context::strand m_strand; + + struct Request { + const LookupTable* l1_table; + + uint64_t l2_table_offset; + + // get_cluster_offset request + uint32_t l2_table_index; + uint64_t* cluster_offset = nullptr; + + // get_l2_table request + std::shared_ptr<const LookupTable>* l2_table; + + Context* on_finish; + + Request(const LookupTable* l1_table, uint64_t l2_table_offset, + uint32_t l2_table_index, uint64_t* cluster_offset, + Context* on_finish) + : l1_table(l1_table), l2_table_offset(l2_table_offset), + l2_table_index(l2_table_index), cluster_offset(cluster_offset), + on_finish(on_finish) { + } + Request(const LookupTable* l1_table, uint64_t l2_table_offset, + std::shared_ptr<const LookupTable>* l2_table, Context* on_finish) + : l1_table(l1_table), l2_table_offset(l2_table_offset), + l2_table(l2_table), on_finish(on_finish) { + } + }; + + typedef std::deque<Request> Requests; + + struct L2Cache { + uint64_t l2_offset = 0; + std::shared_ptr<LookupTable> l2_table; + + utime_t timestamp; + uint32_t count = 0; + bool in_flight = false; + + int ret_val = 0; + }; + std::vector<L2Cache> l2_cache_entries; + + Requests requests; + + void dispatch_request() { + boost::asio::dispatch(m_strand, [this]() { execute_request(); }); + } + + void execute_request() { + auto cct = qcow_format->m_image_ctx->cct; + if (requests.empty()) { + return; + } + + auto request = requests.front(); + ldout(cct, 20) << "l2_table_offset=" << request.l2_table_offset << dendl; + + std::shared_ptr<LookupTable> l2_table; + int r = l2_table_lookup(request.l2_table_offset, &l2_table); + if (r < 0) { + lderr(cct) << "failed to load L2 table: l2_table_offset=" + << request.l2_table_offset << ": " + << cpp_strerror(r) << dendl; + } else if (l2_table == nullptr) { + // table not in cache -- will restart once its loaded + return; + } else if (request.cluster_offset != nullptr) { + auto cluster_offset = l2_table->cluster_offsets[request.l2_table_index]; + if (!l2_table->decoded) { + // table hasn't been byte-swapped + cluster_offset = big_to_native(cluster_offset); + } + + *request.cluster_offset = cluster_offset & qcow_format->m_cluster_mask; + if (*request.cluster_offset == QCOW_OFLAG_ZERO) { + ldout(cct, 20) << "l2_table_offset=" << request.l2_table_offset << ", " + << "l2_table_index=" << request.l2_table_index << ", " + << "cluster_offset=zeroed" << dendl; + } else { + ldout(cct, 20) << "l2_table_offset=" << request.l2_table_offset << ", " + << "l2_table_index=" << request.l2_table_index << ", " + << "cluster_offset=" << *request.cluster_offset + << dendl; + } + } else if (request.l2_table != nullptr) { + // ensure it's in the correct byte-order + l2_table->decode(); + *request.l2_table = l2_table; + } else { + ceph_assert(false); + } + + // complete the L2 cache request + boost::asio::post(*qcow_format->m_image_ctx->asio_engine, + [r, ctx=request.on_finish]() { ctx->complete(r); }); + requests.pop_front(); + + // process next request (if any) + dispatch_request(); + } + + int l2_table_lookup(uint64_t l2_offset, + std::shared_ptr<LookupTable>* l2_table) { + auto cct = qcow_format->m_image_ctx->cct; + + l2_table->reset(); + + // find a match in the existing cache + for (auto idx = 0U; idx < l2_cache_entries.size(); ++idx) { + auto& l2_cache = l2_cache_entries[idx]; + if (l2_cache.l2_offset == l2_offset) { + if (l2_cache.in_flight) { + ldout(cct, 20) << "l2_offset=" << l2_offset << ", " + << "index=" << idx << " (in-flight)" << dendl; + return 0; + } + + if (l2_cache.ret_val < 0) { + ldout(cct, 20) << "l2_offset=" << l2_offset << ", " + << "index=" << idx << " (error): " + << cpp_strerror(l2_cache.ret_val) << dendl; + int r = l2_cache.ret_val; + l2_cache = L2Cache{}; + + return r; + } + + ++l2_cache.count; + if (l2_cache.count == std::numeric_limits<uint32_t>::max()) { + for (auto& entry : l2_cache_entries) { + entry.count >>= 1; + } + } + + ldout(cct, 20) << "l2_offset=" << l2_offset << ", " << "index=" << idx + << dendl; + *l2_table = l2_cache.l2_table; + return 0; + } + } + + // find the least used entry + int32_t min_idx = -1; + uint32_t min_count = std::numeric_limits<uint32_t>::max(); + utime_t min_timestamp; + for (uint32_t idx = 0U; idx < l2_cache_entries.size(); ++idx) { + auto& l2_cache = l2_cache_entries[idx]; + if (l2_cache.in_flight) { + continue; + } + + if (l2_cache.count > 0) { + --l2_cache.count; + } + + if (l2_cache.count <= min_count) { + if (min_idx == -1 || l2_cache.timestamp < min_timestamp) { + min_timestamp = l2_cache.timestamp; + min_count = l2_cache.count; + min_idx = idx; + } + } + } + + if (min_idx == -1) { + // no space in the cache due to in-flight requests + ldout(cct, 20) << "l2_offset=" << l2_offset << ", " + << "index=DNE (cache busy)" << dendl; + return 0; + } + + ldout(cct, 20) << "l2_offset=" << l2_offset << ", " + << "index=" << min_idx << " (loading)" << dendl; + auto& l2_cache = l2_cache_entries[min_idx]; + l2_cache.l2_table = std::make_shared<LookupTable>(qcow_format->m_l2_size); + l2_cache.l2_offset = l2_offset; + l2_cache.timestamp = ceph_clock_now(); + l2_cache.count = 1; + l2_cache.in_flight = true; + + // read the L2 table into the L2 cache entry + auto ctx = new LambdaContext([this, index=min_idx, l2_offset](int r) { + boost::asio::post(m_strand, [this, index, l2_offset, r]() { + handle_l2_table_lookup(r, index, l2_offset); }); }); + qcow_format->m_stream->read( + {{l2_offset, qcow_format->m_l2_size * sizeof(uint64_t)}}, + &l2_cache.l2_table->bl, ctx); + return 0; + } + + void handle_l2_table_lookup(int r, uint32_t index, uint64_t l2_offset) { + auto cct = qcow_format->m_image_ctx->cct; + ldout(cct, 20) << "r=" << r << ", " + << "l2_offset=" << l2_offset << ", " + << "index=" << index << dendl; + + auto& l2_cache = l2_cache_entries[index]; + ceph_assert(l2_cache.in_flight); + l2_cache.in_flight = false; + + if (r < 0) { + lderr(cct) << "failed to load L2 table: " + << "l2_offset=" << l2_cache.l2_offset << ": " + << cpp_strerror(r) << dendl; + l2_cache.ret_val = r; + } else { + // keep the L2 table in big-endian byte-order until the full table + // is requested + l2_cache.l2_table->init(); + } + + // restart the state machine + dispatch_request(); + } + +}; + +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::QCOWFormat::ReadRequest: " \ + << this << " " << __func__ << ": " + +template <typename I> +class QCOWFormat<I>::ReadRequest { +public: + ReadRequest(QCOWFormat* qcow_format, io::AioCompletion* aio_comp, + const LookupTable* l1_table, io::Extents&& image_extents) + : qcow_format(qcow_format), aio_comp(aio_comp), l1_table(l1_table), + image_extents(std::move(image_extents)) { + } + + void send() { + get_cluster_offsets(); + } + +private: + QCOWFormat* qcow_format; + io::AioCompletion* aio_comp; + + const LookupTable* l1_table; + io::Extents image_extents; + + size_t image_extents_idx = 0; + uint32_t image_extent_offset = 0; + + ClusterExtents cluster_extents; + + void get_cluster_offsets() { + auto cct = qcow_format->m_image_ctx->cct; + populate_cluster_extents(cct, qcow_format->m_cluster_size, image_extents, + &cluster_extents); + + ldout(cct, 20) << dendl; + auto ctx = new LambdaContext([this](int r) { + handle_get_cluster_offsets(r); }); + auto gather_ctx = new C_Gather(cct, ctx); + + for (auto& cluster_extent : cluster_extents) { + auto sub_ctx = new LambdaContext( + [this, &cluster_extent, on_finish=gather_ctx->new_sub()](int r) { + handle_get_cluster_offset(r, cluster_extent, on_finish); }); + qcow_format->m_l2_table_cache->get_cluster_offset( + l1_table, cluster_extent.image_offset, + &cluster_extent.cluster_offset, sub_ctx); + } + + gather_ctx->activate(); + } + + void handle_get_cluster_offset(int r, const ClusterExtent& cluster_extent, + Context* on_finish) { + auto cct = qcow_format->m_image_ctx->cct; + ldout(cct, 20) << "r=" << r << ", " + << "image_offset=" << cluster_extent.image_offset << ", " + << "cluster_offset=" << cluster_extent.cluster_offset + << dendl; + + if (r == -ENOENT) { + ldout(cct, 20) << "image offset DNE in QCOW image" << dendl; + r = 0; + } else if (r < 0) { + lderr(cct) << "failed to map image offset " << cluster_extent.image_offset + << ": " << cpp_strerror(r) << dendl; + } + + on_finish->complete(r); + } + + void handle_get_cluster_offsets(int r) { + auto cct = qcow_format->m_image_ctx->cct; + ldout(cct, 20) << "r=" << r << dendl; + + if (r < 0) { + lderr(cct) << "failed to retrieve cluster extents: " << cpp_strerror(r) + << dendl; + aio_comp->fail(r); + delete this; + return; + } + + read_clusters(); + } + + void read_clusters() { + auto cct = qcow_format->m_image_ctx->cct; + ldout(cct, 20) << dendl; + + aio_comp->set_request_count(cluster_extents.size()); + for (auto& cluster_extent : cluster_extents) { + auto read_ctx = new io::ReadResult::C_ImageReadRequest( + aio_comp, cluster_extent.buffer_offset, + {{cluster_extent.image_offset, cluster_extent.cluster_length}}); + read_ctx->ignore_enoent = true; + + auto log_ctx = new LambdaContext( + [this, cct=qcow_format->m_image_ctx->cct, + image_offset=cluster_extent.image_offset, + image_length=cluster_extent.cluster_length, ctx=read_ctx](int r) { + handle_read_cluster(cct, r, image_offset, image_length, ctx); + }); + + if (cluster_extent.cluster_offset == 0) { + // QCOW header is at offset 0, implies cluster DNE + log_ctx->complete(-ENOENT); + } else if (cluster_extent.cluster_offset == QCOW_OFLAG_ZERO) { + // explicitly zeroed section + read_ctx->bl.append_zero(cluster_extent.cluster_length); + log_ctx->complete(0); + } else { + // request the (sub)cluster from the cluster cache + qcow_format->m_cluster_cache->get_cluster( + cluster_extent.cluster_offset, cluster_extent.cluster_length, + cluster_extent.intra_cluster_offset, &read_ctx->bl, log_ctx); + } + } + + delete this; + } + + void handle_read_cluster(CephContext* cct, int r, uint64_t image_offset, + uint64_t image_length, Context* on_finish) const { + // NOTE: treat as static function, expect object has been deleted + + ldout(cct, 20) << "r=" << r << ", " + << "image_offset=" << image_offset << ", " + << "image_length=" << image_length << dendl; + + if (r != -ENOENT && r < 0) { + lderr(cct) << "failed to read image extent " << image_offset << "~" + << image_length << ": " << cpp_strerror(r) << dendl; + } + + on_finish->complete(r); + } +}; + +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::QCOWFormat::" \ + << "ListSnapsRequest: " << this << " " \ + << __func__ << ": " + +template <typename I> +class QCOWFormat<I>::ListSnapsRequest { +public: + ListSnapsRequest( + QCOWFormat* qcow_format, uint32_t l1_table_index, + ClusterExtents&& cluster_extents, + const std::map<uint64_t, const LookupTable*>& snap_id_to_l1_table, + io::SnapshotDelta* snapshot_delta, Context* on_finish) + : qcow_format(qcow_format), l1_table_index(l1_table_index), + cluster_extents(std::move(cluster_extents)), + snap_id_to_l1_table(snap_id_to_l1_table), snapshot_delta(snapshot_delta), + on_finish(on_finish) { + } + + void send() { + get_l2_table(); + } + +private: + QCOWFormat* qcow_format; + uint32_t l1_table_index; + ClusterExtents cluster_extents; + std::map<uint64_t, const LookupTable*> snap_id_to_l1_table; + io::SnapshotDelta* snapshot_delta; + Context* on_finish; + + std::shared_ptr<const LookupTable> previous_l2_table; + std::shared_ptr<const LookupTable> l2_table; + + void get_l2_table() { + auto cct = qcow_format->m_image_ctx->cct; + if (snap_id_to_l1_table.empty()) { + finish(0); + return; + } + + auto it = snap_id_to_l1_table.begin(); + auto [snap_id, l1_table] = *it; + snap_id_to_l1_table.erase(it); + + previous_l2_table = l2_table; + l2_table.reset(); + + auto ctx = new LambdaContext([this, snap_id = snap_id](int r) { + boost::asio::post(qcow_format->m_strand, [this, snap_id, r]() { + handle_get_l2_table(r, snap_id); + }); + }); + + if (l1_table_index >= l1_table->size || + l1_table->cluster_offsets[l1_table_index] == 0) { + ldout(cct, 20) << "l1_table_index=" << l1_table_index << ", " + << "snap_id=" << snap_id << ": DNE" << dendl; + ctx->complete(-ENOENT); + return; + } + + uint64_t l2_table_offset = l1_table->cluster_offsets[l1_table_index] & + qcow_format->m_cluster_mask; + + ldout(cct, 20) << "l1_table_index=" << l1_table_index << ", " + << "snap_id=" << snap_id << ", " + << "l2_table_offset=" << l2_table_offset << dendl; + qcow_format->m_l2_table_cache->get_l2_table(l1_table, l2_table_offset, + &l2_table, ctx); + } + + void handle_get_l2_table(int r, uint64_t snap_id) { + ceph_assert(qcow_format->m_strand.running_in_this_thread()); + + auto cct = qcow_format->m_image_ctx->cct; + ldout(cct, 20) << "r=" << r << ", " + << "snap_id=" << snap_id << dendl; + + if (r == -ENOENT) { + l2_table.reset(); + } else if (r < 0) { + lderr(cct) << "failed to retrieve L2 table for snapshot " << snap_id + << ": " << cpp_strerror(r) << dendl; + finish(r); + return; + } + + // compare the cluster offsets at each requested L2 offset between + // the previous snapshot's L2 table and the current L2 table. + auto& sparse_extents = (*snapshot_delta)[{snap_id, snap_id}]; + for (auto& cluster_extent : cluster_extents) { + uint32_t l2_table_index = + (cluster_extent.image_offset >> qcow_format->m_cluster_bits) & + (qcow_format->m_l2_size - 1); + + std::optional<uint64_t> cluster_offset; + if (l2_table && l2_table_index < l2_table->size) { + cluster_offset = l2_table->cluster_offsets[l2_table_index] & + qcow_format->m_cluster_offset_mask; + } + + std::optional<uint64_t> prev_cluster_offset; + if (previous_l2_table && l2_table_index < previous_l2_table->size) { + prev_cluster_offset = + previous_l2_table->cluster_offsets[l2_table_index] & + qcow_format->m_cluster_offset_mask; + } + + ldout(cct, 20) << "l1_table_index=" << l1_table_index << ", " + << "snap_id=" << snap_id << ", " + << "image_offset=" << cluster_extent.image_offset << ", " + << "l2_table_index=" << l2_table_index << ", " + << "cluster_offset=" << cluster_offset << ", " + << "prev_cluster_offset=" << prev_cluster_offset << dendl; + + auto state = io::SPARSE_EXTENT_STATE_DATA; + if (cluster_offset == prev_cluster_offset) { + continue; + } else if ((prev_cluster_offset && !cluster_offset) || + *cluster_offset == QCOW_OFLAG_ZERO) { + // explicitly zeroed or deallocated + state = io::SPARSE_EXTENT_STATE_ZEROED; + } + + sparse_extents.insert( + cluster_extent.image_offset, cluster_extent.cluster_length, + {state, cluster_extent.cluster_length}); + } + + ldout(cct, 20) << "l1_table_index=" << l1_table_index << ", " + << "snap_id=" << snap_id << ", " + << "sparse_extents=" << sparse_extents << dendl; + + // continue processing the L2 table at this index for all snapshots + boost::asio::post(*qcow_format->m_image_ctx->asio_engine, + [this]() { get_l2_table(); }); + } + + + void finish(int r) { + auto cct = qcow_format->m_image_ctx->cct; + ldout(cct, 20) << "r=" << r << dendl; + + on_finish->complete(r); + delete this; + } +}; + +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::QCOWFormat: " << this \ + << " " << __func__ << ": " + +template <typename I> +QCOWFormat<I>::QCOWFormat( + I* image_ctx, const json_spirit::mObject& json_object, + const SourceSpecBuilder<I>* source_spec_builder) + : m_image_ctx(image_ctx), m_json_object(json_object), + m_source_spec_builder(source_spec_builder), + m_strand(*image_ctx->asio_engine) { +} + +template <typename I> +void QCOWFormat<I>::open(Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + int r = m_source_spec_builder->build_stream(m_json_object, &m_stream); + if (r < 0) { + lderr(cct) << "failed to build migration stream handler" << cpp_strerror(r) + << dendl; + on_finish->complete(r); + return; + } + + auto ctx = new LambdaContext([this, on_finish](int r) { + handle_open(r, on_finish); }); + m_stream->open(ctx); +} + +template <typename I> +void QCOWFormat<I>::handle_open(int r, Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << "r=" << r << dendl; + + if (r < 0) { + lderr(cct) << "failed to open QCOW image: " << cpp_strerror(r) + << dendl; + on_finish->complete(r); + return; + } + + probe(on_finish); +} + +template <typename I> +void QCOWFormat<I>::probe(Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + auto ctx = new LambdaContext([this, on_finish](int r) { + handle_probe(r, on_finish); }); + m_bl.clear(); + m_stream->read({{0, 8}}, &m_bl, ctx); +} + +template <typename I> +void QCOWFormat<I>::handle_probe(int r, Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << "r=" << r << dendl; + + if (r < 0) { + lderr(cct) << "failed to probe QCOW image: " << cpp_strerror(r) + << dendl; + on_finish->complete(r); + return; + } + + auto header_probe = *reinterpret_cast<QCowHeaderProbe*>( + m_bl.c_str()); + header_probe.magic = big_to_native(header_probe.magic); + header_probe.version = big_to_native(header_probe.version); + + if (header_probe.magic != QCOW_MAGIC) { + lderr(cct) << "invalid QCOW header magic" << dendl; + on_finish->complete(-EINVAL); + return; + } + + m_bl.clear(); + if (header_probe.version == 1) { +#ifdef WITH_RBD_MIGRATION_FORMAT_QCOW_V1 + read_v1_header(on_finish); +#else // WITH_RBD_MIGRATION_FORMAT_QCOW_V1 + lderr(cct) << "QCOW is not supported" << dendl; + on_finish->complete(-ENOTSUP); +#endif // WITH_RBD_MIGRATION_FORMAT_QCOW_V1 + return; + } else if (header_probe.version >= 2 && header_probe.version <= 3) { + read_v2_header(on_finish); + return; + } else { + lderr(cct) << "invalid QCOW header version " << header_probe.version + << dendl; + on_finish->complete(-EINVAL); + return; + } +} + +#ifdef WITH_RBD_MIGRATION_FORMAT_QCOW_V1 + +template <typename I> +void QCOWFormat<I>::read_v1_header(Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + auto ctx = new LambdaContext([this, on_finish](int r) { + handle_read_v1_header(r, on_finish); }); + m_bl.clear(); + m_stream->read({{0, sizeof(QCowHeaderV1)}}, &m_bl, ctx); +} + +template <typename I> +void QCOWFormat<I>::handle_read_v1_header(int r, Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << "r=" << r << dendl; + + if (r < 0) { + lderr(cct) << "failed to read QCOW header: " << cpp_strerror(r) << dendl; + on_finish->complete(r); + return; + } + + auto header = *reinterpret_cast<QCowHeaderV1*>(m_bl.c_str()); + + // byte-swap important fields + header.magic = big_to_native(header.magic); + header.version = big_to_native(header.version); + header.backing_file_offset = big_to_native(header.backing_file_offset); + header.backing_file_size = big_to_native(header.backing_file_size); + header.size = big_to_native(header.size); + header.crypt_method = big_to_native(header.crypt_method); + header.l1_table_offset = big_to_native(header.l1_table_offset); + + if (header.magic != QCOW_MAGIC || header.version != 1) { + // honestly shouldn't happen since we've already validated it + lderr(cct) << "header is not QCOW" << dendl; + on_finish->complete(-EINVAL); + return; + } + + if (header.cluster_bits < QCOW_MIN_CLUSTER_BITS || + header.cluster_bits > QCOW_MAX_CLUSTER_BITS) { + lderr(cct) << "invalid cluster bits: " << header.cluster_bits << dendl; + on_finish->complete(-EINVAL); + return; + } + + if (header.l2_bits < (QCOW_MIN_CLUSTER_BITS - 3) || + header.l2_bits > (QCOW_MAX_CLUSTER_BITS - 3)) { + lderr(cct) << "invalid L2 bits: " << header.l2_bits << dendl; + on_finish->complete(-EINVAL); + return; + } + + if (header.crypt_method != QCOW_CRYPT_NONE) { + lderr(cct) << "invalid or unsupported encryption method" << dendl; + on_finish->complete(-EINVAL); + return; + } + + m_size = header.size; + if (p2roundup(m_size, static_cast<uint64_t>(512)) != m_size) { + lderr(cct) << "image size is not a multiple of block size" << dendl; + on_finish->complete(-EINVAL); + return; + } + + m_backing_file_offset = header.backing_file_offset; + m_backing_file_size = header.backing_file_size; + + m_cluster_bits = header.cluster_bits; + m_cluster_size = 1UL << header.cluster_bits; + m_cluster_offset_mask = (1ULL << (63 - header.cluster_bits)) - 1; + m_cluster_mask = ~QCOW_OFLAG_COMPRESSED; + + m_l2_bits = header.l2_bits; + m_l2_size = (1UL << m_l2_bits); + + m_l1_shift = m_cluster_bits + m_l2_bits; + m_l1_table.size = (m_size + (1LL << m_l1_shift) - 1) >> m_l1_shift; + m_l1_table_offset = header.l1_table_offset; + if (m_size > (std::numeric_limits<uint64_t>::max() - (1ULL << m_l1_shift)) || + m_l1_table.size > + (std::numeric_limits<int32_t>::max() / sizeof(uint64_t))) { + lderr(cct) << "image size too big: " << m_size << dendl; + on_finish->complete(-EINVAL); + return; + } + + ldout(cct, 15) << "size=" << m_size << ", " + << "cluster_bits=" << m_cluster_bits << ", " + << "l2_bits=" << m_l2_bits << dendl; + + // allocate memory for L1 table and L2 + cluster caches + m_l2_table_cache = std::make_unique<L2TableCache>(this); + m_cluster_cache = std::make_unique<ClusterCache>(this); + + read_l1_table(on_finish); +} + +#endif // WITH_RBD_MIGRATION_FORMAT_QCOW_V1 + +template <typename I> +void QCOWFormat<I>::read_v2_header(Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + auto ctx = new LambdaContext([this, on_finish](int r) { + handle_read_v2_header(r, on_finish); }); + m_bl.clear(); + m_stream->read({{0, sizeof(QCowHeader)}}, &m_bl, ctx); +} + +template <typename I> +void QCOWFormat<I>::handle_read_v2_header(int r, Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << "r=" << r << dendl; + + if (r < 0) { + lderr(cct) << "failed to read QCOW2 header: " << cpp_strerror(r) << dendl; + on_finish->complete(r); + return; + } + + auto header = *reinterpret_cast<QCowHeader*>(m_bl.c_str()); + + // byte-swap important fields + header.magic = big_to_native(header.magic); + header.version = big_to_native(header.version); + header.backing_file_offset = big_to_native(header.backing_file_offset); + header.backing_file_size = big_to_native(header.backing_file_size); + header.cluster_bits = big_to_native(header.cluster_bits); + header.size = big_to_native(header.size); + header.crypt_method = big_to_native(header.crypt_method); + header.l1_size = big_to_native(header.l1_size); + header.l1_table_offset = big_to_native(header.l1_table_offset); + header.nb_snapshots = big_to_native(header.nb_snapshots); + header.snapshots_offset = big_to_native(header.snapshots_offset); + + if (header.version == 2) { + // valid only for version >= 3 + header.incompatible_features = 0; + header.compatible_features = 0; + header.autoclear_features = 0; + header.header_length = 72; + header.compression_type = 0; + } else { + header.incompatible_features = big_to_native(header.incompatible_features); + header.compatible_features = big_to_native(header.compatible_features); + header.autoclear_features = big_to_native(header.autoclear_features); + header.header_length = big_to_native(header.header_length); + } + + if (header.magic != QCOW_MAGIC || header.version < 2 || header.version > 3) { + // honestly shouldn't happen since we've already validated it + lderr(cct) << "header is not QCOW2" << dendl; + on_finish->complete(-EINVAL); + return; + } + + if (header.cluster_bits < QCOW_MIN_CLUSTER_BITS || + header.cluster_bits > QCOW_MAX_CLUSTER_BITS) { + lderr(cct) << "invalid cluster bits: " << header.cluster_bits << dendl; + on_finish->complete(-EINVAL); + return; + } + + if (header.crypt_method != QCOW_CRYPT_NONE) { + lderr(cct) << "invalid or unsupported encryption method" << dendl; + on_finish->complete(-EINVAL); + return; + } + + m_size = header.size; + if (p2roundup(m_size, static_cast<uint64_t>(512)) != m_size) { + lderr(cct) << "image size is not a multiple of block size" << dendl; + on_finish->complete(-EINVAL); + return; + } + + if (header.header_length <= offsetof(QCowHeader, compression_type)) { + header.compression_type = 0; + } + + if ((header.compression_type != 0) || + ((header.incompatible_features & QCOW2_INCOMPAT_COMPRESSION) != 0)) { + lderr(cct) << "invalid or unsupported compression type" << dendl; + on_finish->complete(-EINVAL); + return; + } + + if ((header.incompatible_features & QCOW2_INCOMPAT_DATA_FILE) != 0) { + lderr(cct) << "external data file feature not supported" << dendl; + on_finish->complete(-ENOTSUP); + } + + if ((header.incompatible_features & QCOW2_INCOMPAT_EXTL2) != 0) { + lderr(cct) << "extended L2 table feature not supported" << dendl; + on_finish->complete(-ENOTSUP); + return; + } + + header.incompatible_features &= ~QCOW2_INCOMPAT_MASK; + if (header.incompatible_features != 0) { + lderr(cct) << "unknown incompatible feature enabled" << dendl; + on_finish->complete(-EINVAL); + return; + } + + m_backing_file_offset = header.backing_file_offset; + m_backing_file_size = header.backing_file_size; + + m_cluster_bits = header.cluster_bits; + m_cluster_size = 1UL << header.cluster_bits; + m_cluster_offset_mask = (1ULL << (63 - header.cluster_bits)) - 1; + m_cluster_mask = ~(QCOW_OFLAG_COMPRESSED | QCOW_OFLAG_COPIED); + + // L2 table is fixed a (1) cluster block to hold 8-byte (3 bit) offsets + m_l2_bits = m_cluster_bits - 3; + m_l2_size = (1UL << m_l2_bits); + + m_l1_shift = m_cluster_bits + m_l2_bits; + m_l1_table.size = (m_size + (1LL << m_l1_shift) - 1) >> m_l1_shift; + m_l1_table_offset = header.l1_table_offset; + if (m_size > (std::numeric_limits<uint64_t>::max() - (1ULL << m_l1_shift)) || + m_l1_table.size > + (std::numeric_limits<int32_t>::max() / sizeof(uint64_t))) { + lderr(cct) << "image size too big: " << m_size << dendl; + on_finish->complete(-EINVAL); + return; + } else if (m_l1_table.size > header.l1_size) { + lderr(cct) << "invalid L1 table size in header (" << header.l1_size + << " < " << m_l1_table.size << ")" << dendl; + on_finish->complete(-EINVAL); + return; + } + + m_snapshot_count = header.nb_snapshots; + m_snapshots_offset = header.snapshots_offset; + + ldout(cct, 15) << "size=" << m_size << ", " + << "cluster_bits=" << m_cluster_bits << ", " + << "l1_table_offset=" << m_l1_table_offset << ", " + << "snapshot_count=" << m_snapshot_count << ", " + << "snapshots_offset=" << m_snapshots_offset << dendl; + + // allocate memory for L1 table and L2 + cluster caches + m_l2_table_cache = std::make_unique<L2TableCache>(this); + m_cluster_cache = std::make_unique<ClusterCache>(this); + + read_snapshot(on_finish); +} + +template <typename I> +void QCOWFormat<I>::read_snapshot(Context* on_finish) { + if (m_snapshots_offset == 0 || m_snapshots.size() == m_snapshot_count) { + read_l1_table(on_finish); + return; + } + + // header is always aligned on 8 byte boundary + m_snapshots_offset = p2roundup(m_snapshots_offset, static_cast<uint64_t>(8)); + + auto cct = m_image_ctx->cct; + ldout(cct, 10) << "snap_id=" << (m_snapshots.size() + 1) << ", " + << "offset=" << m_snapshots_offset << dendl; + + auto ctx = new LambdaContext([this, on_finish](int r) { + handle_read_snapshot(r, on_finish); }); + m_bl.clear(); + m_stream->read({{m_snapshots_offset, sizeof(QCowSnapshotHeader)}}, &m_bl, + ctx); +} + +template <typename I> +void QCOWFormat<I>::handle_read_snapshot(int r, Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << "r=" << r << ", " + << "index=" << m_snapshots.size() << dendl; + + if (r < 0) { + lderr(cct) << "failed to read QCOW2 snapshot header: " << cpp_strerror(r) + << dendl; + on_finish->complete(r); + return; + } + + m_snapshots_offset += m_bl.length(); + auto header = *reinterpret_cast<QCowSnapshotHeader*>(m_bl.c_str()); + + auto& snapshot = m_snapshots[m_snapshots.size() + 1]; + snapshot.id.resize(big_to_native(header.id_str_size)); + snapshot.name.resize(big_to_native(header.name_size)); + snapshot.l1_table_offset = big_to_native(header.l1_table_offset); + snapshot.l1_table.size = big_to_native(header.l1_size); + snapshot.timestamp.sec_ref() = big_to_native(header.date_sec); + snapshot.timestamp.nsec_ref() = big_to_native(header.date_nsec); + snapshot.extra_data_size = big_to_native(header.extra_data_size); + + ldout(cct, 10) << "snap_id=" << m_snapshots.size() << ", " + << "id_str_len=" << snapshot.id.size() << ", " + << "name_str_len=" << snapshot.name.size() << ", " + << "l1_table_offset=" << snapshot.l1_table_offset << ", " + << "l1_size=" << snapshot.l1_table.size << ", " + << "extra_data_size=" << snapshot.extra_data_size << dendl; + + read_snapshot_extra(on_finish); +} + +template <typename I> +void QCOWFormat<I>::read_snapshot_extra(Context* on_finish) { + ceph_assert(!m_snapshots.empty()); + auto& snapshot = m_snapshots.rbegin()->second; + + uint32_t length = snapshot.extra_data_size + + snapshot.id.size() + + snapshot.name.size(); + if (length == 0) { + uuid_d uuid_gen; + uuid_gen.generate_random(); + snapshot.name = uuid_gen.to_string(); + + read_snapshot(on_finish); + return; + } + + auto cct = m_image_ctx->cct; + ldout(cct, 10) << "snap_id=" << m_snapshots.size() << ", " + << "offset=" << m_snapshots_offset << ", " + << "length=" << length << dendl; + + auto offset = m_snapshots_offset; + m_snapshots_offset += length; + + auto ctx = new LambdaContext([this, on_finish](int r) { + handle_read_snapshot_extra(r, on_finish); }); + m_bl.clear(); + m_stream->read({{offset, length}}, &m_bl, ctx); +} + +template <typename I> +void QCOWFormat<I>::handle_read_snapshot_extra(int r, Context* on_finish) { + ceph_assert(!m_snapshots.empty()); + auto& snapshot = m_snapshots.rbegin()->second; + + auto cct = m_image_ctx->cct; + ldout(cct, 10) << "r=" << r << ", " + << "snap_id=" << m_snapshots.size() << dendl; + + if (r < 0) { + lderr(cct) << "failed to read QCOW2 snapshot header extra: " + << cpp_strerror(r) << dendl; + on_finish->complete(r); + return; + } + + if (snapshot.extra_data_size >= + offsetof(QCowSnapshotExtraData, disk_size) + sizeof(uint64_t)) { + auto extra = reinterpret_cast<const QCowSnapshotExtraData*>(m_bl.c_str()); + snapshot.size = big_to_native(extra->disk_size); + } else { + snapshot.size = m_size; + } + + auto data = reinterpret_cast<const char*>(m_bl.c_str()); + data += snapshot.extra_data_size; + + if (!snapshot.id.empty()) { + snapshot.id = std::string(data, snapshot.id.size()); + data += snapshot.id.size(); + } + + if (!snapshot.name.empty()) { + snapshot.name = std::string(data, snapshot.name.size()); + data += snapshot.name.size(); + } else { + uuid_d uuid_gen; + uuid_gen.generate_random(); + snapshot.name = uuid_gen.to_string(); + } + + ldout(cct, 10) << "snap_id=" << m_snapshots.size() << ", " + << "name=" << snapshot.name << ", " + << "size=" << snapshot.size << dendl; + read_snapshot_l1_table(on_finish); +} + +template <typename I> +void QCOWFormat<I>::read_snapshot_l1_table(Context* on_finish) { + ceph_assert(!m_snapshots.empty()); + auto& snapshot = m_snapshots.rbegin()->second; + + auto cct = m_image_ctx->cct; + ldout(cct, 10) << "snap_id=" << m_snapshots.size() << ", " + << "l1_table_offset=" << snapshot.l1_table_offset + << dendl; + + auto ctx = new LambdaContext([this, on_finish](int r) { + handle_read_snapshot_l1_table(r, on_finish); }); + m_stream->read({{snapshot.l1_table_offset, + snapshot.l1_table.size * sizeof(uint64_t)}}, + &snapshot.l1_table.bl, ctx); +} + +template <typename I> +void QCOWFormat<I>::handle_read_snapshot_l1_table(int r, Context* on_finish) { + ceph_assert(!m_snapshots.empty()); + auto& snapshot = m_snapshots.rbegin()->second; + + auto cct = m_image_ctx->cct; + ldout(cct, 10) << "r=" << r << ", " + << "snap_id=" << m_snapshots.size() << dendl; + + if (r < 0) { + lderr(cct) << "failed to read snapshot L1 table: " << cpp_strerror(r) + << dendl; + on_finish->complete(r); + return; + } + + snapshot.l1_table.decode(); + read_snapshot(on_finish); +} + +template <typename I> +void QCOWFormat<I>::read_l1_table(Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + auto ctx = new LambdaContext([this, on_finish](int r) { + handle_read_l1_table(r, on_finish); }); + m_stream->read({{m_l1_table_offset, + m_l1_table.size * sizeof(uint64_t)}}, + &m_l1_table.bl, ctx); +} + +template <typename I> +void QCOWFormat<I>::handle_read_l1_table(int r, Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << "r=" << r << dendl; + + if (r < 0) { + lderr(cct) << "failed to read L1 table: " << cpp_strerror(r) << dendl; + on_finish->complete(r); + return; + } + + m_l1_table.decode(); + read_backing_file(on_finish); +} + +template <typename I> +void QCOWFormat<I>::read_backing_file(Context* on_finish) { + if (m_backing_file_offset == 0 || m_backing_file_size == 0) { + // all data is within the specified file + on_finish->complete(0); + return; + } + + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + // TODO add support for backing files + on_finish->complete(-ENOTSUP); +} + +template <typename I> +void QCOWFormat<I>::close(Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + m_stream->close(on_finish); +} + +template <typename I> +void QCOWFormat<I>::get_snapshots(SnapInfos* snap_infos, Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + snap_infos->clear(); + for (auto& [snap_id, snapshot] : m_snapshots) { + SnapInfo snap_info(snapshot.name, cls::rbd::UserSnapshotNamespace{}, + snapshot.size, {}, 0, 0, snapshot.timestamp); + snap_infos->emplace(snap_id, snap_info); + } + + on_finish->complete(0); +} + +template <typename I> +void QCOWFormat<I>::get_image_size(uint64_t snap_id, uint64_t* size, + Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << "snap_id=" << snap_id << dendl; + + if (snap_id == CEPH_NOSNAP) { + *size = m_size; + } else { + auto snapshot_it = m_snapshots.find(snap_id); + if (snapshot_it == m_snapshots.end()) { + on_finish->complete(-ENOENT); + return; + } + + auto& snapshot = snapshot_it->second; + *size = snapshot.size; + } + + on_finish->complete(0); +} + +template <typename I> +bool QCOWFormat<I>::read( + io::AioCompletion* aio_comp, uint64_t snap_id, io::Extents&& image_extents, + io::ReadResult&& read_result, int op_flags, int read_flags, + const ZTracer::Trace &parent_trace) { + auto cct = m_image_ctx->cct; + ldout(cct, 20) << "snap_id=" << snap_id << ", " + << "image_extents=" << image_extents << dendl; + + const LookupTable* l1_table = nullptr; + if (snap_id == CEPH_NOSNAP) { + l1_table = &m_l1_table; + } else { + auto snapshot_it = m_snapshots.find(snap_id); + if (snapshot_it == m_snapshots.end()) { + aio_comp->fail(-ENOENT); + return true; + } + + auto& snapshot = snapshot_it->second; + l1_table = &snapshot.l1_table; + } + + aio_comp->read_result = std::move(read_result); + aio_comp->read_result.set_image_extents(image_extents); + + auto read_request = new ReadRequest(this, aio_comp, l1_table, + std::move(image_extents)); + read_request->send(); + + return true; +} + +template <typename I> +void QCOWFormat<I>::list_snaps(io::Extents&& image_extents, + io::SnapIds&& snap_ids, int list_snaps_flags, + io::SnapshotDelta* snapshot_delta, + const ZTracer::Trace &parent_trace, + Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 20) << "image_extents=" << image_extents << dendl; + + ClusterExtents cluster_extents; + populate_cluster_extents(cct, m_cluster_size, image_extents, + &cluster_extents); + + // map L1 table indexes to cluster extents + std::map<uint64_t, ClusterExtents> l1_cluster_extents; + for (auto& cluster_extent : cluster_extents) { + uint32_t l1_table_index = cluster_extent.image_offset >> m_l1_shift; + auto& l1_cluster_extent = l1_cluster_extents[l1_table_index]; + l1_cluster_extent.reserve(cluster_extents.size()); + l1_cluster_extent.push_back(cluster_extent); + } + + std::map<uint64_t, const LookupTable*> snap_id_to_l1_table; + for (auto& [snap_id, snapshot] : m_snapshots) { + snap_id_to_l1_table[snap_id] = &snapshot.l1_table; + } + snap_id_to_l1_table[CEPH_NOSNAP] = &m_l1_table; + + on_finish = new LambdaContext([this, image_extents, + snap_ids=std::move(snap_ids), + snapshot_delta, on_finish](int r) mutable { + handle_list_snaps(r, std::move(image_extents), std::move(snap_ids), + snapshot_delta, on_finish); + }); + + auto gather_ctx = new C_Gather(cct, on_finish); + + for (auto& [l1_table_index, cluster_extents] : l1_cluster_extents) { + auto list_snaps_request = new ListSnapsRequest( + this, l1_table_index, std::move(cluster_extents), snap_id_to_l1_table, + snapshot_delta, gather_ctx->new_sub()); + list_snaps_request->send(); + } + + gather_ctx->activate(); +} + +template <typename I> +void QCOWFormat<I>::handle_list_snaps(int r, io::Extents&& image_extents, + io::SnapIds&& snap_ids, + io::SnapshotDelta* snapshot_delta, + Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 20) << "r=" << r << ", " + << "snapshot_delta=" << *snapshot_delta << dendl; + + std::optional<uint64_t> previous_size = std::nullopt; + for (auto& [snap_id, snapshot] : m_snapshots) { + auto sparse_extents = &(*snapshot_delta)[{snap_id, snap_id}]; + util::zero_shrunk_snapshot(cct, image_extents, snap_id, snapshot.size, + &previous_size, sparse_extents); + } + + auto sparse_extents = &(*snapshot_delta)[{CEPH_NOSNAP, CEPH_NOSNAP}]; + util::zero_shrunk_snapshot(cct, image_extents, CEPH_NOSNAP, m_size, + &previous_size, sparse_extents); + + util::merge_snapshot_delta(snap_ids, snapshot_delta); + on_finish->complete(r); +} + +} // namespace migration +} // namespace librbd + +template class librbd::migration::QCOWFormat<librbd::ImageCtx>; diff --git a/src/librbd/migration/QCOWFormat.h b/src/librbd/migration/QCOWFormat.h new file mode 100644 index 000000000..b36506716 --- /dev/null +++ b/src/librbd/migration/QCOWFormat.h @@ -0,0 +1,211 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_MIGRATION_QCOW_FORMAT_H +#define CEPH_LIBRBD_MIGRATION_QCOW_FORMAT_H + +#include "include/int_types.h" +#include "librbd/Types.h" +#include "librbd/migration/FormatInterface.h" +#include "librbd/migration/QCOW.h" +#include "acconfig.h" +#include "json_spirit/json_spirit.h" +#include <boost/asio/io_context_strand.hpp> +#include <boost/iostreams/filter/zlib.hpp> +#include <deque> +#include <vector> +#include <memory> + +struct Context; + +namespace librbd { + +struct AsioEngine; +struct ImageCtx; + +namespace migration { + +template <typename> struct SourceSpecBuilder; +struct StreamInterface; + +namespace qcow_format { + +struct LookupTable { + LookupTable() {} + LookupTable(uint32_t size) : size(size) {} + + bufferlist bl; + uint64_t* cluster_offsets = nullptr; + uint32_t size = 0; + bool decoded = false; + + void init(); + void decode(); +}; + +} // namespace qcow_format + +template <typename ImageCtxT> +class QCOWFormat : public FormatInterface { +public: + static QCOWFormat* create( + ImageCtxT* image_ctx, const json_spirit::mObject& json_object, + const SourceSpecBuilder<ImageCtxT>* source_spec_builder) { + return new QCOWFormat(image_ctx, json_object, source_spec_builder); + } + + QCOWFormat(ImageCtxT* image_ctx, const json_spirit::mObject& json_object, + const SourceSpecBuilder<ImageCtxT>* source_spec_builder); + QCOWFormat(const QCOWFormat&) = delete; + QCOWFormat& operator=(const QCOWFormat&) = delete; + + void open(Context* on_finish) override; + void close(Context* on_finish) override; + + void get_snapshots(SnapInfos* snap_infos, Context* on_finish) override; + void get_image_size(uint64_t snap_id, uint64_t* size, + Context* on_finish) override; + + bool read(io::AioCompletion* aio_comp, uint64_t snap_id, + io::Extents&& image_extents, io::ReadResult&& read_result, + int op_flags, int read_flags, + const ZTracer::Trace &parent_trace) override; + + void list_snaps(io::Extents&& image_extents, io::SnapIds&& snap_ids, + int list_snaps_flags, io::SnapshotDelta* snapshot_delta, + const ZTracer::Trace &parent_trace, + Context* on_finish) override; + +private: + /** + * @verbatim + * + * <start> + * | + * v + * OPEN + * | + * v + * PROBE + * | + * |\---> READ V1 HEADER ----------\ + * | | + * \----> READ V2 HEADER | + * | | + * | /----------\ | + * | | | | + * v v | | + * READ SNAPSHOT | | + * | | | + * v | | + * READ SNAPSHOT EXTRA | | + * | | | + * v | | + * READ SNAPSHOT L1 TABLE | + * | | + * \--------------------\| + * | + * v + * READ L1 TABLE + * | + * v + * READ BACKING FILE + * | + * /-------------------------------/ + * | + * v + * <opened> + * + * @endverbatim + */ + + struct Cluster; + struct ClusterCache; + struct L2TableCache; + struct ReadRequest; + struct ListSnapsRequest; + + struct Snapshot { + std::string id; + std::string name; + + utime_t timestamp; + uint64_t size = 0; + + uint64_t l1_table_offset = 0; + qcow_format::LookupTable l1_table; + + uint32_t extra_data_size = 0; + }; + + ImageCtxT* m_image_ctx; + json_spirit::mObject m_json_object; + const SourceSpecBuilder<ImageCtxT>* m_source_spec_builder; + + boost::asio::io_context::strand m_strand; + std::shared_ptr<StreamInterface> m_stream; + + bufferlist m_bl; + + uint64_t m_size = 0; + + uint64_t m_backing_file_offset = 0; + uint32_t m_backing_file_size = 0; + + uint32_t m_cluster_bits = 0; + uint32_t m_cluster_size = 0; + uint64_t m_cluster_offset_mask = 0; + uint64_t m_cluster_mask = 0; + + uint32_t m_l1_shift = 0; + uint64_t m_l1_table_offset = 0; + qcow_format::LookupTable m_l1_table; + + uint32_t m_l2_bits = 0; + uint32_t m_l2_size = 0; + + uint32_t m_snapshot_count = 0; + uint64_t m_snapshots_offset = 0; + std::map<uint64_t, Snapshot> m_snapshots; + + std::unique_ptr<L2TableCache> m_l2_table_cache; + std::unique_ptr<ClusterCache> m_cluster_cache; + + void handle_open(int r, Context* on_finish); + + void probe(Context* on_finish); + void handle_probe(int r, Context* on_finish); + +#ifdef WITH_RBD_MIGRATION_FORMAT_QCOW_V1 + void read_v1_header(Context* on_finish); + void handle_read_v1_header(int r, Context* on_finish); +#endif // WITH_RBD_MIGRATION_FORMAT_QCOW_V1 + + void read_v2_header(Context* on_finish); + void handle_read_v2_header(int r, Context* on_finish); + + void read_snapshot(Context* on_finish); + void handle_read_snapshot(int r, Context* on_finish); + + void read_snapshot_extra(Context* on_finish); + void handle_read_snapshot_extra(int r, Context* on_finish); + + void read_snapshot_l1_table(Context* on_finish); + void handle_read_snapshot_l1_table(int r, Context* on_finish); + + void read_l1_table(Context* on_finish); + void handle_read_l1_table(int r, Context* on_finish); + + void read_backing_file(Context* on_finish); + + void handle_list_snaps(int r, io::Extents&& image_extents, + io::SnapIds&& snap_ids, + io::SnapshotDelta* snapshot_delta, Context* on_finish); +}; + +} // namespace migration +} // namespace librbd + +extern template class librbd::migration::QCOWFormat<librbd::ImageCtx>; + +#endif // CEPH_LIBRBD_MIGRATION_QCOW_FORMAT_H diff --git a/src/librbd/migration/RawFormat.cc b/src/librbd/migration/RawFormat.cc new file mode 100644 index 000000000..0b655d368 --- /dev/null +++ b/src/librbd/migration/RawFormat.cc @@ -0,0 +1,235 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "librbd/migration/RawFormat.h" +#include "common/dout.h" +#include "common/errno.h" +#include "librbd/ImageCtx.h" +#include "librbd/ImageState.h" +#include "librbd/Utils.h" +#include "librbd/io/AioCompletion.h" +#include "librbd/io/ReadResult.h" +#include "librbd/migration/SnapshotInterface.h" +#include "librbd/migration/SourceSpecBuilder.h" +#include "librbd/migration/Utils.h" + +namespace librbd { +namespace migration { + +namespace { + +static const std::string SNAPSHOTS_KEY {"snapshots"}; + + +} // anonymous namespace + +#define dout_subsys ceph_subsys_rbd +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::RawFormat: " << this \ + << " " << __func__ << ": " + +template <typename I> +RawFormat<I>::RawFormat( + I* image_ctx, const json_spirit::mObject& json_object, + const SourceSpecBuilder<I>* source_spec_builder) + : m_image_ctx(image_ctx), m_json_object(json_object), + m_source_spec_builder(source_spec_builder) { +} + +template <typename I> +void RawFormat<I>::open(Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + on_finish = new LambdaContext([this, on_finish](int r) { + handle_open(r, on_finish); }); + + // treat the base image as a HEAD-revision snapshot + Snapshots snapshots; + int r = m_source_spec_builder->build_snapshot(m_json_object, CEPH_NOSNAP, + &snapshots[CEPH_NOSNAP]); + if (r < 0) { + lderr(cct) << "failed to build HEAD revision handler: " << cpp_strerror(r) + << dendl; + on_finish->complete(r); + return; + } + + auto& snapshots_val = m_json_object[SNAPSHOTS_KEY]; + if (snapshots_val.type() == json_spirit::array_type) { + auto& snapshots_arr = snapshots_val.get_array(); + for (auto& snapshot_val : snapshots_arr) { + uint64_t index = snapshots.size(); + if (snapshot_val.type() != json_spirit::obj_type) { + lderr(cct) << "invalid snapshot " << index << " JSON: " + << cpp_strerror(r) << dendl; + on_finish->complete(-EINVAL); + return; + } + + auto& snapshot_obj = snapshot_val.get_obj(); + r = m_source_spec_builder->build_snapshot(snapshot_obj, index, + &snapshots[index]); + if (r < 0) { + lderr(cct) << "failed to build snapshot " << index << " handler: " + << cpp_strerror(r) << dendl; + on_finish->complete(r); + return; + } + } + } else if (snapshots_val.type() != json_spirit::null_type) { + lderr(cct) << "invalid snapshots array" << dendl; + on_finish->complete(-EINVAL); + return; + } + + m_snapshots = std::move(snapshots); + + auto gather_ctx = new C_Gather(cct, on_finish); + SnapshotInterface* previous_snapshot = nullptr; + for (auto& [_, snapshot] : m_snapshots) { + snapshot->open(previous_snapshot, gather_ctx->new_sub()); + previous_snapshot = snapshot.get(); + } + gather_ctx->activate(); +} + +template <typename I> +void RawFormat<I>::handle_open(int r, Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << "r=" << r << dendl; + + if (r < 0) { + lderr(cct) << "failed to open raw image: " << cpp_strerror(r) + << dendl; + + auto gather_ctx = new C_Gather(cct, on_finish); + for (auto& [_, snapshot] : m_snapshots) { + snapshot->close(gather_ctx->new_sub()); + } + + m_image_ctx->state->close(new LambdaContext( + [r, on_finish=gather_ctx->new_sub()](int _) { on_finish->complete(r); })); + + gather_ctx->activate(); + return; + } + + on_finish->complete(0); +} + +template <typename I> +void RawFormat<I>::close(Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + auto gather_ctx = new C_Gather(cct, on_finish); + for (auto& [snap_id, snapshot] : m_snapshots) { + snapshot->close(gather_ctx->new_sub()); + } + + gather_ctx->activate(); +} + +template <typename I> +void RawFormat<I>::get_snapshots(SnapInfos* snap_infos, Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + snap_infos->clear(); + for (auto& [snap_id, snapshot] : m_snapshots) { + if (snap_id == CEPH_NOSNAP) { + continue; + } + snap_infos->emplace(snap_id, snapshot->get_snap_info()); + } + on_finish->complete(0); +} + +template <typename I> +void RawFormat<I>::get_image_size(uint64_t snap_id, uint64_t* size, + Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + auto snapshot_it = m_snapshots.find(snap_id); + if (snapshot_it == m_snapshots.end()) { + on_finish->complete(-ENOENT); + return; + } + + *size = snapshot_it->second->get_snap_info().size; + on_finish->complete(0); +} + +template <typename I> +bool RawFormat<I>::read( + io::AioCompletion* aio_comp, uint64_t snap_id, io::Extents&& image_extents, + io::ReadResult&& read_result, int op_flags, int read_flags, + const ZTracer::Trace &parent_trace) { + auto cct = m_image_ctx->cct; + ldout(cct, 20) << "snap_id=" << snap_id << ", " + << "image_extents=" << image_extents << dendl; + + auto snapshot_it = m_snapshots.find(snap_id); + if (snapshot_it == m_snapshots.end()) { + aio_comp->fail(-ENOENT); + return true; + } + + snapshot_it->second->read(aio_comp, std::move(image_extents), + std::move(read_result), op_flags, read_flags, + parent_trace); + return true; +} + +template <typename I> +void RawFormat<I>::list_snaps(io::Extents&& image_extents, + io::SnapIds&& snap_ids, int list_snaps_flags, + io::SnapshotDelta* snapshot_delta, + const ZTracer::Trace &parent_trace, + Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 20) << "image_extents=" << image_extents << dendl; + + on_finish = new LambdaContext([this, snap_ids=std::move(snap_ids), + snapshot_delta, on_finish](int r) mutable { + handle_list_snaps(r, std::move(snap_ids), snapshot_delta, on_finish); + }); + + auto gather_ctx = new C_Gather(cct, on_finish); + + std::optional<uint64_t> previous_size = std::nullopt; + for (auto& [snap_id, snapshot] : m_snapshots) { + auto& sparse_extents = (*snapshot_delta)[{snap_id, snap_id}]; + + // zero out any space between the previous snapshot end and this + // snapshot's end + auto& snap_info = snapshot->get_snap_info(); + util::zero_shrunk_snapshot(cct, image_extents, snap_id, snap_info.size, + &previous_size, &sparse_extents); + + // build set of data/zeroed extents for the current snapshot + snapshot->list_snap(io::Extents{image_extents}, list_snaps_flags, + &sparse_extents, parent_trace, gather_ctx->new_sub()); + } + + gather_ctx->activate(); +} + +template <typename I> +void RawFormat<I>::handle_list_snaps(int r, io::SnapIds&& snap_ids, + io::SnapshotDelta* snapshot_delta, + Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 20) << "r=" << r << ", " + << "snapshot_delta=" << snapshot_delta << dendl; + + util::merge_snapshot_delta(snap_ids, snapshot_delta); + on_finish->complete(r); +} + +} // namespace migration +} // namespace librbd + +template class librbd::migration::RawFormat<librbd::ImageCtx>; diff --git a/src/librbd/migration/RawFormat.h b/src/librbd/migration/RawFormat.h new file mode 100644 index 000000000..a20c0814f --- /dev/null +++ b/src/librbd/migration/RawFormat.h @@ -0,0 +1,78 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_MIGRATION_RAW_FORMAT_H +#define CEPH_LIBRBD_MIGRATION_RAW_FORMAT_H + +#include "include/int_types.h" +#include "librbd/Types.h" +#include "librbd/migration/FormatInterface.h" +#include "json_spirit/json_spirit.h" +#include <map> +#include <memory> + +struct Context; + +namespace librbd { + +struct AsioEngine; +struct ImageCtx; + +namespace migration { + +template <typename> struct SourceSpecBuilder; +struct SnapshotInterface; + +template <typename ImageCtxT> +class RawFormat : public FormatInterface { +public: + static RawFormat* create( + ImageCtxT* image_ctx, const json_spirit::mObject& json_object, + const SourceSpecBuilder<ImageCtxT>* source_spec_builder) { + return new RawFormat(image_ctx, json_object, source_spec_builder); + } + + RawFormat(ImageCtxT* image_ctx, const json_spirit::mObject& json_object, + const SourceSpecBuilder<ImageCtxT>* source_spec_builder); + RawFormat(const RawFormat&) = delete; + RawFormat& operator=(const RawFormat&) = delete; + + void open(Context* on_finish) override; + void close(Context* on_finish) override; + + void get_snapshots(SnapInfos* snap_infos, Context* on_finish) override; + void get_image_size(uint64_t snap_id, uint64_t* size, + Context* on_finish) override; + + bool read(io::AioCompletion* aio_comp, uint64_t snap_id, + io::Extents&& image_extents, io::ReadResult&& read_result, + int op_flags, int read_flags, + const ZTracer::Trace &parent_trace) override; + + void list_snaps(io::Extents&& image_extents, io::SnapIds&& snap_ids, + int list_snaps_flags, io::SnapshotDelta* snapshot_delta, + const ZTracer::Trace &parent_trace, + Context* on_finish) override; + +private: + typedef std::shared_ptr<SnapshotInterface> Snapshot; + typedef std::map<uint64_t, Snapshot> Snapshots; + + ImageCtxT* m_image_ctx; + json_spirit::mObject m_json_object; + const SourceSpecBuilder<ImageCtxT>* m_source_spec_builder; + + Snapshots m_snapshots; + + void handle_open(int r, Context* on_finish); + + void handle_list_snaps(int r, io::SnapIds&& snap_ids, + io::SnapshotDelta* snapshot_delta, Context* on_finish); +}; + +} // namespace migration +} // namespace librbd + +extern template class librbd::migration::RawFormat<librbd::ImageCtx>; + +#endif // CEPH_LIBRBD_MIGRATION_RAW_FORMAT_H diff --git a/src/librbd/migration/RawSnapshot.cc b/src/librbd/migration/RawSnapshot.cc new file mode 100644 index 000000000..4a83fd8ad --- /dev/null +++ b/src/librbd/migration/RawSnapshot.cc @@ -0,0 +1,220 @@ +// -*- mode:c++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "librbd/migration/RawSnapshot.h" +#include "common/dout.h" +#include "common/errno.h" +#include "librbd/ImageCtx.h" +#include "librbd/Utils.h" +#include "librbd/io/AioCompletion.h" +#include "librbd/io/ReadResult.h" +#include "librbd/migration/SourceSpecBuilder.h" +#include "librbd/migration/StreamInterface.h" + +namespace librbd { +namespace migration { + +namespace { + +const std::string NAME_KEY{"name"}; + +} // anonymous namespace + +#define dout_subsys ceph_subsys_rbd +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::RawSnapshot::OpenRequest " \ + << this << " " << __func__ << ": " + +template <typename I> +struct RawSnapshot<I>::OpenRequest { + RawSnapshot* raw_snapshot; + Context* on_finish; + + OpenRequest(RawSnapshot* raw_snapshot, Context* on_finish) + : raw_snapshot(raw_snapshot), on_finish(on_finish) { + } + + void send() { + open_stream(); + } + + void open_stream() { + auto cct = raw_snapshot->m_image_ctx->cct; + ldout(cct, 10) << dendl; + + auto ctx = util::create_context_callback< + OpenRequest, &OpenRequest::handle_open_stream>(this); + raw_snapshot->m_stream->open(ctx); + } + + void handle_open_stream(int r) { + auto cct = raw_snapshot->m_image_ctx->cct; + ldout(cct, 10) << "r=" << r << dendl; + + if (r < 0) { + lderr(cct) << "failed to open stream: " << cpp_strerror(r) << dendl; + finish(r); + return; + } + + get_image_size(); + } + + void get_image_size() { + auto cct = raw_snapshot->m_image_ctx->cct; + ldout(cct, 10) << dendl; + + auto ctx = util::create_context_callback< + OpenRequest, &OpenRequest::handle_get_image_size>(this); + raw_snapshot->m_stream->get_size(&raw_snapshot->m_snap_info.size, ctx); + } + + void handle_get_image_size(int r) { + auto cct = raw_snapshot->m_image_ctx->cct; + ldout(cct, 10) << "r=" << r << ", " + << "image_size=" << raw_snapshot->m_snap_info.size << dendl; + + if (r < 0) { + lderr(cct) << "failed to open stream: " << cpp_strerror(r) << dendl; + close_stream(r); + return; + } + + finish(0); + } + + void close_stream(int r) { + auto cct = raw_snapshot->m_image_ctx->cct; + ldout(cct, 10) << dendl; + + auto ctx = new LambdaContext([this, r](int) { + handle_close_stream(r); + }); + raw_snapshot->m_stream->close(ctx); + } + + void handle_close_stream(int r) { + auto cct = raw_snapshot->m_image_ctx->cct; + ldout(cct, 10) << "r=" << r << dendl; + + raw_snapshot->m_stream.reset(); + + finish(r); + } + + void finish(int r) { + auto cct = raw_snapshot->m_image_ctx->cct; + ldout(cct, 10) << "r=" << r << dendl; + + on_finish->complete(r); + delete this; + } +}; + +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::RawSnapshot: " << this \ + << " " << __func__ << ": " + +template <typename I> +RawSnapshot<I>::RawSnapshot(I* image_ctx, + const json_spirit::mObject& json_object, + const SourceSpecBuilder<I>* source_spec_builder, + uint64_t index) + : m_image_ctx(image_ctx), m_json_object(json_object), + m_source_spec_builder(source_spec_builder), m_index(index), + m_snap_info({}, {}, 0, {}, 0, 0, {}) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; +} + +template <typename I> +void RawSnapshot<I>::open(SnapshotInterface* previous_snapshot, + Context* on_finish) { + auto cct = m_image_ctx->cct; + + // special-case for treating the HEAD revision as a snapshot + if (m_index != CEPH_NOSNAP) { + auto& name_val = m_json_object[NAME_KEY]; + if (name_val.type() == json_spirit::str_type) { + m_snap_info.name = name_val.get_str(); + } else if (name_val.type() == json_spirit::null_type) { + uuid_d uuid_gen; + uuid_gen.generate_random(); + + m_snap_info.name = uuid_gen.to_string(); + } else { + lderr(cct) << "invalid snapshot name" << dendl; + on_finish->complete(-EINVAL); + return; + } + } + + ldout(cct, 10) << "name=" << m_snap_info.name << dendl; + + int r = m_source_spec_builder->build_stream(m_json_object, &m_stream); + if (r < 0) { + lderr(cct) << "failed to build migration stream handler" << cpp_strerror(r) + << dendl; + on_finish->complete(r); + return; + } + + auto req = new OpenRequest(this, on_finish); + req->send(); +} + +template <typename I> +void RawSnapshot<I>::close(Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + if (!m_stream) { + on_finish->complete(0); + return; + } + + m_stream->close(on_finish); +} + +template <typename I> +void RawSnapshot<I>::read(io::AioCompletion* aio_comp, + io::Extents&& image_extents, + io::ReadResult&& read_result, int op_flags, + int read_flags, + const ZTracer::Trace &parent_trace) { + auto cct = m_image_ctx->cct; + ldout(cct, 20) << "image_extents=" << image_extents << dendl; + + aio_comp->read_result = std::move(read_result); + aio_comp->read_result.set_image_extents(image_extents); + + aio_comp->set_request_count(1); + auto ctx = new io::ReadResult::C_ImageReadRequest(aio_comp, + 0, image_extents); + + // raw directly maps the image-extent IO down to a byte IO extent + m_stream->read(std::move(image_extents), &ctx->bl, ctx); +} + +template <typename I> +void RawSnapshot<I>::list_snap(io::Extents&& image_extents, + int list_snaps_flags, + io::SparseExtents* sparse_extents, + const ZTracer::Trace &parent_trace, + Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 20) << "image_extents=" << image_extents << dendl; + + // raw does support sparse extents so list the full IO extent as a delta + for (auto& [image_offset, image_length] : image_extents) { + sparse_extents->insert(image_offset, image_length, + {io::SPARSE_EXTENT_STATE_DATA, image_length}); + } + + on_finish->complete(0); +} + +} // namespace migration +} // namespace librbd + +template class librbd::migration::RawSnapshot<librbd::ImageCtx>; diff --git a/src/librbd/migration/RawSnapshot.h b/src/librbd/migration/RawSnapshot.h new file mode 100644 index 000000000..9f76d6878 --- /dev/null +++ b/src/librbd/migration/RawSnapshot.h @@ -0,0 +1,75 @@ +// -*- mode:c++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_MIGRATION_RAW_SNAPSHOT_H +#define CEPH_LIBRBD_MIGRATION_RAW_SNAPSHOT_H + +#include "include/buffer_fwd.h" +#include "include/int_types.h" +#include "common/zipkin_trace.h" +#include "librbd/Types.h" +#include "librbd/io/Types.h" +#include "librbd/migration/SnapshotInterface.h" +#include "json_spirit/json_spirit.h" +#include <memory> + +namespace librbd { + +struct ImageCtx; + +namespace migration { + +template <typename> struct SourceSpecBuilder; +struct StreamInterface; + +template <typename ImageCtxT> +class RawSnapshot : public SnapshotInterface { +public: + static RawSnapshot* create( + ImageCtx* image_ctx, const json_spirit::mObject& json_object, + const SourceSpecBuilder<ImageCtxT>* source_spec_builder, uint64_t index) { + return new RawSnapshot(image_ctx, json_object, source_spec_builder, index); + } + + RawSnapshot(ImageCtxT* image_ctx, const json_spirit::mObject& json_object, + const SourceSpecBuilder<ImageCtxT>* source_spec_builder, + uint64_t index); + RawSnapshot(const RawSnapshot&) = delete; + RawSnapshot& operator=(const RawSnapshot&) = delete; + + void open(SnapshotInterface* previous_snapshot, Context* on_finish) override; + void close(Context* on_finish) override; + + const SnapInfo& get_snap_info() const override { + return m_snap_info; + } + + void read(io::AioCompletion* aio_comp, io::Extents&& image_extents, + io::ReadResult&& read_result, int op_flags, int read_flags, + const ZTracer::Trace &parent_trace) override; + + void list_snap(io::Extents&& image_extents, int list_snaps_flags, + io::SparseExtents* sparse_extents, + const ZTracer::Trace &parent_trace, + Context* on_finish) override; + +private: + struct OpenRequest; + + ImageCtxT* m_image_ctx; + json_spirit::mObject m_json_object; + const SourceSpecBuilder<ImageCtxT>* m_source_spec_builder; + uint64_t m_index = 0; + + SnapInfo m_snap_info; + + std::shared_ptr<StreamInterface> m_stream; + +}; + +} // namespace migration +} // namespace librbd + +extern template class librbd::migration::RawSnapshot<librbd::ImageCtx>; + +#endif // CEPH_LIBRBD_MIGRATION_RAW_SNAPSHOT_H diff --git a/src/librbd/migration/S3Stream.cc b/src/librbd/migration/S3Stream.cc new file mode 100644 index 000000000..a611e274a --- /dev/null +++ b/src/librbd/migration/S3Stream.cc @@ -0,0 +1,200 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "librbd/migration/S3Stream.h" +#include "common/armor.h" +#include "common/ceph_crypto.h" +#include "common/ceph_time.h" +#include "common/dout.h" +#include "common/errno.h" +#include "librbd/AsioEngine.h" +#include "librbd/ImageCtx.h" +#include "librbd/Utils.h" +#include "librbd/asio/Utils.h" +#include "librbd/io/AioCompletion.h" +#include "librbd/io/ReadResult.h" +#include "librbd/migration/HttpClient.h" +#include "librbd/migration/HttpProcessorInterface.h" +#include <boost/beast/http.hpp> + +#include <fmt/chrono.h> +#include <fmt/format.h> + +#include <time.h> + +namespace librbd { +namespace migration { + +using HttpRequest = boost::beast::http::request<boost::beast::http::empty_body>; + +namespace { + +const std::string URL_KEY {"url"}; +const std::string ACCESS_KEY {"access_key"}; +const std::string SECRET_KEY {"secret_key"}; + +} // anonymous namespace + +template <typename I> +struct S3Stream<I>::HttpProcessor : public HttpProcessorInterface { + S3Stream* s3stream; + + HttpProcessor(S3Stream* s3stream) : s3stream(s3stream) { + } + + void process_request(EmptyRequest& request) override { + s3stream->process_request(request); + } +}; + +#define dout_subsys ceph_subsys_rbd +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::S3Stream: " << this \ + << " " << __func__ << ": " + +template <typename I> +S3Stream<I>::S3Stream(I* image_ctx, const json_spirit::mObject& json_object) + : m_image_ctx(image_ctx), m_cct(image_ctx->cct), + m_asio_engine(image_ctx->asio_engine), m_json_object(json_object), + m_http_processor(std::make_unique<HttpProcessor>(this)) { +} + +template <typename I> +S3Stream<I>::~S3Stream() { +} + +template <typename I> +void S3Stream<I>::open(Context* on_finish) { + auto& url_value = m_json_object[URL_KEY]; + if (url_value.type() != json_spirit::str_type) { + lderr(m_cct) << "failed to locate '" << URL_KEY << "' key" << dendl; + on_finish->complete(-EINVAL); + return; + } + + auto& access_key = m_json_object[ACCESS_KEY]; + if (access_key.type() != json_spirit::str_type) { + lderr(m_cct) << "failed to locate '" << ACCESS_KEY << "' key" << dendl; + on_finish->complete(-EINVAL); + return; + } + + auto& secret_key = m_json_object[SECRET_KEY]; + if (secret_key.type() != json_spirit::str_type) { + lderr(m_cct) << "failed to locate '" << SECRET_KEY << "' key" << dendl; + on_finish->complete(-EINVAL); + return; + } + + m_url = url_value.get_str(); + + librados::Rados rados(m_image_ctx->md_ctx); + int r = 0; + m_access_key = access_key.get_str(); + if (util::is_config_key_uri(m_access_key)) { + r = util::get_config_key(rados, m_access_key, &m_access_key); + if (r < 0) { + lderr(m_cct) << "failed to retrieve access key from config: " + << cpp_strerror(r) << dendl; + on_finish->complete(r); + return; + } + } + + m_secret_key = secret_key.get_str(); + if (util::is_config_key_uri(m_secret_key)) { + r = util::get_config_key(rados, m_secret_key, &m_secret_key); + if (r < 0) { + lderr(m_cct) << "failed to retrieve secret key from config: " + << cpp_strerror(r) << dendl; + on_finish->complete(r); + return; + } + } + + ldout(m_cct, 10) << "url=" << m_url << ", " + << "access_key=" << m_access_key << dendl; + + m_http_client.reset(HttpClient<I>::create(m_image_ctx, m_url)); + m_http_client->set_http_processor(m_http_processor.get()); + m_http_client->open(on_finish); +} + +template <typename I> +void S3Stream<I>::close(Context* on_finish) { + ldout(m_cct, 10) << dendl; + + if (!m_http_client) { + on_finish->complete(0); + return; + } + + m_http_client->close(on_finish); +} + +template <typename I> +void S3Stream<I>::get_size(uint64_t* size, Context* on_finish) { + ldout(m_cct, 10) << dendl; + + m_http_client->get_size(size, on_finish); +} + +template <typename I> +void S3Stream<I>::read(io::Extents&& byte_extents, bufferlist* data, + Context* on_finish) { + ldout(m_cct, 20) << "byte_extents=" << byte_extents << dendl; + + m_http_client->read(std::move(byte_extents), data, on_finish); +} + +template <typename I> +void S3Stream<I>::process_request(HttpRequest& http_request) { + ldout(m_cct, 20) << dendl; + + // format RFC 1123 date/time + auto time = ceph::real_clock::to_time_t(ceph::real_clock::now()); + struct tm timeInfo; + gmtime_r(&time, &timeInfo); + + std::string date = fmt::format("{:%a, %d %b %Y %H:%M:%S %z}", timeInfo); + http_request.set(boost::beast::http::field::date, date); + + // note: we don't support S3 subresources + std::string canonicalized_resource = std::string(http_request.target()); + + std::string string_to_sign = fmt::format( + "{}\n\n\n{}\n{}", + std::string(boost::beast::http::to_string(http_request.method())), + date, canonicalized_resource); + + // create HMAC-SHA1 signature from secret key + string-to-sign + sha1_digest_t digest; + ceph::crypto::HMACSHA1 hmac( + reinterpret_cast<const unsigned char*>(m_secret_key.data()), + m_secret_key.size()); + hmac.Update(reinterpret_cast<const unsigned char*>(string_to_sign.data()), + string_to_sign.size()); + hmac.Final(reinterpret_cast<unsigned char*>(digest.v)); + + // base64 encode the result + char buf[64]; + int r = ceph_armor(std::begin(buf), std::begin(buf) + sizeof(buf), + reinterpret_cast<const char *>(digest.v), + reinterpret_cast<const char *>(digest.v + digest.SIZE)); + if (r < 0) { + ceph_abort("ceph_armor failed"); + } + + // store the access-key + signature in the HTTP authorization header + std::string signature = std::string(std::begin(buf), std::begin(buf) + r); + std::string authorization = fmt::format("AWS {}:{}", m_access_key, signature); + http_request.set(boost::beast::http::field::authorization, authorization); + + ldout(m_cct, 20) << "string_to_sign=" << string_to_sign << ", " + << "authorization=" << authorization << dendl; +} + +} // namespace migration +} // namespace librbd + +template class librbd::migration::S3Stream<librbd::ImageCtx>; diff --git a/src/librbd/migration/S3Stream.h b/src/librbd/migration/S3Stream.h new file mode 100644 index 000000000..586b21787 --- /dev/null +++ b/src/librbd/migration/S3Stream.h @@ -0,0 +1,78 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_MIGRATION_S3_STREAM_H +#define CEPH_LIBRBD_MIGRATION_S3_STREAM_H + +#include "include/int_types.h" +#include "librbd/migration/StreamInterface.h" +#include <boost/beast/http/empty_body.hpp> +#include <boost/beast/http/message.hpp> +#include <boost/beast/http/string_body.hpp> +#include <json_spirit/json_spirit.h> +#include <memory> +#include <string> + +struct Context; + +namespace librbd { + +struct AsioEngine; +struct ImageCtx; + +namespace migration { + +template <typename> class HttpClient; + +template <typename ImageCtxT> +class S3Stream : public StreamInterface { +public: + static S3Stream* create(ImageCtxT* image_ctx, + const json_spirit::mObject& json_object) { + return new S3Stream(image_ctx, json_object); + } + + S3Stream(ImageCtxT* image_ctx, const json_spirit::mObject& json_object); + ~S3Stream() override; + + S3Stream(const S3Stream&) = delete; + S3Stream& operator=(const S3Stream&) = delete; + + void open(Context* on_finish) override; + void close(Context* on_finish) override; + + void get_size(uint64_t* size, Context* on_finish) override; + + void read(io::Extents&& byte_extents, bufferlist* data, + Context* on_finish) override; + +private: + using HttpRequest = boost::beast::http::request< + boost::beast::http::empty_body>; + using HttpResponse = boost::beast::http::response< + boost::beast::http::string_body>; + + struct HttpProcessor; + + ImageCtxT* m_image_ctx; + CephContext* m_cct; + std::shared_ptr<AsioEngine> m_asio_engine; + json_spirit::mObject m_json_object; + + std::string m_url; + std::string m_access_key; + std::string m_secret_key; + + std::unique_ptr<HttpProcessor> m_http_processor; + std::unique_ptr<HttpClient<ImageCtxT>> m_http_client; + + void process_request(HttpRequest& http_request); + +}; + +} // namespace migration +} // namespace librbd + +extern template class librbd::migration::S3Stream<librbd::ImageCtx>; + +#endif // CEPH_LIBRBD_MIGRATION_S3_STREAM_H diff --git a/src/librbd/migration/SnapshotInterface.h b/src/librbd/migration/SnapshotInterface.h new file mode 100644 index 000000000..9990802c5 --- /dev/null +++ b/src/librbd/migration/SnapshotInterface.h @@ -0,0 +1,48 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_MIGRATION_SNAPSHOT_INTERFACE_H +#define CEPH_LIBRBD_MIGRATION_SNAPSHOT_INTERFACE_H + +#include "include/buffer_fwd.h" +#include "include/int_types.h" +#include "common/zipkin_trace.h" +#include "librbd/Types.h" +#include "librbd/io/Types.h" +#include <string> + +struct Context; + +namespace librbd { + +namespace io { +struct AioCompletion; +struct ReadResult; +} // namespace io + +namespace migration { + +struct SnapshotInterface { + virtual ~SnapshotInterface() { + } + + virtual void open(SnapshotInterface* previous_snapshot, + Context* on_finish) = 0; + virtual void close(Context* on_finish) = 0; + + virtual const SnapInfo& get_snap_info() const = 0; + + virtual void read(io::AioCompletion* aio_comp, io::Extents&& image_extents, + io::ReadResult&& read_result, int op_flags, int read_flags, + const ZTracer::Trace &parent_trace) = 0; + + virtual void list_snap(io::Extents&& image_extents, int list_snaps_flags, + io::SparseExtents* sparse_extents, + const ZTracer::Trace &parent_trace, + Context* on_finish) = 0; +}; + +} // namespace migration +} // namespace librbd + +#endif // CEPH_LIBRBD_MIGRATION_SNAPSHOT_INTERFACE_H diff --git a/src/librbd/migration/SourceSpecBuilder.cc b/src/librbd/migration/SourceSpecBuilder.cc new file mode 100644 index 000000000..214d7ce0e --- /dev/null +++ b/src/librbd/migration/SourceSpecBuilder.cc @@ -0,0 +1,147 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "librbd/migration/SourceSpecBuilder.h" +#include "common/dout.h" +#include "librbd/ImageCtx.h" +#include "librbd/migration/FileStream.h" +#include "librbd/migration/HttpStream.h" +#include "librbd/migration/S3Stream.h" +#include "librbd/migration/NativeFormat.h" +#include "librbd/migration/QCOWFormat.h" +#include "librbd/migration/RawFormat.h" +#include "librbd/migration/RawSnapshot.h" + +#define dout_subsys ceph_subsys_rbd +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::SourceSpecBuilder: " << this \ + << " " << __func__ << ": " + +namespace librbd { +namespace migration { + +namespace { + +const std::string STREAM_KEY{"stream"}; +const std::string TYPE_KEY{"type"}; + +} // anonymous namespace + +template <typename I> +int SourceSpecBuilder<I>::parse_source_spec( + const std::string& source_spec, + json_spirit::mObject* source_spec_object) const { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + json_spirit::mValue json_root; + if(json_spirit::read(source_spec, json_root)) { + try { + *source_spec_object = json_root.get_obj(); + return 0; + } catch (std::runtime_error&) { + } + } + + lderr(cct) << "invalid source-spec JSON" << dendl; + return -EBADMSG; +} + +template <typename I> +int SourceSpecBuilder<I>::build_format( + const json_spirit::mObject& source_spec_object, bool import_only, + std::unique_ptr<FormatInterface>* format) const { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + auto type_value_it = source_spec_object.find(TYPE_KEY); + if (type_value_it == source_spec_object.end() || + type_value_it->second.type() != json_spirit::str_type) { + lderr(cct) << "failed to locate format type value" << dendl; + return -EINVAL; + } + + auto& type = type_value_it->second.get_str(); + if (type == "native") { + format->reset(NativeFormat<I>::create(m_image_ctx, source_spec_object, + import_only)); + } else if (type == "qcow") { + format->reset(QCOWFormat<I>::create(m_image_ctx, source_spec_object, this)); + } else if (type == "raw") { + format->reset(RawFormat<I>::create(m_image_ctx, source_spec_object, this)); + } else { + lderr(cct) << "unknown or unsupported format type '" << type << "'" + << dendl; + return -ENOSYS; + } + return 0; +} + +template <typename I> +int SourceSpecBuilder<I>::build_snapshot( + const json_spirit::mObject& source_spec_object, uint64_t index, + std::shared_ptr<SnapshotInterface>* snapshot) const { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + auto type_value_it = source_spec_object.find(TYPE_KEY); + if (type_value_it == source_spec_object.end() || + type_value_it->second.type() != json_spirit::str_type) { + lderr(cct) << "failed to locate snapshot type value" << dendl; + return -EINVAL; + } + + auto& type = type_value_it->second.get_str(); + if (type == "raw") { + snapshot->reset(RawSnapshot<I>::create(m_image_ctx, source_spec_object, + this, index)); + } else { + lderr(cct) << "unknown or unsupported format type '" << type << "'" + << dendl; + return -ENOSYS; + } + return 0; +} + +template <typename I> +int SourceSpecBuilder<I>::build_stream( + const json_spirit::mObject& source_spec_object, + std::shared_ptr<StreamInterface>* stream) const { + auto cct = m_image_ctx->cct; + ldout(cct, 10) << dendl; + + auto stream_value_it = source_spec_object.find(STREAM_KEY); + if (stream_value_it == source_spec_object.end() || + stream_value_it->second.type() != json_spirit::obj_type) { + lderr(cct) << "failed to locate stream object" << dendl; + return -EINVAL; + } + + auto& stream_obj = stream_value_it->second.get_obj(); + auto type_value_it = stream_obj.find(TYPE_KEY); + if (type_value_it == stream_obj.end() || + type_value_it->second.type() != json_spirit::str_type) { + lderr(cct) << "failed to locate stream type value" << dendl; + return -EINVAL; + } + + auto& type = type_value_it->second.get_str(); + if (type == "file") { + stream->reset(FileStream<I>::create(m_image_ctx, stream_obj)); + } else if (type == "http") { + stream->reset(HttpStream<I>::create(m_image_ctx, stream_obj)); + } else if (type == "s3") { + stream->reset(S3Stream<I>::create(m_image_ctx, stream_obj)); + } else { + lderr(cct) << "unknown or unsupported stream type '" << type << "'" + << dendl; + return -ENOSYS; + } + + return 0; +} + +} // namespace migration +} // namespace librbd + +template class librbd::migration::SourceSpecBuilder<librbd::ImageCtx>; diff --git a/src/librbd/migration/SourceSpecBuilder.h b/src/librbd/migration/SourceSpecBuilder.h new file mode 100644 index 000000000..191cb1cbd --- /dev/null +++ b/src/librbd/migration/SourceSpecBuilder.h @@ -0,0 +1,54 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_MIGRATION_SOURCE_SPEC_BUILDER_H +#define CEPH_LIBRBD_MIGRATION_SOURCE_SPEC_BUILDER_H + +#include "include/int_types.h" +#include <json_spirit/json_spirit.h> +#include <memory> +#include <optional> +#include <string> + +struct Context; + +namespace librbd { + +struct ImageCtx; + +namespace migration { + +struct FormatInterface; +struct SnapshotInterface; +struct StreamInterface; + +template <typename ImageCtxT> +class SourceSpecBuilder { +public: + SourceSpecBuilder(ImageCtxT* image_ctx) : m_image_ctx(image_ctx) { + } + + int parse_source_spec(const std::string& source_spec, + json_spirit::mObject* source_spec_object) const; + + int build_format(const json_spirit::mObject& format_object, bool import_only, + std::unique_ptr<FormatInterface>* format) const; + + int build_snapshot(const json_spirit::mObject& source_spec_object, + uint64_t index, + std::shared_ptr<SnapshotInterface>* snapshot) const; + + int build_stream(const json_spirit::mObject& source_spec_object, + std::shared_ptr<StreamInterface>* stream) const; + +private: + ImageCtxT* m_image_ctx; + +}; + +} // namespace migration +} // namespace librbd + +extern template class librbd::migration::SourceSpecBuilder<librbd::ImageCtx>; + +#endif // CEPH_LIBRBD_MIGRATION_SOURCE_SPEC_BUILDER_H diff --git a/src/librbd/migration/StreamInterface.h b/src/librbd/migration/StreamInterface.h new file mode 100644 index 000000000..782a9a5f8 --- /dev/null +++ b/src/librbd/migration/StreamInterface.h @@ -0,0 +1,32 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_MIGRATION_STREAM_INTERFACE_H +#define CEPH_LIBRBD_MIGRATION_STREAM_INTERFACE_H + +#include "include/buffer_fwd.h" +#include "include/int_types.h" +#include "librbd/io/Types.h" + +struct Context; + +namespace librbd { +namespace migration { + +struct StreamInterface { + virtual ~StreamInterface() { + } + + virtual void open(Context* on_finish) = 0; + virtual void close(Context* on_finish) = 0; + + virtual void get_size(uint64_t* size, Context* on_finish) = 0; + + virtual void read(io::Extents&& byte_extents, bufferlist* data, + Context* on_finish) = 0; +}; + +} // namespace migration +} // namespace librbd + +#endif // CEPH_LIBRBD_MIGRATION_STREAM_INTERFACE_H diff --git a/src/librbd/migration/Types.h b/src/librbd/migration/Types.h new file mode 100644 index 000000000..244dc28b7 --- /dev/null +++ b/src/librbd/migration/Types.h @@ -0,0 +1,42 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_MIGRATION_TYPES_H +#define CEPH_LIBRBD_MIGRATION_TYPES_H + +#include <string> +#include <utility> + +namespace librbd { +namespace migration { + +enum UrlScheme { + URL_SCHEME_HTTP, + URL_SCHEME_HTTPS, +}; + +struct UrlSpec { + UrlSpec() {} + UrlSpec(UrlScheme scheme, const std::string& host, const std::string& port, + const std::string& path) + : scheme(scheme), host(host), port(port), path(path) { + } + + UrlScheme scheme = URL_SCHEME_HTTP; + std::string host; + std::string port = "80"; + std::string path = "/"; + +}; + +inline bool operator==(const UrlSpec& lhs, const UrlSpec& rhs) { + return (lhs.scheme == rhs.scheme && + lhs.host == rhs.host && + lhs.port == rhs.port && + lhs.path == rhs.path); +} + +} // namespace migration +} // namespace librbd + +#endif // CEPH_LIBRBD_MIGRATION_TYPES_H diff --git a/src/librbd/migration/Utils.cc b/src/librbd/migration/Utils.cc new file mode 100644 index 000000000..c5c1279d8 --- /dev/null +++ b/src/librbd/migration/Utils.cc @@ -0,0 +1,133 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "librbd/migration/Utils.h" +#include "common/dout.h" +#include "common/errno.h" +#include <boost/lexical_cast.hpp> +#include <regex> + +namespace librbd { +namespace migration { +namespace util { + +#define dout_subsys ceph_subsys_rbd +#undef dout_prefix +#define dout_prefix *_dout << "librbd::migration::util::" << __func__ << ": " + +int parse_url(CephContext* cct, const std::string& url, UrlSpec* url_spec) { + ldout(cct, 10) << "url=" << url << dendl; + *url_spec = UrlSpec{}; + + // parse the provided URL (scheme, user, password, host, port, path, + // parameters, query, and fragment) + std::regex url_regex( + R"(^(?:([^:/]*)://)?(?:(\w+)(?::(\w+))?@)?([^/;\?:#]+)(?::([^/;\?#]+))?)" + R"((?:/([^;\?#]*))?(?:;([^\?#]+))?(?:\?([^#]+))?(?:#(\w+))?$)"); + std::smatch match; + if(!std::regex_match(url, match, url_regex)) { + lderr(cct) << "invalid url: '" << url << "'" << dendl; + return -EINVAL; + } + + auto& scheme = match[1]; + if (scheme == "http" || scheme == "") { + url_spec->scheme = URL_SCHEME_HTTP; + } else if (scheme == "https") { + url_spec->scheme = URL_SCHEME_HTTPS; + url_spec->port = "443"; + } else { + lderr(cct) << "invalid url scheme: '" << url << "'" << dendl; + return -EINVAL; + } + + url_spec->host = match[4]; + auto& port = match[5]; + if (port.matched) { + try { + boost::lexical_cast<uint16_t>(port); + } catch (boost::bad_lexical_cast&) { + lderr(cct) << "invalid url port: '" << url << "'" << dendl; + return -EINVAL; + } + url_spec->port = port; + } + + auto& path = match[6]; + if (path.matched) { + url_spec->path += path; + } + return 0; +} + +void zero_shrunk_snapshot(CephContext* cct, const io::Extents& image_extents, + uint64_t snap_id, uint64_t new_size, + std::optional<uint64_t> *previous_size, + io::SparseExtents* sparse_extents) { + if (*previous_size && **previous_size > new_size) { + ldout(cct, 20) << "snapshot resize " << **previous_size << " -> " + << new_size << dendl; + interval_set<uint64_t> zero_interval; + zero_interval.insert(new_size, **previous_size - new_size); + + for (auto& image_extent : image_extents) { + interval_set<uint64_t> image_interval; + image_interval.insert(image_extent.first, image_extent.second); + + image_interval.intersection_of(zero_interval); + for (auto [image_offset, image_length] : image_interval) { + ldout(cct, 20) << "zeroing extent " << image_offset << "~" + << image_length << " at snapshot " << snap_id << dendl; + sparse_extents->insert(image_offset, image_length, + {io::SPARSE_EXTENT_STATE_ZEROED, image_length}); + } + } + } + *previous_size = new_size; +} + +void merge_snapshot_delta(const io::SnapIds& snap_ids, + io::SnapshotDelta* snapshot_delta) { + io::SnapshotDelta orig_snapshot_delta = std::move(*snapshot_delta); + snapshot_delta->clear(); + + auto snap_id_it = snap_ids.begin(); + ceph_assert(snap_id_it != snap_ids.end()); + + // merge any snapshot intervals that were not requested + std::list<io::SparseExtents*> pending_sparse_extents; + for (auto& [snap_key, sparse_extents] : orig_snapshot_delta) { + // advance to next valid requested snap id + while (snap_id_it != snap_ids.end() && *snap_id_it < snap_key.first) { + ++snap_id_it; + } + if (snap_id_it == snap_ids.end()) { + break; + } + + // loop through older write/read snapshot sparse extents to remove any + // overlaps with the current sparse extent + for (auto prev_sparse_extents : pending_sparse_extents) { + for (auto& sparse_extent : sparse_extents) { + prev_sparse_extents->erase(sparse_extent.get_off(), + sparse_extent.get_len()); + } + } + + auto write_read_snap_ids = std::make_pair(*snap_id_it, snap_key.second); + (*snapshot_delta)[write_read_snap_ids] = std::move(sparse_extents); + + if (write_read_snap_ids.first > snap_key.first) { + // the current snapshot wasn't requested so it might need to get + // merged with a later snapshot + pending_sparse_extents.push_back(&(*snapshot_delta)[write_read_snap_ids]); + } else { + // we don't merge results passed a valid requested snapshot + pending_sparse_extents.clear(); + } + } +} + +} // namespace util +} // namespace migration +} // namespace librbd diff --git a/src/librbd/migration/Utils.h b/src/librbd/migration/Utils.h new file mode 100644 index 000000000..afbadde7d --- /dev/null +++ b/src/librbd/migration/Utils.h @@ -0,0 +1,30 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_MIGRATION_UTILS_H +#define CEPH_LIBRBD_MIGRATION_UTILS_H + +#include "include/common_fwd.h" +#include "librbd/io/Types.h" +#include "librbd/migration/Types.h" +#include <optional> +#include <string> + +namespace librbd { +namespace migration { +namespace util { + +int parse_url(CephContext* cct, const std::string& url, UrlSpec* url_spec); + +void zero_shrunk_snapshot(CephContext* cct, const io::Extents& image_extents, + uint64_t snap_id, uint64_t new_size, + std::optional<uint64_t> *previous_size, + io::SparseExtents* sparse_extents); +void merge_snapshot_delta(const io::SnapIds& snap_ids, + io::SnapshotDelta* snapshot_delta); + +} // namespace util +} // namespace migration +} // namespace librbd + +#endif // CEPH_LIBRBD_MIGRATION_UTILS_H |