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
  18 #define dout_context g_ceph_context
  19 #define dout_subsys ceph_subsys_mgr
  20 #undef dout_prefix
  21 #define dout_prefix *_dout << "mgr " << __func__ << " "
  22
  23 void DaemonStateIndex::insert(DaemonStatePtr dm)
  24 {
  25   RWLock::WLocker l(lock);
  26
  27   if (all.count(dm->key)) {
  28     _erase(dm->key);
  29   }
  30
  31   by_server[dm->hostname][dm->key] = dm;
  32   all[dm->key] = dm;
  33 }
  34
  35 void DaemonStateIndex::_erase(const DaemonKey& dmk)
  36 {
  37   assert(lock.is_wlocked());
  38
  39   const auto to_erase = all.find(dmk);
  40   assert(to_erase != all.end());
  41   const auto dm = to_erase->second;
  42   auto &server_collection = by_server[dm->hostname];
  43   server_collection.erase(dm->key);
  44   if (server_collection.empty()) {
  45     by_server.erase(dm->hostname);
  46   }
  47
  48   all.erase(to_erase);
  49 }
  50
  51 DaemonStateCollection DaemonStateIndex::get_by_service(
  52   const std::string& svc) const
  53 {
  54   RWLock::RLocker l(lock);
  55
  56   DaemonStateCollection result;
  57
  58   for (const auto &i : all) {
  59     if (i.first.first == svc) {
  60       result[i.first] = i.second;
  61     }
  62   }
  63
  64   return result;
  65 }
  66
  67 DaemonStateCollection DaemonStateIndex::get_by_server(
  68   const std::string &hostname) const
  69 {
  70   RWLock::RLocker l(lock);
  71
  72   if (by_server.count(hostname)) {
  73     return by_server.at(hostname);
  74   } else {
  75     return {};
  76   }
  77 }
  78
  79 bool DaemonStateIndex::exists(const DaemonKey &key) const
  80 {
  81   RWLock::RLocker l(lock);
  82
  83   return all.count(key) > 0;
  84 }
  85
  86 DaemonStatePtr DaemonStateIndex::get(const DaemonKey &key)
  87 {
  88   RWLock::RLocker l(lock);
  89
  90   auto iter = all.find(key);
  91   if (iter != all.end()) {
  92     return iter->second;
  93   } else {
  94     return nullptr;
  95   }
  96 }
  97
  98 void DaemonStateIndex::cull(const std::string& svc_name,
  99                             const std::set<std::string>& names_exist)
 100 {
 101   std::vector<string> victims;
 102
 103   RWLock::WLocker l(lock);
 104   auto begin = all.lower_bound({svc_name, ""});
 105   auto end = all.end();
 106   for (auto &i = begin; i != end; ++i) {
 107     const auto& daemon_key = i->first;
 108     if (daemon_key.first != svc_name)
 109       break;
 110     if (names_exist.count(daemon_key.second) == 0) {
 111       victims.push_back(daemon_key.second);
 112     }
 113   }
 114
 115   for (auto &i : victims) {
 116     dout(4) << "Removing data for " << i << dendl;
 117     _erase({svc_name, i});
 118   }
 119 }
 120
 121 void DaemonPerfCounters::update(MMgrReport *report)
 122 {
 123   dout(20) << "loading " << report->declare_types.size() << " new types, "
 124            << report->undeclare_types.size() << " old types, had "
 125            << types.size() << " types, got "
 126            << report->packed.length() << " bytes of data" << dendl;
 127
 128   // Retrieve session state
 129   MgrSessionRef session(static_cast<MgrSession*>(
 130         report->get_connection()->get_priv()));
 131
 132   // Load any newly declared types
 133   for (const auto &t : report->declare_types) {
 134     types.insert(std::make_pair(t.path, t));
 135     session->declared_types.insert(t.path);
 136     instances.insert(std::pair<std::string, PerfCounterInstance>(
 137                      t.path, PerfCounterInstance(t.type)));
 138   }
 139   // Remove any old types
 140   for (const auto &t : report->undeclare_types) {
 141     session->declared_types.erase(t);
 142   }
 143
 144   const auto now = ceph_clock_now();
 145
 146   // Parse packed data according to declared set of types
 147   bufferlist::iterator p = report->packed.begin();
 148   DECODE_START(1, p);
 149   for (const auto &t_path : session->declared_types) {
 150     const auto &t = types.at(t_path);
 151     uint64_t val = 0;
 152     uint64_t avgcount = 0;
 153     uint64_t avgcount2 = 0;
 154
 155     ::decode(val, p);
 156     if (t.type & PERFCOUNTER_LONGRUNAVG) {
 157       ::decode(avgcount, p);
 158       ::decode(avgcount2, p);
 159       instances.at(t_path).push_avg(now, val, avgcount);
 160     } else {
 161       instances.at(t_path).push(now, val);
 162     }
 163   }
 164   DECODE_FINISH(p);
 165 }
 166
 167 uint64_t PerfCounterInstance::get_current() const
 168 {
 169   return buffer.front().v;
 170 }
 171
 172 void PerfCounterInstance::push(utime_t t, uint64_t const &v)
 173 {
 174   buffer.push_back({t, v});
 175 }
 176
 177 void PerfCounterInstance::push_avg(utime_t t, uint64_t const &s,
 178                                    uint64_t const &c)
 179 {
 180   avg_buffer.push_back({t, s, c});
 181 }