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