osd: ignore old/stale heartbeat messages

If we get heartbeat messages from old epochs from peers that are not
current, drop them and mark the connection down.  Even if they are peers
we _should_ have (because we haven't gotten a notify yet to learn about
a pg we should have but don't yet) we have a newer map epoch and will learn
about them shortly, reopening the connection.

Fixes: #1107
Signed-off-by: Sage Weil <sage@newdream.net>

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index 3bd6520a6df7b868e06a1c813d4262e39c5e0d26..31c45677ddcd334803d677db70b6b53b8c9e248f 100644 (file)
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -1589,10 +1589,28 @@ void OSD::handle_osd_ping(MOSDPing *m)
   int from = m->get_source().num();
 
   bool locked = map_lock.try_get_read();
+  
+  // ignore (and mark down connection for) old messages 
+  epoch_t e = m->map_epoch;
+  if (!e)
+    e = m->peer_as_of_epoch;
+  if (e <= osdmap->get_epoch() &&
+      ((heartbeat_to.count(from) == 0 && heartbeat_from.count(from) == 0) ||
+       heartbeat_con[from] != m->get_connection())) {
+    dout(5) << "handle_osd_ping marking down peer " << m->get_source_inst()
+           << " after old message from epoch " << e
+           << " <= current " << osdmap->get_epoch() << dendl;
+    heartbeat_messenger->mark_down(m->get_connection());
+    goto out;
+  } 
 
   switch (m->op) {
   case MOSDPing::REQUEST_HEARTBEAT:
-    if (heartbeat_to.count(from) && m->peer_as_of_epoch <= heartbeat_to[from]) {
+    if (m->peer_as_of_epoch <= osdmap->get_epoch()) {
+      dout(5) << "handle_osd_ping ignoring peer " << m->get_source_inst()
+             << " request for heartbeats as_of " << m->peer_as_of_epoch
+             << " <= current " << osdmap->get_epoch() << dendl;
+    } else if (heartbeat_to.count(from) && m->peer_as_of_epoch <= heartbeat_to[from]) {
       dout(5) << "handle_osd_ping ignoring peer " << m->get_source_inst()
              << " request for heartbeats as_of " << m->peer_as_of_epoch
              << " <= current _to as_of " << heartbeat_to[from] << dendl;
@@ -1647,6 +1665,7 @@ void OSD::handle_osd_ping(MOSDPing *m)
     break;
   }
 
+out:
   if (locked) 
     map_lock.put_read();