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
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2015 Red Hat
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15 #ifndef CEPH_RGW_ORPHAN_H
  16 #define CEPH_RGW_ORPHAN_H
  17
  18 #include "common/config.h"
  19 #include "common/Formatter.h"
  20 #include "common/errno.h"
  21
  22 #include "rgw_rados.h"
  23
  24 #define dout_subsys ceph_subsys_rgw
  25
  26 #define RGW_ORPHAN_INDEX_OID "orphan.index"
  27 #define RGW_ORPHAN_INDEX_PREFIX "orphan.scan"
  28
  29
  30 enum RGWOrphanSearchStageId {
  31   ORPHAN_SEARCH_STAGE_UNKNOWN = 0,
  32   ORPHAN_SEARCH_STAGE_INIT = 1,
  33   ORPHAN_SEARCH_STAGE_LSPOOL = 2,
  34   ORPHAN_SEARCH_STAGE_LSBUCKETS = 3,
  35   ORPHAN_SEARCH_STAGE_ITERATE_BI = 4,
  36   ORPHAN_SEARCH_STAGE_COMPARE = 5,
  37 };
  38
  39
  40 struct RGWOrphanSearchStage {
  41   RGWOrphanSearchStageId stage;
  42   int shard;
  43   string marker;
  44
  45   RGWOrphanSearchStage() : stage(ORPHAN_SEARCH_STAGE_UNKNOWN), shard(0) {}
  46   explicit RGWOrphanSearchStage(RGWOrphanSearchStageId _stage) : stage(_stage), shard(0) {}
  47   RGWOrphanSearchStage(RGWOrphanSearchStageId _stage, int _shard, const string& _marker) : stage(_stage), shard(_shard), marker(_marker) {}
  48
  49   void encode(bufferlist& bl) const {
  50     ENCODE_START(1, 1, bl);
  51     ::encode((int)stage, bl);
  52     ::encode(shard, bl);
  53     ::encode(marker, bl);
  54     ENCODE_FINISH(bl);
  55   }
  56
  57   void decode(bufferlist::iterator& bl) {
  58     DECODE_START(1, bl);
  59     int s;
  60     ::decode(s, bl);
  61     stage = (RGWOrphanSearchStageId)s;
  62     ::decode(shard, bl);
  63     ::decode(marker, bl);
  64     DECODE_FINISH(bl);
  65   }
  66
  67   void dump(Formatter *f) const;
  68 };
  69 WRITE_CLASS_ENCODER(RGWOrphanSearchStage)
  70
  71 struct RGWOrphanSearchInfo {
  72   string job_name;
  73   rgw_pool pool;
  74   uint16_t num_shards;
  75   utime_t start_time;
  76
  77   void encode(bufferlist& bl) const {
  78     ENCODE_START(2, 1, bl);
  79     ::encode(job_name, bl);
  80     ::encode(pool.to_str(), bl);
  81     ::encode(num_shards, bl);
  82     ::encode(start_time, bl);
  83     ENCODE_FINISH(bl);
  84   }
  85
  86   void decode(bufferlist::iterator& bl) {
  87     DECODE_START(2, bl);
  88     ::decode(job_name, bl);
  89     string s;
  90     ::decode(s, bl);
  91     pool.from_str(s);
  92     ::decode(num_shards, bl);
  93     ::decode(start_time, bl);
  94     DECODE_FINISH(bl);
  95   }
  96
  97   void dump(Formatter *f) const;
  98 };
  99 WRITE_CLASS_ENCODER(RGWOrphanSearchInfo)
 100
 101 struct RGWOrphanSearchState {
 102   RGWOrphanSearchInfo info;
 103   RGWOrphanSearchStage stage;
 104
 105   RGWOrphanSearchState() : stage(ORPHAN_SEARCH_STAGE_UNKNOWN) {}
 106
 107   void encode(bufferlist& bl) const {
 108     ENCODE_START(1, 1, bl);
 109     ::encode(info, bl);
 110     ::encode(stage, bl);
 111     ENCODE_FINISH(bl);
 112   }
 113
 114   void decode(bufferlist::iterator& bl) {
 115     DECODE_START(1, bl);
 116     ::decode(info, bl);
 117     ::decode(stage, bl);
 118     DECODE_FINISH(bl);
 119   }
 120
 121   void dump(Formatter *f) const;
 122 };
 123 WRITE_CLASS_ENCODER(RGWOrphanSearchState)
 124
 125 class RGWOrphanStore {
 126   RGWRados *store;
 127   librados::IoCtx ioctx;
 128
 129   string oid;
 130
 131 public:
 132   explicit RGWOrphanStore(RGWRados *_store) : store(_store), oid(RGW_ORPHAN_INDEX_OID) {}
 133
 134   librados::IoCtx& get_ioctx() { return ioctx; }
 135
 136   int init();
 137
 138   int read_job(const string& job_name, RGWOrphanSearchState& state);
 139   int write_job(const string& job_name, const RGWOrphanSearchState& state);
 140   int remove_job(const string& job_name);
 141   int list_jobs(map<string,RGWOrphanSearchState> &job_list);
 142
 143
 144   int store_entries(const string& oid, const map<string, bufferlist>& entries);
 145   int read_entries(const string& oid, const string& marker, map<string, bufferlist> *entries, bool *truncated);
 146 };
 147
 148
 149 class RGWOrphanSearch {
 150   RGWRados *store;
 151
 152   RGWOrphanStore orphan_store;
 153
 154   RGWOrphanSearchInfo search_info;
 155   RGWOrphanSearchStage search_stage;
 156
 157   map<int, string> all_objs_index;
 158   map<int, string> buckets_instance_index;
 159   map<int, string> linked_objs_index;
 160
 161   string index_objs_prefix;
 162
 163   uint16_t max_concurrent_ios;
 164   uint64_t stale_secs;
 165
 166   struct log_iter_info {
 167     string oid;
 168     list<string>::iterator cur;
 169     list<string>::iterator end;
 170   };
 171
 172   int log_oids(map<int, string>& log_shards, map<int, list<string> >& oids);
 173
 174 #define RGW_ORPHANSEARCH_HASH_PRIME 7877
 175   int orphan_shard(const string& str) {
 176     return ceph_str_hash_linux(str.c_str(), str.size()) % RGW_ORPHANSEARCH_HASH_PRIME % search_info.num_shards;
 177   }
 178
 179   int handle_stat_result(map<int, list<string> >& oids, RGWRados::Object::Stat::Result& result);
 180   int pop_and_handle_stat_op(map<int, list<string> >& oids, std::deque<RGWRados::Object::Stat>& ops);
 181
 182
 183   int remove_index(map<int, string>& index);
 184 public:
 185   RGWOrphanSearch(RGWRados *_store, int _max_ios, uint64_t _stale_secs) : store(_store), orphan_store(store), max_concurrent_ios(_max_ios), stale_secs(_stale_secs) {}
 186
 187   int save_state() {
 188     RGWOrphanSearchState state;
 189     state.info = search_info;
 190     state.stage = search_stage;
 191     return orphan_store.write_job(search_info.job_name, state);
 192   }
 193
 194   int init(const string& job_name, RGWOrphanSearchInfo *info);
 195
 196   int create(const string& job_name, int num_shards);
 197
 198   int build_all_oids_index();
 199   int build_buckets_instance_index();
 200   int build_linked_oids_for_bucket(const string& bucket_instance_id, map<int, list<string> >& oids);
 201   int build_linked_oids_index();
 202   int compare_oid_indexes();
 203
 204   int run();
 205   int finish();
 206 };
 207
 208
 209
 210 #endif