ceph/src/osd/DynamicPerfStats.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #ifndef DYNAMIC_PERF_STATS_H
   5 #define DYNAMIC_PERF_STATS_H
   6
   7 #include "include/random.h"
   8 #include "messages/MOSDOp.h"
   9 #include "mgr/OSDPerfMetricTypes.h"
  10 #include "osd/OSD.h"
  11 #include "osd/OpRequest.h"
  12
  13 class DynamicPerfStats {
  14 public:
  15   DynamicPerfStats() {
  16   }
  17
  18   DynamicPerfStats(const std::list<OSDPerfMetricQuery> &queries) {
  19     for (auto &query : queries) {
  20       data[query];
  21     }
  22   }
  23
  24   void merge(const DynamicPerfStats &dps) {
  25     for (auto &query_it : dps.data) {
  26       auto &query = query_it.first;
  27       for (auto &key_it : query_it.second) {
  28         auto &key = key_it.first;
  29         auto counter_it = key_it.second.begin();
  30         auto update_counter_fnc =
  31             [&counter_it](const PerformanceCounterDescriptor &d,
  32                           PerformanceCounter *c) {
  33               c->first  += counter_it->first;
  34               c->second += counter_it->second;
  35               counter_it++;
  36             };
  37
  38         ceph_assert(key_it.second.size() >= data[query][key].size());
  39         query.update_counters(update_counter_fnc, &data[query][key]);
  40       }
  41     }
  42   }
  43
  44   void set_queries(const std::list<OSDPerfMetricQuery> &queries) {
  45     std::map<OSDPerfMetricQuery,
  46              std::map<OSDPerfMetricKey, PerformanceCounters>> new_data;
  47     for (auto &query : queries) {
  48       std::swap(new_data[query], data[query]);
  49     }
  50     std::swap(data, new_data);
  51   }
  52
  53   bool is_enabled() {
  54     return !data.empty();
  55   }
  56
  57   void add(const OSDService *osd, const pg_info_t &pg_info, const OpRequest& op,
  58            uint64_t inb, uint64_t outb, const utime_t &latency) {
  59
  60     auto update_counter_fnc =
  61         [&op, inb, outb, &latency](const PerformanceCounterDescriptor &d,
  62                                    PerformanceCounter *c) {
  63           ceph_assert(d.is_supported());
  64
  65           switch(d.type) {
  66           case PerformanceCounterType::OPS:
  67             c->first++;
  68             return;
  69           case PerformanceCounterType::WRITE_OPS:
  70             if (op.may_write() || op.may_cache()) {
  71               c->first++;
  72             }
  73             return;
  74           case PerformanceCounterType::READ_OPS:
  75             if (op.may_read()) {
  76               c->first++;
  77             }
  78             return;
  79           case PerformanceCounterType::BYTES:
  80             c->first += inb + outb;
  81             return;
  82           case PerformanceCounterType::WRITE_BYTES:
  83             if (op.may_write() || op.may_cache()) {
  84               c->first += inb;
  85             }
  86             return;
  87           case PerformanceCounterType::READ_BYTES:
  88             if (op.may_read()) {
  89               c->first += outb;
  90             }
  91             return;
  92           case PerformanceCounterType::LATENCY:
  93             c->first += latency.to_nsec();
  94             c->second++;
  95             return;
  96           case PerformanceCounterType::WRITE_LATENCY:
  97             if (op.may_write() || op.may_cache()) {
  98               c->first += latency.to_nsec();
  99               c->second++;
 100             }
 101             return;
 102           case PerformanceCounterType::READ_LATENCY:
 103             if (op.may_read()) {
 104               c->first += latency.to_nsec();
 105               c->second++;
 106             }
 107             return;
 108           default:
 109             ceph_abort_msg("unknown counter type");
 110           }
 111         };
 112
 113     auto get_subkey_fnc =
 114         [&osd, &pg_info, &op](const OSDPerfMetricSubKeyDescriptor &d,
 115                               OSDPerfMetricSubKey *sub_key) {
 116           ceph_assert(d.is_supported());
 117
 118           auto m = op.get_req<MOSDOp>();
 119           std::string match_string;
 120           switch(d.type) {
 121           case OSDPerfMetricSubKeyType::CLIENT_ID:
 122             match_string = stringify(m->get_reqid().name);
 123             break;
 124           case OSDPerfMetricSubKeyType::CLIENT_ADDRESS:
 125             match_string = stringify(m->get_connection()->get_peer_addr());
 126             break;
 127           case OSDPerfMetricSubKeyType::POOL_ID:
 128             match_string = stringify(m->get_spg().pool());
 129             break;
 130           case OSDPerfMetricSubKeyType::NAMESPACE:
 131             match_string = m->get_hobj().nspace;
 132             break;
 133           case OSDPerfMetricSubKeyType::OSD_ID:
 134             match_string = stringify(osd->get_nodeid());
 135             break;
 136           case OSDPerfMetricSubKeyType::PG_ID:
 137             match_string = stringify(pg_info.pgid);
 138             break;
 139           case OSDPerfMetricSubKeyType::OBJECT_NAME:
 140             match_string = m->get_oid().name;
 141             break;
 142           case OSDPerfMetricSubKeyType::SNAP_ID:
 143             match_string = stringify(m->get_snapid());
 144             break;
 145           default:
 146             ceph_abort_msg("unknown counter type");
 147           }
 148
 149           std::smatch match;
 150           if (!std::regex_search(match_string, match, d.regex)) {
 151             return false;
 152           }
 153           if (match.size() <= 1) {
 154             return false;
 155           }
 156           for (size_t i = 1; i < match.size(); i++) {
 157             sub_key->push_back(match[i].str());
 158           }
 159           return true;
 160         };
 161
 162     for (auto &it : data) {
 163       auto &query = it.first;
 164       OSDPerfMetricKey key;
 165       if (query.get_key(get_subkey_fnc, &key)) {
 166         query.update_counters(update_counter_fnc, &it.second[key]);
 167       }
 168     }
 169   }
 170
 171   void add_to_reports(
 172       const std::map<OSDPerfMetricQuery, OSDPerfMetricLimits> &limits,
 173       std::map<OSDPerfMetricQuery, OSDPerfMetricReport> *reports) {
 174     for (auto &it : data) {
 175       auto &query = it.first;
 176       auto limit_it = limits.find(query);
 177       if (limit_it == limits.end()) {
 178         continue;
 179       }
 180       auto &query_limits = limit_it->second;
 181       auto &counters = it.second;
 182       auto &report = (*reports)[query];
 183
 184       query.get_performance_counter_descriptors(
 185           &report.performance_counter_descriptors);
 186
 187       auto &descriptors = report.performance_counter_descriptors;
 188       ceph_assert(descriptors.size() > 0);
 189
 190       if (!is_limited(query_limits, counters.size())) {
 191         for (auto &it_counters : counters) {
 192           auto &bl = report.group_packed_performance_counters[it_counters.first];
 193           query.pack_counters(it_counters.second, &bl);
 194         }
 195         continue;
 196       }
 197
 198       for (auto &limit : query_limits) {
 199         size_t index = 0;
 200         for (; index < descriptors.size(); index++) {
 201           if (descriptors[index] == limit.order_by) {
 202             break;
 203           }
 204         }
 205         if (index == descriptors.size()) {
 206           // should not happen
 207           continue;
 208         }
 209
 210         // Weighted Random Sampling (Algorithm A-Chao):
 211         // Select the first [0, max_count) samples, randomly replace
 212         // with samples from [max_count, end) using weighted
 213         // probability, and return [0, max_count) as the result.
 214
 215         ceph_assert(limit.max_count < counters.size());
 216         typedef std::map<OSDPerfMetricKey, PerformanceCounters>::iterator
 217             Iterator;
 218         std::vector<Iterator> counter_iterators;
 219         counter_iterators.reserve(limit.max_count);
 220
 221         Iterator it_counters = counters.begin();
 222         uint64_t wsum = 0;
 223         for (size_t i = 0; i < limit.max_count; i++) {
 224           wsum += it_counters->second[index].first;
 225           counter_iterators.push_back(it_counters++);
 226         }
 227         for (; it_counters != counters.end(); it_counters++) {
 228           wsum += it_counters->second[index].first;
 229           if (ceph::util::generate_random_number(0, wsum) <=
 230               it_counters->second[index].first) {
 231             auto i = ceph::util::generate_random_number(0, limit.max_count - 1);
 232             counter_iterators[i] = it_counters;
 233           }
 234         }
 235
 236         for (auto it_counters : counter_iterators) {
 237           auto &bl =
 238               report.group_packed_performance_counters[it_counters->first];
 239           if (bl.length() == 0) {
 240             query.pack_counters(it_counters->second, &bl);
 241           }
 242         }
 243       }
 244     }
 245   }
 246
 247 private:
 248   static bool is_limited(const OSDPerfMetricLimits &limits,
 249                          size_t counters_size) {
 250     if (limits.empty()) {
 251       return false;
 252     }
 253
 254     for (auto &limit : limits) {
 255       if (limit.max_count >= counters_size) {
 256         return false;
 257       }
 258     }
 259
 260     return true;
 261   }
 262
 263   std::map<OSDPerfMetricQuery,
 264            std::map<OSDPerfMetricKey, PerformanceCounters>> data;
 265 };
 266
 267 #endif // DYNAMIC_PERF_STATS_H