improvement: add logging on rocksdb write stalls (#529)

src/server/pegasus_event_listener.cpp

@@ -3,11 +3,14 @@
 // can be found in the LICENSE file in the root directory of this source tree.
 
 #include "pegasus_event_listener.h"
+#include "logging_utils.h"
+
+#include <dsn/c/api_utilities.h>
 
 namespace pegasus {
 namespace server {
 
-pegasus_event_listener::pegasus_event_listener()
+pegasus_event_listener::pegasus_event_listener(replica_base *r) : replica_base(r)
 {
     _pfc_recent_flush_completed_count.init_app_counter("app.pegasus",
                                                        "recent.flush.completed.count",
@@ -42,8 +45,6 @@ pegasus_event_listener::pegasus_event_listener()
         "rocksdb recent write change stopped count");
 }
 
-pegasus_event_listener::~pegasus_event_listener() {}
-
 void pegasus_event_listener::OnFlushCompleted(rocksdb::DB *db,
                                               const rocksdb::FlushJobInfo &flush_job_info)
 {
@@ -61,10 +62,13 @@ void pegasus_event_listener::OnCompactionCompleted(rocksdb::DB *db,
 
 void pegasus_event_listener::OnStallConditionsChanged(const rocksdb::WriteStallInfo &info)
 {
-    if (info.condition.cur == rocksdb::WriteStallCondition::kDelayed)
+    if (info.condition.cur == rocksdb::WriteStallCondition::kDelayed) {
+        derror_replica("rocksdb write delayed");
         _pfc_recent_write_change_delayed_count->increment();
-    else if (info.condition.cur == rocksdb::WriteStallCondition::kStopped)
+    } else if (info.condition.cur == rocksdb::WriteStallCondition::kStopped) {
+        derror_replica("rocksdb write stopped");
         _pfc_recent_write_change_stopped_count->increment();
+    }
 }
 
 } // namespace server

src/server/pegasus_event_listener.h

@@ -7,23 +7,22 @@
 #include <rocksdb/db.h>
 #include <rocksdb/listener.h>
 #include <dsn/perf_counter/perf_counter_wrapper.h>
+#include <dsn/dist/replication/replica_base.h>
 
 namespace pegasus {
 namespace server {
 
-class pegasus_event_listener : public rocksdb::EventListener
+class pegasus_event_listener : public rocksdb::EventListener, dsn::replication::replica_base
 {
 public:
-    pegasus_event_listener();
-    virtual ~pegasus_event_listener();
+    explicit pegasus_event_listener(replica_base *r);
+    ~pegasus_event_listener() override = default;
 
-    virtual void OnFlushCompleted(rocksdb::DB *db,
-                                  const rocksdb::FlushJobInfo &flush_job_info) override;
+    void OnFlushCompleted(rocksdb::DB *db, const rocksdb::FlushJobInfo &flush_job_info) override;
 
-    virtual void OnCompactionCompleted(rocksdb::DB *db,
-                                       const rocksdb::CompactionJobInfo &ci) override;
+    void OnCompactionCompleted(rocksdb::DB *db, const rocksdb::CompactionJobInfo &ci) override;
 
-    virtual void OnStallConditionsChanged(const rocksdb::WriteStallInfo &info) override;
+    void OnStallConditionsChanged(const rocksdb::WriteStallInfo &info) override;
 
 private:
     ::dsn::perf_counter_wrapper _pfc_recent_flush_completed_count;

src/server/pegasus_server_impl_init.cpp

@@ -122,7 +122,7 @@ pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
     _statistics->set_stats_level(rocksdb::kExceptDetailedTimers);
     _db_opts.statistics = _statistics;
 
-    _db_opts.listeners.emplace_back(new pegasus_event_listener());
+    _db_opts.listeners.emplace_back(new pegasus_event_listener(this));
 
     // flush threads are shared among all rocksdb instances in one process.
     _db_opts.max_background_flushes =