src/ceph/src/rgw/rgw_http_client.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_HTTP_CLIENT_H
   5 #define CEPH_RGW_HTTP_CLIENT_H
   6
   7 #include "common/RWLock.h"
   8 #include "common/Cond.h"
   9 #include "rgw_common.h"
  10 #include "rgw_string.h"
  11
  12 #include <atomic>
  13
  14 using param_pair_t = pair<string, string>;
  15 using param_vec_t = vector<param_pair_t>;
  16
  17 struct rgw_http_req_data;
  18
  19 class RGWHTTPClient
  20 {
  21   friend class RGWHTTPManager;
  22
  23   bufferlist send_bl;
  24   bufferlist::iterator send_iter;
  25   size_t send_len;
  26   bool has_send_len;
  27   long http_status;
  28
  29   rgw_http_req_data *req_data;
  30
  31   void *user_info;
  32
  33   string last_method;
  34   string last_url;
  35   bool verify_ssl; // Do not validate self signed certificates, default to false
  36
  37   std::atomic<unsigned> stopped { 0 };
  38
  39 protected:
  40   CephContext *cct;
  41   param_vec_t headers;
  42
  43   int init_request(const char *method,
  44                    const char *url,
  45                    rgw_http_req_data *req_data,
  46                    bool send_data_hint = false);
  47
  48   virtual int receive_header(void *ptr, size_t len) {
  49     return 0;
  50   }
  51   virtual int receive_data(void *ptr, size_t len) {
  52     return 0;
  53   }
  54   virtual int send_data(void *ptr, size_t len) {
  55     return 0;
  56   }
  57
  58   /* Callbacks for libcurl. */
  59   static size_t simple_receive_http_header(void *ptr,
  60                                            size_t size,
  61                                            size_t nmemb,
  62                                            void *_info);
  63   static size_t receive_http_header(void *ptr,
  64                                     size_t size,
  65                                     size_t nmemb,
  66                                     void *_info);
  67
  68   static size_t simple_receive_http_data(void *ptr,
  69                                          size_t size,
  70                                          size_t nmemb,
  71                                          void *_info);
  72   static size_t receive_http_data(void *ptr,
  73                                   size_t size,
  74                                   size_t nmemb,
  75                                   void *_info);
  76
  77   static size_t simple_send_http_data(void *ptr,
  78                                       size_t size,
  79                                       size_t nmemb,
  80                                       void *_info);
  81   static size_t send_http_data(void *ptr,
  82                                size_t size,
  83                                size_t nmemb,
  84                                void *_info);
  85 public:
  86   static const long HTTP_STATUS_NOSTATUS     = 0;
  87   static const long HTTP_STATUS_UNAUTHORIZED = 401;
  88   static const long HTTP_STATUS_NOTFOUND     = 404;
  89
  90   virtual ~RGWHTTPClient();
  91   explicit RGWHTTPClient(CephContext *cct)
  92     : send_len(0),
  93       has_send_len(false),
  94       http_status(HTTP_STATUS_NOSTATUS),
  95       req_data(nullptr),
  96       user_info(nullptr),
  97       verify_ssl(cct->_conf->rgw_verify_ssl),
  98       cct(cct) {
  99   }
 100
 101   void set_user_info(void *info) {
 102     user_info = info;
 103   }
 104
 105   void *get_user_info() {
 106     return user_info;
 107   }
 108
 109   void append_header(const string& name, const string& val) {
 110     headers.push_back(pair<string, string>(name, val));
 111   }
 112
 113   void set_send_length(size_t len) {
 114     send_len = len;
 115     has_send_len = true;
 116   }
 117
 118
 119   long get_http_status() const {
 120     return http_status;
 121   }
 122
 123   void set_verify_ssl(bool flag) {
 124     verify_ssl = flag;
 125   }
 126
 127   int process(const char *method, const char *url);
 128   int process(const char *url) { return process("GET", url); }
 129
 130   int wait();
 131   rgw_http_req_data *get_req_data() { return req_data; }
 132
 133   string to_str();
 134
 135   int get_req_retcode();
 136 };
 137
 138
 139 class RGWHTTPHeadersCollector : public RGWHTTPClient {
 140 public:
 141   typedef std::string header_name_t;
 142   typedef std::string header_value_t;
 143   typedef std::set<header_name_t, ltstr_nocase> header_spec_t;
 144
 145   RGWHTTPHeadersCollector(CephContext * const cct,
 146                           const header_spec_t relevant_headers)
 147     : RGWHTTPClient(cct),
 148       relevant_headers(relevant_headers) {
 149   }
 150
 151   std::map<header_name_t, header_value_t, ltstr_nocase> get_headers() const {
 152     return found_headers;
 153   }
 154
 155   /* Throws std::out_of_range */
 156   const header_value_t& get_header_value(const header_name_t& name) const {
 157     return found_headers.at(name);
 158   }
 159
 160 protected:
 161   int receive_header(void *ptr, size_t len) override;
 162
 163   int receive_data(void *ptr, size_t len) override {
 164     return 0;
 165   }
 166
 167   int send_data(void *ptr, size_t len) override {
 168     return 0;
 169   }
 170
 171 private:
 172   const std::set<header_name_t, ltstr_nocase> relevant_headers;
 173   std::map<header_name_t, header_value_t, ltstr_nocase> found_headers;
 174 };
 175
 176
 177 class RGWHTTPTransceiver : public RGWHTTPHeadersCollector {
 178   bufferlist * const read_bl;
 179   std::string post_data;
 180   size_t post_data_index;
 181
 182 public:
 183   RGWHTTPTransceiver(CephContext * const cct,
 184                      bufferlist * const read_bl,
 185                      const header_spec_t intercept_headers = {})
 186     : RGWHTTPHeadersCollector(cct, intercept_headers),
 187       read_bl(read_bl),
 188       post_data_index(0) {
 189   }
 190
 191   RGWHTTPTransceiver(CephContext * const cct,
 192                      bufferlist * const read_bl,
 193                      const bool verify_ssl,
 194                      const header_spec_t intercept_headers = {})
 195     : RGWHTTPHeadersCollector(cct, intercept_headers),
 196       read_bl(read_bl),
 197       post_data_index(0) {
 198     set_verify_ssl(verify_ssl);
 199   }
 200
 201   void set_post_data(const std::string& _post_data) {
 202     this->post_data = _post_data;
 203   }
 204
 205 protected:
 206   int send_data(void* ptr, size_t len) override;
 207
 208   int receive_data(void *ptr, size_t len) override {
 209     read_bl->append((char *)ptr, len);
 210     return 0;
 211   }
 212 };
 213
 214 typedef RGWHTTPTransceiver RGWPostHTTPData;
 215
 216
 217 class RGWCompletionManager;
 218
 219 class RGWHTTPManager {
 220   CephContext *cct;
 221   RGWCompletionManager *completion_mgr;
 222   void *multi_handle;
 223   bool is_threaded;
 224   std::atomic<unsigned> going_down { 0 };
 225   std::atomic<unsigned> is_stopped { 0 };
 226
 227   RWLock reqs_lock;
 228   map<uint64_t, rgw_http_req_data *> reqs;
 229   list<rgw_http_req_data *> unregistered_reqs;
 230   map<uint64_t, rgw_http_req_data *> complete_reqs;
 231   int64_t num_reqs;
 232   int64_t max_threaded_req;
 233   int thread_pipe[2];
 234
 235   void register_request(rgw_http_req_data *req_data);
 236   void complete_request(rgw_http_req_data *req_data);
 237   void _complete_request(rgw_http_req_data *req_data);
 238   void unregister_request(rgw_http_req_data *req_data);
 239   void _unlink_request(rgw_http_req_data *req_data);
 240   void unlink_request(rgw_http_req_data *req_data);
 241   void finish_request(rgw_http_req_data *req_data, int r);
 242   void _finish_request(rgw_http_req_data *req_data, int r);
 243   int link_request(rgw_http_req_data *req_data);
 244
 245   void manage_pending_requests();
 246
 247   class ReqsThread : public Thread {
 248     RGWHTTPManager *manager;
 249
 250   public:
 251     ReqsThread(RGWHTTPManager *_m) : manager(_m) {}
 252     void *entry() override;
 253   };
 254
 255   ReqsThread *reqs_thread;
 256
 257   void *reqs_thread_entry();
 258
 259   int signal_thread();
 260
 261 public:
 262   RGWHTTPManager(CephContext *_cct, RGWCompletionManager *completion_mgr = NULL);
 263   ~RGWHTTPManager();
 264
 265   int set_threaded();
 266   void stop();
 267
 268   int add_request(RGWHTTPClient *client, const char *method, const char *url,
 269                   bool send_data_hint = false);
 270   int remove_request(RGWHTTPClient *client);
 271
 272   /* only for non threaded case */
 273   int process_requests(bool wait_for_data, bool *done);
 274
 275   int complete_requests();
 276 };
 277
 278 #endif