ceph/src/mgr/MgrClient.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2016 John Spray <john.spray@redhat.com>
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  */
  13
  14
  15 #include "MgrClient.h"
  16
  17 #include "mgr/MgrContext.h"
  18
  19 #include "msg/Messenger.h"
  20 #include "messages/MMgrMap.h"
  21 #include "messages/MMgrReport.h"
  22 #include "messages/MMgrOpen.h"
  23 #include "messages/MMgrConfigure.h"
  24 #include "messages/MCommand.h"
  25 #include "messages/MCommandReply.h"
  26 #include "messages/MPGStats.h"
  27
  28 #define dout_subsys ceph_subsys_mgrc
  29 #undef dout_prefix
  30 #define dout_prefix *_dout << "mgrc " << __func__ << " "
  31
  32 MgrClient::MgrClient(CephContext *cct_, Messenger *msgr_)
  33     : Dispatcher(cct_), cct(cct_), msgr(msgr_),
  34       timer(cct_, lock)
  35 {
  36   assert(cct != nullptr);
  37 }
  38
  39 void MgrClient::init()
  40 {
  41   Mutex::Locker l(lock);
  42
  43   assert(msgr != nullptr);
  44
  45   timer.init();
  46 }
  47
  48 void MgrClient::shutdown()
  49 {
  50   Mutex::Locker l(lock);
  51
  52   if (connect_retry_callback) {
  53     timer.cancel_event(connect_retry_callback);
  54     connect_retry_callback = nullptr;
  55   }
  56
  57   // forget about in-flight commands if we are prematurely shut down
  58   // (e.g., by control-C)
  59   command_table.clear();
  60
  61   timer.shutdown();
  62   if (session) {
  63     session->con->mark_down();
  64     session.reset();
  65   }
  66 }
  67
  68 bool MgrClient::ms_dispatch(Message *m)
  69 {
  70   Mutex::Locker l(lock);
  71
  72   switch(m->get_type()) {
  73   case MSG_MGR_MAP:
  74     return handle_mgr_map(static_cast<MMgrMap*>(m));
  75   case MSG_MGR_CONFIGURE:
  76     return handle_mgr_configure(static_cast<MMgrConfigure*>(m));
  77   case MSG_COMMAND_REPLY:
  78     if (m->get_source().type() == CEPH_ENTITY_TYPE_MGR) {
  79       handle_command_reply(static_cast<MCommandReply*>(m));
  80       return true;
  81     } else {
  82       return false;
  83     }
  84   default:
  85     ldout(cct, 30) << "Not handling " << *m << dendl;
  86     return false;
  87   }
  88 }
  89
  90 void MgrClient::reconnect()
  91 {
  92   assert(lock.is_locked_by_me());
  93
  94   if (session) {
  95     ldout(cct, 4) << "Terminating session with "
  96                   << session->con->get_peer_addr() << dendl;
  97     session->con->mark_down();
  98     session.reset();
  99     stats_period = 0;
 100     if (report_callback != nullptr) {
 101       timer.cancel_event(report_callback);
 102       report_callback = nullptr;
 103     }
 104   }
 105
 106   if (!map.get_available()) {
 107     ldout(cct, 4) << "No active mgr available yet" << dendl;
 108     return;
 109   }
 110
 111   if (last_connect_attempt != utime_t()) {
 112     utime_t now = ceph_clock_now();
 113     utime_t when = last_connect_attempt;
 114     when += cct->_conf->mgr_connect_retry_interval;
 115     if (now < when) {
 116       if (!connect_retry_callback) {
 117         connect_retry_callback = new FunctionContext([this](int r){
 118             connect_retry_callback = nullptr;
 119             reconnect();
 120           });
 121         timer.add_event_at(when, connect_retry_callback);
 122       }
 123       ldout(cct, 4) << "waiting to retry connect until " << when << dendl;
 124       return;
 125     }
 126   }
 127
 128   if (connect_retry_callback) {
 129     timer.cancel_event(connect_retry_callback);
 130     connect_retry_callback = nullptr;
 131   }
 132
 133   ldout(cct, 4) << "Starting new session with " << map.get_active_addr()
 134                 << dendl;
 135   entity_inst_t inst;
 136   inst.addr = map.get_active_addr();
 137   inst.name = entity_name_t::MGR(map.get_active_gid());
 138   last_connect_attempt = ceph_clock_now();
 139
 140   session.reset(new MgrSessionState());
 141   session->con = msgr->get_connection(inst);
 142
 143   // Don't send an open if we're just a client (i.e. doing
 144   // command-sending, not stats etc)
 145   if (g_conf && !g_conf->name.is_client()) {
 146     auto open = new MMgrOpen();
 147     open->daemon_name = g_conf->name.get_id();
 148     session->con->send_message(open);
 149   }
 150
 151   // resend any pending commands
 152   for (const auto &p : command_table.get_commands()) {
 153     MCommand *m = p.second.get_message({});
 154     assert(session);
 155     assert(session->con);
 156     session->con->send_message(m);
 157   }
 158 }
 159
 160 bool MgrClient::handle_mgr_map(MMgrMap *m)
 161 {
 162   assert(lock.is_locked_by_me());
 163
 164   ldout(cct, 20) << *m << dendl;
 165
 166   map = m->get_map();
 167   ldout(cct, 4) << "Got map version " << map.epoch << dendl;
 168   m->put();
 169
 170   ldout(cct, 4) << "Active mgr is now " << map.get_active_addr() << dendl;
 171
 172   // Reset session?
 173   if (!session ||
 174       session->con->get_peer_addr() != map.get_active_addr()) {
 175     reconnect();
 176   }
 177
 178   return true;
 179 }
 180
 181 bool MgrClient::ms_handle_reset(Connection *con)
 182 {
 183   Mutex::Locker l(lock);
 184   if (session && con == session->con) {
 185     ldout(cct, 4) << __func__ << " con " << con << dendl;
 186     reconnect();
 187     return true;
 188   }
 189   return false;
 190 }
 191
 192 bool MgrClient::ms_handle_refused(Connection *con)
 193 {
 194   // do nothing for now
 195   return false;
 196 }
 197
 198 void MgrClient::send_report()
 199 {
 200   assert(lock.is_locked_by_me());
 201   assert(session);
 202   report_callback = nullptr;
 203
 204   auto report = new MMgrReport();
 205   auto pcc = cct->get_perfcounters_collection();
 206
 207   pcc->with_counters([this, report](
 208         const PerfCountersCollection::CounterMap &by_path)
 209   {
 210     ENCODE_START(1, 1, report->packed);
 211     for (auto p = session->declared.begin(); p != session->declared.end(); ) {
 212       if (by_path.count(*p) == 0) {
 213         report->undeclare_types.push_back(*p);
 214         ldout(cct,20) << __func__ << " undeclare " << *p << dendl;
 215         p = session->declared.erase(p);
 216       } else {
 217         ++p;
 218       }
 219     }
 220     for (const auto &i : by_path) {
 221       auto& path = i.first;
 222       auto& data = *(i.second);
 223
 224       if (session->declared.count(path) == 0) {
 225         ldout(cct,20) << __func__ << " declare " << path << dendl;
 226         PerfCounterType type;
 227         type.path = path;
 228         if (data.description) {
 229           type.description = data.description;
 230         }
 231         if (data.nick) {
 232           type.nick = data.nick;
 233         }
 234         type.type = data.type;
 235         report->declare_types.push_back(std::move(type));
 236         session->declared.insert(path);
 237       }
 238
 239       ::encode(static_cast<uint64_t>(data.u64), report->packed);
 240       if (data.type & PERFCOUNTER_LONGRUNAVG) {
 241         ::encode(static_cast<uint64_t>(data.avgcount), report->packed);
 242         ::encode(static_cast<uint64_t>(data.avgcount2), report->packed);
 243       }
 244     }
 245     ENCODE_FINISH(report->packed);
 246
 247     ldout(cct, 20) << by_path.size() << " counters, of which "
 248                    << report->declare_types.size() << " new" << dendl;
 249   });
 250
 251   ldout(cct, 20) << "encoded " << report->packed.length() << " bytes" << dendl;
 252
 253   report->daemon_name = g_conf->name.get_id();
 254
 255   session->con->send_message(report);
 256
 257   if (stats_period != 0) {
 258     report_callback = new FunctionContext([this](int r){send_report();});
 259     timer.add_event_after(stats_period, report_callback);
 260   }
 261
 262   send_pgstats();
 263 }
 264
 265 void MgrClient::send_pgstats()
 266 {
 267   if (pgstats_cb && session) {
 268     session->con->send_message(pgstats_cb());
 269   }
 270 }
 271
 272 bool MgrClient::handle_mgr_configure(MMgrConfigure *m)
 273 {
 274   assert(lock.is_locked_by_me());
 275
 276   ldout(cct, 20) << *m << dendl;
 277
 278   if (!session) {
 279     lderr(cct) << "dropping unexpected configure message" << dendl;
 280     m->put();
 281     return true;
 282   }
 283
 284   ldout(cct, 4) << "stats_period=" << m->stats_period << dendl;
 285
 286   bool starting = (stats_period == 0) && (m->stats_period != 0);
 287   stats_period = m->stats_period;
 288   if (starting) {
 289     send_report();
 290   }
 291
 292   m->put();
 293   return true;
 294 }
 295
 296 int MgrClient::start_command(const vector<string>& cmd, const bufferlist& inbl,
 297                   bufferlist *outbl, string *outs,
 298                   Context *onfinish)
 299 {
 300   Mutex::Locker l(lock);
 301
 302   ldout(cct, 20) << "cmd: " << cmd << dendl;
 303
 304   if (map.epoch == 0) {
 305     ldout(cct,20) << " no MgrMap, assuming EACCES" << dendl;
 306     return -EACCES;
 307   }
 308
 309   auto &op = command_table.start_command();
 310   op.cmd = cmd;
 311   op.inbl = inbl;
 312   op.outbl = outbl;
 313   op.outs = outs;
 314   op.on_finish = onfinish;
 315
 316   if (session && session->con) {
 317     // Leaving fsid argument null because it isn't used.
 318     MCommand *m = op.get_message({});
 319     session->con->send_message(m);
 320   }
 321   return 0;
 322 }
 323
 324 bool MgrClient::handle_command_reply(MCommandReply *m)
 325 {
 326   assert(lock.is_locked_by_me());
 327
 328   ldout(cct, 20) << *m << dendl;
 329
 330   const auto tid = m->get_tid();
 331   if (!command_table.exists(tid)) {
 332     ldout(cct, 4) << "handle_command_reply tid " << m->get_tid()
 333             << " not found" << dendl;
 334     m->put();
 335     return true;
 336   }
 337
 338   auto &op = command_table.get_command(tid);
 339   if (op.outbl) {
 340     op.outbl->claim(m->get_data());
 341   }
 342
 343   if (op.outs) {
 344     *(op.outs) = m->rs;
 345   }
 346
 347   if (op.on_finish) {
 348     op.on_finish->complete(m->r);
 349   }
 350
 351   command_table.erase(tid);
 352
 353   m->put();
 354   return true;
 355 }
 356