ceph/src/rgw/rgw_frontend.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 RGW_FRONTEND_H
   5 #define RGW_FRONTEND_H
   6
   7 #include <map>
   8 #include <string>
   9
  10 #include "rgw_request.h"
  11 #include "rgw_process.h"
  12 #include "rgw_realm_reloader.h"
  13
  14 #include "rgw_civetweb.h"
  15 #include "rgw_civetweb_log.h"
  16 #include "civetweb/civetweb.h"
  17
  18 #include "rgw_auth_registry.h"
  19
  20 #define dout_context g_ceph_context
  21 #define dout_subsys ceph_subsys_rgw
  22
  23 class RGWFrontendConfig {
  24   std::string config;
  25   std::map<std::string, std::string> config_map;
  26   std::string framework;
  27
  28   int parse_config(const std::string& config,
  29                    std::map<std::string, std::string>& config_map);
  30
  31 public:
  32   RGWFrontendConfig(const std::string& config)
  33     : config(config) {
  34   }
  35
  36   int init() {
  37     const int ret = parse_config(config, config_map);
  38     return ret < 0 ? ret : 0;
  39   }
  40
  41   bool get_val(const std::string& key,
  42                const std::string& def_val,
  43                std::string* out);
  44   bool get_val(const std::string& key, int def_val, int *out);
  45
  46   std::string get_val(const std::string& key,
  47                       const std::string& def_val) {
  48     std::string out;
  49     get_val(key, def_val, &out);
  50     return out;
  51   }
  52
  53   std::map<std::string, std::string>& get_config_map() {
  54     return config_map;
  55   }
  56
  57   std::string get_framework() const {
  58     return framework;
  59  }
  60 };
  61
  62 class RGWFrontend {
  63 public:
  64   virtual ~RGWFrontend() {}
  65
  66   virtual int init() = 0;
  67
  68   virtual int run() = 0;
  69   virtual void stop() = 0;
  70   virtual void join() = 0;
  71
  72   virtual void pause_for_new_config() = 0;
  73   virtual void unpause_with_new_config(RGWRados* store,
  74                                        rgw_auth_registry_ptr_t auth_registry) = 0;
  75 };
  76
  77
  78 struct RGWMongooseEnv : public RGWProcessEnv {
  79   // every request holds a read lock, so we need to prioritize write locks to
  80   // avoid starving pause_for_new_config()
  81   static constexpr bool prioritize_write = true;
  82   RWLock mutex;
  83
  84   RGWMongooseEnv(const RGWProcessEnv &env)
  85     : RGWProcessEnv(env),
  86       mutex("RGWCivetWebFrontend", false, true, prioritize_write) {
  87   }
  88 };
  89
  90
  91 class RGWCivetWebFrontend : public RGWFrontend {
  92   RGWFrontendConfig* conf;
  93   struct mg_context* ctx;
  94   RGWMongooseEnv env;
  95
  96   void set_conf_default(std::map<std::string, std::string>& m,
  97                         const std::string& key,
  98                         const std::string& def_val) {
  99     if (m.find(key) == std::end(m)) {
 100       m[key] = def_val;
 101     }
 102   }
 103
 104 public:
 105   RGWCivetWebFrontend(RGWProcessEnv& env,
 106                       RGWFrontendConfig* conf)
 107     : conf(conf),
 108       ctx(nullptr),
 109       env(env) {
 110   }
 111
 112   int init() override {
 113     return 0;
 114   }
 115
 116   int run() override;
 117
 118   int process(struct mg_connection* conn);
 119
 120   void stop() override {
 121     if (ctx) {
 122       mg_stop(ctx);
 123     }
 124   }
 125
 126   void join() override {
 127     return;
 128   }
 129
 130   void pause_for_new_config() override {
 131     // block callbacks until unpause
 132     env.mutex.get_write();
 133   }
 134
 135   void unpause_with_new_config(RGWRados* const store,
 136                                rgw_auth_registry_ptr_t auth_registry) override {
 137     env.store = store;
 138     env.auth_registry = std::move(auth_registry);
 139     // unpause callbacks
 140     env.mutex.put_write();
 141   }
 142 }; /* RGWCivetWebFrontend */
 143
 144 class RGWProcessFrontend : public RGWFrontend {
 145 protected:
 146   RGWFrontendConfig* conf;
 147   RGWProcess* pprocess;
 148   RGWProcessEnv env;
 149   RGWProcessControlThread* thread;
 150
 151 public:
 152   RGWProcessFrontend(RGWProcessEnv& pe, RGWFrontendConfig* _conf)
 153     : conf(_conf), pprocess(nullptr), env(pe), thread(nullptr) {
 154   }
 155
 156   ~RGWProcessFrontend() override {
 157     delete thread;
 158     delete pprocess;
 159   }
 160
 161   int run() override {
 162     assert(pprocess); /* should have initialized by init() */
 163     thread = new RGWProcessControlThread(pprocess);
 164     thread->create("rgw_frontend");
 165     return 0;
 166   }
 167
 168   void stop() override;
 169
 170   void join() override {
 171     thread->join();
 172   }
 173
 174   void pause_for_new_config() override {
 175     pprocess->pause();
 176   }
 177
 178   void unpause_with_new_config(RGWRados* const store,
 179                                rgw_auth_registry_ptr_t auth_registry) override {
 180     env.store = store;
 181     env.auth_registry = auth_registry;
 182     pprocess->unpause_with_new_config(store, std::move(auth_registry));
 183   }
 184 }; /* RGWProcessFrontend */
 185
 186 class RGWFCGXFrontend : public RGWProcessFrontend {
 187 public:
 188   RGWFCGXFrontend(RGWProcessEnv& pe, RGWFrontendConfig* _conf)
 189     : RGWProcessFrontend(pe, _conf) {}
 190
 191   int init() override {
 192     pprocess = new RGWFCGXProcess(g_ceph_context, &env,
 193                                   g_conf->rgw_thread_pool_size, conf);
 194     return 0;
 195   }
 196 }; /* RGWFCGXFrontend */
 197
 198 class RGWLoadGenFrontend : public RGWProcessFrontend {
 199 public:
 200   RGWLoadGenFrontend(RGWProcessEnv& pe, RGWFrontendConfig *_conf)
 201     : RGWProcessFrontend(pe, _conf) {}
 202
 203   int init() override {
 204     int num_threads;
 205     conf->get_val("num_threads", g_conf->rgw_thread_pool_size, &num_threads);
 206     RGWLoadGenProcess *pp = new RGWLoadGenProcess(g_ceph_context, &env,
 207                                                   num_threads, conf);
 208
 209     pprocess = pp;
 210
 211     string uid_str;
 212     conf->get_val("uid", "", &uid_str);
 213     if (uid_str.empty()) {
 214       derr << "ERROR: uid param must be specified for loadgen frontend"
 215            << dendl;
 216       return EINVAL;
 217     }
 218
 219     rgw_user uid(uid_str);
 220
 221     RGWUserInfo user_info;
 222     int ret = rgw_get_user_info_by_uid(env.store, uid, user_info, NULL);
 223     if (ret < 0) {
 224       derr << "ERROR: failed reading user info: uid=" << uid << " ret="
 225            << ret << dendl;
 226       return ret;
 227     }
 228
 229     map<string, RGWAccessKey>::iterator aiter = user_info.access_keys.begin();
 230     if (aiter == user_info.access_keys.end()) {
 231       derr << "ERROR: user has no S3 access keys set" << dendl;
 232       return -EINVAL;
 233     }
 234
 235     pp->set_access_key(aiter->second);
 236
 237     return 0;
 238   }
 239 }; /* RGWLoadGenFrontend */
 240
 241 // FrontendPauser implementation for RGWRealmReloader
 242 class RGWFrontendPauser : public RGWRealmReloader::Pauser {
 243   std::list<RGWFrontend*> &frontends;
 244   RGWRealmReloader::Pauser* pauser;
 245
 246  public:
 247   RGWFrontendPauser(std::list<RGWFrontend*> &frontends,
 248                     RGWRealmReloader::Pauser* pauser = nullptr)
 249     : frontends(frontends), pauser(pauser) {}
 250
 251   void pause() override {
 252     for (auto frontend : frontends)
 253       frontend->pause_for_new_config();
 254     if (pauser)
 255       pauser->pause();
 256   }
 257   void resume(RGWRados *store) override {
 258     /* Initialize the registry of auth strategies which will coordinate
 259      * the dynamic reconfiguration. */
 260     auto auth_registry = \
 261       rgw::auth::StrategyRegistry::create(g_ceph_context, store);
 262
 263     for (auto frontend : frontends)
 264       frontend->unpause_with_new_config(store, auth_registry);
 265     if (pauser)
 266       pauser->resume(store);
 267   }
 268 };
 269
 270 #endif /* RGW_FRONTEND_H */