Merge pull request ceph#61196 from xxhdx1985126/wip-crimson-mgr-dynamic-perf-report

crimson/osd: support mgr's dynamic perf stats

Reviewed-by: Samuel Just <[email protected]>

Author: Matan-B

src/crimson/mgr/client.cc

@@ -26,10 +26,14 @@ namespace crimson::mgr
 {
 
 Client::Client(crimson::net::Messenger& msgr,
-                 WithStats& with_stats)
+	       WithStats& with_stats,
+	       set_perf_queries_cb_t cb_set,
+	       get_perf_report_cb_t cb_get)
   : msgr{msgr},
     with_stats{with_stats},
-    report_timer{[this] {report();}}
+    report_timer{[this] {report();}},
+    set_perf_queries_cb(cb_set),
+    get_perf_report_cb(cb_get)
 {}
 
 seastar::future<> Client::start()
@@ -152,6 +156,10 @@ seastar::future<> Client::handle_mgr_conf(crimson::net::ConnectionRef,
   } else {
     report_timer.cancel();
   }
+  if (!m->osd_perf_metric_queries.empty()) {
+    ceph_assert(set_perf_queries_cb);
+    return set_perf_queries_cb(m->osd_perf_metric_queries);
+  }
   return seastar::now();
 }
 
@@ -202,6 +210,13 @@ void Client::_send_report()
     report->daemon_health_metrics = std::move(daemon_health_metrics);
     local_conf().get_config_bl(last_config_bl_version, &report->config_bl,
 	                      &last_config_bl_version);
+    if (get_perf_report_cb) {
+      return get_perf_report_cb(
+      ).then([report=std::move(report), this](auto payload) mutable {
+	report->metric_report_message = MetricReportMessage(std::move(payload));
+	return conn->send(std::move(report));
+      });
+    }
     return conn->send(std::move(report));
   });
 }

src/crimson/mgr/client.h

@@ -10,6 +10,7 @@
 #include "crimson/net/Fwd.h"
 #include "mgr/DaemonHealthMetric.h"
 #include "mon/MgrMap.h"
+#include "mgr/MetricTypes.h"
 
 template<typename Message> using Ref = boost::intrusive_ptr<Message>;
 namespace crimson::net {
@@ -30,9 +31,14 @@ class WithStats {
 };
 
 class Client : public crimson::net::Dispatcher {
+  using get_perf_report_cb_t = std::function<seastar::future<MetricPayload> ()>;
+  using set_perf_queries_cb_t =
+    std::function<seastar::future<> (const ConfigPayload &)>;
 public:
   Client(crimson::net::Messenger& msgr,
-	 WithStats& with_stats);
+	 WithStats& with_stats,
+	 set_perf_queries_cb_t cb_set,
+	 get_perf_report_cb_t cb_get);
   seastar::future<> start();
   seastar::future<> stop();
   void report();
@@ -60,6 +66,8 @@ class Client : public crimson::net::Dispatcher {
   crimson::common::gate_per_shard gates;
   uint64_t last_config_bl_version = 0;
   std::string service_name, daemon_name;
+  set_perf_queries_cb_t set_perf_queries_cb;
+  get_perf_report_cb_t get_perf_report_cb;
 
   std::vector<DaemonHealthMetric> daemon_health_metrics;
 

src/crimson/osd/CMakeLists.txt

@@ -56,6 +56,7 @@ add_executable(crimson-osd
   ${PROJECT_SOURCE_DIR}/src/osd/SnapMapper.cc
   ${PROJECT_SOURCE_DIR}/src/osd/recovery_types.cc
   ${PROJECT_SOURCE_DIR}/src/osd/osd_perf_counters.cc
+  ${PROJECT_SOURCE_DIR}/src/mgr/OSDPerfMetricTypes.cc
   watch.cc
   )
 if(HAS_VTA)

src/crimson/osd/ops_executer.h

@@ -169,6 +169,10 @@ class OpsExecuter {
       IOInterruptCondition, osd_op_errorator>;
 
   object_stat_sum_t delta_stats;
+
+  size_t get_bytes_written() {
+    return txn.get_num_bytes();
+  }
 private:
   // with_effect can be used to schedule operations to be performed
   // at commit time.  effects will be discarded if the operation does

src/crimson/osd/osd.cc

@@ -95,7 +95,16 @@ OSD::OSD(int id, uint32_t nonce,
     hb_front_msgr{hb_front_msgr},
     hb_back_msgr{hb_back_msgr},
     monc{new crimson::mon::Client{*public_msgr, *this}},
-    mgrc{new crimson::mgr::Client{*public_msgr, *this}},
+    mgrc{new crimson::mgr::Client{
+      *public_msgr,
+      *this,
+      [this](const ConfigPayload &config_payload) {
+	return set_perf_queries(config_payload);
+      },
+      [this] {
+	return get_perf_reports();
+      }
+    }},
     store{store},
     pg_shard_manager{osd_singleton_state,
                      shard_services,
@@ -176,6 +185,70 @@ seastar::future<> OSD::open_meta_coll()
   });
 }
 
+seastar::future<> OSD::set_perf_queries(const ConfigPayload &config_payload) {
+  LOG_PREFIX(OSD::set_perf_queries);
+  const OSDConfigPayload &osd_config_payload =
+    boost::get<OSDConfigPayload>(config_payload);
+  const std::map<OSDPerfMetricQuery, OSDPerfMetricLimits> &queries =
+    osd_config_payload.config;
+  DEBUG("setting {} queries", queries.size());
+
+  std::list<OSDPerfMetricQuery> supported_queries;
+  for (auto &it : queries) {
+    auto &query = it.first;
+    if (!query.key_descriptor.empty()) {
+      supported_queries.push_back(query);
+    }
+  }
+  if (supported_queries.size() < queries.size()) {
+    DEBUG("{} unsupported queries", queries.size() - supported_queries.size());
+  }
+
+  return shard_services.invoke_on_all(
+    [supported_queries, queries](auto &local_service) {
+    auto &pgs = local_service.local_state.pg_map.get_pgs();
+    local_service.local_state.m_perf_queries = supported_queries;
+    local_service.local_state.m_perf_limits = queries;
+    for (auto &[id, pg] : pgs) {
+      pg->set_dynamic_perf_stats_queries(supported_queries);
+    }
+  });
+}
+
+seastar::future<MetricPayload> OSD::get_perf_reports() {
+  LOG_PREFIX(OSD::get_perf_reports);
+  OSDMetricPayload payload;
+  std::map<OSDPerfMetricQuery, OSDPerfMetricReport> &reports = payload.report;
+
+  auto dps = co_await shard_services.map_reduce0(
+    [FNAME](auto &local_service) {
+      auto &pgs = local_service.local_state.pg_map.get_pgs();
+      auto &m_perf_queries = local_service.local_state.m_perf_queries;
+      DynamicPerfStats dps;
+      for (auto &[id, pg] : pgs) {
+	// m_perf_queries can be modified only in set_perf_queries by mgr client
+	// request, and it is protected by by mgr client's lock, which is held
+	// when set_perf_queries/get_perf_reports are called, so we may not hold
+	// m_perf_queries_lock here.
+	DynamicPerfStats pg_dps(m_perf_queries);
+	pg->get_dynamic_perf_stats(&pg_dps);
+	dps.merge(pg_dps);
+	DEBUG("reporting for pg {}", pg->get_pgid());
+      }
+      return dps;
+    },
+    DynamicPerfStats(shard_services.local().local_state.m_perf_queries),
+    [](auto left, auto right) {
+      left.merge(right);
+      return left;
+    });
+
+  dps.add_to_reports(shard_services.local().local_state.m_perf_limits, &reports);
+  DEBUG("reports for {} queries", reports.size());
+
+  co_return payload;
+}
+
 seastar::future<OSDMeta> OSD::open_or_create_meta_coll(FuturizedStore &store)
 {
   return store.get_sharded_store().open_collection(coll_t::meta()).then([&store](auto ch) {

src/crimson/osd/osd.h

@@ -236,6 +236,8 @@ class OSD final : public crimson::net::Dispatcher,
 
   std::vector<DaemonHealthMetric> get_health_metrics();
 
+  seastar::future<> set_perf_queries(const ConfigPayload &config_payload);
+  seastar::future<MetricPayload> get_perf_reports();
 private:
   crimson::common::gate_per_shard gate;
 

src/crimson/osd/osd_operations/client_request.cc

@@ -63,6 +63,7 @@ ClientRequest::ClientRequest(
   : shard_services(&_shard_services),
     l_conn(std::move(conn)),
     m(std::move(m)),
+    begin_time(std::chrono::steady_clock::now()),
     instance_handle(new instance_handle_t)
 {}
 
@@ -525,6 +526,7 @@ ClientRequest::do_process(
     co_return;
   }
 
+  size_t inb = 0, outb = 0;
   {
     auto all_completed = interruptor::now();
     if (ret) {
@@ -540,6 +542,7 @@ ClientRequest::do_process(
       // simply return the error below, leaving all_completed alone
     } else {
       auto submitted = interruptor::now();
+      inb = ox.get_bytes_written();
       std::tie(submitted, all_completed) = co_await pg->submit_executer(
 	std::move(ox), m->ops);
       co_await std::move(submitted);
@@ -573,9 +576,15 @@ ClientRequest::do_process(
 	  reply->set_result(osdop.rval);
 	  break;
 	}
+        outb += osdop.outdata.length();
       }
     }
 
+    pg->add_client_request_lat(
+      *this,
+      inb,
+      outb,
+      utime_t{std::chrono::steady_clock::now() - begin_time});
     reply->set_enoent_reply_versions(
       pg->peering_state.get_info().last_update,
       pg->peering_state.get_info().last_user_version);

src/crimson/osd/osd_operations/client_request.h

@@ -41,12 +41,25 @@ class ClientRequest final : public PhasedOperationT<ClientRequest>,
   OpInfo op_info;
   seastar::promise<> on_complete;
   unsigned instance_id = 0;
+  std::chrono::time_point<std::chrono::steady_clock> begin_time;
 
 public:
   epoch_t get_epoch_sent_at() const {
     return m->get_map_epoch();
   }
 
+  bool may_write() const { return op_info.may_write(); }
+  bool may_cache() const { return op_info.may_cache(); }
+  bool may_read() const { return op_info.may_read(); }
+  template <typename T>
+  T* get_req() const {
+    static_assert(std::is_same_v<T, MOSDOp>);
+    return m.get();
+  }
+  const crimson::net::ConnectionRef &get_connection() const {
+    return l_conn;
+  }
+
   /**
    * instance_handle_t
    *

src/crimson/osd/pg.h

@@ -21,6 +21,7 @@
 #include "crimson/osd/object_context.h"
 #include "osd/PeeringState.h"
 #include "osd/SnapMapper.h"
+#include "osd/DynamicPerfStats.h"
 
 #include "crimson/common/interruptible_future.h"
 #include "crimson/common/log.h"
@@ -764,8 +765,25 @@ class PG : public boost::intrusive_ref_counter<
 
 private:
   std::optional<pg_stat_t> pg_stats;
+  DynamicPerfStats dp_stats;
 
 public:
+  void add_client_request_lat(
+    const ClientRequest& req,
+    size_t inb,
+    size_t outb,
+    const utime_t &lat) {
+    if (dp_stats.is_enabled()) {
+      dp_stats.add(pg_whoami.osd, get_info(), req, inb, outb, lat);
+    }
+  }
+  void set_dynamic_perf_stats_queries(
+    const std::list<OSDPerfMetricQuery> &queries) {
+    dp_stats.set_queries(queries);
+  }
+  void get_dynamic_perf_stats(DynamicPerfStats *stats) {
+    std::swap(dp_stats, *stats);
+  }
   OSDriver &get_osdriver() final {
     return osdriver;
   }

src/crimson/osd/shard_services.h

@@ -25,6 +25,7 @@
 #include "crimson/osd/state.h"
 #include "common/AsyncReserver.h"
 #include "crimson/net/Connection.h"
+#include "mgr/OSDPerfMetricTypes.h"
 
 namespace crimson::net {
   class Messenger;
@@ -197,6 +198,8 @@ class PerShardState {
   }
 
   OSDSuperblock per_shard_superblock;
+  std::list<OSDPerfMetricQuery> m_perf_queries;
+  std::map<OSDPerfMetricQuery, OSDPerfMetricLimits> m_perf_limits;
 
 public:
   PerShardState(