src/ceph/src/rgw/rgw_rest_conn.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include "rgw_rados.h"
   5 #include "rgw_rest_conn.h"
   6
   7 #define dout_subsys ceph_subsys_rgw
   8
   9 RGWRESTConn::RGWRESTConn(CephContext *_cct, RGWRados *store,
  10                          const string& _remote_id,
  11                          const list<string>& remote_endpoints)
  12   : cct(_cct),
  13     endpoints(remote_endpoints.begin(), remote_endpoints.end()),
  14     remote_id(_remote_id)
  15 {
  16   if (store) {
  17     key = store->get_zone_params().system_key;
  18     self_zone_group = store->get_zonegroup().get_id();
  19   }
  20 }
  21
  22 RGWRESTConn::RGWRESTConn(RGWRESTConn&& other)
  23   : cct(other.cct),
  24     endpoints(std::move(other.endpoints)),
  25     key(std::move(other.key)),
  26     self_zone_group(std::move(other.self_zone_group)),
  27     remote_id(std::move(other.remote_id)),
  28     counter(other.counter.load())
  29 {
  30 }
  31
  32 RGWRESTConn& RGWRESTConn::operator=(RGWRESTConn&& other)
  33 {
  34   cct = other.cct;
  35   endpoints = std::move(other.endpoints);
  36   key = std::move(other.key);
  37   self_zone_group = std::move(other.self_zone_group);
  38   remote_id = std::move(other.remote_id);
  39   counter = other.counter.load();
  40   return *this;
  41 }
  42
  43 int RGWRESTConn::get_url(string& endpoint)
  44 {
  45   if (endpoints.empty()) {
  46     ldout(cct, 0) << "ERROR: endpoints not configured for upstream zone" << dendl;
  47     return -EIO;
  48   }
  49
  50   int i = ++counter;
  51   endpoint = endpoints[i % endpoints.size()];
  52
  53   return 0;
  54 }
  55
  56 string RGWRESTConn::get_url()
  57 {
  58   string endpoint;
  59   if (endpoints.empty()) {
  60     ldout(cct, 0) << "WARNING: endpoints not configured for upstream zone" << dendl; /* we'll catch this later */
  61     return endpoint;
  62   }
  63
  64   int i = ++counter;
  65   endpoint = endpoints[i % endpoints.size()];
  66
  67   return endpoint;
  68 }
  69
  70 static void populate_params(param_vec_t& params, const rgw_user *uid, const string& zonegroup)
  71 {
  72   if (uid) {
  73     string uid_str = uid->to_str();
  74     if (!uid->empty()) {
  75       params.push_back(param_pair_t(RGW_SYS_PARAM_PREFIX "uid", uid_str));
  76     }
  77   }
  78   if (!zonegroup.empty()) {
  79     params.push_back(param_pair_t(RGW_SYS_PARAM_PREFIX "zonegroup", zonegroup));
  80   }
  81 }
  82
  83 int RGWRESTConn::forward(const rgw_user& uid, req_info& info, obj_version *objv, size_t max_response, bufferlist *inbl, bufferlist *outbl)
  84 {
  85   string url;
  86   int ret = get_url(url);
  87   if (ret < 0)
  88     return ret;
  89   param_vec_t params;
  90   populate_params(params, &uid, self_zone_group);
  91   if (objv) {
  92     params.push_back(param_pair_t(RGW_SYS_PARAM_PREFIX "tag", objv->tag));
  93     char buf[16];
  94     snprintf(buf, sizeof(buf), "%lld", (long long)objv->ver);
  95     params.push_back(param_pair_t(RGW_SYS_PARAM_PREFIX "ver", buf));
  96   }
  97   RGWRESTSimpleRequest req(cct, url, NULL, &params);
  98   return req.forward_request(key, info, max_response, inbl, outbl);
  99 }
 100
 101 class StreamObjData : public RGWGetDataCB {
 102   rgw_obj obj;
 103 public:
 104     explicit StreamObjData(rgw_obj& _obj) : obj(_obj) {}
 105 };
 106
 107 int RGWRESTConn::put_obj_init(const rgw_user& uid, rgw_obj& obj, uint64_t obj_size,
 108                                       map<string, bufferlist>& attrs, RGWRESTStreamWriteRequest **req)
 109 {
 110   string url;
 111   int ret = get_url(url);
 112   if (ret < 0)
 113     return ret;
 114
 115   param_vec_t params;
 116   populate_params(params, &uid, self_zone_group);
 117   RGWRESTStreamWriteRequest *wr = new RGWRESTStreamWriteRequest(cct, url, NULL, &params);
 118   ret = wr->put_obj_init(key, obj, obj_size, attrs);
 119   if (ret < 0) {
 120     delete wr;
 121     return ret;
 122   }
 123   *req = wr;
 124   return 0;
 125 }
 126
 127 int RGWRESTConn::complete_request(RGWRESTStreamWriteRequest *req, string& etag, real_time *mtime)
 128 {
 129   int ret = req->complete(etag, mtime);
 130   delete req;
 131
 132   return ret;
 133 }
 134
 135 static void set_date_header(const real_time *t, map<string, string>& headers, const string& header_name)
 136 {
 137   if (!t) {
 138     return;
 139   }
 140   stringstream s;
 141   utime_t tm = utime_t(*t);
 142   tm.gmtime_nsec(s);
 143   headers[header_name] = s.str();
 144 }
 145
 146 template <class T>
 147 static void set_header(T val, map<string, string>& headers, const string& header_name)
 148 {
 149   stringstream s;
 150   s << val;
 151   headers[header_name] = s.str();
 152 }
 153
 154
 155 int RGWRESTConn::get_obj(const rgw_user& uid, req_info *info /* optional */, rgw_obj& obj,
 156                          const real_time *mod_ptr, const real_time *unmod_ptr,
 157                          uint32_t mod_zone_id, uint64_t mod_pg_ver,
 158                          bool prepend_metadata, bool get_op, bool rgwx_stat,
 159                          bool sync_manifest, bool skip_decrypt, RGWGetDataCB *cb,
 160                          RGWRESTStreamRWRequest **req)
 161 {
 162   string url;
 163   int ret = get_url(url);
 164   if (ret < 0)
 165     return ret;
 166
 167   param_vec_t params;
 168   populate_params(params, &uid, self_zone_group);
 169   if (prepend_metadata) {
 170     params.push_back(param_pair_t(RGW_SYS_PARAM_PREFIX "prepend-metadata", self_zone_group));
 171   }
 172   if (rgwx_stat) {
 173     params.push_back(param_pair_t(RGW_SYS_PARAM_PREFIX "stat", "true"));
 174   }
 175   if (sync_manifest) {
 176     params.push_back(param_pair_t(RGW_SYS_PARAM_PREFIX "sync-manifest", ""));
 177   }
 178   if (skip_decrypt) {
 179     params.push_back(param_pair_t(RGW_SYS_PARAM_PREFIX "skip-decrypt", ""));
 180   }
 181   if (!obj.key.instance.empty()) {
 182     const string& instance = obj.key.instance;
 183     params.push_back(param_pair_t("versionId", instance));
 184   }
 185   if (get_op) {
 186     *req = new RGWRESTStreamReadRequest(cct, url, cb, NULL, &params);
 187   } else {
 188     *req = new RGWRESTStreamHeadRequest(cct, url, cb, NULL, &params);
 189   }
 190   map<string, string> extra_headers;
 191   if (info) {
 192     const auto& orig_map = info->env->get_map();
 193
 194     /* add original headers that start with HTTP_X_AMZ_ */
 195     static constexpr char SEARCH_AMZ_PREFIX[] = "HTTP_X_AMZ_";
 196     for (auto iter= orig_map.lower_bound(SEARCH_AMZ_PREFIX); iter != orig_map.end(); ++iter) {
 197       const string& name = iter->first;
 198       if (name == "HTTP_X_AMZ_DATE") /* dont forward date from original request */
 199         continue;
 200       if (name.compare(0, strlen(SEARCH_AMZ_PREFIX), SEARCH_AMZ_PREFIX) != 0)
 201         break;
 202       extra_headers[iter->first] = iter->second;
 203     }
 204   }
 205
 206   set_date_header(mod_ptr, extra_headers, "HTTP_IF_MODIFIED_SINCE");
 207   set_date_header(unmod_ptr, extra_headers, "HTTP_IF_UNMODIFIED_SINCE");
 208   if (mod_zone_id != 0) {
 209     set_header(mod_zone_id, extra_headers, "HTTP_DEST_ZONE_SHORT_ID");
 210   }
 211   if (mod_pg_ver != 0) {
 212     set_header(mod_pg_ver, extra_headers, "HTTP_DEST_PG_VER");
 213   }
 214
 215   int r = (*req)->send_request(key, extra_headers, obj);
 216   if (r < 0) {
 217     delete *req;
 218     *req = nullptr;
 219   }
 220
 221   return r;
 222 }
 223
 224 int RGWRESTConn::complete_request(RGWRESTStreamRWRequest *req, string& etag, real_time *mtime,
 225                                   uint64_t *psize, map<string, string>& attrs)
 226 {
 227   int ret = req->complete_request(etag, mtime, psize, attrs);
 228   delete req;
 229
 230   return ret;
 231 }
 232
 233 int RGWRESTConn::get_resource(const string& resource,
 234                      param_vec_t *extra_params,
 235                      map<string, string> *extra_headers,
 236                      bufferlist& bl,
 237                      bufferlist *send_data,
 238                      RGWHTTPManager *mgr)
 239 {
 240   string url;
 241   int ret = get_url(url);
 242   if (ret < 0)
 243     return ret;
 244
 245   param_vec_t params;
 246
 247   if (extra_params) {
 248     params.insert(params.end(), extra_params->begin(), extra_params->end());
 249   }
 250
 251   populate_params(params, nullptr, self_zone_group);
 252
 253   RGWStreamIntoBufferlist cb(bl);
 254
 255   RGWRESTStreamReadRequest req(cct, url, &cb, NULL, &params);
 256
 257   map<string, string> headers;
 258   if (extra_headers) {
 259     headers.insert(extra_headers->begin(), extra_headers->end());
 260   }
 261
 262   ret = req.send_request(&key, headers, resource, send_data, mgr);
 263   if (ret < 0) {
 264     ldout(cct, 5) << __func__ << ": send_request() resource=" << resource << " returned ret=" << ret << dendl;
 265     return ret;
 266   }
 267
 268   string etag;
 269   map<string, string> attrs;
 270   return req.complete_request(etag, NULL, NULL, attrs);
 271 }
 272
 273 RGWRESTReadResource::RGWRESTReadResource(RGWRESTConn *_conn,
 274                                          const string& _resource,
 275                                          const rgw_http_param_pair *pp,
 276                                          param_vec_t *extra_headers,
 277                                          RGWHTTPManager *_mgr)
 278   : cct(_conn->get_ctx()), conn(_conn), resource(_resource),
 279     params(make_param_list(pp)), cb(bl), mgr(_mgr),
 280     req(cct, conn->get_url(), &cb, NULL, NULL)
 281 {
 282   init_common(extra_headers);
 283 }
 284
 285 RGWRESTReadResource::RGWRESTReadResource(RGWRESTConn *_conn,
 286                                          const string& _resource,
 287                                          param_vec_t& _params,
 288                                          param_vec_t *extra_headers,
 289                                          RGWHTTPManager *_mgr)
 290   : cct(_conn->get_ctx()), conn(_conn), resource(_resource), params(_params),
 291     cb(bl), mgr(_mgr), req(cct, conn->get_url(), &cb, NULL, NULL)
 292 {
 293   init_common(extra_headers);
 294 }
 295
 296 void RGWRESTReadResource::init_common(param_vec_t *extra_headers)
 297 {
 298   populate_params(params, nullptr, conn->get_self_zonegroup());
 299
 300   if (extra_headers) {
 301     headers.insert(extra_headers->begin(), extra_headers->end());
 302   }
 303
 304   req.set_params(&params);
 305 }
 306
 307 int RGWRESTReadResource::read()
 308 {
 309   int ret = req.send_request(&conn->get_key(), headers, resource, nullptr, mgr);
 310   if (ret < 0) {
 311     ldout(cct, 5) << __func__ << ": send_request() resource=" << resource << " returned ret=" << ret << dendl;
 312     return ret;
 313   }
 314
 315   string etag;
 316   map<string, string> attrs;
 317   return req.complete_request(etag, NULL, NULL, attrs);
 318 }
 319
 320 int RGWRESTReadResource::aio_read()
 321 {
 322   int ret = req.send_request(&conn->get_key(), headers, resource, nullptr, mgr);
 323   if (ret < 0) {
 324     ldout(cct, 5) << __func__ << ": send_request() resource=" << resource << " returned ret=" << ret << dendl;
 325     return ret;
 326   }
 327
 328   return 0;
 329 }
 330
 331 RGWRESTSendResource::RGWRESTSendResource(RGWRESTConn *_conn,
 332                                          const string& _method,
 333                                          const string& _resource,
 334                                          const rgw_http_param_pair *pp,
 335                                          param_vec_t *extra_headers,
 336                                          RGWHTTPManager *_mgr)
 337   : cct(_conn->get_ctx()), conn(_conn), method(_method), resource(_resource),
 338     params(make_param_list(pp)), cb(bl), mgr(_mgr),
 339     req(cct, method.c_str(), conn->get_url(), &cb, NULL, NULL)
 340 {
 341   init_common(extra_headers);
 342 }
 343
 344 RGWRESTSendResource::RGWRESTSendResource(RGWRESTConn *_conn,
 345                                          const string& _method,
 346                                          const string& _resource,
 347                                          param_vec_t& params,
 348                                          param_vec_t *extra_headers,
 349                                          RGWHTTPManager *_mgr)
 350   : cct(_conn->get_ctx()), conn(_conn), method(_method), resource(_resource), params(params),
 351     cb(bl), mgr(_mgr), req(cct, method.c_str(), conn->get_url(), &cb, NULL, NULL)
 352 {
 353   init_common(extra_headers);
 354 }
 355
 356 void RGWRESTSendResource::init_common(param_vec_t *extra_headers)
 357 {
 358   populate_params(params, nullptr, conn->get_self_zonegroup());
 359
 360   if (extra_headers) {
 361     headers.insert(extra_headers->begin(), extra_headers->end());
 362   }
 363
 364   req.set_params(&params);
 365 }
 366
 367 int RGWRESTSendResource::send(bufferlist& outbl)
 368 {
 369   req.set_outbl(outbl);
 370   int ret = req.send_request(&conn->get_key(), headers, resource, nullptr, mgr);
 371   if (ret < 0) {
 372     ldout(cct, 5) << __func__ << ": send_request() resource=" << resource << " returned ret=" << ret << dendl;
 373     return ret;
 374   }
 375
 376   string etag;
 377   map<string, string> attrs;
 378   return req.complete_request(etag, NULL, NULL, attrs);
 379 }
 380
 381 int RGWRESTSendResource::aio_send(bufferlist& outbl)
 382 {
 383   req.set_outbl(outbl);
 384   int ret = req.send_request(&conn->get_key(), headers, resource, nullptr, mgr);
 385   if (ret < 0) {
 386     ldout(cct, 5) << __func__ << ": send_request() resource=" << resource << " returned ret=" << ret << dendl;
 387     return ret;
 388   }
 389
 390   return 0;
 391 }
 392