ceph/src/rgw/rgw_period_pusher.cc

   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 #include <map>
   5 #include <thread>
   6
   7 #include "rgw_period_pusher.h"
   8 #include "rgw_cr_rest.h"
   9 #include "rgw_zone.h"
  10 #include "rgw_sal.h"
  11 #include "rgw_sal_rados.h"
  12
  13 #include "services/svc_zone.h"
  14
  15 #include "common/errno.h"
  16
  17 #include <boost/asio/yield.hpp>
  18
  19 #define dout_subsys ceph_subsys_rgw
  20
  21 #undef dout_prefix
  22 #define dout_prefix (*_dout << "rgw period pusher: ")
  23
  24 /// A coroutine to post the period over the given connection.
  25 using PushCR = RGWPostRESTResourceCR<RGWPeriod, int>;
  26
  27 /// A coroutine that calls PushCR, and retries with backoff until success.
  28 class PushAndRetryCR : public RGWCoroutine {
  29   const std::string& zone;
  30   RGWRESTConn *const conn;
  31   RGWHTTPManager *const http;
  32   RGWPeriod& period;
  33   const std::string epoch; //< epoch string for params
  34   double timeout; //< current interval between retries
  35   const double timeout_max; //< maximum interval between retries
  36   uint32_t counter; //< number of failures since backoff increased
  37
  38  public:
  39   PushAndRetryCR(CephContext* cct, const std::string& zone, RGWRESTConn* conn,
  40                  RGWHTTPManager* http, RGWPeriod& period)
  41     : RGWCoroutine(cct), zone(zone), conn(conn), http(http), period(period),
  42       epoch(std::to_string(period.get_epoch())),
  43       timeout(cct->_conf->rgw_period_push_interval),
  44       timeout_max(cct->_conf->rgw_period_push_interval_max),
  45       counter(0)
  46   {}
  47
  48   int operate(const DoutPrefixProvider *dpp) override;
  49 };
  50
  51 int PushAndRetryCR::operate(const DoutPrefixProvider *dpp)
  52 {
  53   reenter(this) {
  54     for (;;) {
  55       yield {
  56         ldpp_dout(dpp, 10) << "pushing period " << period.get_id()
  57             << " to " << zone << dendl;
  58         // initialize the http params
  59         rgw_http_param_pair params[] = {
  60           { "period", period.get_id().c_str() },
  61           { "epoch", epoch.c_str() },
  62           { nullptr, nullptr }
  63         };
  64         call(new PushCR(cct, conn, http, "/admin/realm/period",
  65                         params, period, nullptr));
  66       }
  67
  68       // stop on success
  69       if (get_ret_status() == 0) {
  70         ldpp_dout(dpp, 10) << "push to " << zone << " succeeded" << dendl;
  71         return set_cr_done();
  72       }
  73
  74       // try each endpoint in the connection before waiting
  75       if (++counter < conn->get_endpoint_count())
  76         continue;
  77       counter = 0;
  78
  79       // wait with exponential backoff up to timeout_max
  80       yield {
  81         utime_t dur;
  82         dur.set_from_double(timeout);
  83
  84         ldpp_dout(dpp, 10) << "waiting " << dur << "s for retry.." << dendl;
  85         wait(dur);
  86
  87         timeout *= 2;
  88         if (timeout > timeout_max)
  89           timeout = timeout_max;
  90       }
  91     }
  92   }
  93   return 0;
  94 }
  95
  96 /**
  97  * PushAllCR is a coroutine that sends the period over all of the given
  98  * connections, retrying until they are all marked as completed.
  99  */
 100 class PushAllCR : public RGWCoroutine {
 101   RGWHTTPManager *const http;
 102   RGWPeriod period; //< period object to push
 103   std::map<std::string, RGWRESTConn> conns; //< zones that need the period
 104
 105  public:
 106   PushAllCR(CephContext* cct, RGWHTTPManager* http, RGWPeriod&& period,
 107             std::map<std::string, RGWRESTConn>&& conns)
 108     : RGWCoroutine(cct), http(http),
 109       period(std::move(period)),
 110       conns(std::move(conns))
 111   {}
 112
 113   int operate(const DoutPrefixProvider *dpp) override;
 114 };
 115
 116 int PushAllCR::operate(const DoutPrefixProvider *dpp)
 117 {
 118   reenter(this) {
 119     // spawn a coroutine to push the period over each connection
 120     yield {
 121       ldpp_dout(dpp, 4) << "sending " << conns.size() << " periods" << dendl;
 122       for (auto& c : conns)
 123         spawn(new PushAndRetryCR(cct, c.first, &c.second, http, period), false);
 124     }
 125     // wait for all to complete
 126     drain_all();
 127     return set_cr_done();
 128   }
 129   return 0;
 130 }
 131
 132 /// A background thread to run the PushAllCR coroutine and exit.
 133 class RGWPeriodPusher::CRThread : public DoutPrefixProvider {
 134   CephContext* cct;
 135   RGWCoroutinesManager coroutines;
 136   RGWHTTPManager http;
 137   boost::intrusive_ptr<PushAllCR> push_all;
 138   std::thread thread;
 139
 140  public:
 141   CRThread(CephContext* cct, RGWPeriod&& period,
 142            std::map<std::string, RGWRESTConn>&& conns)
 143     : cct(cct), coroutines(cct, NULL),
 144       http(cct, coroutines.get_completion_mgr()),
 145       push_all(new PushAllCR(cct, &http, std::move(period), std::move(conns)))
 146   {
 147     http.start();
 148     // must spawn the CR thread after start
 149     thread = std::thread([this]() noexcept { coroutines.run(this, push_all.get()); });
 150   }
 151   ~CRThread()
 152   {
 153     push_all.reset();
 154     coroutines.stop();
 155     http.stop();
 156     if (thread.joinable())
 157       thread.join();
 158   }
 159
 160   CephContext *get_cct() const override { return cct; }
 161   unsigned get_subsys() const override { return dout_subsys; }
 162   std::ostream& gen_prefix(std::ostream& out) const override { return out << "rgw period pusher CR thread: "; }
 163 };
 164
 165
 166 RGWPeriodPusher::RGWPeriodPusher(const DoutPrefixProvider *dpp, rgw::sal::Store* store,
 167                                  optional_yield y)
 168   : cct(store->ctx()), store(store)
 169 {
 170   const auto& realm = store->get_zone()->get_realm();
 171   auto& realm_id = realm.get_id();
 172   if (realm_id.empty()) // no realm configuration
 173     return;
 174
 175   // always send out the current period on startup
 176   RGWPeriod period;
 177   // XXX dang
 178   int r = period.init(dpp, cct, static_cast<rgw::sal::RadosStore* >(store)->svc()->sysobj, realm_id, y, realm.get_name());
 179   if (r < 0) {
 180     ldpp_dout(dpp, -1) << "failed to load period for realm " << realm_id << dendl;
 181     return;
 182   }
 183
 184   std::lock_guard<std::mutex> lock(mutex);
 185   handle_notify(std::move(period));
 186 }
 187
 188 // destructor is here because CRThread is incomplete in the header
 189 RGWPeriodPusher::~RGWPeriodPusher() = default;
 190
 191 void RGWPeriodPusher::handle_notify(RGWRealmNotify type,
 192                                     bufferlist::const_iterator& p)
 193 {
 194   // decode the period
 195   RGWZonesNeedPeriod info;
 196   try {
 197     decode(info, p);
 198   } catch (buffer::error& e) {
 199     lderr(cct) << "Failed to decode the period: " << e.what() << dendl;
 200     return;
 201   }
 202
 203   std::lock_guard<std::mutex> lock(mutex);
 204
 205   // we can't process this notification without access to our current realm
 206   // configuration. queue it until resume()
 207   if (store == nullptr) {
 208     pending_periods.emplace_back(std::move(info));
 209     return;
 210   }
 211
 212   handle_notify(std::move(info));
 213 }
 214
 215 // expects the caller to hold a lock on mutex
 216 void RGWPeriodPusher::handle_notify(RGWZonesNeedPeriod&& period)
 217 {
 218   if (period.get_realm_epoch() < realm_epoch) {
 219     ldout(cct, 10) << "period's realm epoch " << period.get_realm_epoch()
 220         << " is not newer than current realm epoch " << realm_epoch
 221         << ", discarding update" << dendl;
 222     return;
 223   }
 224   if (period.get_realm_epoch() == realm_epoch &&
 225       period.get_epoch() <= period_epoch) {
 226     ldout(cct, 10) << "period epoch " << period.get_epoch() << " is not newer "
 227         "than current epoch " << period_epoch << ", discarding update" << dendl;
 228     return;
 229   }
 230
 231   // find our zonegroup in the new period
 232   auto& zonegroups = period.get_map().zonegroups;
 233   auto i = zonegroups.find(store->get_zone()->get_zonegroup().get_id());
 234   if (i == zonegroups.end()) {
 235     lderr(cct) << "The new period does not contain my zonegroup!" << dendl;
 236     return;
 237   }
 238   auto& my_zonegroup = i->second;
 239
 240   // if we're not a master zone, we're not responsible for pushing any updates
 241   if (my_zonegroup.master_zone != store->get_zone()->get_params().get_id())
 242     return;
 243
 244   // construct a map of the zones that need this period. the map uses the same
 245   // keys/ordering as the zone[group] map, so we can use a hint for insertions
 246   std::map<std::string, RGWRESTConn> conns;
 247   auto hint = conns.end();
 248
 249   // are we the master zonegroup in this period?
 250   if (period.get_map().master_zonegroup == store->get_zone()->get_zonegroup().get_id()) {
 251     // update other zonegroup endpoints
 252     for (auto& zg : zonegroups) {
 253       auto& zonegroup = zg.second;
 254       if (zonegroup.get_id() == store->get_zone()->get_zonegroup().get_id())
 255         continue;
 256       if (zonegroup.endpoints.empty())
 257         continue;
 258
 259       hint = conns.emplace_hint(
 260           hint, std::piecewise_construct,
 261           std::forward_as_tuple(zonegroup.get_id()),
 262           std::forward_as_tuple(cct, store, zonegroup.get_id(), zonegroup.endpoints, zonegroup.api_name));
 263     }
 264   }
 265
 266   // update other zone endpoints
 267   for (auto& z : my_zonegroup.zones) {
 268     auto& zone = z.second;
 269     if (zone.id == store->get_zone()->get_params().get_id())
 270       continue;
 271     if (zone.endpoints.empty())
 272       continue;
 273
 274     hint = conns.emplace_hint(
 275         hint, std::piecewise_construct,
 276         std::forward_as_tuple(zone.id),
 277         std::forward_as_tuple(cct, store, zone.id, zone.endpoints, my_zonegroup.api_name));
 278   }
 279
 280   if (conns.empty()) {
 281     ldout(cct, 4) << "No zones to update" << dendl;
 282     return;
 283   }
 284
 285   realm_epoch = period.get_realm_epoch();
 286   period_epoch = period.get_epoch();
 287
 288   ldout(cct, 4) << "Zone master pushing period " << period.get_id()
 289       << " epoch " << period_epoch << " to "
 290       << conns.size() << " other zones" << dendl;
 291
 292   // spawn a new coroutine thread, destroying the previous one
 293   cr_thread.reset(new CRThread(cct, std::move(period), std::move(conns)));
 294 }
 295
 296 void RGWPeriodPusher::pause()
 297 {
 298   ldout(cct, 4) << "paused for realm update" << dendl;
 299   std::lock_guard<std::mutex> lock(mutex);
 300   store = nullptr;
 301 }
 302
 303 void RGWPeriodPusher::resume(rgw::sal::Store* store)
 304 {
 305   std::lock_guard<std::mutex> lock(mutex);
 306   this->store = store;
 307
 308   ldout(cct, 4) << "resume with " << pending_periods.size()
 309       << " periods pending" << dendl;
 310
 311   // process notification queue
 312   for (auto& info : pending_periods) {
 313     handle_notify(std::move(info));
 314   }
 315   pending_periods.clear();
 316 }