[ceph.git] / ceph / src / rgw / rgw_orphan.h

// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
// vim: ts=8 sw=2 smarttab ft=cpp

/*
 * Ceph - scalable distributed file system
 *
 * Copyright (C) 2015 Red Hat
 *
 * This is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License version 2.1, as published by the Free Software 
 * Foundation.  See file COPYING.
 * 
 */

#pragma once

#include "common/config.h"
#include "common/Formatter.h"
#include "common/errno.h"

#include "rgw_sal.h"

#define dout_subsys ceph_subsys_rgw

#define RGW_ORPHAN_INDEX_OID "orphan.index"
#define RGW_ORPHAN_INDEX_PREFIX "orphan.scan"


enum RGWOrphanSearchStageId {
  ORPHAN_SEARCH_STAGE_UNKNOWN = 0,
  ORPHAN_SEARCH_STAGE_INIT = 1,
  ORPHAN_SEARCH_STAGE_LSPOOL = 2,
  ORPHAN_SEARCH_STAGE_LSBUCKETS = 3,
  ORPHAN_SEARCH_STAGE_ITERATE_BI = 4,
  ORPHAN_SEARCH_STAGE_COMPARE = 5,
};


struct RGWOrphanSearchStage {
  RGWOrphanSearchStageId stage;
  int shard;
  string marker;

  RGWOrphanSearchStage() : stage(ORPHAN_SEARCH_STAGE_UNKNOWN), shard(0) {}
  explicit RGWOrphanSearchStage(RGWOrphanSearchStageId _stage) : stage(_stage), shard(0) {}
  RGWOrphanSearchStage(RGWOrphanSearchStageId _stage, int _shard, const string& _marker) : stage(_stage), shard(_shard), marker(_marker) {}

  void encode(bufferlist& bl) const {
    ENCODE_START(1, 1, bl);
    encode((int)stage, bl);
    encode(shard, bl);
    encode(marker, bl);
    ENCODE_FINISH(bl);
  }

  void decode(bufferlist::const_iterator& bl) {
    DECODE_START(1, bl);
    int s;
    decode(s, bl);
    stage = (RGWOrphanSearchStageId)s;
    decode(shard, bl);
    decode(marker, bl);
    DECODE_FINISH(bl);
  }

  void dump(Formatter *f) const;
};
WRITE_CLASS_ENCODER(RGWOrphanSearchStage)
  
struct RGWOrphanSearchInfo {
  string job_name;
  rgw_pool pool;
  uint16_t num_shards;
  utime_t start_time;

  void encode(bufferlist& bl) const {
    ENCODE_START(2, 1, bl);
    encode(job_name, bl);
    encode(pool.to_str(), bl);
    encode(num_shards, bl);
    encode(start_time, bl);
    ENCODE_FINISH(bl);
  }

  void decode(bufferlist::const_iterator& bl) {
    DECODE_START(2, bl);
    decode(job_name, bl);
    string s;
    decode(s, bl);
    pool.from_str(s);
    decode(num_shards, bl);
    decode(start_time, bl);
    DECODE_FINISH(bl);
  }

  void dump(Formatter *f) const;
};
WRITE_CLASS_ENCODER(RGWOrphanSearchInfo)

struct RGWOrphanSearchState {
  RGWOrphanSearchInfo info;
  RGWOrphanSearchStage stage;

  RGWOrphanSearchState() : stage(ORPHAN_SEARCH_STAGE_UNKNOWN) {}

  void encode(bufferlist& bl) const {
    ENCODE_START(1, 1, bl);
    encode(info, bl);
    encode(stage, bl);
    ENCODE_FINISH(bl);
  }

  void decode(bufferlist::const_iterator& bl) {
    DECODE_START(1, bl);
    decode(info, bl);
    decode(stage, bl);
    DECODE_FINISH(bl);
  }

  void dump(Formatter *f) const;
};
WRITE_CLASS_ENCODER(RGWOrphanSearchState)

class RGWOrphanStore {
  rgw::sal::RGWRadosStore *store;
  librados::IoCtx ioctx;

  string oid;

public:
  explicit RGWOrphanStore(rgw::sal::RGWRadosStore *_store) : store(_store), oid(RGW_ORPHAN_INDEX_OID) {}

  librados::IoCtx& get_ioctx() { return ioctx; }

  int init(const DoutPrefixProvider *dpp);

  int read_job(const string& job_name, RGWOrphanSearchState& state);
  int write_job(const string& job_name, const RGWOrphanSearchState& state);
  int remove_job(const string& job_name);
  int list_jobs(map<string,RGWOrphanSearchState> &job_list);


  int store_entries(const DoutPrefixProvider *dpp, const string& oid, const map<string, bufferlist>& entries);
  int read_entries(const string& oid, const string& marker, map<string, bufferlist> *entries, bool *truncated);
};


class RGWOrphanSearch {
  rgw::sal::RGWRadosStore *store;

  RGWOrphanStore orphan_store;

  RGWOrphanSearchInfo search_info;
  RGWOrphanSearchStage search_stage;

  map<int, string> all_objs_index;
  map<int, string> buckets_instance_index;
  map<int, string> linked_objs_index;

  string index_objs_prefix;

  uint16_t max_concurrent_ios;
  uint64_t stale_secs;
  int64_t max_list_bucket_entries;

  bool detailed_mode;

  struct log_iter_info {
    string oid;
    list<string>::iterator cur;
    list<string>::iterator end;
  };

  int log_oids(const DoutPrefixProvider *dpp, map<int, string>& log_shards, map<int, list<string> >& oids);

#define RGW_ORPHANSEARCH_HASH_PRIME 7877
  int orphan_shard(const string& str) {
    return ceph_str_hash_linux(str.c_str(), str.size()) % RGW_ORPHANSEARCH_HASH_PRIME % search_info.num_shards;
  }

  int handle_stat_result(const DoutPrefixProvider *dpp, map<int, list<string> >& oids, RGWRados::Object::Stat::Result& result);
  int pop_and_handle_stat_op(const DoutPrefixProvider *dpp, map<int, list<string> >& oids, std::deque<RGWRados::Object::Stat>& ops);


  int remove_index(map<int, string>& index);
public:
  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) {}

  int save_state() {
    RGWOrphanSearchState state;
    state.info = search_info;
    state.stage = search_stage;
    return orphan_store.write_job(search_info.job_name, state);
  }

  int init(const DoutPrefixProvider *dpp, const string& job_name, RGWOrphanSearchInfo *info, bool _detailed_mode=false);

  int create(const string& job_name, int num_shards);

  int build_all_oids_index(const DoutPrefixProvider *dpp);
  int build_buckets_instance_index(const DoutPrefixProvider *dpp);
  int build_linked_oids_for_bucket(const DoutPrefixProvider *dpp, const string& bucket_instance_id, map<int, list<string> >& oids);
  int build_linked_oids_index(const DoutPrefixProvider *dpp);
  int compare_oid_indexes(const DoutPrefixProvider *dpp);

  int run(const DoutPrefixProvider *dpp);
  int finish();
};


class RGWRadosList {

  /*
   * process_t describes how to process a irectory, we will either
   * process the whole thing (entire_container == true) or a portion
   * of it (entire_container == false). When we only process a
   * portion, we will list the specific keys and/or specific lexical
   * prefixes.
   */
  struct process_t {
    bool entire_container;
    std::set<rgw_obj_key> filter_keys;
    std::set<std::string> prefixes;

    process_t() :
      entire_container(false)
    {}
  };

  std::map<std::string,process_t> bucket_process_map;
  std::set<std::string> visited_oids;

  void add_bucket_entire(const std::string& bucket_name) {
    auto p = bucket_process_map.emplace(std::make_pair(bucket_name,
						       process_t()));
    p.first->second.entire_container = true;
  }

  void add_bucket_prefix(const std::string& bucket_name,
			 const std::string& prefix) {
    auto p = bucket_process_map.emplace(std::make_pair(bucket_name,
						       process_t()));
    p.first->second.prefixes.insert(prefix);
  }

  void add_bucket_filter(const std::string& bucket_name,
			 const rgw_obj_key& obj_key) {
    auto p = bucket_process_map.emplace(std::make_pair(bucket_name,
						       process_t()));
    p.first->second.filter_keys.insert(obj_key);
  }

  rgw::sal::RGWRadosStore* store;

  uint16_t max_concurrent_ios;
  uint64_t stale_secs;
  std::string tenant_name;

  bool include_rgw_obj_name;
  std::string field_separator;

  int handle_stat_result(const DoutPrefixProvider *dpp, 
                         RGWRados::Object::Stat::Result& result,
			 std::string& bucket_name,
			 rgw_obj_key& obj_key,
			 std::set<string>& obj_oids);
  int pop_and_handle_stat_op(const DoutPrefixProvider *dpp,
                             RGWObjectCtx& obj_ctx,
			     std::deque<RGWRados::Object::Stat>& ops);

public:

  RGWRadosList(rgw::sal::RGWRadosStore* _store,
	       int _max_ios,
	       uint64_t _stale_secs,
	       const std::string& _tenant_name) :
    store(_store),
    max_concurrent_ios(_max_ios),
    stale_secs(_stale_secs),
    tenant_name(_tenant_name),
    include_rgw_obj_name(false)
  {}

  int process_bucket(const DoutPrefixProvider *dpp, 
                     const std::string& bucket_instance_id,
		     const std::string& prefix,
		     const std::set<rgw_obj_key>& entries_filter);

  int do_incomplete_multipart(const DoutPrefixProvider *dpp, 
                              rgw::sal::RGWRadosStore* store,
			      RGWBucketInfo& bucket_info);

  int build_linked_oids_index();

  int run(const DoutPrefixProvider *dpp, const std::string& bucket_id);
  int run(const DoutPrefixProvider *dpp);

  // if there's a non-empty field separator, that means we'll display
  // bucket and object names
  void set_field_separator(const std::string& fs) {
    field_separator = fs;
    include_rgw_obj_name = !field_separator.empty();
  }
}; // class RGWRadosList
Commit	Line	Data
7c673cae	1	// -- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t --
9f95a23c	2	// vim: ts=8 sw=2 smarttab ft=cpp
11fdf7f2	3
7c673cae FG	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
f67539c2	16	#pragma once
7c673cae FG	17
	18	#include "common/config.h"
	19	#include "common/Formatter.h"
	20	#include "common/errno.h"
	21
9f95a23c	22	#include "rgw_sal.h"
7c673cae FG	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);
11fdf7f2 TL	51	encode((int)stage, bl);
	52	encode(shard, bl);
	53	encode(marker, bl);
7c673cae FG	54	ENCODE_FINISH(bl);
	55	}
	56
11fdf7f2	57	void decode(bufferlist::const_iterator& bl) {
7c673cae FG	58	DECODE_START(1, bl);
7c673cae FG	59	int s;
11fdf7f2	60	decode(s, bl);
7c673cae	61	stage = (RGWOrphanSearchStageId)s;
11fdf7f2 TL	62	decode(shard, bl);
11fdf7f2 TL	63	decode(marker, bl);
7c673cae FG	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);
11fdf7f2 TL	79	encode(job_name, bl);
	80	encode(pool.to_str(), bl);
	81	encode(num_shards, bl);
	82	encode(start_time, bl);
7c673cae FG	83	ENCODE_FINISH(bl);
	84	}
	85
11fdf7f2	86	void decode(bufferlist::const_iterator& bl) {
7c673cae	87	DECODE_START(2, bl);
11fdf7f2	88	decode(job_name, bl);
7c673cae	89	string s;
11fdf7f2	90	decode(s, bl);
7c673cae	91	pool.from_str(s);
11fdf7f2 TL	92	decode(num_shards, bl);
11fdf7f2 TL	93	decode(start_time, bl);
7c673cae FG	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);
11fdf7f2 TL	109	encode(info, bl);
11fdf7f2 TL	110	encode(stage, bl);
7c673cae FG	111	ENCODE_FINISH(bl);
	112	}
	113
11fdf7f2	114	void decode(bufferlist::const_iterator& bl) {
7c673cae	115	DECODE_START(1, bl);
11fdf7f2 TL	116	decode(info, bl);
11fdf7f2 TL	117	decode(stage, bl);
7c673cae FG	118	DECODE_FINISH(bl);
	119	}
	120
	121	void dump(Formatter *f) const;
	122	};
	123	WRITE_CLASS_ENCODER(RGWOrphanSearchState)
	124
	125	class RGWOrphanStore {
9f95a23c	126	rgw::sal::RGWRadosStore *store;
7c673cae FG	127	librados::IoCtx ioctx;
	128
	129	string oid;
	130
	131	public:
9f95a23c	132	explicit RGWOrphanStore(rgw::sal::RGWRadosStore *_store) : store(_store), oid(RGW_ORPHAN_INDEX_OID) {}
7c673cae FG	133
	134	librados::IoCtx& get_ioctx() { return ioctx; }
	135
b3b6e05e	136	int init(const DoutPrefixProvider *dpp);
7c673cae FG	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
b3b6e05e	144	int store_entries(const DoutPrefixProvider *dpp, const string& oid, const map<string, bufferlist>& entries);
7c673cae FG	145	int read_entries(const string& oid, const string& marker, map<string, bufferlist> entries, bool truncated);
	146	};
	147
	148
	149	class RGWOrphanSearch {
9f95a23c	150	rgw::sal::RGWRadosStore *store;
7c673cae FG	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;
11fdf7f2 TL	165	int64_t max_list_bucket_entries;
	166
	167	bool detailed_mode;
7c673cae FG	168
	169	struct log_iter_info {
	170	string oid;
	171	list<string>::iterator cur;
	172	list<string>::iterator end;
	173	};
	174
b3b6e05e	175	int log_oids(const DoutPrefixProvider *dpp, map<int, string>& log_shards, map<int, list<string> >& oids);
7c673cae FG	176
	177	#define RGW_ORPHANSEARCH_HASH_PRIME 7877
	178	int orphan_shard(const string& str) {
	179	return ceph_str_hash_linux(str.c_str(), str.size()) % RGW_ORPHANSEARCH_HASH_PRIME % search_info.num_shards;
	180	}
	181
b3b6e05e TL	182	int handle_stat_result(const DoutPrefixProvider *dpp, map<int, list<string> >& oids, RGWRados::Object::Stat::Result& result);
b3b6e05e TL	183	int pop_and_handle_stat_op(const DoutPrefixProvider *dpp, map<int, list<string> >& oids, std::deque<RGWRados::Object::Stat>& ops);
7c673cae FG	184
	185
	186	int remove_index(map<int, string>& index);
	187	public:
9f95a23c	188	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) {}
7c673cae FG	189
	190	int save_state() {
	191	RGWOrphanSearchState state;
	192	state.info = search_info;
	193	state.stage = search_stage;
	194	return orphan_store.write_job(search_info.job_name, state);
	195	}
	196
b3b6e05e	197	int init(const DoutPrefixProvider dpp, const string& job_name, RGWOrphanSearchInfo info, bool _detailed_mode=false);
7c673cae FG	198
	199	int create(const string& job_name, int num_shards);
	200
b3b6e05e TL	201	int build_all_oids_index(const DoutPrefixProvider *dpp);
	202	int build_buckets_instance_index(const DoutPrefixProvider *dpp);
	203	int build_linked_oids_for_bucket(const DoutPrefixProvider *dpp, const string& bucket_instance_id, map<int, list<string> >& oids);
	204	int build_linked_oids_index(const DoutPrefixProvider *dpp);
	205	int compare_oid_indexes(const DoutPrefixProvider *dpp);
7c673cae	206
b3b6e05e	207	int run(const DoutPrefixProvider *dpp);
7c673cae FG	208	int finish();
	209	};
	210
	211
e306af50 TL	212	class RGWRadosList {
	213
	214	/*
	215	* process_t describes how to process a irectory, we will either
	216	* process the whole thing (entire_container == true) or a portion
	217	* of it (entire_container == false). When we only process a
	218	* portion, we will list the specific keys and/or specific lexical
	219	* prefixes.
	220	*/
	221	struct process_t {
	222	bool entire_container;
	223	std::set<rgw_obj_key> filter_keys;
	224	std::set<std::string> prefixes;
	225
	226	process_t() :
	227	entire_container(false)
	228	{}
	229	};
	230
	231	std::map<std::string,process_t> bucket_process_map;
	232	std::set<std::string> visited_oids;
	233
	234	void add_bucket_entire(const std::string& bucket_name) {
	235	auto p = bucket_process_map.emplace(std::make_pair(bucket_name,
	236	process_t()));
	237	p.first->second.entire_container = true;
	238	}
	239
	240	void add_bucket_prefix(const std::string& bucket_name,
	241	const std::string& prefix) {
	242	auto p = bucket_process_map.emplace(std::make_pair(bucket_name,
	243	process_t()));
	244	p.first->second.prefixes.insert(prefix);
	245	}
	246
	247	void add_bucket_filter(const std::string& bucket_name,
	248	const rgw_obj_key& obj_key) {
	249	auto p = bucket_process_map.emplace(std::make_pair(bucket_name,
	250	process_t()));
	251	p.first->second.filter_keys.insert(obj_key);
	252	}
	253
	254	rgw::sal::RGWRadosStore* store;
	255
	256	uint16_t max_concurrent_ios;
	257	uint64_t stale_secs;
	258	std::string tenant_name;
	259
f67539c2 TL	260	bool include_rgw_obj_name;
	261	std::string field_separator;
	262
b3b6e05e TL	263	int handle_stat_result(const DoutPrefixProvider *dpp,
b3b6e05e TL	264	RGWRados::Object::Stat::Result& result,
f67539c2 TL	265	std::string& bucket_name,
f67539c2 TL	266	rgw_obj_key& obj_key,
e306af50	267	std::set<string>& obj_oids);
b3b6e05e TL	268	int pop_and_handle_stat_op(const DoutPrefixProvider *dpp,
b3b6e05e TL	269	RGWObjectCtx& obj_ctx,
e306af50 TL	270	std::deque<RGWRados::Object::Stat>& ops);
	271
	272	public:
	273
	274	RGWRadosList(rgw::sal::RGWRadosStore* _store,
	275	int _max_ios,
	276	uint64_t _stale_secs,
	277	const std::string& _tenant_name) :
	278	store(_store),
	279	max_concurrent_ios(_max_ios),
	280	stale_secs(_stale_secs),
f67539c2 TL	281	tenant_name(_tenant_name),
f67539c2 TL	282	include_rgw_obj_name(false)
e306af50 TL	283	{}
e306af50 TL	284
b3b6e05e TL	285	int process_bucket(const DoutPrefixProvider *dpp,
b3b6e05e TL	286	const std::string& bucket_instance_id,
e306af50 TL	287	const std::string& prefix,
	288	const std::set<rgw_obj_key>& entries_filter);
	289
b3b6e05e TL	290	int do_incomplete_multipart(const DoutPrefixProvider *dpp,
b3b6e05e TL	291	rgw::sal::RGWRadosStore* store,
e306af50 TL	292	RGWBucketInfo& bucket_info);
	293
	294	int build_linked_oids_index();
	295
b3b6e05e TL	296	int run(const DoutPrefixProvider *dpp, const std::string& bucket_id);
b3b6e05e TL	297	int run(const DoutPrefixProvider *dpp);
7c673cae	298
f67539c2 TL	299	// if there's a non-empty field separator, that means we'll display
	300	// bucket and object names
	301	void set_field_separator(const std::string& fs) {
	302	field_separator = fs;
	303	include_rgw_obj_name = !field_separator.empty();
	304	}
	305	}; // class RGWRadosList