ceph/src/rgw/rgw_rest_client.h

   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 #pragma once
   5
   6 #include "rgw_http_client.h"
   7
   8 class RGWGetDataCB;
   9
  10 class RGWHTTPSimpleRequest : public RGWHTTPClient {
  11 protected:
  12   int http_status;
  13   int status;
  14
  15   using unique_lock = std::unique_lock<std::mutex>;
  16
  17   std::mutex out_headers_lock;
  18   map<string, string> out_headers;
  19   param_vec_t params;
  20
  21   bufferlist::iterator *send_iter;
  22
  23   size_t max_response; /* we need this as we don't stream out response */
  24   bufferlist response;
  25
  26   virtual int handle_header(const string& name, const string& val);
  27   void get_params_str(map<string, string>& extra_args, string& dest);
  28
  29 public:
  30   RGWHTTPSimpleRequest(CephContext *_cct, const string& _method, const string& _url,
  31                        param_vec_t *_headers, param_vec_t *_params) : RGWHTTPClient(_cct, _method, _url),
  32                 http_status(0), status(0),
  33                 send_iter(NULL),
  34                 max_response(0) {
  35     set_headers(_headers);
  36     set_params(_params);
  37   }
  38
  39   void set_headers(param_vec_t *_headers) {
  40     if (_headers)
  41       headers = *_headers;
  42   }
  43
  44   void set_params(param_vec_t *_params) {
  45     if (_params)
  46       params = *_params;
  47   }
  48
  49   int receive_header(void *ptr, size_t len) override;
  50   int receive_data(void *ptr, size_t len, bool *pause) override;
  51   int send_data(void *ptr, size_t len, bool* pause=nullptr) override;
  52
  53   bufferlist& get_response() { return response; }
  54
  55   void get_out_headers(map<string, string> *pheaders); /* modifies out_headers */
  56
  57   int get_http_status() { return http_status; }
  58   int get_status();
  59 };
  60
  61 class RGWRESTSimpleRequest : public RGWHTTPSimpleRequest {
  62 public:
  63   RGWRESTSimpleRequest(CephContext *_cct, const string& _method, const string& _url,
  64                        param_vec_t *_headers, param_vec_t *_params) : RGWHTTPSimpleRequest(_cct, _method, _url, _headers, _params) {}
  65
  66   int execute(const DoutPrefixProvider *dpp, RGWAccessKey& key, const char *method, const char *resource, optional_yield y);
  67   int forward_request(const DoutPrefixProvider *dpp, RGWAccessKey& key, req_info& info, size_t max_response, bufferlist *inbl, bufferlist *outbl, optional_yield y);
  68 };
  69
  70 class RGWWriteDrainCB {
  71 public:
  72   RGWWriteDrainCB() = default;
  73   virtual ~RGWWriteDrainCB() = default;
  74   virtual void notify(uint64_t pending_size) = 0;
  75 };
  76
  77 class RGWRESTGenerateHTTPHeaders {
  78   CephContext *cct;
  79   RGWEnv *new_env;
  80   req_info *new_info;
  81   string method;
  82   string url;
  83   string resource;
  84
  85 public:
  86   RGWRESTGenerateHTTPHeaders(CephContext *_cct, RGWEnv *_env, req_info *_info) : cct(_cct), new_env(_env), new_info(_info) {}
  87   void init(const string& method, const string& url, const string& resource, const param_vec_t& params);
  88   void set_extra_headers(const map<string, string>& extra_headers);
  89   int set_obj_attrs(const DoutPrefixProvider *dpp, map<string, bufferlist>& rgw_attrs);
  90   void set_http_attrs(const map<string, string>& http_attrs);
  91   void set_policy(RGWAccessControlPolicy& policy);
  92   int sign(const DoutPrefixProvider *dpp, RGWAccessKey& key);
  93
  94   const string& get_url() { return url; }
  95 };
  96
  97 class RGWHTTPStreamRWRequest : public RGWHTTPSimpleRequest {
  98 public:
  99     class ReceiveCB;
 100
 101 private:
 102   ceph::mutex lock =
 103     ceph::make_mutex("RGWHTTPStreamRWRequest");
 104   ceph::mutex write_lock =
 105     ceph::make_mutex("RGWHTTPStreamRWRequest::write_lock");
 106   ReceiveCB *cb{nullptr};
 107   RGWWriteDrainCB *write_drain_cb{nullptr};
 108   bufferlist outbl;
 109   bufferlist in_data;
 110   size_t chunk_ofs{0};
 111   size_t ofs{0};
 112   uint64_t write_ofs{0};
 113   bool read_paused{false};
 114   bool send_paused{false};
 115   bool stream_writes{false};
 116   bool write_stream_complete{false};
 117 protected:
 118   int handle_header(const string& name, const string& val) override;
 119 public:
 120   int send_data(void *ptr, size_t len, bool *pause) override;
 121   int receive_data(void *ptr, size_t len, bool *pause) override;
 122
 123   class ReceiveCB {
 124     protected:
 125       uint64_t extra_data_len{0};
 126     public:
 127       ReceiveCB() = default;
 128       virtual ~ReceiveCB() = default;
 129       virtual int handle_data(bufferlist& bl, bool *pause = nullptr) = 0;
 130       virtual void set_extra_data_len(uint64_t len) {
 131         extra_data_len = len;
 132       }
 133   };
 134
 135   RGWHTTPStreamRWRequest(CephContext *_cct, const string& _method, const string& _url,
 136                          param_vec_t *_headers, param_vec_t *_params) : RGWHTTPSimpleRequest(_cct, _method, _url, _headers, _params) {
 137   }
 138   RGWHTTPStreamRWRequest(CephContext *_cct, const string& _method, const string& _url, ReceiveCB *_cb,
 139                          param_vec_t *_headers, param_vec_t *_params) : RGWHTTPSimpleRequest(_cct, _method, _url, _headers, _params),
 140                                                                         cb(_cb) {
 141   }
 142   virtual ~RGWHTTPStreamRWRequest() override {}
 143
 144   void set_outbl(bufferlist& _outbl) {
 145     outbl.swap(_outbl);
 146   }
 147
 148   void set_in_cb(ReceiveCB *_cb) { cb = _cb; }
 149   void set_write_drain_cb(RGWWriteDrainCB *_cb) { write_drain_cb = _cb; }
 150
 151   void unpause_receive();
 152
 153   void add_send_data(bufferlist& bl);
 154
 155   void set_stream_write(bool s);
 156
 157   uint64_t get_pending_send_size();
 158
 159   /* finish streaming writes */
 160   void finish_write();
 161 };
 162
 163 class RGWRESTStreamRWRequest : public RGWHTTPStreamRWRequest {
 164 protected:
 165   HostStyle host_style;
 166 public:
 167   RGWRESTStreamRWRequest(CephContext *_cct, const string& _method, const string& _url, RGWHTTPStreamRWRequest::ReceiveCB *_cb,
 168                 param_vec_t *_headers, param_vec_t *_params, HostStyle _host_style = PathStyle) : RGWHTTPStreamRWRequest(_cct, _method, _url, _cb, _headers, _params), host_style(_host_style) {
 169   }
 170   virtual ~RGWRESTStreamRWRequest() override {}
 171
 172   int send_prepare(const DoutPrefixProvider *dpp, RGWAccessKey *key, map<string, string>& extra_headers, const string& resource, bufferlist *send_data = nullptr /* optional input data */);
 173   int send_prepare(const DoutPrefixProvider *dpp, RGWAccessKey& key, map<string, string>& extra_headers, const rgw_obj& obj);
 174   int send(RGWHTTPManager *mgr);
 175
 176   int send_request(const DoutPrefixProvider *dpp, RGWAccessKey& key, map<string, string>& extra_headers, const rgw_obj& obj, RGWHTTPManager *mgr);
 177   int send_request(const DoutPrefixProvider *dpp, RGWAccessKey *key, map<string, string>& extra_headers, const string& resource, RGWHTTPManager *mgr, bufferlist *send_data = nullptr /* optional input data */);
 178
 179   int complete_request(optional_yield y,
 180                        string *etag = nullptr,
 181                        real_time *mtime = nullptr,
 182                        uint64_t *psize = nullptr,
 183                        map<string, string> *pattrs = nullptr,
 184                        map<string, string> *pheaders = nullptr);
 185
 186   void add_params(param_vec_t *params);
 187
 188 private:
 189   int do_send_prepare(const DoutPrefixProvider *dpp, RGWAccessKey *key, map<string, string>& extra_headers, const string& resource, bufferlist *send_data = nullptr /* optional input data */);
 190 };
 191
 192 class RGWRESTStreamReadRequest : public RGWRESTStreamRWRequest {
 193 public:
 194   RGWRESTStreamReadRequest(CephContext *_cct, const string& _url, ReceiveCB *_cb, param_vec_t *_headers,
 195                 param_vec_t *_params, HostStyle _host_style = PathStyle) : RGWRESTStreamRWRequest(_cct, "GET", _url, _cb, _headers, _params, _host_style) {}
 196 };
 197
 198 class RGWRESTStreamHeadRequest : public RGWRESTStreamRWRequest {
 199 public:
 200   RGWRESTStreamHeadRequest(CephContext *_cct, const string& _url, ReceiveCB *_cb, param_vec_t *_headers,
 201                 param_vec_t *_params) : RGWRESTStreamRWRequest(_cct, "HEAD", _url, _cb, _headers, _params) {}
 202 };
 203
 204 class RGWRESTStreamS3PutObj : public RGWRESTStreamRWRequest {
 205   RGWGetDataCB *out_cb;
 206   RGWEnv new_env;
 207   req_info new_info;
 208   RGWRESTGenerateHTTPHeaders headers_gen;
 209 public:
 210   RGWRESTStreamS3PutObj(CephContext *_cct, const string& _method, const string& _url, param_vec_t *_headers,
 211                 param_vec_t *_params, HostStyle _host_style) : RGWRESTStreamRWRequest(_cct, _method, _url, nullptr, _headers, _params, _host_style),
 212                 out_cb(NULL), new_info(cct, &new_env), headers_gen(_cct, &new_env, &new_info) {}
 213   ~RGWRESTStreamS3PutObj() override;
 214
 215   void send_init(rgw::sal::RGWObject* obj);
 216   int send_ready(const DoutPrefixProvider *dpp, RGWAccessKey& key, map<string, bufferlist>& rgw_attrs, bool send);
 217   int send_ready(const DoutPrefixProvider *dpp, RGWAccessKey& key, const map<string, string>& http_attrs,
 218                  RGWAccessControlPolicy& policy, bool send);
 219   int send_ready(const DoutPrefixProvider *dpp, RGWAccessKey& key, bool send);
 220
 221   int put_obj_init(const DoutPrefixProvider *dpp, RGWAccessKey& key, rgw::sal::RGWObject* obj, uint64_t obj_size, map<string, bufferlist>& attrs, bool send);
 222
 223   RGWGetDataCB *get_out_cb() { return out_cb; }
 224 };