From: Jason Dillaman <dillaman@redhat.com>
Date: Fri, 13 Nov 2020 15:51:51 +0000 (-0500)
Subject: librbd/migration: implement a basic S3 stream source
X-Git-Tag: v16.1.0~479^2~1
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=b58bb486d639ee18e87d83cb9ae29d73c7585fa1;p=ceph.git

librbd/migration: implement a basic S3 stream source

The new S3 stream requires a fully qualified URL to a bucket
(i.e. no virtual host names) along with the access and secret
keys to the bucket.

Signed-off-by: Jason Dillaman <dillaman@redhat.com>
---

diff --git a/src/librbd/CMakeLists.txt b/src/librbd/CMakeLists.txt
index 6f7977fdae5b..eac8fc13243d 100644
--- a/src/librbd/CMakeLists.txt
+++ b/src/librbd/CMakeLists.txt
@@ -133,6 +133,7 @@ set(librbd_internal_srcs
   migration/NativeFormat.cc
   migration/OpenSourceImageRequest.cc
   migration/RawFormat.cc
+  migration/S3Stream.cc
   migration/SourceSpecBuilder.cc
   migration/Utils.cc
   mirror/DemoteRequest.cc
diff --git a/src/librbd/migration/S3Stream.cc b/src/librbd/migration/S3Stream.cc
new file mode 100644
index 000000000000..f812ef0294ad
--- /dev/null
+++ b/src/librbd/migration/S3Stream.cc
@@ -0,0 +1,178 @@
+// -*- 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>
+
+#undef FMT_HEADER_ONLY
+#define FMT_HEADER_ONLY 1
+#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();
+  m_access_key = access_key.get_str();
+  m_secret_key = secret_key.get_str();
+  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);
+  }
+
+  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;
+  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 000000000000..586b217878c6
--- /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/SourceSpecBuilder.cc b/src/librbd/migration/SourceSpecBuilder.cc
index 45fdce039809..3500238f8864 100644
--- a/src/librbd/migration/SourceSpecBuilder.cc
+++ b/src/librbd/migration/SourceSpecBuilder.cc
@@ -6,6 +6,7 @@
 #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/RawFormat.h"
 
@@ -99,6 +100,8 @@ int SourceSpecBuilder<I>::build_stream(
     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;
diff --git a/src/test/librbd/CMakeLists.txt b/src/test/librbd/CMakeLists.txt
index 12d80ccecba3..227419712e01 100644
--- a/src/test/librbd/CMakeLists.txt
+++ b/src/test/librbd/CMakeLists.txt
@@ -93,6 +93,7 @@ set(unittest_librbd_srcs
   migration/test_mock_HttpClient.cc
   migration/test_mock_HttpStream.cc
   migration/test_mock_RawFormat.cc
+  migration/test_mock_S3Stream.cc
   migration/test_mock_Utils.cc
   mirror/snapshot/test_mock_CreateNonPrimaryRequest.cc
   mirror/snapshot/test_mock_CreatePrimaryRequest.cc
diff --git a/src/test/librbd/migration/test_mock_S3Stream.cc b/src/test/librbd/migration/test_mock_S3Stream.cc
new file mode 100644
index 000000000000..2f2097f7926a
--- /dev/null
+++ b/src/test/librbd/migration/test_mock_S3Stream.cc
@@ -0,0 +1,238 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "test/librbd/test_mock_fixture.h"
+#include "test/librbd/test_support.h"
+#include "include/rbd_types.h"
+#include "common/ceph_mutex.h"
+#include "librbd/migration/HttpClient.h"
+#include "librbd/migration/S3Stream.h"
+#include "gtest/gtest.h"
+#include "gmock/gmock.h"
+#include "json_spirit/json_spirit.h"
+#include <boost/algorithm/string/predicate.hpp>
+#include <boost/beast/http.hpp>
+
+namespace librbd {
+namespace {
+
+struct MockTestImageCtx : public MockImageCtx {
+  MockTestImageCtx(ImageCtx &image_ctx) : MockImageCtx(image_ctx) {
+  }
+};
+
+} // anonymous namespace
+
+namespace migration {
+
+template <>
+struct HttpClient<MockTestImageCtx> {
+  static HttpClient* s_instance;
+  static HttpClient* create(MockTestImageCtx*, const std::string&) {
+    ceph_assert(s_instance != nullptr);
+    return s_instance;
+  }
+
+  HttpProcessorInterface* http_processor = nullptr;
+  void set_http_processor(HttpProcessorInterface* http_processor) {
+    this->http_processor = http_processor;
+  }
+
+  MOCK_METHOD1(open, void(Context*));
+  MOCK_METHOD1(close, void(Context*));
+  MOCK_METHOD2(get_size, void(uint64_t*, Context*));
+  MOCK_METHOD3(do_read, void(const io::Extents&, bufferlist*, Context*));
+  void read(io::Extents&& extents, bufferlist* bl, Context* ctx) {
+    do_read(extents, bl, ctx);
+  }
+
+  HttpClient() {
+    s_instance = this;
+  }
+};
+
+HttpClient<MockTestImageCtx>* HttpClient<MockTestImageCtx>::s_instance = nullptr;
+
+} // namespace migration
+} // namespace librbd
+
+#include "librbd/migration/S3Stream.cc"
+
+namespace librbd {
+namespace migration {
+
+using ::testing::_;
+using ::testing::Invoke;
+using ::testing::InSequence;
+using ::testing::WithArgs;
+
+class TestMockMigrationS3Stream : public TestMockFixture {
+public:
+  typedef S3Stream<MockTestImageCtx> MockS3Stream;
+  typedef HttpClient<MockTestImageCtx> MockHttpClient;
+
+  using EmptyBody = boost::beast::http::empty_body;
+  using EmptyRequest = boost::beast::http::request<EmptyBody>;
+
+  librbd::ImageCtx *m_image_ctx;
+
+  void SetUp() override {
+    TestMockFixture::SetUp();
+
+    ASSERT_EQ(0, open_image(m_image_name, &m_image_ctx));
+    json_object["url"] = "http://some.site/bucket/file";
+    json_object["access_key"] = "0555b35654ad1656d804";
+    json_object["secret_key"] = "h7GhxuBLTrlhVUyxSPUKUV8r/2EI4ngqJxD7iBdBYLhwluN30JaT3Q==";
+  }
+
+  void expect_open(MockHttpClient& mock_http_client, int r) {
+    EXPECT_CALL(mock_http_client, open(_))
+      .WillOnce(Invoke([r](Context* ctx) { ctx->complete(r); }));
+  }
+
+  void expect_close(MockHttpClient& mock_http_client, int r) {
+    EXPECT_CALL(mock_http_client, close(_))
+      .WillOnce(Invoke([r](Context* ctx) { ctx->complete(r); }));
+  }
+
+  void expect_get_size(MockHttpClient& mock_http_client, uint64_t size, int r) {
+    EXPECT_CALL(mock_http_client, get_size(_, _))
+      .WillOnce(Invoke([size, r](uint64_t* out_size, Context* ctx) {
+        *out_size = size;
+        ctx->complete(r);
+      }));
+  }
+
+  void expect_read(MockHttpClient& mock_http_client, io::Extents byte_extents,
+                   const bufferlist& bl, int r) {
+    uint64_t len = 0;
+    for (auto [_, byte_len] : byte_extents) {
+      len += byte_len;
+    }
+    EXPECT_CALL(mock_http_client, do_read(byte_extents, _, _))
+      .WillOnce(WithArgs<1, 2>(Invoke(
+        [len, bl, r](bufferlist* out_bl, Context* ctx) {
+          *out_bl = bl;
+          ctx->complete(r < 0 ? r : len);
+        })));
+  }
+
+  json_spirit::mObject json_object;
+};
+
+TEST_F(TestMockMigrationS3Stream, OpenClose) {
+  MockTestImageCtx mock_image_ctx(*m_image_ctx);
+
+  InSequence seq;
+
+  auto mock_http_client = new MockHttpClient();
+  expect_open(*mock_http_client, 0);
+
+  expect_close(*mock_http_client, 0);
+
+  MockS3Stream mock_http_stream(&mock_image_ctx, json_object);
+
+  C_SaferCond ctx1;
+  mock_http_stream.open(&ctx1);
+  ASSERT_EQ(0, ctx1.wait());
+
+  C_SaferCond ctx2;
+  mock_http_stream.close(&ctx2);
+  ASSERT_EQ(0, ctx2.wait());
+}
+
+TEST_F(TestMockMigrationS3Stream, GetSize) {
+  MockTestImageCtx mock_image_ctx(*m_image_ctx);
+
+  InSequence seq;
+
+  auto mock_http_client = new MockHttpClient();
+  expect_open(*mock_http_client, 0);
+
+  expect_get_size(*mock_http_client, 128, 0);
+
+  expect_close(*mock_http_client, 0);
+
+  MockS3Stream mock_http_stream(&mock_image_ctx, json_object);
+
+  C_SaferCond ctx1;
+  mock_http_stream.open(&ctx1);
+  ASSERT_EQ(0, ctx1.wait());
+
+  C_SaferCond ctx2;
+  uint64_t size;
+  mock_http_stream.get_size(&size, &ctx2);
+  ASSERT_EQ(0, ctx2.wait());
+  ASSERT_EQ(128, size);
+
+  C_SaferCond ctx3;
+  mock_http_stream.close(&ctx3);
+  ASSERT_EQ(0, ctx3.wait());
+}
+
+TEST_F(TestMockMigrationS3Stream, Read) {
+  MockTestImageCtx mock_image_ctx(*m_image_ctx);
+
+  InSequence seq;
+
+  auto mock_http_client = new MockHttpClient();
+  expect_open(*mock_http_client, 0);
+
+  bufferlist expect_bl;
+  expect_bl.append(std::string(192, '1'));
+  expect_read(*mock_http_client, {{0, 128}, {256, 64}}, expect_bl, 0);
+
+  expect_close(*mock_http_client, 0);
+
+  MockS3Stream mock_http_stream(&mock_image_ctx, json_object);
+
+  C_SaferCond ctx1;
+  mock_http_stream.open(&ctx1);
+  ASSERT_EQ(0, ctx1.wait());
+
+  C_SaferCond ctx2;
+  bufferlist bl;
+  mock_http_stream.read({{0, 128}, {256, 64}}, &bl, &ctx2);
+  ASSERT_EQ(192, ctx2.wait());
+  ASSERT_EQ(expect_bl, bl);
+
+  C_SaferCond ctx3;
+  mock_http_stream.close(&ctx3);
+  ASSERT_EQ(0, ctx3.wait());
+}
+
+TEST_F(TestMockMigrationS3Stream, ProcessRequest) {
+  MockTestImageCtx mock_image_ctx(*m_image_ctx);
+
+  InSequence seq;
+
+  auto mock_http_client = new MockHttpClient();
+  expect_open(*mock_http_client, 0);
+
+  expect_close(*mock_http_client, 0);
+
+  MockS3Stream mock_http_stream(&mock_image_ctx, json_object);
+
+  C_SaferCond ctx1;
+  mock_http_stream.open(&ctx1);
+  ASSERT_EQ(0, ctx1.wait());
+
+  EmptyRequest request;
+  request.method(boost::beast::http::verb::get);
+  request.target("/bucket/resource");
+  mock_http_client->http_processor->process_request(request);
+
+  // basic test for date and known portion of authorization
+  ASSERT_EQ(1U, request.count(boost::beast::http::field::date));
+  ASSERT_EQ(1U, request.count(boost::beast::http::field::authorization));
+  ASSERT_TRUE(boost::algorithm::starts_with(
+    request[boost::beast::http::field::authorization],
+    "AWS 0555b35654ad1656d804:"));
+
+  C_SaferCond ctx2;
+  mock_http_stream.close(&ctx2);
+  ASSERT_EQ(0, ctx2.wait());
+}
+
+} // namespace migration
+} // namespace librbd