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 ft=cpp
   3
   4 #ifndef RGW_FRONTEND_H
   5 #define RGW_FRONTEND_H
   6
   7 #include <map>
   8 #include <string>
   9
  10 #include "common/RWLock.h"
  11
  12 #include "rgw_request.h"
  13 #include "rgw_process.h"
  14 #include "rgw_realm_reloader.h"
  15
  16 #include "rgw_auth_registry.h"
  17 #include "rgw_sal_rados.h"
  18
  19 #define dout_context g_ceph_context
  20
  21 namespace rgw::dmclock {
  22   class SyncScheduler;
  23   class ClientConfig;
  24   class SchedulerCtx;
  25 }
  26
  27 class RGWFrontendConfig {
  28   std::string config;
  29   std::multimap<std::string, std::string> config_map;
  30   std::string framework;
  31
  32   int parse_config(const std::string& config,
  33                    std::multimap<std::string, std::string>& config_map);
  34
  35 public:
  36   explicit RGWFrontendConfig(const std::string& config)
  37     : config(config) {
  38   }
  39
  40   int init() {
  41     const int ret = parse_config(config, config_map);
  42     return ret < 0 ? ret : 0;
  43   }
  44
  45   void set_default_config(RGWFrontendConfig& def_conf);
  46
  47   std::optional<std::string> get_val(const std::string& key);
  48
  49   bool get_val(const std::string& key,
  50                const std::string& def_val,
  51                std::string* out);
  52   bool get_val(const std::string& key, int def_val, int *out);
  53
  54   std::string get_val(const std::string& key,
  55                       const std::string& def_val) {
  56     std::string out;
  57     get_val(key, def_val, &out);
  58     return out;
  59   }
  60
  61   const std::string& get_config() {
  62     return config;
  63   }
  64
  65   std::multimap<std::string, std::string>& get_config_map() {
  66     return config_map;
  67   }
  68
  69   std::string get_framework() const {
  70     return framework;
  71  }
  72 };
  73
  74 class RGWFrontend {
  75 public:
  76   virtual ~RGWFrontend() {}
  77
  78   virtual int init() = 0;
  79
  80   virtual int run() = 0;
  81   virtual void stop() = 0;
  82   virtual void join() = 0;
  83
  84   virtual void pause_for_new_config() = 0;
  85   virtual void unpause_with_new_config(rgw::sal::Store* store,
  86                                        rgw_auth_registry_ptr_t auth_registry) = 0;
  87 };
  88
  89
  90 class RGWProcessFrontend : public RGWFrontend {
  91 protected:
  92   RGWFrontendConfig* conf;
  93   RGWProcess* pprocess;
  94   RGWProcessEnv env;
  95   RGWProcessControlThread* thread;
  96
  97 public:
  98   RGWProcessFrontend(RGWProcessEnv& pe, RGWFrontendConfig* _conf)
  99     : conf(_conf), pprocess(nullptr), env(pe), thread(nullptr) {
 100   }
 101
 102   ~RGWProcessFrontend() override {
 103     delete thread;
 104     delete pprocess;
 105   }
 106
 107   int run() override {
 108     ceph_assert(pprocess); /* should have initialized by init() */
 109     thread = new RGWProcessControlThread(pprocess);
 110     thread->create("rgw_frontend");
 111     return 0;
 112   }
 113
 114   void stop() override;
 115
 116   void join() override {
 117     thread->join();
 118   }
 119
 120   void pause_for_new_config() override {
 121     pprocess->pause();
 122   }
 123
 124   void unpause_with_new_config(rgw::sal::Store* const store,
 125                                rgw_auth_registry_ptr_t auth_registry) override {
 126     env.store = store;
 127     env.auth_registry = auth_registry;
 128     pprocess->unpause_with_new_config(store, std::move(auth_registry));
 129   }
 130 }; /* RGWProcessFrontend */
 131
 132 class RGWLoadGenFrontend : public RGWProcessFrontend, public DoutPrefixProvider {
 133 public:
 134   RGWLoadGenFrontend(RGWProcessEnv& pe, RGWFrontendConfig *_conf)
 135     : RGWProcessFrontend(pe, _conf) {}
 136
 137   CephContext *get_cct() const {
 138     return env.store->ctx();
 139   }
 140
 141   unsigned get_subsys() const
 142   {
 143     return ceph_subsys_rgw;
 144   }
 145
 146   std::ostream& gen_prefix(std::ostream& out) const
 147   {
 148     return out << "rgw loadgen frontend: ";
 149   }
 150
 151   int init() override {
 152     int num_threads;
 153     conf->get_val("num_threads", g_conf()->rgw_thread_pool_size, &num_threads);
 154     RGWLoadGenProcess *pp = new RGWLoadGenProcess(g_ceph_context, &env,
 155                                                   num_threads, conf);
 156
 157     pprocess = pp;
 158
 159     std::string uid_str;
 160     conf->get_val("uid", "", &uid_str);
 161     if (uid_str.empty()) {
 162       derr << "ERROR: uid param must be specified for loadgen frontend"
 163            << dendl;
 164       return -EINVAL;
 165     }
 166
 167     rgw_user uid(uid_str);
 168     std::unique_ptr<rgw::sal::User> user = env.store->get_user(uid);
 169
 170     int ret = user->load_user(this, null_yield);
 171     if (ret < 0) {
 172       derr << "ERROR: failed reading user info: uid=" << uid << " ret="
 173            << ret << dendl;
 174       return ret;
 175     }
 176
 177     auto aiter = user->get_info().access_keys.begin();
 178     if (aiter == user->get_info().access_keys.end()) {
 179       derr << "ERROR: user has no S3 access keys set" << dendl;
 180       return -EINVAL;
 181     }
 182
 183     pp->set_access_key(aiter->second);
 184
 185     return 0;
 186   }
 187 }; /* RGWLoadGenFrontend */
 188
 189 // FrontendPauser implementation for RGWRealmReloader
 190 class RGWFrontendPauser : public RGWRealmReloader::Pauser {
 191   std::list<RGWFrontend*> &frontends;
 192   RGWRealmReloader::Pauser* pauser;
 193   rgw::auth::ImplicitTenants& implicit_tenants;
 194
 195  public:
 196   RGWFrontendPauser(std::list<RGWFrontend*> &frontends,
 197                     rgw::auth::ImplicitTenants& implicit_tenants,
 198                     RGWRealmReloader::Pauser* pauser = nullptr)
 199     : frontends(frontends),
 200       pauser(pauser),
 201       implicit_tenants(implicit_tenants) {
 202   }
 203
 204   void pause() override {
 205     for (auto frontend : frontends)
 206       frontend->pause_for_new_config();
 207     if (pauser)
 208       pauser->pause();
 209   }
 210   void resume(rgw::sal::Store* store) override {
 211     /* Initialize the registry of auth strategies which will coordinate
 212      * the dynamic reconfiguration. */
 213     auto auth_registry = \
 214       rgw::auth::StrategyRegistry::create(g_ceph_context, implicit_tenants, store);
 215
 216     for (auto frontend : frontends)
 217       frontend->unpause_with_new_config(store, auth_registry);
 218     if (pauser)
 219       pauser->resume(store);
 220   }
 221 };
 222
 223 #endif /* RGW_FRONTEND_H */