ceph/src/mds/SessionMap.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) 2004-2006 Sage Weil <sage@newdream.net>
   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 "MDSRank.h"
  16 #include "MDCache.h"
  17 #include "Mutation.h"
  18 #include "SessionMap.h"
  19 #include "osdc/Filer.h"
  20 #include "common/Finisher.h"
  21
  22 #include "common/config.h"
  23 #include "common/errno.h"
  24 #include "common/DecayCounter.h"
  25 #include "include/ceph_assert.h"
  26 #include "include/stringify.h"
  27
  28 #define dout_context g_ceph_context
  29 #define dout_subsys ceph_subsys_mds
  30 #undef dout_prefix
  31 #define dout_prefix *_dout << "mds." << rank << ".sessionmap "
  32
  33 namespace {
  34 class SessionMapIOContext : public MDSIOContextBase
  35 {
  36   protected:
  37     SessionMap *sessionmap;
  38     MDSRank *get_mds() override {return sessionmap->mds;}
  39   public:
  40     explicit SessionMapIOContext(SessionMap *sessionmap_) : sessionmap(sessionmap_) {
  41       ceph_assert(sessionmap != NULL);
  42     }
  43 };
  44 };
  45
  46 void SessionMap::register_perfcounters()
  47 {
  48   PerfCountersBuilder plb(g_ceph_context, "mds_sessions",
  49       l_mdssm_first, l_mdssm_last);
  50
  51   plb.add_u64(l_mdssm_session_count, "session_count",
  52       "Session count", "sess", PerfCountersBuilder::PRIO_INTERESTING);
  53
  54   plb.set_prio_default(PerfCountersBuilder::PRIO_USEFUL);
  55   plb.add_u64_counter(l_mdssm_session_add, "session_add",
  56       "Sessions added");
  57   plb.add_u64_counter(l_mdssm_session_remove, "session_remove",
  58       "Sessions removed");
  59   plb.add_u64(l_mdssm_session_open, "sessions_open",
  60               "Sessions currently open");
  61   plb.add_u64(l_mdssm_session_stale, "sessions_stale",
  62               "Sessions currently stale");
  63   plb.add_u64(l_mdssm_total_load, "total_load", "Total Load");
  64   plb.add_u64(l_mdssm_avg_load, "average_load", "Average Load");
  65   plb.add_u64(l_mdssm_avg_session_uptime, "avg_session_uptime",
  66                "Average session uptime");
  67
  68   logger = plb.create_perf_counters();
  69   g_ceph_context->get_perfcounters_collection()->add(logger);
  70 }
  71
  72 void SessionMap::dump()
  73 {
  74   dout(10) << "dump" << dendl;
  75   for (ceph::unordered_map<entity_name_t,Session*>::iterator p = session_map.begin();
  76        p != session_map.end();
  77        ++p)
  78     dout(10) << p->first << " " << p->second
  79              << " state " << p->second->get_state_name()
  80              << " completed " << p->second->info.completed_requests
  81              << " prealloc_inos " << p->second->info.prealloc_inos
  82              << " used_inos " << p->second->info.used_inos
  83              << dendl;
  84 }
  85
  86
  87 // ----------------
  88 // LOAD
  89
  90
  91 object_t SessionMap::get_object_name() const
  92 {
  93   char s[30];
  94   snprintf(s, sizeof(s), "mds%d_sessionmap", int(mds->get_nodeid()));
  95   return object_t(s);
  96 }
  97
  98 namespace {
  99 class C_IO_SM_Load : public SessionMapIOContext {
 100 public:
 101   const bool first;  //< Am I the initial (header) load?
 102   int header_r;  //< Return value from OMAP header read
 103   int values_r;  //< Return value from OMAP value read
 104   bufferlist header_bl;
 105   std::map<std::string, bufferlist> session_vals;
 106   bool more_session_vals = false;
 107
 108   C_IO_SM_Load(SessionMap *cm, const bool f)
 109     : SessionMapIOContext(cm), first(f), header_r(0), values_r(0) {}
 110
 111   void finish(int r) override {
 112     sessionmap->_load_finish(r, header_r, values_r, first, header_bl, session_vals,
 113       more_session_vals);
 114   }
 115   void print(ostream& out) const override {
 116     out << "session_load";
 117   }
 118 };
 119 }
 120
 121
 122 /**
 123  * Decode OMAP header.  Call this once when loading.
 124  */
 125 void SessionMapStore::decode_header(
 126       bufferlist &header_bl)
 127 {
 128   auto q = header_bl.cbegin();
 129   DECODE_START(1, q)
 130   decode(version, q);
 131   DECODE_FINISH(q);
 132 }
 133
 134 void SessionMapStore::encode_header(
 135     bufferlist *header_bl)
 136 {
 137   ENCODE_START(1, 1, *header_bl);
 138   encode(version, *header_bl);
 139   ENCODE_FINISH(*header_bl);
 140 }
 141
 142 /**
 143  * Decode and insert some serialized OMAP values.  Call this
 144  * repeatedly to insert batched loads.
 145  */
 146 void SessionMapStore::decode_values(std::map<std::string, bufferlist> &session_vals)
 147 {
 148   for (std::map<std::string, bufferlist>::iterator i = session_vals.begin();
 149        i != session_vals.end(); ++i) {
 150
 151     entity_inst_t inst;
 152
 153     bool parsed = inst.name.parse(i->first);
 154     if (!parsed) {
 155       derr << "Corrupt entity name '" << i->first << "' in sessionmap" << dendl;
 156       throw buffer::malformed_input("Corrupt entity name in sessionmap");
 157     }
 158
 159     Session *s = get_or_add_session(inst);
 160     if (s->is_closed()) {
 161       s->set_state(Session::STATE_OPEN);
 162       s->set_load_avg_decay_rate(decay_rate);
 163     }
 164     auto q = i->second.cbegin();
 165     s->decode(q);
 166   }
 167 }
 168
 169 /**
 170  * An OMAP read finished.
 171  */
 172 void SessionMap::_load_finish(
 173     int operation_r,
 174     int header_r,
 175     int values_r,
 176     bool first,
 177     bufferlist &header_bl,
 178     std::map<std::string, bufferlist> &session_vals,
 179     bool more_session_vals)
 180 {
 181   if (operation_r < 0) {
 182     derr << "_load_finish got " << cpp_strerror(operation_r) << dendl;
 183     mds->clog->error() << "error reading sessionmap '" << get_object_name()
 184                        << "' " << operation_r << " ("
 185                        << cpp_strerror(operation_r) << ")";
 186     mds->damaged();
 187     ceph_abort();  // Should be unreachable because damaged() calls respawn()
 188   }
 189
 190   // Decode header
 191   if (first) {
 192     if (header_r != 0) {
 193       derr << __func__ << ": header error: " << cpp_strerror(header_r) << dendl;
 194       mds->clog->error() << "error reading sessionmap header "
 195                          << header_r << " (" << cpp_strerror(header_r) << ")";
 196       mds->damaged();
 197       ceph_abort();  // Should be unreachable because damaged() calls respawn()
 198     }
 199
 200     if(header_bl.length() == 0) {
 201       dout(4) << __func__ << ": header missing, loading legacy..." << dendl;
 202       load_legacy();
 203       return;
 204     }
 205
 206     try {
 207       decode_header(header_bl);
 208     } catch (buffer::error &e) {
 209       mds->clog->error() << "corrupt sessionmap header: " << e.what();
 210       mds->damaged();
 211       ceph_abort();  // Should be unreachable because damaged() calls respawn()
 212     }
 213     dout(10) << __func__ << " loaded version " << version << dendl;
 214   }
 215
 216   if (values_r != 0) {
 217     derr << __func__ << ": error reading values: "
 218       << cpp_strerror(values_r) << dendl;
 219     mds->clog->error() << "error reading sessionmap values: "
 220                        << values_r << " (" << cpp_strerror(values_r) << ")";
 221     mds->damaged();
 222     ceph_abort();  // Should be unreachable because damaged() calls respawn()
 223   }
 224
 225   // Decode session_vals
 226   try {
 227     decode_values(session_vals);
 228   } catch (buffer::error &e) {
 229     mds->clog->error() << "corrupt sessionmap values: " << e.what();
 230     mds->damaged();
 231     ceph_abort();  // Should be unreachable because damaged() calls respawn()
 232   }
 233
 234   if (more_session_vals) {
 235     // Issue another read if we're not at the end of the omap
 236     const std::string last_key = session_vals.rbegin()->first;
 237     dout(10) << __func__ << ": continue omap load from '"
 238              << last_key << "'" << dendl;
 239     object_t oid = get_object_name();
 240     object_locator_t oloc(mds->mdsmap->get_metadata_pool());
 241     C_IO_SM_Load *c = new C_IO_SM_Load(this, false);
 242     ObjectOperation op;
 243     op.omap_get_vals(last_key, "", g_conf()->mds_sessionmap_keys_per_op,
 244                      &c->session_vals, &c->more_session_vals, &c->values_r);
 245     mds->objecter->read(oid, oloc, op, CEPH_NOSNAP, NULL, 0,
 246         new C_OnFinisher(c, mds->finisher));
 247   } else {
 248     // I/O is complete.  Update `by_state`
 249     dout(10) << __func__ << ": omap load complete" << dendl;
 250     for (ceph::unordered_map<entity_name_t, Session*>::iterator i = session_map.begin();
 251          i != session_map.end(); ++i) {
 252       Session *s = i->second;
 253       auto by_state_entry = by_state.find(s->get_state());
 254       if (by_state_entry == by_state.end())
 255         by_state_entry = by_state.emplace(s->get_state(),
 256                                           new xlist<Session*>).first;
 257       by_state_entry->second->push_back(&s->item_session_list);
 258     }
 259
 260     // Population is complete.  Trigger load waiters.
 261     dout(10) << __func__ << ": v " << version
 262            << ", " << session_map.size() << " sessions" << dendl;
 263     projected = committing = committed = version;
 264     dump();
 265     finish_contexts(g_ceph_context, waiting_for_load);
 266   }
 267 }
 268
 269 /**
 270  * Populate session state from OMAP records in this
 271  * rank's sessionmap object.
 272  */
 273 void SessionMap::load(MDSContext *onload)
 274 {
 275   dout(10) << "load" << dendl;
 276
 277   if (onload)
 278     waiting_for_load.push_back(onload);
 279
 280   C_IO_SM_Load *c = new C_IO_SM_Load(this, true);
 281   object_t oid = get_object_name();
 282   object_locator_t oloc(mds->mdsmap->get_metadata_pool());
 283
 284   ObjectOperation op;
 285   op.omap_get_header(&c->header_bl, &c->header_r);
 286   op.omap_get_vals("", "", g_conf()->mds_sessionmap_keys_per_op,
 287                    &c->session_vals, &c->more_session_vals, &c->values_r);
 288
 289   mds->objecter->read(oid, oloc, op, CEPH_NOSNAP, NULL, 0, new C_OnFinisher(c, mds->finisher));
 290 }
 291
 292 namespace {
 293 class C_IO_SM_LoadLegacy : public SessionMapIOContext {
 294 public:
 295   bufferlist bl;
 296   explicit C_IO_SM_LoadLegacy(SessionMap *cm) : SessionMapIOContext(cm) {}
 297   void finish(int r) override {
 298     sessionmap->_load_legacy_finish(r, bl);
 299   }
 300   void print(ostream& out) const override {
 301     out << "session_load_legacy";
 302   }
 303 };
 304 }
 305
 306
 307 /**
 308  * Load legacy (object data blob) SessionMap format, assuming
 309  * that waiting_for_load has already been populated with
 310  * the relevant completion.  This is the fallback if we do not
 311  * find an OMAP header when attempting to load normally.
 312  */
 313 void SessionMap::load_legacy()
 314 {
 315   dout(10) << __func__ << dendl;
 316
 317   C_IO_SM_LoadLegacy *c = new C_IO_SM_LoadLegacy(this);
 318   object_t oid = get_object_name();
 319   object_locator_t oloc(mds->mdsmap->get_metadata_pool());
 320
 321   mds->objecter->read_full(oid, oloc, CEPH_NOSNAP, &c->bl, 0,
 322                            new C_OnFinisher(c, mds->finisher));
 323 }
 324
 325 void SessionMap::_load_legacy_finish(int r, bufferlist &bl)
 326 {
 327   auto blp = bl.cbegin();
 328   if (r < 0) {
 329     derr << "_load_finish got " << cpp_strerror(r) << dendl;
 330     ceph_abort_msg("failed to load sessionmap");
 331   }
 332   dump();
 333   decode_legacy(blp);  // note: this sets last_cap_renew = now()
 334   dout(10) << "_load_finish v " << version
 335            << ", " << session_map.size() << " sessions, "
 336            << bl.length() << " bytes"
 337            << dendl;
 338   projected = committing = committed = version;
 339   dump();
 340
 341   // Mark all sessions dirty, so that on next save() we will write
 342   // a complete OMAP version of the data loaded from the legacy format
 343   for (ceph::unordered_map<entity_name_t, Session*>::iterator i = session_map.begin();
 344        i != session_map.end(); ++i) {
 345     // Don't use mark_dirty because on this occasion we want to ignore the
 346     // keys_per_op limit and do one big write (upgrade must be atomic)
 347     dirty_sessions.insert(i->first);
 348   }
 349   loaded_legacy = true;
 350
 351   finish_contexts(g_ceph_context, waiting_for_load);
 352 }
 353
 354
 355 // ----------------
 356 // SAVE
 357
 358 namespace {
 359 class C_IO_SM_Save : public SessionMapIOContext {
 360   version_t version;
 361 public:
 362   C_IO_SM_Save(SessionMap *cm, version_t v) : SessionMapIOContext(cm), version(v) {}
 363   void finish(int r) override {
 364     if (r != 0) {
 365       get_mds()->handle_write_error(r);
 366     } else {
 367       sessionmap->_save_finish(version);
 368     }
 369   }
 370   void print(ostream& out) const override {
 371     out << "session_save";
 372   }
 373 };
 374 }
 375
 376 void SessionMap::save(MDSContext *onsave, version_t needv)
 377 {
 378   dout(10) << __func__ << ": needv " << needv << ", v " << version << dendl;
 379
 380   if (needv && committing >= needv) {
 381     ceph_assert(committing > committed);
 382     commit_waiters[committing].push_back(onsave);
 383     return;
 384   }
 385
 386   commit_waiters[version].push_back(onsave);
 387
 388   committing = version;
 389   SnapContext snapc;
 390   object_t oid = get_object_name();
 391   object_locator_t oloc(mds->mdsmap->get_metadata_pool());
 392
 393   ObjectOperation op;
 394
 395   /* Compose OSD OMAP transaction for full write */
 396   bufferlist header_bl;
 397   encode_header(&header_bl);
 398   op.omap_set_header(header_bl);
 399
 400   /* If we loaded a legacy sessionmap, then erase the old data.  If
 401    * an old-versioned MDS tries to read it, it'll fail out safely
 402    * with an end_of_buffer exception */
 403   if (loaded_legacy) {
 404     dout(4) << __func__ << " erasing legacy sessionmap" << dendl;
 405     op.truncate(0);
 406     loaded_legacy = false;  // only need to truncate once.
 407   }
 408
 409   dout(20) << " updating keys:" << dendl;
 410   map<string, bufferlist> to_set;
 411   for(std::set<entity_name_t>::iterator i = dirty_sessions.begin();
 412       i != dirty_sessions.end(); ++i) {
 413     const entity_name_t name = *i;
 414     Session *session = session_map[name];
 415
 416     if (session->is_open() ||
 417         session->is_closing() ||
 418         session->is_stale() ||
 419         session->is_killing()) {
 420       dout(20) << "  " << name << dendl;
 421       // Serialize K
 422       std::ostringstream k;
 423       k << name;
 424
 425       // Serialize V
 426       bufferlist bl;
 427       session->info.encode(bl, mds->mdsmap->get_up_features());
 428
 429       // Add to RADOS op
 430       to_set[k.str()] = bl;
 431
 432       session->clear_dirty_completed_requests();
 433     } else {
 434       dout(20) << "  " << name << " (ignoring)" << dendl;
 435     }
 436   }
 437   if (!to_set.empty()) {
 438     op.omap_set(to_set);
 439   }
 440
 441   dout(20) << " removing keys:" << dendl;
 442   set<string> to_remove;
 443   for(std::set<entity_name_t>::const_iterator i = null_sessions.begin();
 444       i != null_sessions.end(); ++i) {
 445     dout(20) << "  " << *i << dendl;
 446     std::ostringstream k;
 447     k << *i;
 448     to_remove.insert(k.str());
 449   }
 450   if (!to_remove.empty()) {
 451     op.omap_rm_keys(to_remove);
 452   }
 453
 454   dirty_sessions.clear();
 455   null_sessions.clear();
 456
 457   mds->objecter->mutate(oid, oloc, op, snapc,
 458                         ceph::real_clock::now(),
 459                         0,
 460                         new C_OnFinisher(new C_IO_SM_Save(this, version),
 461                                          mds->finisher));
 462 }
 463
 464 void SessionMap::_save_finish(version_t v)
 465 {
 466   dout(10) << "_save_finish v" << v << dendl;
 467   committed = v;
 468
 469   finish_contexts(g_ceph_context, commit_waiters[v]);
 470   commit_waiters.erase(v);
 471 }
 472
 473
 474 /**
 475  * Deserialize sessions, and update by_state index
 476  */
 477 void SessionMap::decode_legacy(bufferlist::const_iterator &p)
 478 {
 479   // Populate `sessions`
 480   SessionMapStore::decode_legacy(p);
 481
 482   // Update `by_state`
 483   for (ceph::unordered_map<entity_name_t, Session*>::iterator i = session_map.begin();
 484        i != session_map.end(); ++i) {
 485     Session *s = i->second;
 486     auto by_state_entry = by_state.find(s->get_state());
 487     if (by_state_entry == by_state.end())
 488       by_state_entry = by_state.emplace(s->get_state(),
 489                                         new xlist<Session*>).first;
 490     by_state_entry->second->push_back(&s->item_session_list);
 491   }
 492 }
 493
 494 uint64_t SessionMap::set_state(Session *session, int s) {
 495   if (session->state != s) {
 496     session->set_state(s);
 497     auto by_state_entry = by_state.find(s);
 498     if (by_state_entry == by_state.end())
 499       by_state_entry = by_state.emplace(s, new xlist<Session*>).first;
 500     by_state_entry->second->push_back(&session->item_session_list);
 501
 502     if (session->is_open() || session->is_stale()) {
 503       session->set_load_avg_decay_rate(decay_rate);
 504     }
 505
 506     // refresh number of sessions for states which have perf
 507     // couters associated
 508     logger->set(l_mdssm_session_open,
 509                 get_session_count_in_state(Session::STATE_OPEN));
 510     logger->set(l_mdssm_session_stale,
 511                 get_session_count_in_state(Session::STATE_STALE));
 512   }
 513
 514   return session->get_state_seq();
 515 }
 516
 517 void SessionMapStore::decode_legacy(bufferlist::const_iterator& p)
 518 {
 519   auto now = clock::now();
 520   uint64_t pre;
 521   decode(pre, p);
 522   if (pre == (uint64_t)-1) {
 523     DECODE_START_LEGACY_COMPAT_LEN(3, 3, 3, p);
 524     ceph_assert(struct_v >= 2);
 525
 526     decode(version, p);
 527
 528     while (!p.end()) {
 529       entity_inst_t inst;
 530       decode(inst.name, p);
 531       Session *s = get_or_add_session(inst);
 532       if (s->is_closed()) {
 533         s->set_state(Session::STATE_OPEN);
 534         s->set_load_avg_decay_rate(decay_rate);
 535       }
 536       s->decode(p);
 537     }
 538
 539     DECODE_FINISH(p);
 540   } else {
 541     // --- old format ----
 542     version = pre;
 543
 544     // this is a meaningless upper bound.  can be ignored.
 545     __u32 n;
 546     decode(n, p);
 547
 548     while (n-- && !p.end()) {
 549       auto p2 = p;
 550       Session *s = new Session(ConnectionRef());
 551       s->info.decode(p);
 552       {
 553         auto& name = s->info.inst.name;
 554         auto it = session_map.find(name);
 555         if (it != session_map.end()) {
 556           // eager client connected too fast!  aie.
 557           dout(10) << " already had session for " << name << ", recovering" << dendl;
 558           delete s;
 559           s = it->second;
 560           p = p2;
 561           s->info.decode(p);
 562         } else {
 563           it->second = s;
 564         }
 565       }
 566       s->set_state(Session::STATE_OPEN);
 567       s->set_load_avg_decay_rate(decay_rate);
 568       s->last_cap_renew = now;
 569     }
 570   }
 571 }
 572
 573 void Session::dump(Formatter *f) const
 574 {
 575   f->dump_int("id", info.inst.name.num());
 576   f->dump_object("entity", info.inst);
 577   f->dump_string("state", get_state_name());
 578   f->dump_int("num_leases", leases.size());
 579   f->dump_int("num_caps", caps.size());
 580   if (is_open() || is_stale()) {
 581     f->dump_unsigned("request_load_avg", get_load_avg());
 582   }
 583   f->dump_float("uptime", get_session_uptime());
 584   f->dump_unsigned("requests_in_flight", get_request_count());
 585   f->dump_unsigned("completed_requests", get_num_completed_requests());
 586   f->dump_bool("reconnecting", reconnecting);
 587   f->dump_object("recall_caps", recall_caps);
 588   f->dump_object("release_caps", release_caps);
 589   f->dump_object("recall_caps_throttle", recall_caps_throttle);
 590   f->dump_object("recall_caps_throttle2o", recall_caps_throttle2o);
 591   f->dump_object("session_cache_liveness", session_cache_liveness);
 592   info.dump(f);
 593 }
 594
 595 void SessionMapStore::dump(Formatter *f) const
 596 {
 597   f->open_array_section("sessions");
 598   for (const auto& p : session_map) {
 599     f->dump_object("session", *p.second);
 600   }
 601   f->close_section(); // Sessions
 602 }
 603
 604 void SessionMapStore::generate_test_instances(list<SessionMapStore*>& ls)
 605 {
 606   // pretty boring for now
 607   ls.push_back(new SessionMapStore());
 608 }
 609
 610 void SessionMap::wipe()
 611 {
 612   dout(1) << "wipe start" << dendl;
 613   dump();
 614   while (!session_map.empty()) {
 615     Session *s = session_map.begin()->second;
 616     remove_session(s);
 617   }
 618   version = ++projected;
 619   dout(1) << "wipe result" << dendl;
 620   dump();
 621   dout(1) << "wipe done" << dendl;
 622 }
 623
 624 void SessionMap::wipe_ino_prealloc()
 625 {
 626   for (ceph::unordered_map<entity_name_t,Session*>::iterator p = session_map.begin();
 627        p != session_map.end();
 628        ++p) {
 629     p->second->pending_prealloc_inos.clear();
 630     p->second->info.prealloc_inos.clear();
 631     p->second->info.used_inos.clear();
 632   }
 633   projected = ++version;
 634 }
 635
 636 void SessionMap::add_session(Session *s)
 637 {
 638   dout(10) << __func__ << " s=" << s << " name=" << s->info.inst.name << dendl;
 639
 640   ceph_assert(session_map.count(s->info.inst.name) == 0);
 641   session_map[s->info.inst.name] = s;
 642   auto by_state_entry = by_state.find(s->state);
 643   if (by_state_entry == by_state.end())
 644     by_state_entry = by_state.emplace(s->state, new xlist<Session*>).first;
 645   by_state_entry->second->push_back(&s->item_session_list);
 646   s->get();
 647
 648   update_average_birth_time(*s);
 649
 650   logger->set(l_mdssm_session_count, session_map.size());
 651   logger->inc(l_mdssm_session_add);
 652 }
 653
 654 void SessionMap::remove_session(Session *s)
 655 {
 656   dout(10) << __func__ << " s=" << s << " name=" << s->info.inst.name << dendl;
 657
 658   update_average_birth_time(*s, false);
 659
 660   s->trim_completed_requests(0);
 661   s->item_session_list.remove_myself();
 662   session_map.erase(s->info.inst.name);
 663   dirty_sessions.erase(s->info.inst.name);
 664   null_sessions.insert(s->info.inst.name);
 665   s->put();
 666
 667   logger->set(l_mdssm_session_count, session_map.size());
 668   logger->inc(l_mdssm_session_remove);
 669 }
 670
 671 void SessionMap::touch_session(Session *session)
 672 {
 673   dout(10) << __func__ << " s=" << session << " name=" << session->info.inst.name << dendl;
 674
 675   // Move to the back of the session list for this state (should
 676   // already be on a list courtesy of add_session and set_state)
 677   ceph_assert(session->item_session_list.is_on_list());
 678   auto by_state_entry = by_state.find(session->state);
 679   if (by_state_entry == by_state.end())
 680     by_state_entry = by_state.emplace(session->state,
 681                                       new xlist<Session*>).first;
 682   by_state_entry->second->push_back(&session->item_session_list);
 683
 684   session->last_cap_renew = clock::now();
 685 }
 686
 687 void SessionMap::_mark_dirty(Session *s, bool may_save)
 688 {
 689   if (dirty_sessions.count(s->info.inst.name))
 690     return;
 691
 692   if (may_save &&
 693       dirty_sessions.size() >= g_conf()->mds_sessionmap_keys_per_op) {
 694     // Pre-empt the usual save() call from journal segment trim, in
 695     // order to avoid building up an oversized OMAP update operation
 696     // from too many sessions modified at once
 697     save(new C_MDSInternalNoop, version);
 698   }
 699
 700   null_sessions.erase(s->info.inst.name);
 701   dirty_sessions.insert(s->info.inst.name);
 702 }
 703
 704 void SessionMap::mark_dirty(Session *s, bool may_save)
 705 {
 706   dout(20) << __func__ << " s=" << s << " name=" << s->info.inst.name
 707     << " v=" << version << dendl;
 708
 709   _mark_dirty(s, may_save);
 710   version++;
 711   s->pop_pv(version);
 712 }
 713
 714 void SessionMap::replay_dirty_session(Session *s)
 715 {
 716   dout(20) << __func__ << " s=" << s << " name=" << s->info.inst.name
 717     << " v=" << version << dendl;
 718
 719   _mark_dirty(s, false);
 720
 721   replay_advance_version();
 722 }
 723
 724 void SessionMap::replay_advance_version()
 725 {
 726   version++;
 727   projected = version;
 728 }
 729
 730 void SessionMap::replay_open_sessions(version_t event_cmapv,
 731                             map<client_t,entity_inst_t>& client_map,
 732                             map<client_t,client_metadata_t>& client_metadata_map)
 733 {
 734   unsigned already_saved;
 735
 736   if (version + client_map.size() < event_cmapv)
 737     goto bad;
 738
 739   // Server::finish_force_open_sessions() marks sessions dirty one by one.
 740   // Marking a session dirty may flush all existing dirty sessions. So it's
 741   // possible that some sessions are already saved in sessionmap.
 742   already_saved = client_map.size() - (event_cmapv - version);
 743   for (const auto& p : client_map) {
 744     Session *s = get_or_add_session(p.second);
 745     auto q = client_metadata_map.find(p.first);
 746     if (q != client_metadata_map.end())
 747       s->info.client_metadata.merge(q->second);
 748
 749     if (already_saved > 0) {
 750       if (s->is_closed())
 751         goto bad;
 752
 753       --already_saved;
 754       continue;
 755     }
 756
 757     set_state(s, Session::STATE_OPEN);
 758     replay_dirty_session(s);
 759   }
 760   return;
 761
 762 bad:
 763   mds->clog->error() << "error replaying open sessions(" << client_map.size()
 764                      << ") sessionmap v " << event_cmapv << " table " << version;
 765   ceph_assert(g_conf()->mds_wipe_sessions);
 766   mds->sessionmap.wipe();
 767   mds->sessionmap.set_version(event_cmapv);
 768 }
 769
 770 version_t SessionMap::mark_projected(Session *s)
 771 {
 772   dout(20) << __func__ << " s=" << s << " name=" << s->info.inst.name
 773     << " pv=" << projected << " -> " << projected + 1 << dendl;
 774   ++projected;
 775   s->push_pv(projected);
 776   return projected;
 777 }
 778
 779 namespace {
 780 class C_IO_SM_Save_One : public SessionMapIOContext {
 781   MDSContext *on_safe;
 782 public:
 783   C_IO_SM_Save_One(SessionMap *cm, MDSContext *on_safe_)
 784     : SessionMapIOContext(cm), on_safe(on_safe_) {}
 785   void finish(int r) override {
 786     if (r != 0) {
 787       get_mds()->handle_write_error(r);
 788     } else {
 789       on_safe->complete(r);
 790     }
 791   }
 792   void print(ostream& out) const override {
 793     out << "session_save_one";
 794   }
 795 };
 796 }
 797
 798
 799 void SessionMap::save_if_dirty(const std::set<entity_name_t> &tgt_sessions,
 800                                MDSGatherBuilder *gather_bld)
 801 {
 802   ceph_assert(gather_bld != NULL);
 803
 804   std::vector<entity_name_t> write_sessions;
 805
 806   // Decide which sessions require a write
 807   for (std::set<entity_name_t>::iterator i = tgt_sessions.begin();
 808        i != tgt_sessions.end(); ++i) {
 809     const entity_name_t &session_id = *i;
 810
 811     if (session_map.count(session_id) == 0) {
 812       // Session isn't around any more, never mind.
 813       continue;
 814     }
 815
 816     Session *session = session_map[session_id];
 817     if (!session->has_dirty_completed_requests()) {
 818       // Session hasn't had completed_requests
 819       // modified since last write, no need to
 820       // write it now.
 821       continue;
 822     }
 823
 824     if (dirty_sessions.count(session_id) > 0) {
 825       // Session is already dirtied, will be written, no
 826       // need to pre-empt that.
 827       continue;
 828     }
 829     // Okay, passed all our checks, now we write
 830     // this session out.  The version we write
 831     // into the OMAP may now be higher-versioned
 832     // than the version in the header, but that's
 833     // okay because it's never a problem to have
 834     // an overly-fresh copy of a session.
 835     write_sessions.push_back(*i);
 836   }
 837
 838   dout(4) << __func__ << ": writing " << write_sessions.size() << dendl;
 839
 840   // Batch writes into mds_sessionmap_keys_per_op
 841   const uint32_t kpo = g_conf()->mds_sessionmap_keys_per_op;
 842   map<string, bufferlist> to_set;
 843   for (uint32_t i = 0; i < write_sessions.size(); ++i) {
 844     const entity_name_t &session_id = write_sessions[i];
 845     Session *session = session_map[session_id];
 846     session->clear_dirty_completed_requests();
 847
 848     // Serialize K
 849     std::ostringstream k;
 850     k << session_id;
 851
 852     // Serialize V
 853     bufferlist bl;
 854     session->info.encode(bl, mds->mdsmap->get_up_features());
 855
 856     // Add to RADOS op
 857     to_set[k.str()] = bl;
 858
 859     // Complete this write transaction?
 860     if (i == write_sessions.size() - 1
 861         || i % kpo == kpo - 1) {
 862       ObjectOperation op;
 863       op.omap_set(to_set);
 864       to_set.clear(); // clear to start a new transaction
 865
 866       SnapContext snapc;
 867       object_t oid = get_object_name();
 868       object_locator_t oloc(mds->mdsmap->get_metadata_pool());
 869       MDSContext *on_safe = gather_bld->new_sub();
 870       mds->objecter->mutate(oid, oloc, op, snapc,
 871                             ceph::real_clock::now(), 0,
 872                             new C_OnFinisher(
 873                               new C_IO_SM_Save_One(this, on_safe),
 874                               mds->finisher));
 875     }
 876   }
 877 }
 878
 879 // =================
 880 // Session
 881
 882 #undef dout_prefix
 883 #define dout_prefix *_dout << "Session "
 884
 885 /**
 886  * Calculate the length of the `requests` member list,
 887  * because elist does not have a size() method.
 888  *
 889  * O(N) runtime.
 890  */
 891 size_t Session::get_request_count() const
 892 {
 893   size_t result = 0;
 894
 895   auto it = requests.begin(member_offset(MDRequestImpl, item_session_request));
 896   while (!it.end()) {
 897     ++result;
 898     ++it;
 899   }
 900
 901   return result;
 902 }
 903
 904 /**
 905  * Capped in response to a CEPH_MSG_CLIENT_CAPRELEASE message,
 906  * with n_caps equal to the number of caps that were released
 907  * in the message.  Used to update state about how many caps a
 908  * client has released since it was last instructed to RECALL_STATE.
 909  */
 910 void Session::notify_cap_release(size_t n_caps)
 911 {
 912   recall_caps.hit(-(double)n_caps);
 913   release_caps.hit(n_caps);
 914 }
 915
 916 /**
 917  * Called when a CEPH_MSG_CLIENT_SESSION->CEPH_SESSION_RECALL_STATE
 918  * message is sent to the client.  Update our recall-related state
 919  * in order to generate health metrics if the session doesn't see
 920  * a commensurate number of calls to ::notify_cap_release
 921  */
 922 uint64_t Session::notify_recall_sent(size_t new_limit)
 923 {
 924   const auto num_caps = caps.size();
 925   ceph_assert(new_limit < num_caps);  // Behaviour of Server::recall_client_state
 926   const auto count = num_caps-new_limit;
 927   uint64_t new_change;
 928   if (recall_limit != new_limit) {
 929     new_change = count;
 930   } else {
 931     new_change = 0; /* no change! */
 932   }
 933
 934   /* Always hit the session counter as a RECALL message is still sent to the
 935    * client and we do not want the MDS to burn its global counter tokens on a
 936    * session that is not releasing caps (i.e. allow the session counter to
 937    * throttle future RECALL messages).
 938    */
 939   recall_caps_throttle.hit(count);
 940   recall_caps_throttle2o.hit(count);
 941   recall_caps.hit(count);
 942   return new_change;
 943 }
 944
 945 /**
 946  * Use client metadata to generate a somewhat-friendlier
 947  * name for the client than its session ID.
 948  *
 949  * This is *not* guaranteed to be unique, and any machine
 950  * consumers of session-related output should always use
 951  * the session ID as a primary capacity and use this only
 952  * as a presentation hint.
 953  */
 954 void Session::_update_human_name()
 955 {
 956   auto info_client_metadata_entry = info.client_metadata.find("hostname");
 957   if (info_client_metadata_entry != info.client_metadata.end()) {
 958     // Happy path, refer to clients by hostname
 959     human_name = info_client_metadata_entry->second;
 960     if (!info.auth_name.has_default_id()) {
 961       // When a non-default entity ID is set by the user, assume they
 962       // would like to see it in references to the client, if it's
 963       // reasonable short.  Limit the length because we don't want
 964       // to put e.g. uuid-generated names into a "human readable"
 965       // rendering.
 966       const int arbitrarily_short = 16;
 967       if (info.auth_name.get_id().size() < arbitrarily_short) {
 968         human_name += std::string(":") + info.auth_name.get_id();
 969       }
 970     }
 971   } else {
 972     // Fallback, refer to clients by ID e.g. client.4567
 973     human_name = stringify(info.inst.name.num());
 974   }
 975 }
 976
 977 void Session::decode(bufferlist::const_iterator &p)
 978 {
 979   info.decode(p);
 980
 981   _update_human_name();
 982 }
 983
 984 int Session::check_access(CInode *in, unsigned mask,
 985                           int caller_uid, int caller_gid,
 986                           const vector<uint64_t> *caller_gid_list,
 987                           int new_uid, int new_gid)
 988 {
 989   string path;
 990   CInode *diri = NULL;
 991   if (!in->is_base())
 992     diri = in->get_projected_parent_dn()->get_dir()->get_inode();
 993   if (diri && diri->is_stray()){
 994     path = in->get_projected_inode()->stray_prior_path;
 995     dout(20) << __func__ << " stray_prior_path " << path << dendl;
 996   } else {
 997     in->make_path_string(path, true);
 998     dout(20) << __func__ << " path " << path << dendl;
 999   }
1000   if (path.length())
1001     path = path.substr(1);    // drop leading /
1002
1003   if (in->inode.is_dir() &&
1004       in->inode.has_layout() &&
1005       in->inode.layout.pool_ns.length() &&
1006       !connection->has_feature(CEPH_FEATURE_FS_FILE_LAYOUT_V2)) {
1007     dout(10) << __func__ << " client doesn't support FS_FILE_LAYOUT_V2" << dendl;
1008     return -EIO;
1009   }
1010
1011   if (!auth_caps.is_capable(path, in->inode.uid, in->inode.gid, in->inode.mode,
1012                             caller_uid, caller_gid, caller_gid_list, mask,
1013                             new_uid, new_gid,
1014                             info.inst.addr)) {
1015     return -EACCES;
1016   }
1017   return 0;
1018 }
1019
1020 // track total and per session load
1021 void SessionMap::hit_session(Session *session) {
1022   uint64_t sessions = get_session_count_in_state(Session::STATE_OPEN) +
1023                       get_session_count_in_state(Session::STATE_STALE);
1024   ceph_assert(sessions != 0);
1025
1026   double total_load = total_load_avg.hit();
1027   double avg_load = total_load / sessions;
1028
1029   logger->set(l_mdssm_total_load, (uint64_t)total_load);
1030   logger->set(l_mdssm_avg_load, (uint64_t)avg_load);
1031
1032   session->hit_session();
1033 }
1034
1035 void SessionMap::handle_conf_change(const std::set<std::string>& changed)
1036 {
1037   auto apply_to_open_sessions = [this](auto f) {
1038     if (auto it = by_state.find(Session::STATE_OPEN); it != by_state.end()) {
1039       for (const auto &session : *(it->second)) {
1040         f(session);
1041       }
1042     }
1043     if (auto it = by_state.find(Session::STATE_STALE); it != by_state.end()) {
1044       for (const auto &session : *(it->second)) {
1045         f(session);
1046       }
1047     }
1048   };
1049
1050   if (changed.count("mds_request_load_average_decay_rate")) {
1051     auto d = g_conf().get_val<double>("mds_request_load_average_decay_rate");
1052
1053     decay_rate = d;
1054     total_load_avg = DecayCounter(d);
1055
1056     auto mut = [d](auto s) {
1057       s->set_load_avg_decay_rate(d);
1058     };
1059     apply_to_open_sessions(mut);
1060   }
1061   if (changed.count("mds_recall_max_decay_rate")) {
1062     auto d = g_conf().get_val<double>("mds_recall_max_decay_rate");
1063     auto mut = [d](auto s) {
1064       s->recall_caps_throttle = DecayCounter(d);
1065     };
1066     apply_to_open_sessions(mut);
1067   }
1068   if (changed.count("mds_recall_warning_decay_rate")) {
1069     auto d = g_conf().get_val<double>("mds_recall_warning_decay_rate");
1070     auto mut = [d](auto s) {
1071       s->recall_caps = DecayCounter(d);
1072       s->release_caps = DecayCounter(d);
1073     };
1074     apply_to_open_sessions(mut);
1075   }
1076   if (changed.count("mds_session_cache_liveness_decay_rate")) {
1077     auto d = g_conf().get_val<double>("mds_session_cache_liveness_decay_rate");
1078     auto mut = [d](auto s) {
1079       s->session_cache_liveness = DecayCounter(d);
1080       s->session_cache_liveness.hit(s->caps.size()); /* so the MDS doesn't immediately start trimming a new session */
1081     };
1082     apply_to_open_sessions(mut);
1083   }
1084 }
1085
1086 void SessionMap::update_average_session_age() {
1087   if (!session_map.size()) {
1088     return;
1089   }
1090
1091   double avg_uptime = std::chrono::duration<double>(clock::now()-avg_birth_time).count();
1092   logger->set(l_mdssm_avg_session_uptime, (uint64_t)avg_uptime);
1093 }
1094
1095 int SessionFilter::parse(
1096     const std::vector<std::string> &args,
1097     std::stringstream *ss)
1098 {
1099   ceph_assert(ss != NULL);
1100
1101   for (const auto &s : args) {
1102     dout(20) << __func__ << " parsing filter '" << s << "'" << dendl;
1103
1104     auto eq = s.find("=");
1105     if (eq == std::string::npos || eq == s.size()) {
1106       *ss << "Invalid filter '" << s << "'";
1107       return -EINVAL;
1108     }
1109
1110     // Keys that start with this are to be taken as referring
1111     // to freeform client metadata fields.
1112     const std::string metadata_prefix("client_metadata.");
1113
1114     auto k = s.substr(0, eq);
1115     auto v = s.substr(eq + 1);
1116
1117     dout(20) << __func__ << " parsed k='" << k << "', v='" << v << "'" << dendl;
1118
1119     if (k.compare(0, metadata_prefix.size(), metadata_prefix) == 0
1120         && k.size() > metadata_prefix.size()) {
1121       // Filter on arbitrary metadata key (no fixed schema for this,
1122       // so anything after the dot is a valid field to filter on)
1123       auto metadata_key = k.substr(metadata_prefix.size());
1124       metadata.insert(std::make_pair(metadata_key, v));
1125     } else if (k == "auth_name") {
1126       // Filter on client entity name
1127       auth_name = v;
1128     } else if (k == "state") {
1129       state = v;
1130     } else if (k == "id") {
1131       std::string err;
1132       id = strict_strtoll(v.c_str(), 10, &err);
1133       if (!err.empty()) {
1134         *ss << err;
1135         return -EINVAL;
1136       }
1137     } else if (k == "reconnecting") {
1138
1139       /**
1140        * Strict boolean parser.  Allow true/false/0/1.
1141        * Anything else is -EINVAL.
1142        */
1143       auto is_true = [](std::string_view bstr, bool *out) -> bool
1144       {
1145         ceph_assert(out != nullptr);
1146
1147         if (bstr == "true" || bstr == "1") {
1148           *out = true;
1149           return 0;
1150         } else if (bstr == "false" || bstr == "0") {
1151           *out = false;
1152           return 0;
1153         } else {
1154           return -EINVAL;
1155         }
1156       };
1157
1158       bool bval;
1159       int r = is_true(v, &bval);
1160       if (r == 0) {
1161         set_reconnecting(bval);
1162       } else {
1163         *ss << "Invalid boolean value '" << v << "'";
1164         return -EINVAL;
1165       }
1166     } else {
1167       *ss << "Invalid filter key '" << k << "'";
1168       return -EINVAL;
1169     }
1170   }
1171
1172   return 0;
1173 }
1174
1175 bool SessionFilter::match(
1176     const Session &session,
1177     std::function<bool(client_t)> is_reconnecting) const
1178 {
1179   for (const auto &m : metadata) {
1180     const auto &k = m.first;
1181     const auto &v = m.second;
1182     auto it = session.info.client_metadata.find(k);
1183     if (it == session.info.client_metadata.end()) {
1184       return false;
1185     }
1186     if (it->second != v) {
1187       return false;
1188     }
1189   }
1190
1191   if (!auth_name.empty() && auth_name != session.info.auth_name.get_id()) {
1192     return false;
1193   }
1194
1195   if (!state.empty() && state != session.get_state_name()) {
1196     return false;
1197   }
1198
1199   if (id != 0 && id != session.info.inst.name.num()) {
1200     return false;
1201   }
1202
1203   if (reconnecting.first) {
1204     const bool am_reconnecting = is_reconnecting(session.info.inst.name.num());
1205     if (reconnecting.second != am_reconnecting) {
1206       return false;
1207     }
1208   }
1209
1210   return true;
1211 }
1212
1213 std::ostream& operator<<(std::ostream &out, const Session &s)
1214 {
1215  if (s.get_human_name() == stringify(s.get_client())) {
1216    out << s.get_human_name();
1217  } else {
1218    out << s.get_human_name() << " (" << std::dec << s.get_client() << ")";
1219  }
1220  return out;
1221 }
1222