ceph/src/rgw/rgw_orphan.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab ft=cpp
   3
   4 /*
   5  * Ceph - scalable distributed file system
   6  *
   7  * Copyright (C) 2015 Red Hat
   8  *
   9  * This is free software; you can redistribute it and/or
  10  * modify it under the terms of the GNU Lesser General Public
  11  * License version 2.1, as published by the Free Software
  12  * Foundation.  See file COPYING.
  13  *
  14  */
  15
  16 #ifndef CEPH_RGW_ORPHAN_H
  17 #define CEPH_RGW_ORPHAN_H
  18
  19 #include "common/config.h"
  20 #include "common/Formatter.h"
  21 #include "common/errno.h"
  22
  23 #include "rgw_sal.h"
  24
  25 #define dout_subsys ceph_subsys_rgw
  26
  27 #define RGW_ORPHAN_INDEX_OID "orphan.index"
  28 #define RGW_ORPHAN_INDEX_PREFIX "orphan.scan"
  29
  30
  31 enum RGWOrphanSearchStageId {
  32   ORPHAN_SEARCH_STAGE_UNKNOWN = 0,
  33   ORPHAN_SEARCH_STAGE_INIT = 1,
  34   ORPHAN_SEARCH_STAGE_LSPOOL = 2,
  35   ORPHAN_SEARCH_STAGE_LSBUCKETS = 3,
  36   ORPHAN_SEARCH_STAGE_ITERATE_BI = 4,
  37   ORPHAN_SEARCH_STAGE_COMPARE = 5,
  38 };
  39
  40
  41 struct RGWOrphanSearchStage {
  42   RGWOrphanSearchStageId stage;
  43   int shard;
  44   string marker;
  45
  46   RGWOrphanSearchStage() : stage(ORPHAN_SEARCH_STAGE_UNKNOWN), shard(0) {}
  47   explicit RGWOrphanSearchStage(RGWOrphanSearchStageId _stage) : stage(_stage), shard(0) {}
  48   RGWOrphanSearchStage(RGWOrphanSearchStageId _stage, int _shard, const string& _marker) : stage(_stage), shard(_shard), marker(_marker) {}
  49
  50   void encode(bufferlist& bl) const {
  51     ENCODE_START(1, 1, bl);
  52     encode((int)stage, bl);
  53     encode(shard, bl);
  54     encode(marker, bl);
  55     ENCODE_FINISH(bl);
  56   }
  57
  58   void decode(bufferlist::const_iterator& bl) {
  59     DECODE_START(1, bl);
  60     int s;
  61     decode(s, bl);
  62     stage = (RGWOrphanSearchStageId)s;
  63     decode(shard, bl);
  64     decode(marker, bl);
  65     DECODE_FINISH(bl);
  66   }
  67
  68   void dump(Formatter *f) const;
  69 };
  70 WRITE_CLASS_ENCODER(RGWOrphanSearchStage)
  71
  72 struct RGWOrphanSearchInfo {
  73   string job_name;
  74   rgw_pool pool;
  75   uint16_t num_shards;
  76   utime_t start_time;
  77
  78   void encode(bufferlist& bl) const {
  79     ENCODE_START(2, 1, bl);
  80     encode(job_name, bl);
  81     encode(pool.to_str(), bl);
  82     encode(num_shards, bl);
  83     encode(start_time, bl);
  84     ENCODE_FINISH(bl);
  85   }
  86
  87   void decode(bufferlist::const_iterator& bl) {
  88     DECODE_START(2, bl);
  89     decode(job_name, bl);
  90     string s;
  91     decode(s, bl);
  92     pool.from_str(s);
  93     decode(num_shards, bl);
  94     decode(start_time, bl);
  95     DECODE_FINISH(bl);
  96   }
  97
  98   void dump(Formatter *f) const;
  99 };
 100 WRITE_CLASS_ENCODER(RGWOrphanSearchInfo)
 101
 102 struct RGWOrphanSearchState {
 103   RGWOrphanSearchInfo info;
 104   RGWOrphanSearchStage stage;
 105
 106   RGWOrphanSearchState() : stage(ORPHAN_SEARCH_STAGE_UNKNOWN) {}
 107
 108   void encode(bufferlist& bl) const {
 109     ENCODE_START(1, 1, bl);
 110     encode(info, bl);
 111     encode(stage, bl);
 112     ENCODE_FINISH(bl);
 113   }
 114
 115   void decode(bufferlist::const_iterator& bl) {
 116     DECODE_START(1, bl);
 117     decode(info, bl);
 118     decode(stage, bl);
 119     DECODE_FINISH(bl);
 120   }
 121
 122   void dump(Formatter *f) const;
 123 };
 124 WRITE_CLASS_ENCODER(RGWOrphanSearchState)
 125
 126 class RGWOrphanStore {
 127   rgw::sal::RGWRadosStore *store;
 128   librados::IoCtx ioctx;
 129
 130   string oid;
 131
 132 public:
 133   explicit RGWOrphanStore(rgw::sal::RGWRadosStore *_store) : store(_store), oid(RGW_ORPHAN_INDEX_OID) {}
 134
 135   librados::IoCtx& get_ioctx() { return ioctx; }
 136
 137   int init();
 138
 139   int read_job(const string& job_name, RGWOrphanSearchState& state);
 140   int write_job(const string& job_name, const RGWOrphanSearchState& state);
 141   int remove_job(const string& job_name);
 142   int list_jobs(map<string,RGWOrphanSearchState> &job_list);
 143
 144
 145   int store_entries(const string& oid, const map<string, bufferlist>& entries);
 146   int read_entries(const string& oid, const string& marker, map<string, bufferlist> *entries, bool *truncated);
 147 };
 148
 149
 150 class RGWOrphanSearch {
 151   rgw::sal::RGWRadosStore *store;
 152
 153   RGWOrphanStore orphan_store;
 154
 155   RGWOrphanSearchInfo search_info;
 156   RGWOrphanSearchStage search_stage;
 157
 158   map<int, string> all_objs_index;
 159   map<int, string> buckets_instance_index;
 160   map<int, string> linked_objs_index;
 161
 162   string index_objs_prefix;
 163
 164   uint16_t max_concurrent_ios;
 165   uint64_t stale_secs;
 166   int64_t max_list_bucket_entries;
 167
 168   bool detailed_mode;
 169
 170   struct log_iter_info {
 171     string oid;
 172     list<string>::iterator cur;
 173     list<string>::iterator end;
 174   };
 175
 176   int log_oids(map<int, string>& log_shards, map<int, list<string> >& oids);
 177
 178 #define RGW_ORPHANSEARCH_HASH_PRIME 7877
 179   int orphan_shard(const string& str) {
 180     return ceph_str_hash_linux(str.c_str(), str.size()) % RGW_ORPHANSEARCH_HASH_PRIME % search_info.num_shards;
 181   }
 182
 183   int handle_stat_result(map<int, list<string> >& oids, RGWRados::Object::Stat::Result& result);
 184   int pop_and_handle_stat_op(map<int, list<string> >& oids, std::deque<RGWRados::Object::Stat>& ops);
 185
 186
 187   int remove_index(map<int, string>& index);
 188 public:
 189   RGWOrphanSearch(rgw::sal::RGWRadosStore *_store, int _max_ios, uint64_t _stale_secs) : store(_store), orphan_store(store), max_concurrent_ios(_max_ios), stale_secs(_stale_secs) {}
 190
 191   int save_state() {
 192     RGWOrphanSearchState state;
 193     state.info = search_info;
 194     state.stage = search_stage;
 195     return orphan_store.write_job(search_info.job_name, state);
 196   }
 197
 198   int init(const string& job_name, RGWOrphanSearchInfo *info, bool _detailed_mode=false);
 199
 200   int create(const string& job_name, int num_shards);
 201
 202   int build_all_oids_index();
 203   int build_buckets_instance_index();
 204   int build_linked_oids_for_bucket(const string& bucket_instance_id, map<int, list<string> >& oids);
 205   int build_linked_oids_index();
 206   int compare_oid_indexes();
 207
 208   int run();
 209   int finish();
 210 };
 211
 212
 213 class RGWRadosList {
 214
 215   /*
 216    * process_t describes how to process a irectory, we will either
 217    * process the whole thing (entire_container == true) or a portion
 218    * of it (entire_container == false). When we only process a
 219    * portion, we will list the specific keys and/or specific lexical
 220    * prefixes.
 221    */
 222   struct process_t {
 223     bool entire_container;
 224     std::set<rgw_obj_key> filter_keys;
 225     std::set<std::string> prefixes;
 226
 227     process_t() :
 228       entire_container(false)
 229     {}
 230   };
 231
 232   std::map<std::string,process_t> bucket_process_map;
 233   std::set<std::string> visited_oids;
 234
 235   void add_bucket_entire(const std::string& bucket_name) {
 236     auto p = bucket_process_map.emplace(std::make_pair(bucket_name,
 237                                                        process_t()));
 238     p.first->second.entire_container = true;
 239   }
 240
 241   void add_bucket_prefix(const std::string& bucket_name,
 242                          const std::string& prefix) {
 243     auto p = bucket_process_map.emplace(std::make_pair(bucket_name,
 244                                                        process_t()));
 245     p.first->second.prefixes.insert(prefix);
 246   }
 247
 248   void add_bucket_filter(const std::string& bucket_name,
 249                          const rgw_obj_key& obj_key) {
 250     auto p = bucket_process_map.emplace(std::make_pair(bucket_name,
 251                                                        process_t()));
 252     p.first->second.filter_keys.insert(obj_key);
 253   }
 254
 255   rgw::sal::RGWRadosStore* store;
 256
 257   uint16_t max_concurrent_ios;
 258   uint64_t stale_secs;
 259   std::string tenant_name;
 260
 261   int handle_stat_result(RGWRados::Object::Stat::Result& result,
 262                          std::set<string>& obj_oids);
 263   int pop_and_handle_stat_op(RGWObjectCtx& obj_ctx,
 264                              std::deque<RGWRados::Object::Stat>& ops);
 265
 266 public:
 267
 268   RGWRadosList(rgw::sal::RGWRadosStore* _store,
 269                int _max_ios,
 270                uint64_t _stale_secs,
 271                const std::string& _tenant_name) :
 272     store(_store),
 273     max_concurrent_ios(_max_ios),
 274     stale_secs(_stale_secs),
 275     tenant_name(_tenant_name)
 276   {}
 277
 278   int process_bucket(const std::string& bucket_instance_id,
 279                      const std::string& prefix,
 280                      const std::set<rgw_obj_key>& entries_filter);
 281
 282   int do_incomplete_multipart(rgw::sal::RGWRadosStore* store,
 283                               RGWBucketInfo& bucket_info);
 284
 285   int build_linked_oids_index();
 286
 287   int run(const std::string& bucket_id);
 288   int run();
 289 }; // class RGWRadosList
 290
 291 #endif