storage/engine: log RocksDB compaction stats

pkg/storage/engine/db.cc

@@ -69,6 +69,7 @@ struct DBEngine {
   virtual DBStatus Get(DBKey key, DBString* value) = 0;
   virtual DBIterator* NewIter(rocksdb::ReadOptions*) = 0;
   virtual DBStatus GetStats(DBStatsResult* stats) = 0;
+  virtual DBString GetCompactionStats() = 0;
   virtual DBStatus EnvWriteFile(DBSlice path, DBSlice contents) = 0;
 
   DBSSTable* GetSSTables(int* n);
@@ -110,6 +111,7 @@ struct DBImpl : public DBEngine {
   virtual DBStatus Get(DBKey key, DBString* value);
   virtual DBIterator* NewIter(rocksdb::ReadOptions*);
   virtual DBStatus GetStats(DBStatsResult* stats);
+  virtual DBString GetCompactionStats();
   virtual DBStatus EnvWriteFile(DBSlice path, DBSlice contents);
 };
 
@@ -131,6 +133,7 @@ struct DBBatch : public DBEngine {
   virtual DBStatus Get(DBKey key, DBString* value);
   virtual DBIterator* NewIter(rocksdb::ReadOptions*);
   virtual DBStatus GetStats(DBStatsResult* stats);
+  virtual DBString GetCompactionStats();
   virtual DBStatus EnvWriteFile(DBSlice path, DBSlice contents);
 };
 
@@ -152,6 +155,7 @@ struct DBWriteOnlyBatch : public DBEngine {
   virtual DBStatus Get(DBKey key, DBString* value);
   virtual DBIterator* NewIter(rocksdb::ReadOptions*);
   virtual DBStatus GetStats(DBStatsResult* stats);
+  virtual DBString GetCompactionStats();
   virtual DBStatus EnvWriteFile(DBSlice path, DBSlice contents);
 };
 
@@ -175,6 +179,7 @@ struct DBSnapshot : public DBEngine {
   virtual DBStatus Get(DBKey key, DBString* value);
   virtual DBIterator* NewIter(rocksdb::ReadOptions*);
   virtual DBStatus GetStats(DBStatsResult* stats);
+  virtual DBString GetCompactionStats();
   virtual DBStatus EnvWriteFile(DBSlice path, DBSlice contents);
 };
 
@@ -2007,6 +2012,24 @@ DBStatus DBSnapshot::GetStats(DBStatsResult* stats) {
   return FmtStatus("unsupported");
 }
 
+DBString DBImpl::GetCompactionStats() {
+  std::string tmp;
+  rep->GetProperty("rocksdb.cfstats", &tmp);
+  return ToDBString(tmp);
+}
+
+DBString DBBatch::GetCompactionStats() {
+  return ToDBString("unsupported");
+}
+
+DBString DBWriteOnlyBatch::GetCompactionStats() {
+  return ToDBString("unsupported");
+}
+
+DBString DBSnapshot::GetCompactionStats() {
+  return ToDBString("unsupported");
+}
+
 // EnvWriteFile writes the given data as a new "file" in the given engine.
 DBStatus DBImpl::EnvWriteFile(DBSlice path, DBSlice contents) {
   rocksdb::Status s;
@@ -2323,6 +2346,10 @@ DBStatus DBGetStats(DBEngine* db, DBStatsResult* stats) {
   return db->GetStats(stats);
 }
 
+DBString DBGetCompactionStats(DBEngine* db) {
+  return db->GetCompactionStats();
+}
+
 DBSSTable* DBGetSSTables(DBEngine* db, int* n) {
   return db->GetSSTables(n);
 }

pkg/storage/engine/db.h

@@ -237,6 +237,7 @@ typedef struct {
 } DBStatsResult;
 
 DBStatus DBGetStats(DBEngine* db, DBStatsResult* stats);
+DBString DBGetCompactionStats(DBEngine* db);
 
 typedef struct {
   int level;

pkg/storage/engine/eventlistener.cc

@@ -29,24 +29,27 @@ void DBEventListener::OnFlushCompleted(rocksdb::DB* db, const rocksdb::FlushJobI
 
   if (kDebug) {
     const rocksdb::TableProperties &p = flush_job_info.table_properties;
-    fprintf(stderr, "OnFlushCompleted:\n  %40s:  index=%.1f  filter=%.1f\n",
-            flush_job_info.file_path.c_str(),
-            p.index_size / float(p.num_entries),
-            p.filter_size / float(p.num_entries));
+    fprintf(stderr, "OnFlushCompleted:\n  %40s:  entries=%d  data=%.1fMB  index=%.1fMB  filter=%.1fMB\n",
+            flush_job_info.file_path.c_str(), (int)p.num_entries,
+            float(p.data_size) / (1024.0 * 1024.0),
+            float(p.index_size) / (1024.0 * 1024.0),
+            float(p.filter_size) / (1024.0 * 1024.0));
   }
 }
 
 void DBEventListener::OnCompactionCompleted(rocksdb::DB* db, const rocksdb::CompactionJobInfo& ci) {
   ++compactions_;
 
   if (kDebug) {
-    fprintf(stderr, "OnCompactionCompleted:\n");
+    fprintf(stderr, "OnCompactionCompleted: input=%d output=%d\n",
+            ci.base_input_level, ci.output_level);
     for (auto iter = ci.table_properties.begin(); iter != ci.table_properties.end(); ++iter) {
       const rocksdb::TableProperties &p = *iter->second;
-      fprintf(stderr, "  %40s: index=%.1f  filter=%.1f\n",
-              iter->first.c_str(),
-              p.index_size / float(p.num_entries),
-              p.filter_size / float(p.num_entries));
+      fprintf(stderr, "  %40s: entries=%d  data=%.1fMB  index=%.1fMB  filter=%.1fMB\n",
+              iter->first.c_str(), (int)p.num_entries,
+              float(p.data_size) / (1024.0 * 1024.0),
+              float(p.index_size) / (1024.0 * 1024.0),
+              float(p.filter_size) / (1024.0 * 1024.0));
     }
   }
 }

pkg/storage/engine/rocksdb.go

@@ -829,6 +829,12 @@ func (r *RocksDB) GetStats() (*Stats, error) {
 	}, nil
 }
 
+// GetCompactionStats returns the internal RocksDB compaction stats. See
+// https://github.com/facebook/rocksdb/wiki/RocksDB-Tuning-Guide#rocksdb-statistics.
+func (r *RocksDB) GetCompactionStats() string {
+	return cStringToGoString(C.DBGetCompactionStats(r.rdb))
+}
+
 type rocksDBSnapshot struct {
 	parent *RocksDB
 	handle *C.DBEngine

pkg/storage/store.go

@@ -4142,8 +4142,9 @@ func (s *Store) ComputeMetrics(ctx context.Context, tick int) error {
 		readAmp := sstables.ReadAmplification()
 		s.metrics.RdbReadAmplification.Update(int64(readAmp))
 		// Log this metric infrequently.
-		if tick%100 == 0 {
+		if tick%60 == 0 /* every 10m */ {
 			log.Infof(ctx, "sstables (read amplification = %d):\n%s", readAmp, sstables)
+			log.Info(ctx, rocksdb.GetCompactionStats())
 		}
 	}
 	return nil