ceph/src/rgw/rgw_bucket.cc

   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 #include "rgw_bucket.h"
   5
   6 #include "common/errno.h"
   7
   8 #define dout_subsys ceph_subsys_rgw
   9
  10 // stolen from src/cls/version/cls_version.cc
  11 #define VERSION_ATTR "ceph.objclass.version"
  12
  13 using namespace std;
  14
  15 static void set_err_msg(std::string *sink, std::string msg)
  16 {
  17   if (sink && !msg.empty())
  18     *sink = msg;
  19 }
  20
  21 void init_bucket(rgw_bucket *b, const char *t, const char *n, const char *dp, const char *ip, const char *m, const char *id)
  22 {
  23   b->tenant = t;
  24   b->name = n;
  25   b->marker = m;
  26   b->bucket_id = id;
  27   b->explicit_placement.data_pool = rgw_pool(dp);
  28   b->explicit_placement.index_pool = rgw_pool(ip);
  29 }
  30
  31 // parse key in format: [tenant/]name:instance[:shard_id]
  32 int rgw_bucket_parse_bucket_key(CephContext *cct, const string& key,
  33                                 rgw_bucket *bucket, int *shard_id)
  34 {
  35   std::string_view name{key};
  36   std::string_view instance;
  37
  38   // split tenant/name
  39   auto pos = name.find('/');
  40   if (pos != string::npos) {
  41     auto tenant = name.substr(0, pos);
  42     bucket->tenant.assign(tenant.begin(), tenant.end());
  43     name = name.substr(pos + 1);
  44   } else {
  45     bucket->tenant.clear();
  46   }
  47
  48   // split name:instance
  49   pos = name.find(':');
  50   if (pos != string::npos) {
  51     instance = name.substr(pos + 1);
  52     name = name.substr(0, pos);
  53   }
  54   bucket->name.assign(name.begin(), name.end());
  55
  56   // split instance:shard
  57   pos = instance.find(':');
  58   if (pos == string::npos) {
  59     bucket->bucket_id.assign(instance.begin(), instance.end());
  60     if (shard_id) {
  61       *shard_id = -1;
  62     }
  63     return 0;
  64   }
  65
  66   // parse shard id
  67   auto shard = instance.substr(pos + 1);
  68   string err;
  69   auto id = strict_strtol(shard.data(), 10, &err);
  70   if (!err.empty()) {
  71     if (cct) {
  72       ldout(cct, 0) << "ERROR: failed to parse bucket shard '"
  73           << instance.data() << "': " << err << dendl;
  74     }
  75     return -EINVAL;
  76   }
  77
  78   if (shard_id) {
  79     *shard_id = id;
  80   }
  81   instance = instance.substr(0, pos);
  82   bucket->bucket_id.assign(instance.begin(), instance.end());
  83   return 0;
  84 }
  85
  86 /*
  87  * Note that this is not a reversal of parse_bucket(). That one deals
  88  * with the syntax we need in metadata and such. This one deals with
  89  * the representation in RADOS pools. We chose '/' because it's not
  90  * acceptable in bucket names and thus qualified buckets cannot conflict
  91  * with the legacy or S3 buckets.
  92  */
  93 std::string rgw_make_bucket_entry_name(const std::string& tenant_name,
  94                                        const std::string& bucket_name) {
  95   std::string bucket_entry;
  96
  97   if (bucket_name.empty()) {
  98     bucket_entry.clear();
  99   } else if (tenant_name.empty()) {
 100     bucket_entry = bucket_name;
 101   } else {
 102     bucket_entry = tenant_name + "/" + bucket_name;
 103   }
 104
 105   return bucket_entry;
 106 }
 107
 108 /*
 109  * Tenants are separated from buckets in URLs by a colon in S3.
 110  * This function is not to be used on Swift URLs, not even for COPY arguments.
 111  */
 112 int rgw_parse_url_bucket(const string &bucket, const string& auth_tenant,
 113                          string &tenant_name, string &bucket_name) {
 114
 115   int pos = bucket.find(':');
 116   if (pos >= 0) {
 117     /*
 118      * N.B.: We allow ":bucket" syntax with explicit empty tenant in order
 119      * to refer to the legacy tenant, in case users in new named tenants
 120      * want to access old global buckets.
 121      */
 122     tenant_name = bucket.substr(0, pos);
 123     bucket_name = bucket.substr(pos + 1);
 124     if (bucket_name.empty()) {
 125       return -ERR_INVALID_BUCKET_NAME;
 126     }
 127   } else {
 128     tenant_name = auth_tenant;
 129     bucket_name = bucket;
 130   }
 131   return 0;
 132 }
 133
 134 int rgw_chown_bucket_and_objects(rgw::sal::Driver* driver, rgw::sal::Bucket* bucket,
 135                                  rgw::sal::User* new_user,
 136                                  const std::string& marker, std::string *err_msg,
 137                                  const DoutPrefixProvider *dpp, optional_yield y)
 138 {
 139   /* Chown on the bucket */
 140   int ret = bucket->chown(dpp, *new_user, y);
 141   if (ret < 0) {
 142     set_err_msg(err_msg, "Failed to change object ownership: " + cpp_strerror(-ret));
 143   }
 144
 145   /* Now chown on all the objects in the bucket */
 146   map<string, bool> common_prefixes;
 147
 148   rgw::sal::Bucket::ListParams params;
 149   rgw::sal::Bucket::ListResults results;
 150
 151   params.list_versions = true;
 152   params.allow_unordered = true;
 153   params.marker = marker;
 154
 155   int count = 0;
 156   int max_entries = 1000;
 157
 158   //Loop through objects and update object acls to point to bucket owner
 159
 160   do {
 161     results.objs.clear();
 162     ret = bucket->list(dpp, params, max_entries, results, y);
 163     if (ret < 0) {
 164       ldpp_dout(dpp, 0) << "ERROR: list objects failed: " << cpp_strerror(-ret) << dendl;
 165       return ret;
 166     }
 167
 168     params.marker = results.next_marker;
 169     count += results.objs.size();
 170
 171     for (const auto& obj : results.objs) {
 172       std::unique_ptr<rgw::sal::Object> r_obj = bucket->get_object(obj.key);
 173
 174       ret = r_obj->chown(*new_user, dpp, y);
 175         if (ret < 0) {
 176           ldpp_dout(dpp, 0) << "ERROR: chown failed on " << r_obj << " :" << cpp_strerror(-ret) << dendl;
 177           return ret;
 178         }
 179       }
 180     cerr << count << " objects processed in " << bucket
 181         << ". Next marker " << params.marker.name << std::endl;
 182   } while(results.is_truncated);
 183
 184   return ret;
 185 }
 186