Merge pull request #9039 from dillaman/wip-15820

jewel: librbd: client-side handling for incompatible object map sizes

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

src/cls/rbd/cls_rbd.cc

@@ -138,7 +138,6 @@ cls_method_handle_t h_mirror_image_status_remove_down;
 #define RBD_DIR_ID_KEY_PREFIX "id_"
 #define RBD_DIR_NAME_KEY_PREFIX "name_"
 #define RBD_METADATA_KEY_PREFIX "metadata_"
-#define RBD_MAX_OBJECT_MAP_OBJECT_COUNT 256000000
 
 static int snap_read_header(cls_method_context_t hctx, bufferlist& bl)
 {
@@ -2268,7 +2267,7 @@ int object_map_resize(cls_method_context_t hctx, bufferlist *in, bufferlist *out
   }
 
   // protect against excessive memory requirements
-  if (object_count > RBD_MAX_OBJECT_MAP_OBJECT_COUNT) {
+  if (object_count > cls::rbd::MAX_OBJECT_MAP_OBJECT_COUNT) {
     CLS_ERR("object map too large: %" PRIu64, object_count);
     return -EINVAL;
   }

src/cls/rbd/cls_rbd_types.h

@@ -16,6 +16,8 @@ namespace ceph { class Formatter; }
 namespace cls {
 namespace rbd {
 
+static const uint32_t MAX_OBJECT_MAP_OBJECT_COUNT = 256000000;
+
 enum MirrorMode {
   MIRROR_MODE_DISABLED = 0,
   MIRROR_MODE_IMAGE    = 1,

src/librbd/ObjectMap.cc

@@ -17,8 +17,10 @@
 #include "common/dout.h"
 #include "common/errno.h"
 #include "common/WorkQueue.h"
-#include "include/stringify.h"
 #include "cls/lock/cls_lock_client.h"
+#include "cls/rbd/cls_rbd_types.h"
+#include "include/stringify.h"
+#include "osdc/Striper.h"
 #include <sstream>
 
 #define dout_subsys ceph_subsys_rbd
@@ -48,6 +50,11 @@ std::string ObjectMap::object_map_name(const std::string &image_id,
   return oid;
 }
 
+bool ObjectMap::is_compatible(const file_layout_t& layout, uint64_t size) {
+  uint64_t object_count = Striper::get_num_objects(layout, size);
+  return (object_count <= cls::rbd::MAX_OBJECT_MAP_OBJECT_COUNT);
+}
+
 ceph::BitVector<2u>::Reference ObjectMap::operator[](uint64_t object_no)
 {
   assert(m_image_ctx.object_map_lock.is_wlocked());

src/librbd/ObjectMap.h

@@ -4,6 +4,7 @@
 #define CEPH_LIBRBD_OBJECT_MAP_H
 
 #include "include/int_types.h"
+#include "include/fs_types.h"
 #include "include/rados/librados.hpp"
 #include "include/rbd/object_map_types.h"
 #include "common/bit_vector.hpp"
@@ -24,6 +25,8 @@ class ObjectMap {
   static std::string object_map_name(const std::string &image_id,
 				     uint64_t snap_id);
 
+  static bool is_compatible(const file_layout_t& layout, uint64_t size);
+
   ceph::BitVector<2u>::Reference operator[](uint64_t object_no);
   uint8_t operator[](uint64_t object_no) const;
   inline uint64_t size() const {

src/librbd/Operations.cc

@@ -9,6 +9,7 @@
 #include "librbd/ImageCtx.h"
 #include "librbd/ImageState.h"
 #include "librbd/ImageWatcher.h"
+#include "librbd/ObjectMap.h"
 #include "librbd/Utils.h"
 #include "librbd/operation/FlattenRequest.h"
 #include "librbd/operation/RebuildObjectMapRequest.h"
@@ -499,6 +500,12 @@ int Operations<I>::resize(uint64_t size, ProgressContext& prog_ctx) {
     return r;
   }
 
+  if (m_image_ctx.test_features(RBD_FEATURE_OBJECT_MAP) &&
+      !ObjectMap::is_compatible(m_image_ctx.layout, size)) {
+    lderr(cct) << "New size not compatible with object map" << dendl;
+    return -EINVAL;
+  }
+
   uint64_t request_id = ++m_async_request_seq;
   r = invoke_async_request("resize", false,
                            boost::bind(&Operations<I>::execute_resize, this,
@@ -525,13 +532,17 @@ void Operations<I>::execute_resize(uint64_t size, ProgressContext &prog_ctx,
   ldout(cct, 5) << this << " " << __func__ << ": "
                 << "size=" << m_image_ctx.size << ", "
                 << "new_size=" << size << dendl;
-  m_image_ctx.snap_lock.put_read();
 
-  m_image_ctx.snap_lock.get_read();
   if (m_image_ctx.snap_id != CEPH_NOSNAP || m_image_ctx.read_only) {
     m_image_ctx.snap_lock.put_read();
     on_finish->complete(-EROFS);
     return;
+  } else if (m_image_ctx.test_features(RBD_FEATURE_OBJECT_MAP,
+                                       m_image_ctx.snap_lock) &&
+             !ObjectMap::is_compatible(m_image_ctx.layout, size)) {
+    m_image_ctx.snap_lock.put_read();
+    on_finish->complete(-EINVAL);
+    return;
   }
   m_image_ctx.snap_lock.put_read();
 

src/librbd/exclusive_lock/AcquireRequest.cc

@@ -250,8 +250,15 @@ Context *AcquireRequest<I>::handle_open_object_map(int *ret_val) {
   CephContext *cct = m_image_ctx.cct;
   ldout(cct, 10) << __func__ << ": r=" << *ret_val << dendl;
 
-  // object map should never result in an error
-  assert(*ret_val == 0);
+  if (*ret_val < 0) {
+    lderr(cct) << "failed to open object map: " << cpp_strerror(*ret_val)
+               << dendl;
+
+    *ret_val = 0;
+    delete m_object_map;
+    m_object_map = nullptr;
+  }
+
   return send_open_journal();
 }
 

src/librbd/image/RefreshRequest.cc

@@ -607,7 +607,13 @@ Context *RefreshRequest<I>::handle_v2_open_object_map(int *result) {
   CephContext *cct = m_image_ctx.cct;
   ldout(cct, 10) << this << " " << __func__ << ": r=" << *result << dendl;
 
-  assert(*result == 0);
+  if (*result < 0) {
+    lderr(cct) << "failed to open object map: " << cpp_strerror(*result)
+               << dendl;
+    delete m_object_map;
+    m_object_map = nullptr;
+  }
+
   send_v2_open_journal();
   return nullptr;
 }

src/librbd/image/SetSnapRequest.cc

@@ -272,8 +272,12 @@ Context *SetSnapRequest<I>::handle_open_object_map(int *result) {
   CephContext *cct = m_image_ctx.cct;
   ldout(cct, 10) << __func__ << ": r=" << *result << dendl;
 
-  // object map should never report errors
-  assert(*result == 0);
+  if (*result < 0) {
+    lderr(cct) << "failed to open object map: " << cpp_strerror(*result)
+               << dendl;
+    delete m_object_map;
+    m_object_map = nullptr;
+  }
 
   *result = apply();
   if (*result < 0) {

src/librbd/internal.cc

@@ -95,18 +95,26 @@ int remove_object_map(ImageCtx *ictx) {
   }
   return 0;
 }
+
 int create_object_map(ImageCtx *ictx) {
   assert(ictx->snap_lock.is_locked());
   CephContext *cct = ictx->cct;
 
   int r;
+  uint64_t max_size = ictx->size;
   std::vector<uint64_t> snap_ids;
   snap_ids.push_back(CEPH_NOSNAP);
   for (std::map<snap_t, SnapInfo>::iterator it = ictx->snap_info.begin();
        it != ictx->snap_info.end(); ++it) {
+    max_size = MAX(max_size, it->second.size);
     snap_ids.push_back(it->first);
   }
 
+  if (!ObjectMap::is_compatible(ictx->layout, max_size)) {
+    lderr(cct) << "image size not compatible with object map" << dendl;
+    return -EINVAL;
+  }
+
   for (std::vector<uint64_t>::iterator it = snap_ids.begin();
     it != snap_ids.end(); ++it) {
     librados::ObjectWriteOperation op;
@@ -1098,6 +1106,11 @@ int mirror_image_disable_internal(ImageCtx *ictx, bool force) {
         layout.stripe_count = stripe_count;
       }
 
+      if (!ObjectMap::is_compatible(layout, size)) {
+        lderr(cct) << "image size not compatible with object map" << dendl;
+        goto err_remove_header;
+      }
+
       librados::ObjectWriteOperation op;
       cls_client::object_map_resize(&op, Striper::get_num_objects(layout, size),
                                     OBJECT_NONEXISTENT);

src/librbd/object_map/RefreshRequest.cc

@@ -3,9 +3,11 @@
 
 #include "librbd/object_map/RefreshRequest.h"
 #include "cls/rbd/cls_rbd_client.h"
+#include "cls/rbd/cls_rbd_types.h"
 #include "cls/lock/cls_lock_client.h"
 #include "common/dout.h"
 #include "common/errno.h"
+#include "common/WorkQueue.h"
 #include "librbd/ImageCtx.h"
 #include "librbd/ObjectMap.h"
 #include "librbd/object_map/InvalidateRequest.h"
@@ -42,6 +44,10 @@ void RefreshRequest<I>::send() {
       m_image_ctx.layout, m_image_ctx.get_image_size(m_snap_id));
   }
 
+
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << this << " " << __func__ << ": "
+                 << "object_count=" << m_object_count << dendl;
   send_lock();
 }
 
@@ -60,12 +66,15 @@ void RefreshRequest<I>::apply() {
 
 template <typename I>
 void RefreshRequest<I>::send_lock() {
-  if (m_snap_id != CEPH_NOSNAP) {
+  CephContext *cct = m_image_ctx.cct;
+  if (m_object_count > cls::rbd::MAX_OBJECT_MAP_OBJECT_COUNT) {
+    send_invalidate_and_close();
+    return;
+  } else if (m_snap_id != CEPH_NOSNAP) {
     send_load();
     return;
   }
 
-  CephContext *cct = m_image_ctx.cct;
   std::string oid(ObjectMap::object_map_name(m_image_ctx.id, m_snap_id));
   ldout(cct, 10) << this << " " << __func__ << ": oid=" << oid << dendl;
 
@@ -248,6 +257,35 @@ Context *RefreshRequest<I>::handle_resize(int *ret_val) {
   return m_on_finish;
 }
 
+template <typename I>
+void RefreshRequest<I>::send_invalidate_and_close() {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 10) << this << " " << __func__ << dendl;
+
+  using klass = RefreshRequest<I>;
+  Context *ctx = create_context_callback<
+    klass, &klass::handle_invalidate_and_close>(this);
+  InvalidateRequest<I> *req = InvalidateRequest<I>::create(
+    m_image_ctx, m_snap_id, false, ctx);
+
+  lderr(cct) << "object map too large: " << m_object_count << dendl;
+  RWLock::RLocker owner_locker(m_image_ctx.owner_lock);
+  RWLock::WLocker snap_locker(m_image_ctx.snap_lock);
+  req->send();
+}
+
+template <typename I>
+Context *RefreshRequest<I>::handle_invalidate_and_close(int *ret_val) {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 10) << this << " " << __func__ << ": r=" << *ret_val << dendl;
+
+  assert(*ret_val == 0);
+
+  *ret_val = -EFBIG;
+  m_object_map->clear();
+  return m_on_finish;
+}
+
 } // namespace object_map
 } // namespace librbd
 

src/librbd/object_map/RefreshRequest.h

@@ -28,24 +28,25 @@ class RefreshRequest {
    * @verbatim
    *
    * <start> -----> LOCK (skip if snapshot)
-   *                  |
-   *                  v  (other errors)
-   *                LOAD * * * * * * * > INVALIDATE ------------\
-   *                  |    *                                    |
-   *                  |    * (-EINVAL or too small)             |
-   *                  |    * * * * * * > INVALIDATE_AND_RESIZE  |
-   *                  |                      |              *   |
-   *                  |                      |              *   |
-   *                  |                      v              *   |
-   *                  |                    RESIZE           *   |
-   *                  |                      |              *   |
-   *                  |                      |  * * * * * * *   |
-   *                  |                      |  *               |
-   *                  |                      v  v               |
-   *                  \--------------------> LOCK <-------------/
-   *                                          |
-   *                                          v
-   *                                      <finish>
+   *    *             |
+   *    *             v  (other errors)
+   *    *           LOAD * * * * * * * > INVALIDATE ------------\
+   *    *             |    *                                    |
+   *    *             |    * (-EINVAL or too small)             |
+   *    *             |    * * * * * * > INVALIDATE_AND_RESIZE  |
+   *    *             |                      |              *   |
+   *    *             |                      |              *   |
+   *    *             |                      v              *   |
+   *    *             |                    RESIZE           *   |
+   *    *             |                      |              *   |
+   *    *             |                      |  * * * * * * *   |
+   *    *             |                      |  *               |
+   *    *             |                      v  v               |
+   *    *             \--------------------> LOCK <-------------/
+   *    *                                     |
+   *    v                                     v
+   * INVALIDATE_AND_CLOSE ---------------> <finish>
+   *
    * @endverbatim
    */
 
@@ -74,6 +75,9 @@ class RefreshRequest {
   void send_resize();
   Context *handle_resize(int *ret_val);
 
+  void send_invalidate_and_close();
+  Context *handle_invalidate_and_close(int *ret_val);
+
   void apply();
 };