ceph/src/rgw/rgw_bucket.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #ifndef CEPH_RGW_BUCKET_H
   5 #define CEPH_RGW_BUCKET_H
   6
   7 #include <string>
   8 #include <memory>
   9
  10 #include "include/types.h"
  11 #include "rgw_common.h"
  12 #include "rgw_tools.h"
  13
  14 #include "rgw_rados.h"
  15
  16 #include "rgw_string.h"
  17
  18 #include "common/Formatter.h"
  19 #include "common/lru_map.h"
  20 #include "common/ceph_time.h"
  21 #include "rgw_formats.h"
  22
  23 // define as static when RGWBucket implementation completes
  24 extern void rgw_get_buckets_obj(const rgw_user& user_id, string& buckets_obj_id);
  25
  26 extern int rgw_bucket_store_info(RGWRados *store, const string& bucket_name, bufferlist& bl, bool exclusive,
  27                                  map<string, bufferlist> *pattrs, RGWObjVersionTracker *objv_tracker,
  28                                  real_time mtime);
  29 extern int rgw_bucket_instance_store_info(RGWRados *store, string& oid, bufferlist& bl, bool exclusive,
  30                                  map<string, bufferlist> *pattrs, RGWObjVersionTracker *objv_tracker,
  31                                  real_time mtime);
  32
  33 extern int rgw_bucket_parse_bucket_instance(const string& bucket_instance, string *target_bucket_instance, int *shard_id);
  34 extern int rgw_bucket_parse_bucket_key(CephContext *cct, const string& key,
  35                                        rgw_bucket* bucket, int *shard_id);
  36
  37 extern int rgw_bucket_instance_remove_entry(RGWRados *store, const string& entry,
  38                                             RGWObjVersionTracker *objv_tracker);
  39 extern void rgw_bucket_instance_key_to_oid(string& key);
  40 extern void rgw_bucket_instance_oid_to_key(string& oid);
  41
  42 extern int rgw_bucket_delete_bucket_obj(RGWRados *store,
  43                                         const string& tenant_name,
  44                                         const string& bucket_name,
  45                                         RGWObjVersionTracker& objv_tracker);
  46
  47 extern int rgw_bucket_sync_user_stats(RGWRados *store, const rgw_user& user_id, const RGWBucketInfo& bucket_info);
  48 extern int rgw_bucket_sync_user_stats(RGWRados *store, const string& tenant_name, const string& bucket_name);
  49
  50 extern std::string rgw_make_bucket_entry_name(const std::string& tenant_name,
  51                                               const std::string& bucket_name);
  52 static inline void rgw_make_bucket_entry_name(const string& tenant_name,
  53                                               const string& bucket_name,
  54                                               std::string& bucket_entry) {
  55   bucket_entry = rgw_make_bucket_entry_name(tenant_name, bucket_name);
  56 }
  57
  58 extern void rgw_parse_url_bucket(const string& bucket,
  59                                  const string& auth_tenant,
  60                                  string &tenant_name, string &bucket_name);
  61
  62 struct RGWBucketCompleteInfo {
  63   RGWBucketInfo info;
  64   map<string, bufferlist> attrs;
  65
  66   void dump(Formatter *f) const;
  67   void decode_json(JSONObj *obj);
  68 };
  69
  70 class RGWBucketEntryMetadataObject : public RGWMetadataObject {
  71   RGWBucketEntryPoint ep;
  72 public:
  73   RGWBucketEntryMetadataObject(RGWBucketEntryPoint& _ep, obj_version& v, real_time m) : ep(_ep) {
  74     objv = v;
  75     mtime = m;
  76   }
  77
  78   void dump(Formatter *f) const override {
  79     ep.dump(f);
  80   }
  81 };
  82
  83 class RGWBucketInstanceMetadataObject : public RGWMetadataObject {
  84   RGWBucketCompleteInfo info;
  85 public:
  86   RGWBucketInstanceMetadataObject() {}
  87   RGWBucketInstanceMetadataObject(RGWBucketCompleteInfo& i, obj_version& v, real_time m) : info(i) {
  88     objv = v;
  89     mtime = m;
  90   }
  91
  92   void dump(Formatter *f) const override {
  93     info.dump(f);
  94   }
  95
  96   void decode_json(JSONObj *obj) {
  97     info.decode_json(obj);
  98   }
  99
 100   RGWBucketInfo& get_bucket_info() { return info.info; }
 101 };
 102
 103 /**
 104  * Store a list of the user's buckets, with associated functinos.
 105  */
 106 class RGWUserBuckets
 107 {
 108   std::map<std::string, RGWBucketEnt> buckets;
 109
 110 public:
 111   RGWUserBuckets() = default;
 112   RGWUserBuckets(RGWUserBuckets&&) = default;
 113
 114   RGWUserBuckets& operator=(const RGWUserBuckets&) = default;
 115
 116   void encode(bufferlist& bl) const {
 117     using ceph::encode;
 118     encode(buckets, bl);
 119   }
 120   void decode(bufferlist::const_iterator& bl) {
 121     using ceph::decode;
 122     decode(buckets, bl);
 123   }
 124   /**
 125    * Check if the user owns a bucket by the given name.
 126    */
 127   bool owns(string& name) {
 128     map<string, RGWBucketEnt>::iterator iter;
 129     iter = buckets.find(name);
 130     return (iter != buckets.end());
 131   }
 132
 133   /**
 134    * Add a (created) bucket to the user's bucket list.
 135    */
 136   void add(const RGWBucketEnt& bucket) {
 137     buckets[bucket.bucket.name] = bucket;
 138   }
 139
 140   /**
 141    * Remove a bucket from the user's list by name.
 142    */
 143   void remove(const string& name) {
 144     map<string, RGWBucketEnt>::iterator iter;
 145     iter = buckets.find(name);
 146     if (iter != buckets.end()) {
 147       buckets.erase(iter);
 148     }
 149   }
 150
 151   /**
 152    * Get the user's buckets as a map.
 153    */
 154   map<string, RGWBucketEnt>& get_buckets() { return buckets; }
 155
 156   /**
 157    * Cleanup data structure
 158    */
 159   void clear() { buckets.clear(); }
 160
 161   size_t count() { return buckets.size(); }
 162 };
 163 WRITE_CLASS_ENCODER(RGWUserBuckets)
 164
 165 class RGWMetadataManager;
 166 class RGWMetadataHandler;
 167
 168 class RGWBucketMetaHandlerAllocator {
 169 public:
 170   static RGWMetadataHandler *alloc();
 171 };
 172
 173 class RGWBucketInstanceMetaHandlerAllocator {
 174 public:
 175   static RGWMetadataHandler *alloc();
 176 };
 177
 178 class RGWArchiveBucketMetaHandlerAllocator {
 179 public:
 180   static RGWMetadataHandler *alloc();
 181 };
 182
 183 class RGWArchiveBucketInstanceMetaHandlerAllocator {
 184 public:
 185   static RGWMetadataHandler *alloc();
 186 };
 187
 188 extern void rgw_bucket_init(RGWMetadataManager *mm);
 189 /**
 190  * Get all the buckets owned by a user and fill up an RGWUserBuckets with them.
 191  * Returns: 0 on success, -ERR# on failure.
 192  */
 193 extern int rgw_read_user_buckets(RGWRados *store,
 194                                  const rgw_user& user_id,
 195                                  RGWUserBuckets& buckets,
 196                                  const string& marker,
 197                                  const string& end_marker,
 198                                  uint64_t max,
 199                                  bool need_stats,
 200                                  bool* is_truncated,
 201                                  uint64_t default_amount = 1000);
 202
 203 extern int rgw_link_bucket(RGWRados* store,
 204                            const rgw_user& user_id,
 205                            rgw_bucket& bucket,
 206                            ceph::real_time creation_time,
 207                            bool update_entrypoint = true);
 208 extern int rgw_unlink_bucket(RGWRados *store, const rgw_user& user_id,
 209                              const string& tenant_name, const string& bucket_name, bool update_entrypoint = true);
 210
 211 extern int rgw_remove_object(RGWRados *store, const RGWBucketInfo& bucket_info, const rgw_bucket& bucket, rgw_obj_key& key);
 212 extern int rgw_remove_bucket(RGWRados *store, rgw_bucket& bucket, bool delete_children);
 213 extern int rgw_remove_bucket_bypass_gc(RGWRados *store, rgw_bucket& bucket, int concurrent_max);
 214
 215 extern int rgw_bucket_set_attrs(RGWRados *store, RGWBucketInfo& bucket_info,
 216                                 map<string, bufferlist>& attrs,
 217                                 RGWObjVersionTracker *objv_tracker);
 218 extern int rgw_object_get_attr(RGWRados* store, const RGWBucketInfo& bucket_info,
 219                                const rgw_obj& obj, const char* attr_name,
 220                                bufferlist& out_bl);
 221
 222 extern void check_bad_user_bucket_mapping(RGWRados *store, const rgw_user& user_id, bool fix);
 223
 224 struct RGWBucketAdminOpState {
 225   rgw_user uid;
 226   std::string display_name;
 227   std::string bucket_name;
 228   std::string bucket_id;
 229   std::string object_name;
 230
 231   bool list_buckets;
 232   bool stat_buckets;
 233   bool check_objects;
 234   bool fix_index;
 235   bool delete_child_objects;
 236   bool bucket_stored;
 237   int max_aio = 0;
 238
 239   rgw_bucket bucket;
 240
 241   RGWQuotaInfo quota;
 242
 243   void set_fetch_stats(bool value) { stat_buckets = value; }
 244   void set_check_objects(bool value) { check_objects = value; }
 245   void set_fix_index(bool value) { fix_index = value; }
 246   void set_delete_children(bool value) { delete_child_objects = value; }
 247
 248   void set_max_aio(int value) { max_aio = value; }
 249
 250   void set_user_id(const rgw_user& user_id) {
 251     if (!user_id.empty())
 252       uid = user_id;
 253   }
 254   void set_bucket_name(const std::string& bucket_str) {
 255     bucket_name = bucket_str;
 256   }
 257   void set_object(std::string& object_str) {
 258     object_name = object_str;
 259   }
 260   void set_quota(RGWQuotaInfo& value) {
 261     quota = value;
 262   }
 263
 264
 265   rgw_user& get_user_id() { return uid; }
 266   std::string& get_user_display_name() { return display_name; }
 267   std::string& get_bucket_name() { return bucket_name; }
 268   std::string& get_object_name() { return object_name; }
 269
 270   rgw_bucket& get_bucket() { return bucket; }
 271   void set_bucket(rgw_bucket& _bucket) {
 272     bucket = _bucket;
 273     bucket_stored = true;
 274   }
 275
 276   void set_bucket_id(const string& bi) {
 277     bucket_id = bi;
 278   }
 279   const string& get_bucket_id() { return bucket_id; }
 280
 281   bool will_fetch_stats() { return stat_buckets; }
 282   bool will_fix_index() { return fix_index; }
 283   bool will_delete_children() { return delete_child_objects; }
 284   bool will_check_objects() { return check_objects; }
 285   bool is_user_op() { return !uid.empty(); }
 286   bool is_system_op() { return uid.empty(); }
 287   bool has_bucket_stored() { return bucket_stored; }
 288   int get_max_aio() { return max_aio; }
 289
 290   RGWBucketAdminOpState() : list_buckets(false), stat_buckets(false), check_objects(false),
 291                             fix_index(false), delete_child_objects(false),
 292                             bucket_stored(false)  {}
 293 };
 294
 295 /*
 296  * A simple wrapper class for administrative bucket operations
 297  */
 298
 299 class RGWBucket
 300 {
 301   RGWUserBuckets buckets;
 302   RGWRados *store;
 303   RGWAccessHandle handle;
 304
 305   RGWUserInfo user_info;
 306   std::string tenant;
 307   std::string bucket_name;
 308
 309   bool failure;
 310
 311   RGWBucketInfo bucket_info;
 312
 313 public:
 314   RGWBucket() : store(NULL), handle(NULL), failure(false) {}
 315   int init(RGWRados *storage, RGWBucketAdminOpState& op_state);
 316
 317   int check_bad_index_multipart(RGWBucketAdminOpState& op_state,
 318               RGWFormatterFlusher& flusher, std::string *err_msg = NULL);
 319
 320   int check_object_index(RGWBucketAdminOpState& op_state,
 321                          RGWFormatterFlusher& flusher,
 322                          std::string *err_msg = NULL);
 323
 324   int check_index(RGWBucketAdminOpState& op_state,
 325           map<RGWObjCategory, RGWStorageStats>& existing_stats,
 326           map<RGWObjCategory, RGWStorageStats>& calculated_stats,
 327           std::string *err_msg = NULL);
 328
 329   int remove(RGWBucketAdminOpState& op_state, bool bypass_gc = false, bool keep_index_consistent = true, std::string *err_msg = NULL);
 330   int link(RGWBucketAdminOpState& op_state, std::string *err_msg = NULL);
 331   int unlink(RGWBucketAdminOpState& op_state, std::string *err_msg = NULL);
 332   int set_quota(RGWBucketAdminOpState& op_state, std::string *err_msg = NULL);
 333
 334   int remove_object(RGWBucketAdminOpState& op_state, std::string *err_msg = NULL);
 335   int policy_bl_to_stream(bufferlist& bl, ostream& o);
 336   int get_policy(RGWBucketAdminOpState& op_state, RGWAccessControlPolicy& policy);
 337
 338   void clear_failure() { failure = false; }
 339
 340   const RGWBucketInfo& get_bucket_info() const { return bucket_info; }
 341 };
 342
 343 class RGWBucketAdminOp
 344 {
 345 public:
 346   static int get_policy(RGWRados *store, RGWBucketAdminOpState& op_state,
 347                   RGWFormatterFlusher& flusher);
 348   static int get_policy(RGWRados *store, RGWBucketAdminOpState& op_state,
 349                   RGWAccessControlPolicy& policy);
 350   static int dump_s3_policy(RGWRados *store, RGWBucketAdminOpState& op_state,
 351                   ostream& os);
 352
 353   static int unlink(RGWRados *store, RGWBucketAdminOpState& op_state);
 354   static int link(RGWRados *store, RGWBucketAdminOpState& op_state, string *err_msg = NULL);
 355
 356   static int check_index(RGWRados *store, RGWBucketAdminOpState& op_state,
 357                   RGWFormatterFlusher& flusher);
 358
 359   static int remove_bucket(RGWRados *store, RGWBucketAdminOpState& op_state, bool bypass_gc = false, bool keep_index_consistent = true);
 360   static int remove_object(RGWRados *store, RGWBucketAdminOpState& op_state);
 361   static int info(RGWRados *store, RGWBucketAdminOpState& op_state, RGWFormatterFlusher& flusher);
 362   static int limit_check(RGWRados *store, RGWBucketAdminOpState& op_state,
 363                          const std::list<std::string>& user_ids,
 364                          RGWFormatterFlusher& flusher,
 365                          bool warnings_only = false);
 366   static int set_quota(RGWRados *store, RGWBucketAdminOpState& op_state);
 367
 368   static int list_stale_instances(RGWRados *store, RGWBucketAdminOpState& op_state,
 369                                   RGWFormatterFlusher& flusher);
 370
 371   static int clear_stale_instances(RGWRados *store, RGWBucketAdminOpState& op_state,
 372                                    RGWFormatterFlusher& flusher);
 373   static int fix_lc_shards(RGWRados *store, RGWBucketAdminOpState& op_state,
 374                            RGWFormatterFlusher& flusher);
 375   static int fix_obj_expiry(RGWRados *store, RGWBucketAdminOpState& op_state,
 376                             RGWFormatterFlusher& flusher, bool dry_run = false);
 377 };
 378
 379
 380 enum DataLogEntityType {
 381   ENTITY_TYPE_UNKNOWN = 0,
 382   ENTITY_TYPE_BUCKET = 1,
 383 };
 384
 385 struct rgw_data_change {
 386   DataLogEntityType entity_type;
 387   string key;
 388   real_time timestamp;
 389
 390   void encode(bufferlist& bl) const {
 391     ENCODE_START(1, 1, bl);
 392     uint8_t t = (uint8_t)entity_type;
 393     encode(t, bl);
 394     encode(key, bl);
 395     encode(timestamp, bl);
 396     ENCODE_FINISH(bl);
 397   }
 398
 399   void decode(bufferlist::const_iterator& bl) {
 400      DECODE_START(1, bl);
 401      uint8_t t;
 402      decode(t, bl);
 403      entity_type = (DataLogEntityType)t;
 404      decode(key, bl);
 405      decode(timestamp, bl);
 406      DECODE_FINISH(bl);
 407   }
 408
 409   void dump(Formatter *f) const;
 410   void decode_json(JSONObj *obj);
 411 };
 412 WRITE_CLASS_ENCODER(rgw_data_change)
 413
 414 struct rgw_data_change_log_entry {
 415   string log_id;
 416   real_time log_timestamp;
 417   rgw_data_change entry;
 418
 419   void encode(bufferlist& bl) const {
 420     ENCODE_START(1, 1, bl);
 421     encode(log_id, bl);
 422     encode(log_timestamp, bl);
 423     encode(entry, bl);
 424     ENCODE_FINISH(bl);
 425   }
 426
 427   void decode(bufferlist::const_iterator& bl) {
 428      DECODE_START(1, bl);
 429      decode(log_id, bl);
 430      decode(log_timestamp, bl);
 431      decode(entry, bl);
 432      DECODE_FINISH(bl);
 433   }
 434
 435   void dump(Formatter *f) const;
 436   void decode_json(JSONObj *obj);
 437 };
 438 WRITE_CLASS_ENCODER(rgw_data_change_log_entry)
 439
 440 struct RGWDataChangesLogInfo {
 441   string marker;
 442   real_time last_update;
 443
 444   void dump(Formatter *f) const;
 445   void decode_json(JSONObj *obj);
 446 };
 447
 448 namespace rgw {
 449 struct BucketChangeObserver;
 450 }
 451
 452 class RGWDataChangesLog {
 453   CephContext *cct;
 454   RGWRados *store;
 455   rgw::BucketChangeObserver *observer = nullptr;
 456
 457   int num_shards;
 458   string *oids;
 459
 460   Mutex lock;
 461   RWLock modified_lock;
 462   map<int, set<string> > modified_shards;
 463
 464   std::atomic<bool> down_flag = { false };
 465
 466   struct ChangeStatus {
 467     real_time cur_expiration;
 468     real_time cur_sent;
 469     bool pending;
 470     RefCountedCond *cond;
 471     Mutex *lock;
 472
 473     ChangeStatus() : pending(false), cond(NULL) {
 474       lock = new Mutex("RGWDataChangesLog::ChangeStatus");
 475     }
 476
 477     ~ChangeStatus() {
 478       delete lock;
 479     }
 480   };
 481
 482   typedef std::shared_ptr<ChangeStatus> ChangeStatusPtr;
 483
 484   lru_map<rgw_bucket_shard, ChangeStatusPtr> changes;
 485
 486   map<rgw_bucket_shard, bool> cur_cycle;
 487
 488   void _get_change(const rgw_bucket_shard& bs, ChangeStatusPtr& status);
 489   void register_renew(rgw_bucket_shard& bs);
 490   void update_renewed(rgw_bucket_shard& bs, real_time& expiration);
 491
 492   class ChangesRenewThread : public Thread {
 493     CephContext *cct;
 494     RGWDataChangesLog *log;
 495     Mutex lock;
 496     Cond cond;
 497
 498   public:
 499     ChangesRenewThread(CephContext *_cct, RGWDataChangesLog *_log) : cct(_cct), log(_log), lock("ChangesRenewThread::lock") {}
 500     void *entry() override;
 501     void stop();
 502   };
 503
 504   ChangesRenewThread *renew_thread;
 505
 506 public:
 507
 508   RGWDataChangesLog(CephContext *_cct, RGWRados *_store) : cct(_cct), store(_store),
 509                                                            lock("RGWDataChangesLog::lock"), modified_lock("RGWDataChangesLog::modified_lock"),
 510                                                            changes(cct->_conf->rgw_data_log_changes_size) {
 511     num_shards = cct->_conf->rgw_data_log_num_shards;
 512
 513     oids = new string[num_shards];
 514
 515     string prefix = cct->_conf->rgw_data_log_obj_prefix;
 516
 517     if (prefix.empty()) {
 518       prefix = "data_log";
 519     }
 520
 521     for (int i = 0; i < num_shards; i++) {
 522       char buf[16];
 523       snprintf(buf, sizeof(buf), "%s.%d", prefix.c_str(), i);
 524       oids[i] = buf;
 525     }
 526
 527     renew_thread = new ChangesRenewThread(cct, this);
 528     renew_thread->create("rgw_dt_lg_renew");
 529   }
 530
 531   ~RGWDataChangesLog();
 532
 533   int choose_oid(const rgw_bucket_shard& bs);
 534   const std::string& get_oid(int shard_id) const { return oids[shard_id]; }
 535   int add_entry(rgw_bucket& bucket, int shard_id);
 536   int get_log_shard_id(rgw_bucket& bucket, int shard_id);
 537   int renew_entries();
 538   int list_entries(int shard, const real_time& start_time, const real_time& end_time, int max_entries,
 539                    list<rgw_data_change_log_entry>& entries,
 540                    const string& marker,
 541                    string *out_marker,
 542                    bool *truncated);
 543   int trim_entries(int shard_id, const real_time& start_time, const real_time& end_time,
 544                    const string& start_marker, const string& end_marker);
 545   int get_info(int shard_id, RGWDataChangesLogInfo *info);
 546   int lock_exclusive(int shard_id, timespan duration, string& zone_id, string& owner_id);
 547   int unlock(int shard_id, string& zone_id, string& owner_id);
 548   struct LogMarker {
 549     int shard;
 550     string marker;
 551
 552     LogMarker() : shard(0) {}
 553   };
 554   int list_entries(const real_time& start_time, const real_time& end_time, int max_entries,
 555                list<rgw_data_change_log_entry>& entries, LogMarker& marker, bool *ptruncated);
 556
 557   void mark_modified(int shard_id, const rgw_bucket_shard& bs);
 558   void read_clear_modified(map<int, set<string> > &modified);
 559
 560   void set_observer(rgw::BucketChangeObserver *observer) {
 561     this->observer = observer;
 562   }
 563
 564   bool going_down();
 565 };
 566
 567
 568 #endif