Merge pull request ceph#62028 from Matan-B/wip-matanb-crimson-mgrc-con

crimson/mgr/client: fix _send_report empty conn usage

Reviewed-by: Yingxin Cheng <[email protected]>

Author: Matan-B

src/crimson/mgr/client.cc

@@ -4,21 +4,18 @@
 #include "client.h"
 
 #include <seastar/core/sleep.hh>
+#include <seastar/util/defer.hh>
 
 #include "crimson/common/log.h"
 #include "crimson/net/Connection.h"
 #include "crimson/net/Messenger.h"
+#include "crimson/common/coroutine.h"
 #include "messages/MMgrConfigure.h"
 #include "messages/MMgrMap.h"
 #include "messages/MMgrOpen.h"
 #include "messages/MMgrReport.h"
 
-namespace {
-  seastar::logger& logger()
-  {
-    return crimson::get_logger(ceph_subsys_mgrc);
-  }
-}
+SET_SUBSYS(mgrc);
 
 using crimson::common::local_conf;
 
@@ -38,25 +35,51 @@ Client::Client(crimson::net::Messenger& msgr,
 
 seastar::future<> Client::start()
 {
-  return seastar::now();
+  LOG_PREFIX(Client::start);
+  DEBUGDPP("", *this);
+  co_return;
 }
 
 seastar::future<> Client::stop()
 {
-  logger().info("{}", __func__);
+  LOG_PREFIX(Client::stop);
+  DEBUGDPP("", *this);
   report_timer.cancel();
-  auto fut = gates.close_all();
   if (conn) {
+    DEBUGDPP("marking down", *this);
     conn->mark_down();
   }
-  return fut;
+  co_await gates.close_all();
+}
+
+seastar::future<> Client::send(MessageURef msg)
+{
+  LOG_PREFIX(Client::send);
+  DEBUGDPP("{}", *this, *msg);
+  if (!conn_lock.try_lock_shared()) {
+    WARNDPP("ongoing reconnect, report skipped", *this, *msg);
+    co_return;
+  }
+  auto unlocker = seastar::defer([this] {
+    conn_lock.unlock_shared();
+  });
+  if (!conn) {
+    WARNDPP("no conn available, report skipped", *this, *msg);
+    co_return;
+  }
+  DEBUGDPP("sending {}", *this, *msg);
+  co_await conn->send(std::move(msg));
 }
 
 std::optional<seastar::future<>>
 Client::ms_dispatch(crimson::net::ConnectionRef conn, MessageRef m)
 {
+  LOG_PREFIX(Client::ms_dispatch);
+  DEBUGDPP("{}", *this, *m);
   bool dispatched = true;
-  gates.dispatch_in_background(__func__, *this, [this, conn, &m, &dispatched] {
+  gates.dispatch_in_background(__func__, *this,
+  [this, conn, &m, &dispatched, FNAME] {
+    DEBUGDPP("dispatching in background {}", *this, *m);
     switch(m->get_type()) {
     case MSG_MGR_MAP:
       return handle_mgr_map(conn, boost::static_pointer_cast<MMgrMap>(m));
@@ -74,24 +97,33 @@ void Client::ms_handle_connect(
     crimson::net::ConnectionRef c,
     seastar::shard_id prv_shard)
 {
+  LOG_PREFIX(Client::ms_handle_connect);
+  DEBUGDPP("prev_shard: {}", *this, prv_shard);
   ceph_assert_always(prv_shard == seastar::this_shard_id());
-  gates.dispatch_in_background(__func__, *this, [this, c] {
+  gates.dispatch_in_background(__func__, *this,
+  [this, c, FNAME] {
     if (conn == c) {
+      DEBUGDPP("dispatching in background", *this);
       // ask for the mgrconfigure message
       auto m = crimson::make_message<MMgrOpen>();
       m->daemon_name = local_conf()->name.get_id();
       local_conf().get_config_bl(0, &m->config_bl, &last_config_bl_version);
       local_conf().get_defaults_bl(&m->config_defaults_bl);
-      return conn->send(std::move(m));
+      return send(std::move(m));
     } else {
+      DEBUGDPP("connection changed", *this);
       return seastar::now();
     }
   });
 }
 
 void Client::ms_handle_reset(crimson::net::ConnectionRef c, bool /* is_replace */)
 {
-  gates.dispatch_in_background(__func__, *this, [this, c] {
+  LOG_PREFIX(Client::ms_handle_reset);
+  DEBUGDPP("", *this);
+  gates.dispatch_in_background(__func__, *this,
+  [this, c, FNAME] {
+    DEBUGDPP("dispatching in background", *this);
     if (conn == c) {
       report_timer.cancel();
       return reconnect();
@@ -101,50 +133,64 @@ void Client::ms_handle_reset(crimson::net::ConnectionRef c, bool /* is_replace *
   });
 }
 
+seastar::future<> Client::retry_interval()
+{
+  LOG_PREFIX(Client::retry_interval);
+  auto retry_interval = std::chrono::duration<double>(
+    local_conf().get_val<double>("mgr_connect_retry_interval"));
+  auto a_while = std::chrono::duration_cast<seastar::steady_clock_type::duration>(
+    retry_interval);
+  DEBUGDPP("reconnecting in {} seconds", *this, retry_interval);
+  co_await seastar::sleep(a_while);
+}
+
 seastar::future<> Client::reconnect()
 {
+  LOG_PREFIX(Client::reconnect);
+  DEBUGDPP("", *this);
+  co_await conn_lock.lock();
+  auto unlocker = seastar::defer([this] {
+    conn_lock.unlock();
+  });
   if (conn) {
+    DEBUGDPP("marking down", *this);
     conn->mark_down();
     conn = {};
   }
   if (!mgrmap.get_available()) {
-    logger().warn("No active mgr available yet");
-    return seastar::now();
+    WARNDPP("No active mgr available yet", *this);
+    co_return;
   }
-  auto retry_interval = std::chrono::duration<double>(
-    local_conf().get_val<double>("mgr_connect_retry_interval"));
-  auto a_while = std::chrono::duration_cast<seastar::steady_clock_type::duration>(
-    retry_interval);
-  return seastar::sleep(a_while).then([this] {
-    auto peer = mgrmap.get_active_addrs().pick_addr(msgr.get_myaddr().get_type());
-    if (peer == entity_addr_t{}) {
-      // crimson msgr only uses the first bound addr
-      logger().error("mgr.{} does not have an addr compatible with me",
-                     mgrmap.get_active_name());
-      return;
-    }
-    conn = msgr.connect(peer, CEPH_ENTITY_TYPE_MGR);
-  });
+  co_await retry_interval();
+
+  auto peer = mgrmap.get_active_addrs().pick_addr(msgr.get_myaddr().get_type());
+  if (peer == entity_addr_t{}) {
+    // crimson msgr only uses the first bound addr
+    ERRORDPP("mgr.{} does not have an addr compatible with me",
+             *this, mgrmap.get_active_name());
+    co_return;
+  }
+  conn = msgr.connect(peer, CEPH_ENTITY_TYPE_MGR);
+  DEBUGDPP("reconnected successfully", *this);
 }
 
 seastar::future<> Client::handle_mgr_map(crimson::net::ConnectionRef,
                                          Ref<MMgrMap> m)
 {
+  LOG_PREFIX(Client::handle_mgr_map);
+  DEBUGDPP("", *this);
   mgrmap = m->get_map();
-  if (!conn) {
-    return reconnect();
-  } else if (conn->get_peer_addr() !=
-             mgrmap.get_active_addrs().legacy_addr()) {
-    return reconnect();
-  } else {
-    return seastar::now();
+  if (!conn || conn->get_peer_addr() !=
+               mgrmap.get_active_addrs().legacy_addr()) {
+    co_await reconnect();
   }
 }
 
 seastar::future<> Client::handle_mgr_conf(crimson::net::ConnectionRef,
                                           Ref<MMgrConfigure> m)
 {
-  logger().info("{} {}", __func__, *m);
+  LOG_PREFIX(Client::handle_mgr_conf);
+  DEBUGDPP("{}", *this, *m);
 
   auto report_period = std::chrono::seconds{m->stats_period};
   if (report_period.count()) {
@@ -158,26 +204,20 @@ seastar::future<> Client::handle_mgr_conf(crimson::net::ConnectionRef,
   }
   if (!m->osd_perf_metric_queries.empty()) {
     ceph_assert(set_perf_queries_cb);
-    return set_perf_queries_cb(m->osd_perf_metric_queries);
+    co_await set_perf_queries_cb(m->osd_perf_metric_queries);
   }
-  return seastar::now();
 }
 
 void Client::report()
 {
+  LOG_PREFIX(Client::report);
+  DEBUGDPP("", *this);
   _send_report();
-  gates.dispatch_in_background(__func__, *this, [this] {
-    if (!conn) {
-      logger().warn("report: no conn available; report skipped");
-      return seastar::now();
-    }
+  gates.dispatch_in_background(__func__, *this, [this, FNAME] {
+    DEBUGDPP("dispatching in background", *this);
     return with_stats.get_stats(
     ).then([this](auto &&pg_stats) {
-      if (!conn) {
-        logger().warn("report: no conn available; before sending stats, report skipped");
-        return seastar::now();
-      }
-      return conn->send(std::move(pg_stats));
+      return send(std::move(pg_stats));
     });
   });
 }
@@ -189,11 +229,10 @@ void Client::update_daemon_health(std::vector<DaemonHealthMetric>&& metrics)
 
 void Client::_send_report()
 {
-  gates.dispatch_in_background(__func__, *this, [this] {
-    if (!conn) {
-      logger().warn("cannot send report; no conn available");
-      return seastar::now();
-    }
+  LOG_PREFIX(Client::_send_report);
+  DEBUGDPP("", *this);
+  gates.dispatch_in_background(__func__, *this, [this, FNAME] {
+    DEBUGDPP("dispatching in background", *this);
     auto report = make_message<MMgrReport>();
     // Adding empty information since we don't support perfcounters yet
     report->undeclare_types.emplace_back();
@@ -214,10 +253,10 @@ void Client::_send_report()
       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 send(std::move(report));
       });
     }
-    return conn->send(std::move(report));
+    return send(std::move(report));
   });
 }
 

src/crimson/mgr/client.h

@@ -4,6 +4,7 @@
 #pragma once
 
 #include <seastar/core/timer.hh>
+#include <seastar/core/shared_mutex.hh>
 
 #include "crimson/common/gated.h"
 #include "crimson/net/Dispatcher.h"
@@ -41,6 +42,7 @@ class Client : public crimson::net::Dispatcher {
 	 get_perf_report_cb_t cb_get);
   seastar::future<> start();
   seastar::future<> stop();
+  seastar::future<> send(MessageURef msg);
   void report();
   void update_daemon_health(std::vector<DaemonHealthMetric>&& metrics);
 
@@ -54,6 +56,7 @@ class Client : public crimson::net::Dispatcher {
   seastar::future<> handle_mgr_conf(crimson::net::ConnectionRef conn,
 				    Ref<MMgrConfigure> m);
   seastar::future<> reconnect();
+  seastar::future<> retry_interval();
 
   void print(std::ostream&) const;
   friend std::ostream& operator<<(std::ostream& out, const Client& client);
@@ -62,6 +65,7 @@ class Client : public crimson::net::Dispatcher {
   crimson::net::Messenger& msgr;
   WithStats& with_stats;
   crimson::net::ConnectionRef conn;
+  seastar::shared_mutex conn_lock;
   seastar::timer<seastar::lowres_clock> report_timer;
   crimson::common::gate_per_shard gates;
   uint64_t last_config_bl_version = 0;