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