ceph/src/osd/ScrubStore.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include "ScrubStore.h"
   5 #include "osd_types.h"
   6 #include "common/scrub_types.h"
   7 #include "include/rados/rados_types.hpp"
   8
   9 namespace {
  10 ghobject_t make_scrub_object(const spg_t& pgid)
  11 {
  12   ostringstream ss;
  13   ss << "scrub_" << pgid;
  14   return pgid.make_temp_ghobject(ss.str());
  15 }
  16
  17 string first_object_key(int64_t pool)
  18 {
  19   auto hoid = hobject_t(object_t(),
  20                         "",
  21                         0,
  22                         0x00000000,
  23                         pool,
  24                         "");
  25   hoid.build_hash_cache();
  26   return "SCRUB_OBJ_" + hoid.to_str();
  27 }
  28
  29 // the object_key should be unique across pools
  30 string to_object_key(int64_t pool, const librados::object_id_t& oid)
  31 {
  32   auto hoid = hobject_t(object_t(oid.name),
  33                         oid.locator, // key
  34                         oid.snap,
  35                         0,              // hash
  36                         pool,
  37                         oid.nspace);
  38   hoid.build_hash_cache();
  39   return "SCRUB_OBJ_" + hoid.to_str();
  40 }
  41
  42 string last_object_key(int64_t pool)
  43 {
  44   auto hoid = hobject_t(object_t(),
  45                         "",
  46                         0,
  47                         0xffffffff,
  48                         pool,
  49                         "");
  50   hoid.build_hash_cache();
  51   return "SCRUB_OBJ_" + hoid.to_str();
  52 }
  53
  54 string first_snap_key(int64_t pool)
  55 {
  56   // scrub object is per spg_t object, so we can misuse the hash (pg.seed) for
  57   // the representing the minimal and maximum keys. and this relies on how
  58   // hobject_t::to_str() works: hex(pool).hex(revhash).
  59   auto hoid = hobject_t(object_t(),
  60                         "",
  61                         0,
  62                         0x00000000,
  63                         pool,
  64                         "");
  65   hoid.build_hash_cache();
  66   return "SCRUB_SS_" + hoid.to_str();
  67 }
  68
  69 string to_snap_key(int64_t pool, const librados::object_id_t& oid)
  70 {
  71   auto hoid = hobject_t(object_t(oid.name),
  72                         oid.locator, // key
  73                         oid.snap,
  74                         0x77777777, // hash
  75                         pool,
  76                         oid.nspace);
  77   hoid.build_hash_cache();
  78   return "SCRUB_SS_" + hoid.to_str();
  79 }
  80
  81 string last_snap_key(int64_t pool)
  82 {
  83   auto hoid = hobject_t(object_t(),
  84                         "",
  85                         0,
  86                         0xffffffff,
  87                         pool,
  88                         "");
  89   hoid.build_hash_cache();
  90   return "SCRUB_SS_" + hoid.to_str();
  91 }
  92 }
  93
  94 namespace Scrub {
  95
  96 Store*
  97 Store::create(ObjectStore* store,
  98               ObjectStore::Transaction* t,
  99               const spg_t& pgid,
 100               const coll_t& coll)
 101 {
 102   assert(store);
 103   assert(t);
 104   ghobject_t oid = make_scrub_object(pgid);
 105   t->touch(coll, oid);
 106   return new Store{coll, oid, store};
 107 }
 108
 109 Store::Store(const coll_t& coll, const ghobject_t& oid, ObjectStore* store)
 110   : coll(coll),
 111     hoid(oid),
 112     driver(store, coll, hoid),
 113     backend(&driver)
 114 {}
 115
 116 Store::~Store()
 117 {
 118   assert(results.empty());
 119 }
 120
 121 void Store::add_object_error(int64_t pool, const inconsistent_obj_wrapper& e)
 122 {
 123   bufferlist bl;
 124   e.encode(bl);
 125   results[to_object_key(pool, e.object)] = bl;
 126 }
 127
 128 void Store::add_snap_error(int64_t pool, const inconsistent_snapset_wrapper& e)
 129 {
 130   bufferlist bl;
 131   e.encode(bl);
 132   results[to_snap_key(pool, e.object)] = bl;
 133 }
 134
 135 bool Store::empty() const
 136 {
 137   return results.empty();
 138 }
 139
 140 void Store::flush(ObjectStore::Transaction* t)
 141 {
 142   if (t) {
 143     OSDriver::OSTransaction txn = driver.get_transaction(t);
 144     backend.set_keys(results, &txn);
 145   }
 146   results.clear();
 147 }
 148
 149 void Store::cleanup(ObjectStore::Transaction* t)
 150 {
 151   t->remove(coll, hoid);
 152 }
 153
 154 std::vector<bufferlist>
 155 Store::get_snap_errors(ObjectStore* store,
 156                        int64_t pool,
 157                        const librados::object_id_t& start,
 158                        uint64_t max_return)
 159 {
 160   const string begin = (start.name.empty() ?
 161                         first_snap_key(pool) : to_snap_key(pool, start));
 162   const string end = last_snap_key(pool);
 163   return get_errors(store, begin, end, max_return);
 164 }
 165
 166 std::vector<bufferlist>
 167 Store::get_object_errors(ObjectStore* store,
 168                          int64_t pool,
 169                          const librados::object_id_t& start,
 170                          uint64_t max_return)
 171 {
 172   const string begin = (start.name.empty() ?
 173                         first_object_key(pool) : to_object_key(pool, start));
 174   const string end = last_object_key(pool);
 175   return get_errors(store, begin, end, max_return);
 176 }
 177
 178 std::vector<bufferlist>
 179 Store::get_errors(ObjectStore* store,
 180                   const string& begin,
 181                   const string& end,
 182                   uint64_t max_return)
 183 {
 184   vector<bufferlist> errors;
 185   auto next = std::make_pair(begin, bufferlist{});
 186   while (max_return && !backend.get_next(next.first, &next)) {
 187     if (next.first >= end)
 188       break;
 189     errors.push_back(next.second);
 190     max_return--;
 191   }
 192   return errors;
 193 }
 194
 195 } // namespace Scrub