osd/ReplicatedPG: handle misdirected do_command

We can get a query on a pg we still have but are no longer primary for.  If
that happens, do not reply.  The client will resend to the correct OSD
assuming it has the map.  Send them the latest incremental so that we know
they know there is something new.  We don't know the exact epoch they have,
unfortunately, because MCommand doesn't include it, but a newer inc is
enough to make them request the right incrementals from a mon.  Eventually
they will figure it out and Objecter will resend the request to the
correct target.

It is possible we should include epoch in the MCommand message so that we
can do this mapping "correctly" (as in, the same way MOSDOp does).  That
makes MCommand less general, though... a PG-specific command message might
be the most precise thing.  Another day...

Fixes: #8085
Signed-off-by: Sage Weil <sage@inktank.com>

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index cde9baab686a1158ecf64bc28172ca66c8c5d14b..986eef51b6e19d23c8dc89a1d54268e2361398a2 100644 (file)
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -4279,10 +4279,23 @@ void OSD::do_command(Connection *con, ceph_tid_t tid, vector<string>& cmd, buffe
          _have_pg(pcand)) {
        PG *pg = _lookup_lock_pg(pcand);
        assert(pg);
-       // simulate pg <pgid> cmd= for pg->do-command
-       if (prefix != "pg")
-         cmd_putval(cct, cmdmap, "cmd", prefix);
-       r = pg->do_command(cmdmap, ss, data, odata);
+       if (pg->is_primary()) {
+         // simulate pg <pgid> cmd= for pg->do-command
+         if (prefix != "pg")
+           cmd_putval(cct, cmdmap, "cmd", prefix);
+         r = pg->do_command(cmdmap, ss, data, odata);
+       } else {
+         ss << "not primary for pgid " << pgid;
+
+         // send them the latest diff to ensure they realize the mapping
+         // has changed.
+         send_incremental_map(osdmap->get_epoch() - 1, con);
+
+         // do not reply; they will get newer maps and realize they
+         // need to resend.
+         pg->unlock();
+         return;
+       }
        pg->unlock();
       } else {
        ss << "i don't have pgid " << pgid;