ceph/src/mgr/DaemonState.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 #include "DaemonState.h"
  15
  16 #include "MgrSession.h"
  17 #include "include/stringify.h"
  18 #include "common/Formatter.h"
  19
  20 #define dout_context g_ceph_context
  21 #define dout_subsys ceph_subsys_mgr
  22 #undef dout_prefix
  23 #define dout_prefix *_dout << "mgr " << __func__ << " "
  24
  25 void DeviceState::set_metadata(map<string,string>&& m)
  26 {
  27   metadata = std::move(m);
  28   auto p = metadata.find("life_expectancy_min");
  29   if (p != metadata.end()) {
  30     life_expectancy.first.parse(p->second);
  31   }
  32   p = metadata.find("life_expectancy_max");
  33   if (p != metadata.end()) {
  34     life_expectancy.second.parse(p->second);
  35   }
  36   p = metadata.find("life_expectancy_stamp");
  37   if (p != metadata.end()) {
  38     life_expectancy_stamp.parse(p->second);
  39   }
  40 }
  41
  42 void DeviceState::set_life_expectancy(utime_t from, utime_t to, utime_t now)
  43 {
  44   life_expectancy = make_pair(from, to);
  45   life_expectancy_stamp = now;
  46   if (from != utime_t()) {
  47     metadata["life_expectancy_min"] = from;
  48   } else {
  49     metadata["life_expectancy_min"] = "";
  50   }
  51   if (to != utime_t()) {
  52     metadata["life_expectancy_max"] = to;
  53   } else {
  54     metadata["life_expectancy_max"] = "";
  55   }
  56   if (now != utime_t()) {
  57     metadata["life_expectancy_stamp"] = stringify(now);
  58   } else {
  59     metadata["life_expectancy_stamp"] = "";
  60   }
  61 }
  62
  63 void DeviceState::rm_life_expectancy()
  64 {
  65   life_expectancy = make_pair(utime_t(), utime_t());
  66   life_expectancy_stamp = utime_t();
  67   metadata.erase("life_expectancy_min");
  68   metadata.erase("life_expectancy_max");
  69   metadata.erase("life_expectancy_stamp");
  70 }
  71
  72 string DeviceState::get_life_expectancy_str(utime_t now) const
  73 {
  74   if (life_expectancy.first == utime_t()) {
  75     return string();
  76   }
  77   if (now >= life_expectancy.first) {
  78     return "now";
  79   }
  80   utime_t min = life_expectancy.first - now;
  81   utime_t max = life_expectancy.second - now;
  82   if (life_expectancy.second == utime_t()) {
  83     return string(">") + timespan_str(make_timespan(min));
  84   }
  85   string a = timespan_str(make_timespan(min));
  86   string b = timespan_str(make_timespan(max));
  87   if (a == b) {
  88     return a;
  89   }
  90   return a + " to " + b;
  91 }
  92
  93 void DeviceState::dump(Formatter *f) const
  94 {
  95   f->dump_string("devid", devid);
  96   f->open_array_section("location");
  97   for (auto& i : devnames) {
  98     f->open_object_section("attachment");
  99     f->dump_string("host", i.first);
 100     f->dump_string("dev", i.second);
 101     f->close_section();
 102   }
 103   f->close_section();
 104   f->open_array_section("daemons");
 105   for (auto& i : daemons) {
 106     f->dump_string("daemon", to_string(i));
 107   }
 108   f->close_section();
 109   if (life_expectancy.first != utime_t()) {
 110     f->dump_stream("life_expectancy_min") << life_expectancy.first;
 111     f->dump_stream("life_expectancy_max") << life_expectancy.second;
 112     f->dump_stream("life_expectancy_stamp")
 113       << life_expectancy_stamp;
 114   }
 115 }
 116
 117 void DeviceState::print(ostream& out) const
 118 {
 119   out << "device " << devid << "\n";
 120   for (auto& i : devnames) {
 121     out << "attachment " << i.first << ":" << i.second << "\n";
 122   }
 123   set<string> d;
 124   for (auto& j : daemons) {
 125     d.insert(to_string(j));
 126   }
 127   out << "daemons " << d << "\n";
 128   if (life_expectancy.first != utime_t()) {
 129     out << "life_expectancy " << life_expectancy.first << " to "
 130         << life_expectancy.second
 131         << " (as of " << life_expectancy_stamp << ")\n";
 132   }
 133 }
 134
 135 void DaemonStateIndex::insert(DaemonStatePtr dm)
 136 {
 137   RWLock::WLocker l(lock);
 138   _insert(dm);
 139 }
 140
 141 void DaemonStateIndex::_insert(DaemonStatePtr dm)
 142 {
 143   if (all.count(dm->key)) {
 144     _erase(dm->key);
 145   }
 146
 147   by_server[dm->hostname][dm->key] = dm;
 148   all[dm->key] = dm;
 149
 150   for (auto& i : dm->devices) {
 151     auto d = _get_or_create_device(i.first);
 152     d->daemons.insert(dm->key);
 153     d->devnames.insert(make_pair(dm->hostname, i.second));
 154   }
 155 }
 156
 157 void DaemonStateIndex::_erase(const DaemonKey& dmk)
 158 {
 159   ceph_assert(lock.is_wlocked());
 160
 161   const auto to_erase = all.find(dmk);
 162   ceph_assert(to_erase != all.end());
 163   const auto dm = to_erase->second;
 164
 165   for (auto& i : dm->devices) {
 166     auto d = _get_or_create_device(i.first);
 167     ceph_assert(d->daemons.count(dmk));
 168     d->daemons.erase(dmk);
 169     d->devnames.erase(make_pair(dm->hostname, i.second));
 170     if (d->empty()) {
 171       _erase_device(d);
 172     }
 173   }
 174
 175   auto &server_collection = by_server[dm->hostname];
 176   server_collection.erase(dm->key);
 177   if (server_collection.empty()) {
 178     by_server.erase(dm->hostname);
 179   }
 180
 181   all.erase(to_erase);
 182 }
 183
 184 DaemonStateCollection DaemonStateIndex::get_by_service(
 185   const std::string& svc) const
 186 {
 187   RWLock::RLocker l(lock);
 188
 189   DaemonStateCollection result;
 190
 191   for (const auto &i : all) {
 192     if (i.first.first == svc) {
 193       result[i.first] = i.second;
 194     }
 195   }
 196
 197   return result;
 198 }
 199
 200 DaemonStateCollection DaemonStateIndex::get_by_server(
 201   const std::string &hostname) const
 202 {
 203   RWLock::RLocker l(lock);
 204
 205   if (by_server.count(hostname)) {
 206     return by_server.at(hostname);
 207   } else {
 208     return {};
 209   }
 210 }
 211
 212 bool DaemonStateIndex::exists(const DaemonKey &key) const
 213 {
 214   RWLock::RLocker l(lock);
 215
 216   return all.count(key) > 0;
 217 }
 218
 219 DaemonStatePtr DaemonStateIndex::get(const DaemonKey &key)
 220 {
 221   RWLock::RLocker l(lock);
 222
 223   auto iter = all.find(key);
 224   if (iter != all.end()) {
 225     return iter->second;
 226   } else {
 227     return nullptr;
 228   }
 229 }
 230
 231 void DaemonStateIndex::rm(const DaemonKey &key)
 232 {
 233   RWLock::WLocker l(lock);
 234   _rm(key);
 235 }
 236
 237 void DaemonStateIndex::_rm(const DaemonKey &key)
 238 {
 239   if (all.count(key)) {
 240     _erase(key);
 241   }
 242 }
 243
 244 void DaemonStateIndex::cull(const std::string& svc_name,
 245                             const std::set<std::string>& names_exist)
 246 {
 247   std::vector<string> victims;
 248
 249   RWLock::WLocker l(lock);
 250   auto begin = all.lower_bound({svc_name, ""});
 251   auto end = all.end();
 252   for (auto &i = begin; i != end; ++i) {
 253     const auto& daemon_key = i->first;
 254     if (daemon_key.first != svc_name)
 255       break;
 256     if (names_exist.count(daemon_key.second) == 0) {
 257       victims.push_back(daemon_key.second);
 258     }
 259   }
 260
 261   for (auto &i : victims) {
 262     dout(4) << "Removing data for " << i << dendl;
 263     _erase({svc_name, i});
 264   }
 265 }
 266
 267 void DaemonPerfCounters::update(MMgrReport *report)
 268 {
 269   dout(20) << "loading " << report->declare_types.size() << " new types, "
 270            << report->undeclare_types.size() << " old types, had "
 271            << types.size() << " types, got "
 272            << report->packed.length() << " bytes of data" << dendl;
 273
 274   // Retrieve session state
 275   auto priv = report->get_connection()->get_priv();
 276   auto session = static_cast<MgrSession*>(priv.get());
 277
 278   // Load any newly declared types
 279   for (const auto &t : report->declare_types) {
 280     types.insert(std::make_pair(t.path, t));
 281     session->declared_types.insert(t.path);
 282   }
 283   // Remove any old types
 284   for (const auto &t : report->undeclare_types) {
 285     session->declared_types.erase(t);
 286   }
 287
 288   const auto now = ceph_clock_now();
 289
 290   // Parse packed data according to declared set of types
 291   auto p = report->packed.cbegin();
 292   DECODE_START(1, p);
 293   for (const auto &t_path : session->declared_types) {
 294     const auto &t = types.at(t_path);
 295     auto instances_it = instances.find(t_path);
 296     // Always check the instance exists, as we don't prevent yet
 297     // multiple sessions from daemons with the same name, and one
 298     // session clearing stats created by another on open.
 299     if (instances_it == instances.end()) {
 300       instances_it = instances.insert({t_path, t.type}).first;
 301     }
 302     uint64_t val = 0;
 303     uint64_t avgcount = 0;
 304     uint64_t avgcount2 = 0;
 305
 306     decode(val, p);
 307     if (t.type & PERFCOUNTER_LONGRUNAVG) {
 308       decode(avgcount, p);
 309       decode(avgcount2, p);
 310       instances_it->second.push_avg(now, val, avgcount);
 311     } else {
 312       instances_it->second.push(now, val);
 313     }
 314   }
 315   DECODE_FINISH(p);
 316 }
 317
 318 void PerfCounterInstance::push(utime_t t, uint64_t const &v)
 319 {
 320   buffer.push_back({t, v});
 321 }
 322
 323 void PerfCounterInstance::push_avg(utime_t t, uint64_t const &s,
 324                                    uint64_t const &c)
 325 {
 326   avg_buffer.push_back({t, s, c});
 327 }