ceph/src/mgr/MetricCollector.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include "common/debug.h"
   5 #include "common/errno.h"
   6
   7 #include "mgr/MetricCollector.h"
   8 #include "mgr/OSDPerfMetricTypes.h"
   9 #include "mgr/MDSPerfMetricTypes.h"
  10
  11 #define dout_context g_ceph_context
  12 #define dout_subsys ceph_subsys_mgr
  13 #undef dout_prefix
  14 #define dout_prefix *_dout << "mgr.metric_collector " << __func__ << ": "
  15
  16 template <typename Query, typename Limit, typename Key, typename Report>
  17 MetricCollector<Query, Limit, Key, Report>::MetricCollector(MetricListener &listener)
  18   : listener(listener)
  19 {
  20 }
  21
  22 template <typename Query, typename Limit, typename Key, typename Report>
  23 MetricQueryID MetricCollector<Query, Limit, Key, Report>::add_query(
  24     const Query &query,
  25     const std::optional<Limit> &limit) {
  26   dout(20) << "query=" << query << ", limit=" << limit << dendl;
  27   uint64_t query_id;
  28   bool notify = false;
  29
  30   {
  31     std::lock_guard locker(lock);
  32
  33     query_id = next_query_id++;
  34     auto it = queries.find(query);
  35     if (it == queries.end()) {
  36       it = queries.emplace(query, std::map<MetricQueryID, OptionalLimit>{}).first;
  37       notify = true;
  38     } else if (is_limited(it->second)) {
  39       notify = true;
  40     }
  41
  42     it->second.emplace(query_id, limit);
  43     counters.emplace(query_id, std::map<Key, PerformanceCounters>{});
  44   }
  45
  46   dout(10) << query << " " << (limit ? stringify(*limit) : "unlimited")
  47            << " query_id=" << query_id << dendl;
  48
  49   if (notify) {
  50     listener.handle_query_updated();
  51   }
  52
  53   return query_id;
  54 }
  55
  56 template <typename Query, typename Limit, typename Key, typename Report>
  57 int MetricCollector<Query, Limit, Key, Report>::remove_query(MetricQueryID query_id) {
  58   dout(20) << "query_id=" << query_id << dendl;
  59   bool found = false;
  60   bool notify = false;
  61
  62   {
  63     std::lock_guard locker(lock);
  64
  65     for (auto it = queries.begin() ; it != queries.end();) {
  66       auto iter = it->second.find(query_id);
  67       if (iter == it->second.end()) {
  68         ++it;
  69         continue;
  70       }
  71
  72       it->second.erase(iter);
  73       if (it->second.empty()) {
  74         it = queries.erase(it);
  75         notify = true;
  76       } else if (is_limited(it->second)) {
  77         ++it;
  78         notify = true;
  79       }
  80       found = true;
  81       break;
  82     }
  83     counters.erase(query_id);
  84   }
  85
  86   if (!found) {
  87     dout(10) << query_id << " not found" << dendl;
  88     return -ENOENT;
  89   }
  90
  91   dout(10) << query_id << dendl;
  92
  93   if (notify) {
  94     listener.handle_query_updated();
  95   }
  96
  97   return 0;
  98 }
  99
 100 template <typename Query, typename Limit, typename Key, typename Report>
 101 void MetricCollector<Query, Limit, Key, Report>::remove_all_queries() {
 102   dout(20) << dendl;
 103   bool notify;
 104
 105   {
 106     std::lock_guard locker(lock);
 107
 108     notify = !queries.empty();
 109     queries.clear();
 110   }
 111
 112   if (notify) {
 113     listener.handle_query_updated();
 114   }
 115 }
 116
 117 template <typename Query, typename Limit, typename Key, typename Report>
 118 int MetricCollector<Query, Limit, Key, Report>::get_counters_generic(
 119     MetricQueryID query_id, std::map<Key, PerformanceCounters> *c) {
 120   dout(20) << dendl;
 121   ceph_assert(ceph_mutex_is_locked(lock));
 122
 123   auto it = counters.find(query_id);
 124   if (it == counters.end()) {
 125     dout(10) << "counters for " << query_id << " not found" << dendl;
 126     return -ENOENT;
 127   }
 128
 129   *c = std::move(it->second);
 130   it->second.clear();
 131
 132   return 0;
 133 }
 134
 135 template <typename Query, typename Limit, typename Key, typename Report>
 136 void MetricCollector<Query, Limit, Key, Report>::process_reports_generic(
 137     const std::map<Query, Report> &reports, UpdateCallback callback) {
 138   ceph_assert(ceph_mutex_is_locked(lock));
 139
 140   if (reports.empty()) {
 141     return;
 142   }
 143
 144   for (auto& [query, report] : reports) {
 145     dout(10) << "report for " << query << " query: "
 146              << report.group_packed_performance_counters.size() << " records"
 147              << dendl;
 148
 149     for (auto& [key, bl] : report.group_packed_performance_counters) {
 150       auto bl_it = bl.cbegin();
 151
 152       for (auto& p : queries[query]) {
 153         auto &key_counters = counters[p.first][key];
 154         if (key_counters.empty()) {
 155           key_counters.resize(query.performance_counter_descriptors.size(),
 156                               {0, 0});
 157         }
 158       }
 159
 160       auto desc_it = report.performance_counter_descriptors.begin();
 161       for (size_t i = 0; i < query.performance_counter_descriptors.size(); i++) {
 162         if (desc_it == report.performance_counter_descriptors.end()) {
 163           break;
 164         }
 165         if (*desc_it != query.performance_counter_descriptors[i]) {
 166           continue;
 167         }
 168         PerformanceCounter c;
 169         desc_it->unpack_counter(bl_it, &c);
 170         dout(20) << "counter " << key << " " << *desc_it << ": " << c << dendl;
 171
 172         for (auto& p : queries[query]) {
 173           auto &key_counters = counters[p.first][key];
 174           callback(&key_counters[i], c);
 175         }
 176         desc_it++;
 177       }
 178     }
 179   }
 180 }
 181
 182 template class
 183 MetricCollector<OSDPerfMetricQuery, OSDPerfMetricLimit, OSDPerfMetricKey, OSDPerfMetricReport>;
 184 template class
 185 MetricCollector<MDSPerfMetricQuery, MDSPerfMetricLimit, MDSPerfMetricKey, MDSPerfMetrics>;