Merge pull request ceph#58136 from xxhdx1985126/wip-66372

crimson/osd/osd: mark down connections to downed osds

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

Author: Matan-B

src/crimson/net/Messenger.h

@@ -57,6 +57,8 @@ class Messenger {
 
   virtual void set_auth_server(crimson::auth::AuthServer *) = 0;
 
+  virtual seastar::future<> mark_down(const entity_addr_t &addr) = 0;
+
   using bind_ertr = crimson::errorator<
     crimson::ct_error::address_in_use, // The address (range) is already bound
     crimson::ct_error::address_not_available

src/crimson/net/SocketMessenger.h

@@ -152,6 +152,18 @@ class SocketMessenger final : public Messenger {
   Interceptor *interceptor = nullptr;
 #endif
 
+  seastar::future<> mark_down(const entity_addr_t& a) final {
+    auto conn = lookup_conn(a);
+    if (conn) {
+      return seastar::smp::submit_to(
+	conn->get_shard_id(),
+	[conn=conn.get()] {
+	conn->mark_down();
+	return seastar::now();
+      }).then([conn] { return seastar::now(); });
+    }
+    return seastar::now();
+  }
 private:
   seastar::future<> accept(SocketFRef &&, const entity_addr_t &);
 

src/crimson/osd/osd.cc

@@ -1122,14 +1122,33 @@ seastar::future<> OSD::committed_osd_maps(
   ceph_assert(seastar::this_shard_id() == PRIMARY_CORE);
   INFO("osd.{} ({}, {})", whoami, first, last);
   // advance through the new maps
+  auto old_map = osdmap;
   return seastar::do_for_each(boost::make_counting_iterator(first),
                               boost::make_counting_iterator(last + 1),
-                              [this](epoch_t cur) {
+                              [this, old_map, FNAME](epoch_t cur) {
     return pg_shard_manager.get_local_map(
       cur
-    ).then([this](OSDMapService::local_cached_map_t&& o) {
+    ).then([this, old_map, FNAME](OSDMapService::local_cached_map_t&& o) {
       osdmap = make_local_shared_foreign(OSDMapService::local_cached_map_t(o));
-      return pg_shard_manager.update_map(std::move(o));
+      std::set<int> old_osds;
+      old_map->get_all_osds(old_osds);
+      return seastar::parallel_for_each(
+	old_osds,
+	[this, FNAME, old_map](auto &osd_id) {
+	DEBUG("osd.{}: whoami ? {}, old up ? {} , now down ? {}",
+	  osd_id, osd_id != whoami,
+	  old_map->is_up(osd_id), osdmap->is_down(osd_id));
+	if (osd_id != whoami &&
+	    old_map->is_up(osd_id) &&
+	    osdmap->is_down(osd_id)) {
+	  DEBUG("osd.{}: mark osd.{} down", whoami, osd_id);
+	  return cluster_msgr->mark_down(
+	    osdmap->get_cluster_addrs(osd_id).front());
+	}
+	return seastar::now();
+      }).then([this, o=std::move(o)]() mutable {
+	return pg_shard_manager.update_map(std::move(o));
+      });
     }).then([this] {
       if (get_shard_services().get_up_epoch() == 0 &&
 	  osdmap->is_up(whoami) &&
@@ -1170,7 +1189,9 @@ seastar::future<> OSD::committed_osd_maps(
 	return seastar::now();
       }
     }
-    return fut.then([FNAME, this] {
+    return fut.then([this] {
+      return update_heartbeat_peers();
+    }).then([FNAME, this] {
       return check_osdmap_features().then([FNAME, this] {
         // yay!
         INFO("osd.{}: committed_osd_maps: broadcasting osdmaps up"