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
  10 #define dout_context g_ceph_context
  11 #define dout_subsys ceph_subsys_mgr
  12 #undef dout_prefix
  13 #define dout_prefix *_dout << "mgr.metric_collector " << __func__ << ": "
  14
  15 template <typename Query, typename Limit, typename Key, typename Report>
  16 MetricCollector<Query, Limit, Key, Report>::MetricCollector(MetricListener &listener)
  17   : listener(listener)
  18 {
  19 }
  20
  21 template <typename Query, typename Limit, typename Key, typename Report>
  22 MetricQueryID MetricCollector<Query, Limit, Key, Report>::add_query(
  23     const Query &query,
  24     const std::optional<Limit> &limit) {
  25   dout(20) << "query=" << query << ", limit=" << limit << dendl;
  26   uint64_t query_id;
  27   bool notify = false;
  28
  29   {
  30     std::lock_guard locker(lock);
  31
  32     query_id = next_query_id++;
  33     auto it = queries.find(query);
  34     if (it == queries.end()) {
  35       it = queries.emplace(query, std::map<MetricQueryID, OptionalLimit>{}).first;
  36       notify = true;
  37     } else if (is_limited(it->second)) {
  38       notify = true;
  39     }
  40
  41     it->second.emplace(query_id, limit);
  42     counters.emplace(query_id, std::map<Key, PerformanceCounters>{});
  43   }
  44
  45   dout(10) << query << " " << (limit ? stringify(*limit) : "unlimited")
  46            << " query_id=" << query_id << dendl;
  47
  48   if (notify) {
  49     listener.handle_query_updated();
  50   }
  51
  52   return query_id;
  53 }
  54
  55 template <typename Query, typename Limit, typename Key, typename Report>
  56 int MetricCollector<Query, Limit, Key, Report>::remove_query(MetricQueryID query_id) {
  57   dout(20) << "query_id=" << query_id << dendl;
  58   bool found = false;
  59   bool notify = false;
  60
  61   {
  62     std::lock_guard locker(lock);
  63
  64     for (auto it = queries.begin() ; it != queries.end();) {
  65       auto iter = it->second.find(query_id);
  66       if (iter == it->second.end()) {
  67         ++it;
  68         continue;
  69       }
  70
  71       it->second.erase(iter);
  72       if (it->second.empty()) {
  73         it = queries.erase(it);
  74         notify = true;
  75       } else if (is_limited(it->second)) {
  76         ++it;
  77         notify = true;
  78       }
  79       found = true;
  80       break;
  81     }
  82     counters.erase(query_id);
  83   }
  84
  85   if (!found) {
  86     dout(10) << query_id << " not found" << dendl;
  87     return -ENOENT;
  88   }
  89
  90   dout(10) << query_id << dendl;
  91
  92   if (notify) {
  93     listener.handle_query_updated();
  94   }
  95
  96   return 0;
  97 }
  98
  99 template <typename Query, typename Limit, typename Key, typename Report>
 100 void MetricCollector<Query, Limit, Key, Report>::remove_all_queries() {
 101   dout(20) << dendl;
 102   bool notify;
 103
 104   {
 105     std::lock_guard locker(lock);
 106
 107     notify = !queries.empty();
 108     queries.clear();
 109   }
 110
 111   if (notify) {
 112     listener.handle_query_updated();
 113   }
 114 }
 115
 116 template <typename Query, typename Limit, typename Key, typename Report>
 117 int MetricCollector<Query, Limit, Key, Report>::get_counters(
 118     MetricQueryID query_id, std::map<Key, PerformanceCounters> *c) {
 119   dout(20) << dendl;
 120
 121   std::lock_guard locker(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>;