*: update rdsn; add pegasus_event_listener; add some counters; fix run.sh;...

 *: update rdsn; add pegasus_event_listener; add some counters; fix run.sh; limit kill_test log size (#56)

rdsn @ f19ad49d

-Subproject commit e77b7bb6df1859c324a54804f4e254d5155092d7
+Subproject commit f19ad49dd37420f59f0e72f5bb70d4aaac04e3db

run.sh

@@ -897,7 +897,6 @@ function run_start_kill_test()
     done
 
     run_start_onebox -m $META_COUNT -r $REPLICA_COUNT -a $APP_NAME -p $PARTITION_COUNT
-    echo
 
     cd $ROOT
     CONFIG=config-kill-test.ini
@@ -914,20 +913,22 @@ s+@ONEBOX_RUN_PATH@+`pwd`+g" ${ROOT}/src/test/kill_test/config.ini >$CONFIG
     mkdir -p onebox/verifier && cd onebox/verifier
     ln -s -f ${DSN_ROOT}/bin/pegasus_kill_test/pegasus_kill_test
     ln -s -f ${ROOT}/$CONFIG config.ini
-    echo "./pegasus_kill_test config.ini verifier &>/dev/null &"
-    ./pegasus_kill_test config.ini verifier &>/dev/null &
+    echo "$PWD/pegasus_kill_test config.ini verifier &>/dev/null &"
+    $PWD/pegasus_kill_test config.ini verifier &>/dev/null &
+    PID=$!
+    ps -ef | grep '/pegasus_kill_test config.ini verifier' | grep "\<$PID\>"
     sleep 0.2
-    echo
     cd ${ROOT}
 
     #start killer
     mkdir -p onebox/killer && cd onebox/killer
     ln -s -f ${DSN_ROOT}/bin/pegasus_kill_test/pegasus_kill_test
     ln -s -f ${ROOT}/$CONFIG config.ini
-    echo "./pegasus_kill_test config.ini killer &>/dev/null &"
-    ./pegasus_kill_test config.ini killer &>/dev/null &
+    echo "$PWD/pegasus_kill_test config.ini killer &>/dev/null &"
+    $PWD/pegasus_kill_test config.ini killer &>/dev/null &
+    PID=$!
+    ps -ef | grep '/pegasus_kill_test config.ini killer' | grep "\<$PID\>"
     sleep 0.2
-    echo
     cd ${ROOT}
     run_list_kill_test
 }
@@ -960,7 +961,7 @@ function run_stop_kill_test()
         shift
     done
 
-    ps -ef | grep ' \./pegasus_kill_test ' | awk '{print $2}' | xargs kill &>/dev/null
+    ps -ef | grep '/pegasus_kill_test ' | awk '{print $2}' | xargs kill &>/dev/null
     run_stop_onebox
 }
 
@@ -993,7 +994,7 @@ function run_list_kill_test()
     done
     echo "------------------------------"
     run_list_onebox
-    ps -ef | grep ' \./pegasus_kill_test ' | grep -v grep
+    ps -ef | grep '/pegasus_kill_test ' | grep -v grep
     echo "------------------------------"
     echo "Server dir: ./onebox"
     echo "------------------------------"

src/server/pegasus_event_listener.cpp

+// Copyright (c) 2017, Xiaomi, Inc.  All rights reserved.
+// This source code is licensed under the Apache License Version 2.0, which
+// can be found in the LICENSE file in the root directory of this source tree.
+
+#ifndef __STDC_FORMAT_MACROS
+#define __STDC_FORMAT_MACROS
+#endif
+
+#include "pegasus_event_listener.h"
+
+namespace pegasus {
+namespace server {
+
+pegasus_event_listener::pegasus_event_listener()
+{
+    _pfc_recent_flush_completed_count.init_app_counter("app.pegasus",
+                                                       "recent.flush.completed.count",
+                                                       COUNTER_TYPE_VOLATILE_NUMBER,
+                                                       "rocksdb recent flush completed count");
+    _pfc_recent_flush_output_bytes.init_app_counter("app.pegasus",
+                                                    "recent.flush.output.bytes",
+                                                    COUNTER_TYPE_VOLATILE_NUMBER,
+                                                    "rocksdb recent flush output bytes");
+    _pfc_recent_compaction_completed_count.init_app_counter(
+        "app.pegasus",
+        "recent.compaction.completed.count",
+        COUNTER_TYPE_VOLATILE_NUMBER,
+        "rocksdb recent compaction completed count");
+    _pfc_recent_compaction_input_bytes.init_app_counter("app.pegasus",
+                                                        "recent.compaction.input.bytes",
+                                                        COUNTER_TYPE_VOLATILE_NUMBER,
+                                                        "rocksdb recent compaction input bytes");
+    _pfc_recent_compaction_output_bytes.init_app_counter("app.pegasus",
+                                                         "recent.compaction.output.bytes",
+                                                         COUNTER_TYPE_VOLATILE_NUMBER,
+                                                         "rocksdb recent compaction output bytes");
+    _pfc_recent_write_change_delayed_count.init_app_counter(
+        "app.pegasus",
+        "recent.write.change.delayed.count",
+        COUNTER_TYPE_VOLATILE_NUMBER,
+        "rocksdb recent write change delayed count");
+    _pfc_recent_write_change_stopped_count.init_app_counter(
+        "app.pegasus",
+        "recent.write.change.stopped.count",
+        COUNTER_TYPE_VOLATILE_NUMBER,
+        "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)
+{
+    _pfc_recent_flush_completed_count->increment();
+    _pfc_recent_flush_output_bytes->add(flush_job_info.table_properties.data_size);
+}
+
+void pegasus_event_listener::OnCompactionCompleted(rocksdb::DB *db,
+                                                   const rocksdb::CompactionJobInfo &ci)
+{
+    _pfc_recent_compaction_completed_count->increment();
+    _pfc_recent_compaction_input_bytes->add(ci.stats.total_input_bytes);
+    _pfc_recent_compaction_output_bytes->add(ci.stats.total_output_bytes);
+}
+
+void pegasus_event_listener::OnStallConditionsChanged(const rocksdb::WriteStallInfo &info)
+{
+    if (info.condition.cur == rocksdb::WriteStallCondition::kDelayed)
+        _pfc_recent_write_change_delayed_count->increment();
+    else if (info.condition.cur == rocksdb::WriteStallCondition::kStopped)
+        _pfc_recent_write_change_stopped_count->increment();
+}
+
+} // namespace server
+} // namespace pegasus

src/server/pegasus_event_listener.h

+// Copyright (c) 2017, Xiaomi, Inc.  All rights reserved.
+// This source code is licensed under the Apache License Version 2.0, which
+// can be found in the LICENSE file in the root directory of this source tree.
+
+#pragma once
+
+#include <rocksdb/db.h>
+#include <rocksdb/listener.h>
+#include <dsn/cpp/perf_counter_wrapper.h>
+
+namespace pegasus {
+namespace server {
+
+class pegasus_event_listener : public rocksdb::EventListener
+{
+public:
+    pegasus_event_listener();
+    virtual ~pegasus_event_listener();
+
+    virtual void OnFlushCompleted(rocksdb::DB *db,
+                                  const rocksdb::FlushJobInfo &flush_job_info) override;
+
+    virtual void OnCompactionCompleted(rocksdb::DB *db,
+                                       const rocksdb::CompactionJobInfo &ci) override;
+
+    virtual void OnStallConditionsChanged(const rocksdb::WriteStallInfo &info) override;
+
+private:
+    ::dsn::perf_counter_wrapper _pfc_recent_flush_completed_count;
+    ::dsn::perf_counter_wrapper _pfc_recent_flush_output_bytes;
+    ::dsn::perf_counter_wrapper _pfc_recent_compaction_completed_count;
+    ::dsn::perf_counter_wrapper _pfc_recent_compaction_input_bytes;
+    ::dsn::perf_counter_wrapper _pfc_recent_compaction_output_bytes;
+    ::dsn::perf_counter_wrapper _pfc_recent_write_change_delayed_count;
+    ::dsn::perf_counter_wrapper _pfc_recent_write_change_stopped_count;
+};
+
+} // namespace server
+} // namespace pegasus

src/server/pegasus_server_impl.cpp

@@ -17,6 +17,7 @@
 #include "base/pegasus_key_schema.h"
 #include "base/pegasus_value_schema.h"
 #include "base/pegasus_utils.h"
+#include "pegasus_event_listener.h"
 
 namespace pegasus {
 namespace server {
@@ -224,6 +225,8 @@ pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
 
     _db_opts.table_factory.reset(NewBlockBasedTableFactory(tbl_opts));
 
+    _db_opts.listeners.emplace_back(new pegasus_event_listener());
+
     // disable write ahead logging as replication handles logging instead now
     _wt_opts.disableWAL = true;
 

src/server/pegasus_server_impl.h

@@ -6,6 +6,7 @@
 
 #include <vector>
 #include <rocksdb/db.h>
+#include <rocksdb/listener.h>
 #include <dsn/cpp/perf_counter_wrapper.h>
 #include <dsn/dist/replication/replication.codes.h>
 #include <rrdb/rrdb_types.h>
@@ -136,7 +137,9 @@ public:
     storage_apply_checkpoint(chkpt_apply_mode mode,
                              const dsn::replication::learn_state &state) override;
 
-    virtual int64_t last_durable_decree() const { return _last_durable_decree.load(); }
+    virtual int64_t last_durable_decree() const override { return _last_durable_decree.load(); }
+
+    virtual int64_t last_flushed_decree() const override { return _db->GetLastFlushedDecree(); }
 
 private:
     friend class pagasus_manual_compact_service;

src/server/test/CMakeLists.txt

@@ -4,6 +4,7 @@ set(MY_PROJ_SRC "../pegasus_server_impl.cpp"
                 "../pegasus_perf_counter.cpp"
                 "../pegasus_counter_updater.cpp"
                 "../pagasus_manual_compact_service.cpp"
+                "../pegasus_event_listener.cpp"
 )
 
 set(MY_SRC_SEARCH_MODE "GLOB")

src/test/kill_test/config.ini

@@ -39,12 +39,12 @@ data_dir = ./data
 [tools.simple_logger]
 short_header = false
 fast_flush = true
-max_number_of_log_files_on_disk = 100000
+max_number_of_log_files_on_disk = 1000
 stderr_start_level = LOG_LEVEL_FATAL
 
 [tools.hpc_logger]
 per_thread_buffer_bytes = 8192
-max_number_of_log_files_on_disk = 100000
+max_number_of_log_files_on_disk = 1000
 
 [tools.simulator]
 random_seed = 0