PGMap: maintain a mapping of osd to epoch of most recent stat message

Each up osd will have a mapping since out osds are now included as
empty stats.

Signed-off-by: Samuel Just <sam.just@inktank.com>

2 files changed, 43 insertions, 2 deletions

diff --git a/src/mon/PGMap.cc b/src/mon/PGMap.cc
index 9865fe5f2d7..2cbf1a6c2fb 100644
--- a/src/mon/PGMap.cc
+++ b/src/mon/PGMap.cc
@@ -30,7 +30,7 @@ void PGMap::Incremental::encode(bufferlist &bl, uint64_t features) const
     return;
   }
 
-  ENCODE_START(6, 5, bl);
+  ENCODE_START(7, 5, bl);
   ::encode(version, bl);
   ::encode(pg_stat_updates, bl);
   ::encode(osd_stat_updates, bl);
@@ -41,6 +41,7 @@ void PGMap::Incremental::encode(bufferlist &bl, uint64_t features) const
   ::encode(nearfull_ratio, bl);
   ::encode(pg_remove, bl);
   ::encode(stamp, bl);
+  ::encode(osd_epochs, bl);
   ENCODE_FINISH(bl);
 }
 
@@ -89,6 +90,17 @@ void PGMap::Incremental::decode(bufferlist::iterator &bl)
   }
   if (struct_v >= 6)
     ::decode(stamp, bl);
+  if (struct_v >= 7) {
+    ::decode(osd_epochs, bl);
+  } else {
+    for (map<int32_t, osd_stat_t>::iterator i = osd_stat_updates.begin();
+	 i != osd_stat_updates.end();
+	 ++i) {
+      // This isn't accurate, but will cause trimming to behave like
+      // previously.
+      osd_epochs.insert(make_pair(i->first, osdmap_epoch));
+    }
+  }
   DECODE_FINISH(bl);
 }
 
@@ -211,6 +223,8 @@ void PGMap::apply_incremental(CephContext *cct, const Incremental& inc)
       stat_osd_sub(t->second);
       t->second = new_stats;
     }
+    assert(inc.get_osd_epochs().find(osd) != inc.get_osd_epochs().end());
+    osd_epochs.insert(*(inc.get_osd_epochs().find(osd)));
 
     stat_osd_add(new_stats);
     
@@ -436,7 +450,7 @@ void PGMap::encode(bufferlist &bl, uint64_t features) const
     return;
   }
 
-  ENCODE_START(5, 4, bl);
+  ENCODE_START(6, 4, bl);
   ::encode(version, bl);
   ::encode(pg_stat, bl);
   ::encode(osd_stat, bl);
@@ -445,6 +459,7 @@ void PGMap::encode(bufferlist &bl, uint64_t features) const
   ::encode(full_ratio, bl);
   ::encode(nearfull_ratio, bl);
   ::encode(stamp, bl);
+  ::encode(osd_epochs, bl);
   ENCODE_FINISH(bl);
 }
 
@@ -474,6 +489,17 @@ void PGMap::decode(bufferlist::iterator &bl)
   }
   if (struct_v >= 5)
     ::decode(stamp, bl);
+  if (struct_v >= 6) {
+    ::decode(osd_epochs, bl);
+  } else {
+    for (hash_map<int32_t, osd_stat_t>::iterator i = osd_stat.begin();
+	 i != osd_stat.end();
+	 ++i) {
+      // This isn't accurate, but will cause trimming to behave like
+      // previously.
+      osd_epochs.insert(make_pair(i->first, last_osdmap_epoch));
+    }
+  }
   DECODE_FINISH(bl);
 
   calc_stats();
diff --git a/src/mon/PGMap.h b/src/mon/PGMap.h
index 99e46d13453..b117196b415 100644
--- a/src/mon/PGMap.h
+++ b/src/mon/PGMap.h
@@ -43,6 +43,9 @@ public:
   float full_ratio;
   float nearfull_ratio;
 
+  // mapping of osd to most recently reported osdmap epoch
+  hash_map<int32_t,epoch_t> osd_epochs;
+
   class Incremental {
   public:
     version_t version;
@@ -57,16 +60,27 @@ public:
   private:
     map<int32_t,osd_stat_t> osd_stat_updates;
     set<int32_t> osd_stat_rm;
+
+    // mapping of osd to most recently reported osdmap epoch
+    map<int32_t,epoch_t> osd_epochs;
   public:
+
     const map<int32_t, osd_stat_t> &get_osd_stat_updates() const {
       return osd_stat_updates;
     }
     const set<int32_t> &get_osd_stat_rm() const {
       return osd_stat_rm;
     }
+    const map<int32_t, epoch_t> &get_osd_epochs() const {
+      return osd_epochs;
+    }
+
     void add_stat(int32_t osd, epoch_t epoch, const osd_stat_t &stat) {
       assert(osd_stat_rm.find(osd) == osd_stat_rm.end());
       osd_stat_updates.insert(make_pair(osd, stat));
+      osd_epochs.insert(make_pair(osd, epoch));
+      assert(osd_epochs.size() == osd_stat_updates.size());
+    }
     void stat_osd_out(int32_t osd) {
       // 0 the stats for the osd
       osd_stat_updates.erase(osd);
@@ -74,6 +88,7 @@ public:
     }
     void rm_stat(int32_t osd) {
       osd_stat_rm.insert(osd);
+      osd_epochs.erase(osd);
       osd_stat_updates.erase(osd);
     }
     void encode(bufferlist &bl, uint64_t features=-1) const;