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(
  66   const std::string &hostname) const
  67 {
  68   Mutex::Locker l(lock);
  69
  70   if (by_server.count(hostname)) {
  71     return by_server.at(hostname);
  72   } else {
  73     return {};
  74   }
  75 }
  76
  77 bool DaemonStateIndex::exists(const DaemonKey &key) const
  78 {
  79   Mutex::Locker l(lock);
  80
  81   return all.count(key) > 0;
  82 }
  83
  84 DaemonStatePtr DaemonStateIndex::get(const DaemonKey &key)
  85 {
  86   Mutex::Locker l(lock);
  87
  88   return all.at(key);
  89 }
  90
  91 void DaemonStateIndex::cull(const std::string& svc_name,
  92                             const std::set<std::string>& names_exist)
  93 {
  94   std::vector<string> victims;
  95
  96   Mutex::Locker l(lock);
  97   auto begin = all.lower_bound({svc_name, ""});
  98   auto end = all.end();
  99   for (auto &i = begin; i != end; ++i) {
 100     const auto& daemon_key = i->first;
 101     if (daemon_key.first != svc_name)
 102       break;
 103     if (names_exist.count(daemon_key.second) == 0) {
 104       victims.push_back(daemon_key.second);
 105     }
 106   }
 107
 108   for (auto &i : victims) {
 109     dout(4) << "Removing data for " << i << dendl;
 110     _erase({svc_name, i});
 111   }
 112 }
 113
 114 void DaemonPerfCounters::update(MMgrReport *report)
 115 {
 116   dout(20) << "loading " << report->declare_types.size() << " new types, "
 117            << report->undeclare_types.size() << " old types, had "
 118            << types.size() << " types, got "
 119            << report->packed.length() << " bytes of data" << dendl;
 120
 121   // Load any newly declared types
 122   for (const auto &t : report->declare_types) {
 123     types.insert(std::make_pair(t.path, t));
 124     declared_types.insert(t.path);
 125   }
 126   // Remove any old types
 127   for (const auto &t : report->undeclare_types) {
 128     declared_types.erase(t);
 129   }
 130
 131   const auto now = ceph_clock_now();
 132
 133   // Parse packed data according to declared set of types
 134   bufferlist::iterator p = report->packed.begin();
 135   DECODE_START(1, p);
 136   for (const auto &t_path : declared_types) {
 137     const auto &t = types.at(t_path);
 138     uint64_t val = 0;
 139     uint64_t avgcount = 0;
 140     uint64_t avgcount2 = 0;
 141
 142     ::decode(val, p);
 143     if (t.type & PERFCOUNTER_LONGRUNAVG) {
 144       ::decode(avgcount, p);
 145       ::decode(avgcount2, p);
 146     }
 147     // TODO: interface for insertion of avgs
 148     instances[t_path].push(now, val);
 149   }
 150   DECODE_FINISH(p);
 151 }
 152
 153 uint64_t PerfCounterInstance::get_current() const
 154 {
 155   return buffer.front().v;
 156 }
 157
 158 void PerfCounterInstance::push(utime_t t, uint64_t const &v)
 159 {
 160   buffer.push_back({t, v});
 161 }
 162