Merge pull request ceph#56446 from guojidan/metrics

crimson/mgr/client.cc: daemon_health_metrics support

Reviewed-by: Matan Breizman <[email protected]>

Author: Matan-B

src/crimson/mgr/client.cc

@@ -174,10 +174,13 @@ void Client::report()
   });
 }
 
+void Client::update_daemon_health(std::vector<DaemonHealthMetric>&& metrics)
+{
+  daemon_health_metrics = std::move(metrics);
+}
+
 void Client::_send_report()
 {
-  // TODO: implement daemon_health_metrics support
-  // https://tracker.ceph.com/issues/63766
   gates.dispatch_in_background(__func__, *this, [this] {
     if (!conn) {
       logger().warn("cannot send report; no conn available");
@@ -196,6 +199,7 @@ void Client::_send_report()
       report->daemon_name = local_conf()->name.get_id();
     }
     report->service_name = service_name;
+    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);
     return conn->send(std::move(report));

src/crimson/mgr/client.h

@@ -8,6 +8,7 @@
 #include "crimson/common/gated.h"
 #include "crimson/net/Dispatcher.h"
 #include "crimson/net/Fwd.h"
+#include "mgr/DaemonHealthMetric.h"
 #include "mon/MgrMap.h"
 
 template<typename Message> using Ref = boost::intrusive_ptr<Message>;
@@ -35,6 +36,7 @@ class Client : public crimson::net::Dispatcher {
   seastar::future<> start();
   seastar::future<> stop();
   void report();
+  void update_daemon_health(std::vector<DaemonHealthMetric>&& metrics);
 
 private:
   std::optional<seastar::future<>> ms_dispatch(
@@ -59,6 +61,8 @@ class Client : public crimson::net::Dispatcher {
   uint64_t last_config_bl_version = 0;
   std::string service_name, daemon_name;
 
+  std::vector<DaemonHealthMetric> daemon_health_metrics;
+
   void _send_report();
 };
 

src/crimson/osd/osd.cc

@@ -105,6 +105,7 @@ OSD::OSD(int id, uint32_t nonce,
       std::ignore = update_heartbeat_peers(
       ).then([this] {
 	update_stats();
+        mgrc->update_daemon_health(get_health_metrics());
 	tick_timer.arm(
 	  std::chrono::seconds(TICK_INTERVAL));
       });
@@ -976,7 +977,7 @@ void OSD::handle_conf_change(
   const crimson::common::ConfigProxy& conf,
   const std::set <std::string> &changed)
 {
-  if (changed.count("osd_beacon_report_interval")) {
+  if (changed.contains("osd_beacon_report_interval")) {
     beacon_timer.rearm_periodic(
       std::chrono::seconds(conf->osd_beacon_report_interval));
   }
@@ -1415,6 +1416,74 @@ seastar::future<> OSD::handle_recovery_subreq(
     conn, std::move(m)).second;
 }
 
+vector<DaemonHealthMetric> OSD::get_health_metrics()
+{
+  LOG_PREFIX(OSD::get_health_metrics);
+  vector<DaemonHealthMetric> metrics;
+
+  const utime_t now = ceph_clock_now();
+  utime_t oldest_secs = now;
+  utime_t too_old = now;
+  too_old -= local_conf()->osd_op_complaint_time;
+  int slow = 0;
+  ClientRequest::ICRef oldest_op;
+  map<uint64_t, int> slow_op_pools;
+  bool log_aggregated_slow_op = local_conf()->osd_aggregated_slow_ops_logging;
+  auto count_slow_ops = [&](const ClientRequest& op) {
+    if (op.get_started() < too_old) {
+      std::stringstream ss;
+      ss << "slow request ";
+      op.print(ss);
+      ss << " initiated "
+         << op.get_started();
+      WARN("{}", ss.str());
+      if (log_aggregated_slow_op) {
+        uint64_t pool_id = op.get_pgid().pgid.m_pool;
+        if (pool_id > 0 && pool_id <= (uint64_t) osdmap->get_pool_max()) {
+          slow_op_pools[pool_id]++;
+        }
+      } else {
+        clog->warn() << ss.str();
+      }
+      ++slow;
+      if (!oldest_op || op.get_started() < oldest_op->get_started()) {
+        oldest_op = &op;
+      }
+    }
+  };
+
+  auto& op_registry = get_shard_services().get_registry();
+  op_registry.visit_ops_in_flight(count_slow_ops);
+  if (slow) {
+    std::stringstream ss;
+    ss << __func__ << " reporting " << slow << " slow ops, oldest is ";
+    ceph_assert(oldest_op);
+    oldest_op->print(ss);
+    ERROR("{}", ss.str());
+    if (log_aggregated_slow_op && !slow_op_pools.empty()) {
+      std::stringstream ss;
+      auto slow_pool_it = std::max_element(slow_op_pools.begin(), slow_op_pools.end(),
+                             [](std::pair<uint64_t, int> p1, std::pair<uint64_t, int> p2) {
+                               return p1.second < p2.second;
+                             });
+      if (osdmap->get_pools().find(slow_pool_it->first) != osdmap->get_pools().end()) {
+        string pool_name = osdmap->get_pool_name(slow_pool_it->first);
+        ss << "slow requests (most affected pool [ '"
+           << pool_name
+           << "' : "
+           << slow_pool_it->second
+           << " ])";
+      }
+      WARN("{}", ss.str());
+      clog->warn() << ss.str();
+    }
+    oldest_secs = now - oldest_op->get_started();
+    metrics.emplace_back(daemon_metric::SLOW_OPS, slow, oldest_secs);
+  }
+
+  return metrics;
+}
+
 bool OSD::should_restart() const
 {
   LOG_PREFIX(OSD::should_restart);

src/crimson/osd/osd.h

@@ -234,6 +234,8 @@ class OSD final : public crimson::net::Dispatcher,
     crimson::net::ConnectionRef conn,
     Ref<MOSDPGUpdateLogMissingReply> m);
 
+  std::vector<DaemonHealthMetric> get_health_metrics();
+
 private:
   crimson::common::gate_per_shard gate;
 

src/crimson/osd/osd_operation.cc

@@ -136,6 +136,17 @@ size_t OSDOperationRegistry::dump_slowest_historic_client_requests(ceph::Formatt
   return ops_count;
 }
 
+void OSDOperationRegistry::visit_ops_in_flight(std::function<void(const ClientRequest&)>&& visit)
+{
+  const auto& client_registry =
+    get_registry<static_cast<size_t>(OperationTypeCode::client_request)>();
+  auto it = std::begin(client_registry);
+  for (; it != std::end(client_registry); ++it) {
+    const auto& fastest_historic_op = static_cast<const ClientRequest&>(*it);
+    visit(fastest_historic_op);
+  }
+}
+
 OperationThrottler::OperationThrottler(ConfigProxy &conf)
   : scheduler(crimson::osd::scheduler::make_scheduler(conf))
 {

src/crimson/osd/osd_operation.h

@@ -280,6 +280,7 @@ struct OSDOperationRegistry : OperationRegistryT<
 
   size_t dump_historic_client_requests(ceph::Formatter* f) const;
   size_t dump_slowest_historic_client_requests(ceph::Formatter* f) const;
+  void visit_ops_in_flight(std::function<void(const ClientRequest&)>&& visit);
 
 private:
   size_t num_recent_ops = 0;

src/crimson/osd/osd_operations/client_request.cc

@@ -450,6 +450,9 @@ ClientRequest::do_process(
       co_return;
     }
   }
+
+  co_await maybe_inject_delay();
+
   if (m->get_oid().name.size()
     > crimson::common::local_conf()->osd_max_object_name_len) {
     co_await reply_op_error(pg, -ENAMETOOLONG);

src/crimson/osd/osd_operations/client_request.h

@@ -4,6 +4,7 @@
 #pragma once
 
 #include <seastar/core/future.hh>
+#include <seastar/core/sleep.hh>
 
 #include <boost/intrusive/list.hpp>
 #include <boost/intrusive_ptr.hpp>
@@ -308,6 +309,18 @@ class ClientRequest final : public PhasedOperationT<ClientRequest>,
   };
 
   void put_historic() const;
+  static interruptible_future<> maybe_inject_delay() {
+    if (common::local_conf()->osd_debug_inject_dispatch_delay_probability > 0) {
+      if (rand() % 10000 <
+        common::local_conf()->osd_debug_inject_dispatch_delay_probability * 10000) {
+        auto delay_duration = std::chrono::duration<double>(
+          common::local_conf()->osd_debug_inject_dispatch_delay_duration);
+        auto a_while = std::chrono::duration_cast<std::chrono::seconds>(delay_duration);
+        return seastar::sleep(a_while);
+      }
+    }
+    return seastar::now();
+  }
 };
 
 }