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