add two matric and type key if dup non-idempotent write retried

ninsmiracle · ninsmiracle · commit 61b446f8ee33 · 2024-02-26T20:27:45.000+08:00
diff --git a/src/common/duplication_common.cpp b/src/common/duplication_common.cpp
@@ -40,6 +40,13 @@ DSN_DEFINE_uint32(replication,
                   "send mutation log batch bytes size per rpc");
 DSN_TAG_VARIABLE(duplicate_log_batch_bytes, FT_MUTABLE);
 
+DSN_DEFINE_bool("replication",
+                force_send_no_idempotent_when_duplication,
+                false,
+                "receive client idempotent write requests and send them to backup cluster when "
+                "doing duplication");
+DSN_TAG_VARIABLE(force_send_no_idempotent_when_duplication, FT_MUTABLE);
+
 const std::string duplication_constants::kDuplicationCheckpointRootDir /*NOLINT*/ = "duplication";
 const std::string duplication_constants::kClustersSectionName /*NOLINT*/ = "pegasus.clusters";
 const std::string duplication_constants::kDuplicationEnvMasterClusterKey /*NOLINT*/ =
diff --git a/src/common/duplication_common.h b/src/common/duplication_common.h
@@ -34,6 +34,7 @@ namespace dsn {
 namespace replication {
 
 DSN_DECLARE_uint32(duplicate_log_batch_bytes);
+DSN_DECLARE_bool(force_send_no_idempotent_when_duplication);
 
 typedef rpc_holder<duplication_modify_request, duplication_modify_response> duplication_modify_rpc;
 typedef rpc_holder<duplication_add_request, duplication_add_response> duplication_add_rpc;
diff --git a/src/replica/mutation.h b/src/replica/mutation.h
@@ -34,6 +34,7 @@
 
 #include "common/replication_common.h"
 #include "common/replication_other_types.h"
+#include "common/duplication_common.h"
 #include "consensus_types.h"
 #include "runtime/api_layer1.h"
 #include "runtime/rpc/rpc_message.h"
@@ -42,7 +43,6 @@
 #include "utils/autoref_ptr.h"
 #include "utils/fmt_logging.h"
 #include "utils/link.h"
-#include "utils/flags.h"
 
 namespace dsn {
 class binary_reader;
@@ -55,8 +55,6 @@ class latency_tracer;
 
 namespace replication {
 
-DSN_DECLARE_bool(force_send_no_idempotent_when_duplication);
-
 class mutation;
 
 typedef dsn::ref_ptr<mutation> mutation_ptr;
diff --git a/src/replica/replica_2pc.cpp b/src/replica/replica_2pc.cpp
@@ -85,15 +85,6 @@ DSN_DEFINE_bool(replication,
                 true,
                 "reject client write requests if disk status is space insufficient");
 DSN_TAG_VARIABLE(reject_write_when_disk_insufficient, FT_MUTABLE);
-
-DSN_DEFINE_bool("replication",
-                force_send_no_idempotent_when_duplication,
-                false,
-                "receive client idempotent write requests and send them to backup cluster when "
-                "doing duplication");
-DSN_TAG_VARIABLE(force_send_no_idempotent_when_duplication, FT_MUTABLE);
-
-
 DSN_DEFINE_int32(replication,
                  prepare_timeout_ms_for_secondaries,
                  1000,
diff --git a/src/server/info_collector.cpp b/src/server/info_collector.cpp
@@ -207,9 +207,11 @@ info_collector::app_stat_counters *info_collector::get_app_counters(const std::s
     INIT_COUNTER(incr_qps);
     INIT_COUNTER(check_and_set_qps);
     INIT_COUNTER(check_and_mutate_qps);
+    INIT_COUNTER(force_receive_no_idempotent_duplicate_qps);
     INIT_COUNTER(scan_qps);
     INIT_COUNTER(duplicate_qps);
     INIT_COUNTER(dup_shipped_ops);
+    INIT_COUNTER(dup_retry_no_idempotent_duplicate_qps);
     INIT_COUNTER(dup_failed_shipping_ops);
     INIT_COUNTER(dup_recent_mutation_loss_count);
     INIT_COUNTER(recent_read_cu);
diff --git a/src/server/info_collector.h b/src/server/info_collector.h
@@ -66,10 +66,14 @@ class info_collector
             incr_qps->set(row_stats.incr_qps);
             check_and_set_qps->set(row_stats.check_and_set_qps);
             check_and_mutate_qps->set(row_stats.check_and_mutate_qps);
+            force_receive_no_idempotent_duplicate_qps->set(
+                row_stats.force_receive_no_idempotent_duplicate_qps);
             scan_qps->set(row_stats.scan_qps);
             duplicate_qps->set(row_stats.duplicate_qps);
             dup_shipped_ops->set(row_stats.dup_shipped_ops);
             dup_failed_shipping_ops->set(row_stats.dup_failed_shipping_ops);
+            dup_retry_no_idempotent_duplicate_qps->set(
+                row_stats.dup_retry_no_idempotent_duplicate_qps);
             dup_recent_mutation_loss_count->set(row_stats.dup_recent_mutation_loss_count);
             recent_read_cu->set(row_stats.recent_read_cu);
             recent_write_cu->set(row_stats.recent_write_cu);
@@ -144,10 +148,12 @@ class info_collector
         ::dsn::perf_counter_wrapper incr_qps;
         ::dsn::perf_counter_wrapper check_and_set_qps;
         ::dsn::perf_counter_wrapper check_and_mutate_qps;
+        ::dsn::perf_counter_wrapper force_receive_no_idempotent_duplicate_qps;
         ::dsn::perf_counter_wrapper scan_qps;
         ::dsn::perf_counter_wrapper duplicate_qps;
         ::dsn::perf_counter_wrapper dup_shipped_ops;
         ::dsn::perf_counter_wrapper dup_failed_shipping_ops;
+        ::dsn::perf_counter_wrapper dup_retry_no_idempotent_duplicate_qps;
         ::dsn::perf_counter_wrapper dup_recent_mutation_loss_count;
         ::dsn::perf_counter_wrapper recent_read_cu;
         ::dsn::perf_counter_wrapper recent_write_cu;
diff --git a/src/server/pegasus_mutation_duplicator.cpp b/src/server/pegasus_mutation_duplicator.cpp
@@ -48,6 +48,7 @@
 #include "utils/errors.h"
 #include "utils/fmt_logging.h"
 #include "utils/rand.h"
+#include "pegasus_rpc_types.h"
 
 METRIC_DEFINE_counter(replica,
                       dup_shipped_successful_requests,
@@ -59,6 +60,11 @@ METRIC_DEFINE_counter(replica,
                       dsn::metric_unit::kRequests,
                       "The number of failed DUPLICATE requests sent from client");
 
+METRIC_DEFINE_counter(replica,
+                      dup_retry_no_idempotent_duplicate_qps,
+                      dsn::metric_unit::kRequests,
+                      "The qps of Non-idempotent write when doing DUPLICATE which is Retried");
+
 namespace dsn {
 namespace replication {
 struct replica_base;
@@ -205,6 +211,9 @@ void pegasus_mutation_duplicator::on_duplicate_reply(uint64_t hash,
             // retry this rpc
             _inflights[hash].push_front(rpc);
             _env.schedule([hash, cb, this]() { send(hash, cb); }, 1_s);
+
+            type_force_send_no_idempotent_if_need(rpc);
+
             return;
         }
         if (_inflights[hash].empty()) {
@@ -221,6 +230,74 @@ void pegasus_mutation_duplicator::on_duplicate_reply(uint64_t hash,
     }
 }
 
+void pegasus_mutation_duplicator::type_force_send_no_idempotent_if_need(duplicate_rpc &rpc)
+{
+
+    // there maybe more than one mutation in one dup rpc
+    if (dsn::replication::FLAGS_force_send_no_idempotent_when_duplication) {
+        for (auto entry : rpc.request().entries) {
+            if (entry.task_code == dsn::apps::RPC_RRDB_RRDB_INCR ||
+                entry.task_code == dsn::apps::RPC_RRDB_RRDB_CHECK_AND_SET ||
+                entry.task_code == dsn::apps::RPC_RRDB_RRDB_CHECK_AND_MUTATE) {
+
+                METRIC_VAR_INCREMENT(dup_retry_no_idempotent_duplicate_qps);
+
+                dsn::message_ex *write =
+                    dsn::from_blob_to_received_msg(entry.task_code, entry.raw_message);
+
+                if (entry.task_code == dsn::apps::RPC_RRDB_RRDB_INCR) {
+                    incr_rpc raw_rpc(write);
+                    absl::string_view unmarshall_key(raw_rpc.request().key.data(),
+                                                    raw_rpc.request().key.length());
+
+                    LOG_DEBUG("Non-indempotent write RPC_RRDB_RRDB_INCR has been retried when doing "
+                             "duplication,"
+                             "key is [{}]",
+                             unmarshall_key);
+                    continue;
+                }
+
+                if (entry.task_code == dsn::apps::RPC_RRDB_RRDB_CHECK_AND_SET) {
+                    check_and_set_rpc raw_rpc(write);
+                    absl::string_view unmarshall_hash_key(raw_rpc.request().hash_key.data(),
+                                                         raw_rpc.request().hash_key.length());
+                    absl::string_view unmarshall_ori_sort_key(
+                        raw_rpc.request().check_sort_key.data(),
+                        raw_rpc.request().check_sort_key.length());
+                    absl::string_view unmarshall_set_sort_key(
+                        raw_rpc.request().set_sort_key.data(),
+                        raw_rpc.request().set_sort_key.length());
+
+                    LOG_DEBUG("Non-indempotent write RPC_RRDB_RRDB_CHECK_AND_SET has been retried "
+                             "when doing duplication,"
+                             "hash key [{}], check sort key [{}],"
+                             "set sort key [{}]",
+                             unmarshall_hash_key,
+                             unmarshall_ori_sort_key,
+                             unmarshall_set_sort_key);
+                    continue;
+                }
+
+                if (entry.task_code == dsn::apps::RPC_RRDB_RRDB_CHECK_AND_MUTATE) {
+                    check_and_mutate_rpc raw_rpc(write);
+                    absl::string_view unmarshall_hash_key(raw_rpc.request().hash_key.data(),
+                                                         raw_rpc.request().hash_key.length());
+                    absl::string_view unmarshall_ori_sort_key(
+                        raw_rpc.request().check_sort_key.data(),
+                        raw_rpc.request().check_sort_key.length());
+
+                    LOG_DEBUG("Non-indempotent write RPC_RRDB_RRDB_CHECK_AND_MUTATE has been "
+                             "retried when doing duplication,"
+                             "hash key is [{}] , sort key is [{}] .",
+                             unmarshall_hash_key,
+                             unmarshall_ori_sort_key);
+                    continue;
+                }
+            }
+        }
+    }
+}
+
 void pegasus_mutation_duplicator::duplicate(mutation_tuple_set muts, callback cb)
 {
     _total_shipped_size = 0;
diff --git a/src/server/pegasus_mutation_duplicator.h b/src/server/pegasus_mutation_duplicator.h
@@ -73,6 +73,8 @@ class pegasus_mutation_duplicator : public dsn::replication::mutation_duplicator
 
     void on_duplicate_reply(uint64_t hash, callback, duplicate_rpc, dsn::error_code err);
 
+    void type_force_send_no_idempotent_if_need(duplicate_rpc &rpc);
+
 private:
     friend class pegasus_mutation_duplicator_test;
 
@@ -91,6 +93,7 @@ class pegasus_mutation_duplicator : public dsn::replication::mutation_duplicator
 
     METRIC_VAR_DECLARE_counter(dup_shipped_successful_requests);
     METRIC_VAR_DECLARE_counter(dup_shipped_failed_requests);
+    METRIC_VAR_DECLARE_counter(dup_retry_no_idempotent_duplicate_qps);
 };
 
 // Decodes the binary `request_data` into write request in thrift struct, and
diff --git a/src/server/pegasus_write_service.cpp b/src/server/pegasus_write_service.cpp
@@ -127,9 +127,9 @@ METRIC_DEFINE_counter(replica,
                       "The number of DUPLICATE requests");
 
 METRIC_DEFINE_counter(replica,
-                      no_idempotent_duplicate,
+                      force_receive_no_idempotent_duplicate_qps,
                       dsn::metric_unit::kRequests,
-                      "The number of forced idempotent requests when doing duplication");
+                      "statistic the those no idempotent qps of DUPLICATE requests Force received");
 
 METRIC_DEFINE_percentile_int64(replica,
                                dup_time_lag_ms,
@@ -174,7 +174,7 @@ pegasus_write_service::pegasus_write_service(pegasus_server_impl *server)
       METRIC_VAR_INIT_replica(check_and_set_latency_ns),
       METRIC_VAR_INIT_replica(check_and_mutate_latency_ns),
       METRIC_VAR_INIT_replica(dup_requests),
-      METRIC_VAR_INIT_replica(no_idempotent_duplicate),
+      METRIC_VAR_INIT_replica(force_receive_no_idempotent_duplicate_qps),
       METRIC_VAR_INIT_replica(dup_time_lag_ms),
       METRIC_VAR_INIT_replica(dup_lagging_writes),
       _put_batch_size(0),
@@ -427,7 +427,7 @@ int pegasus_write_service::duplicate(int64_t decree,
             request.task_code == dsn::apps::RPC_RRDB_RRDB_CHECK_AND_SET ||
             request.task_code == dsn::apps::RPC_RRDB_RRDB_CHECK_AND_MUTATE) {
             // receive no idempotent request from master cluster via duplication
-            METRIC_VAR_INCREMENT(no_idempotent_duplicate);
+            METRIC_VAR_INCREMENT(force_receive_no_idempotent_duplicate_qps);
 
             if (request.task_code == dsn::apps::RPC_RRDB_RRDB_INCR) {
                 incr_rpc rpc(write);
diff --git a/src/server/pegasus_write_service.h b/src/server/pegasus_write_service.h
@@ -232,7 +232,7 @@ class pegasus_write_service : dsn::replication::replica_base
     METRIC_VAR_DECLARE_percentile_int64(check_and_mutate_latency_ns);
 
     METRIC_VAR_DECLARE_counter(dup_requests);
-    METRIC_VAR_DECLARE_counter(no_idempotent_duplicate);
+    METRIC_VAR_DECLARE_counter(force_receive_no_idempotent_duplicate_qps);
     METRIC_VAR_DECLARE_percentile_int64(dup_time_lag_ms);
     METRIC_VAR_DECLARE_counter(dup_lagging_writes);
 
diff --git a/src/shell/command_helper.h b/src/shell/command_helper.h
@@ -813,9 +813,11 @@ struct row_data
         incr_qps += row.incr_qps;
         check_and_set_qps += row.check_and_set_qps;
         check_and_mutate_qps += row.check_and_mutate_qps;
+        force_receive_no_idempotent_duplicate_qps += row.force_receive_no_idempotent_duplicate_qps;
         scan_qps += row.scan_qps;
         duplicate_qps += row.duplicate_qps;
         dup_shipped_ops += row.dup_shipped_ops;
+        dup_retry_no_idempotent_duplicate_qps += row.dup_retry_no_idempotent_duplicate_qps;
         dup_failed_shipping_ops += row.dup_failed_shipping_ops;
         dup_recent_mutation_loss_count += row.dup_recent_mutation_loss_count;
         recent_read_cu += row.recent_read_cu;
@@ -880,9 +882,11 @@ struct row_data
     double incr_qps = 0;
     double check_and_set_qps = 0;
     double check_and_mutate_qps = 0;
+    double force_receive_no_idempotent_duplicate_qps = 0;
     double scan_qps = 0;
     double duplicate_qps = 0;
     double dup_shipped_ops = 0;
+    double dup_retry_no_idempotent_duplicate_qps = 0;
     double dup_failed_shipping_ops = 0;
     double dup_recent_mutation_loss_count = 0;
     double recent_read_cu = 0;
@@ -954,12 +958,16 @@ update_app_pegasus_perf_counter(row_data &row, const std::string &counter_name,
         row.check_and_set_qps += value;
     else if (counter_name == "check_and_mutate_qps")
         row.check_and_mutate_qps += value;
+    else if (counter_name == "force_receive_no_idempotent_duplicate_qps")
+        row.force_receive_no_idempotent_duplicate_qps += value;
     else if (counter_name == "scan_qps")
         row.scan_qps += value;
     else if (counter_name == "duplicate_qps")
         row.duplicate_qps += value;
     else if (counter_name == "dup_shipped_ops")
         row.dup_shipped_ops += value;
+    else if (counter_name == "dup_retry_no_idempotent_duplicate_qps")
+        row.dup_retry_no_idempotent_duplicate_qps += value;
     else if (counter_name == "dup_failed_shipping_ops")
         row.dup_failed_shipping_ops += value;
     else if (counter_name == "dup_recent_mutation_loss_count")
diff --git a/src/shell/commands/table_management.cpp b/src/shell/commands/table_management.cpp
@@ -555,6 +555,9 @@ bool app_stat(command_executor *e, shell_context *sc, arguments args)
         sum.incr_qps += row.incr_qps;
         sum.check_and_set_qps += row.check_and_set_qps;
         sum.check_and_mutate_qps += row.check_and_mutate_qps;
+        sum.force_receive_no_idempotent_duplicate_qps +=
+            row.force_receive_no_idempotent_duplicate_qps;
+
         sum.scan_qps += row.scan_qps;
         sum.recent_read_cu += row.recent_read_cu;
         sum.recent_write_cu += row.recent_write_cu;
@@ -651,6 +654,7 @@ bool app_stat(command_executor *e, shell_context *sc, arguments args)
             tp.append_data(row.incr_qps);
             tp.append_data(row.check_and_set_qps);
             tp.append_data(row.check_and_mutate_qps);
+            tp.append_data(row.force_receive_no_idempotent_duplicate_qps);
             tp.append_data(row.scan_qps);
             tp.append_data(row.recent_read_cu);
             tp.append_data(row.recent_write_cu);
