Convert txservice cluster config to dss_config

data_store_service_client.cpp

@@ -19,8 +19,6 @@
  *    <http://www.gnu.org/licenses/>.
  *
  */
-#include "data_store_service_client.h"
-
 #include <glog/logging.h>
 
 #include <boost/lexical_cast.hpp>
@@ -36,7 +34,9 @@
 #include <vector>
 
 #include "cc_req_misc.h"
+#include "data_store_service_client.h"
 #include "data_store_service_client_closure.h"
+#include "data_store_service_config.h"
 #include "data_store_service_scanner.h"
 #include "eloq_data_store_service/object_pool.h"  // ObjectPool
 #include "eloq_data_store_service/thread_worker_pool.h"
@@ -132,8 +132,78 @@ void DataStoreServiceClient::SetupConfig(
                 LOG(INFO) << "UpgradeShardVersion failed, retry";
                 bthread_usleep(1000000);
             }
-            LOG(INFO) << "UpgradeShardVersion success, shard_id:"
-                      << group.shard_id_ << ", version:" << group.version_;
+            LOG(INFO) << "DataStoreServiceCliet UpgradeShardVersion success, "
+                         "shard_id:"
+                      << group.shard_id_ << ", version:" << group.version_
+                      << ", owner_node:" << group.nodes_[0].host_name_ << ":"
+                      << group.nodes_[0].port_;
+        }
+    }
+    else
+    {
+        LOG(INFO)
+            << "DataStoreServiceCliet SetupConfig skipped, current_version:"
+            << current_version << ", new_version:" << new_version;
+    }
+}
+
+void DataStoreServiceClient::TxConfigsToDssClusterConfig(
+    uint32_t dss_node_id,  // = 0,
+    uint32_t ng_id,        // = 0,
+    const std::unordered_map<uint32_t, std::vector<txservice::NodeConfig>>
+        &ng_configs,
+    uint32_t dss_leader_node_id,  // if no leader,set uint32t_max
+    DataStoreServiceClusterManager &cluster_manager)
+{
+    assert(ng_configs.size() == 1);
+    auto it = ng_configs.find(ng_id);
+    assert(it != ng_configs.end());
+    auto &ng_member_configs = it->second;
+
+    const txservice::NodeConfig *this_node = nullptr;
+    const txservice::NodeConfig *leader_node = nullptr;
+    for (auto &node_config : ng_member_configs)
+    {
+        if (node_config.node_id_ == dss_node_id)
+        {
+            this_node = &node_config;
+        }
+        if (node_config.node_id_ == dss_leader_node_id)
+        {
+            leader_node = &node_config;
+        }
+    }
+    assert(this_node != nullptr);
+    assert(dss_leader_node_id == UNKNOWN_DSS_LEADER_NODE_ID ||
+           leader_node != nullptr);
+    cluster_manager.Initialize(this_node->host_name_,
+                               TxPort2DssPort(this_node->port_));
+
+    std::vector<DSSNode> shard_nodes;
+    for (auto &node_config : ng_member_configs)
+    {
+        if (node_config.node_id_ != dss_node_id)
+        {
+            DSSNode dss_node(node_config.host_name_,
+                             TxPort2DssPort(node_config.port_));
+            cluster_manager.AddShardMember(ng_id, dss_node);
+        }
+    }
+
+    if (dss_leader_node_id != dss_node_id)
+    {
+        LOG(INFO) << "cluster_manager change shard status " << ng_id << " from "
+                  << static_cast<int>(
+                         cluster_manager.FetchDSShardStatus(ng_id));
+        cluster_manager.SwitchShardToClosed(ng_id, DSShardStatus::ReadWrite);
+        LOG(INFO) << "cluster_manager change shard status " << ng_id << " to "
+                  << static_cast<int>(
+                         cluster_manager.FetchDSShardStatus(ng_id));
+        if (dss_leader_node_id != UNKNOWN_DSS_LEADER_NODE_ID)
+        {
+            DSSNode dss_node(leader_node->host_name_,
+                             TxPort2DssPort(leader_node->port_));
+            cluster_manager.UpdatePrimaryNode(ng_id, dss_node);
         }
     }
 }
@@ -213,6 +283,8 @@ bool DataStoreServiceClient::PutAll(
                        std::vector<std::unique_ptr<txservice::FlushTaskEntry>>>
         &flush_task)
 {
+    DLOG(INFO) << "DataStoreServiceClient::PutAll called with "
+               << flush_task.size() << " tables to flush.";
     uint64_t now = txservice::LocalCcShards::ClockTsInMillseconds();
 
     // Process each table
@@ -2868,6 +2940,31 @@ void DataStoreServiceClient::RestoreTxCache(txservice::NodeGroupId cc_ng_id,
  */
 bool DataStoreServiceClient::OnLeaderStart(uint32_t *next_leader_node)
 {
+    DLOG(INFO)
+        << "DataStoreServiceClient OnLeaderStart called data_store_service_:"
+        << data_store_service_;
+    if (data_store_service_ != nullptr)
+    {
+        // Now, only support one shard.
+        data_store_service_->OpenDataStore(0);
+    }
+
+    Connect();
+
+    return true;
+}
+
+bool DataStoreServiceClient::OnLeaderStop(int64_t term)
+{
+    DLOG(INFO)
+        << "DataStoreServiceClient OnLeaderStop called data_store_service_:"
+        << data_store_service_;
+    // swith to read only in case of data store status is read write
+    if (data_store_service_ != nullptr)
+    {
+        // Now, only support one shard.
+        data_store_service_->CloseDataStore(0);
+    }
     return true;
 }
 
@@ -2878,8 +2975,53 @@ bool DataStoreServiceClient::OnLeaderStart(uint32_t *next_leader_node)
  * following another leader and can be used to perform follower-specific
  * initialization.
  */
-void DataStoreServiceClient::OnStartFollowing()
+void DataStoreServiceClient::OnStartFollowing(uint32_t leader_node_id,
+                                              int64_t term,
+                                              int64_t standby_term,
+                                              bool resubscribe)
 {
+    DLOG(INFO)
+        << "DataStoreServiceClient OnStartFollowing called data_store_service_:"
+        << data_store_service_;
+    if (data_store_service_ != nullptr)
+    {
+        // Now, only support one shard.
+        data_store_service_->CloseDataStore(0);
+    }
+
+    // Treat leader_node_id as dss_leader_node_id
+    uint32_t dss_leader_node_id = leader_node_id;
+    uint32_t dss_shard_id = txservice::Sharder::Instance().NativeNodeGroup();
+
+    // Update leader node in cluster_manager if necessary
+    auto ng_configs = txservice::Sharder::Instance().GetNodeGroupConfigs();
+    auto ng_config_it = ng_configs.find(dss_shard_id);
+    assert(ng_config_it != ng_configs.end());
+    auto ng_member_configs = ng_config_it->second;
+    const txservice::NodeConfig *dss_leader_node_config = nullptr;
+    for (const auto &node_config : ng_member_configs)
+    {
+        if (node_config.node_id_ == dss_leader_node_id)
+        {
+            dss_leader_node_config = &node_config;
+            break;
+        }
+    }
+    assert(dss_leader_node_config != nullptr);
+    DSSNode dss_leader_node(dss_leader_node_config->host_name_,
+                            TxPort2DssPort(dss_leader_node_config->port_));
+    auto &cluster_manager = data_store_service_->GetClusterManager();
+    cluster_manager.UpdatePrimaryNode(dss_shard_id, dss_leader_node);
+    DLOG(INFO) << "UpdatePrimaryNode, dss_shard_id: " << dss_shard_id
+               << ", DSSNode: " << dss_leader_node.host_name_ << ":"
+               << dss_leader_node.port_;
+    // Pump the dss shard version
+    cluster_manager.UpdateDSShardVersion(
+        dss_shard_id, cluster_manager.FetchDSShardVersion(dss_shard_id) + 1);
+    // Update the client config
+    SetupConfig(cluster_manager);
+
+    Connect();
 }
 
 /**
@@ -4610,4 +4752,4 @@ void DataStoreServiceClient::PrepareRangePartitionBatches(
     }
 }
 
-}  // namespace EloqDS
+}  // namespace EloqD

data_store_service_client.h

@@ -126,6 +126,19 @@ class DataStoreServiceClient : public txservice::store::DataStoreHandler
      */
     void SetupConfig(const DataStoreServiceClusterManager &config);
 
+    static uint16_t TxPort2DssPort(uint16_t tx_port)
+    {
+        return tx_port + 7;
+    }
+
+    static void TxConfigsToDssClusterConfig(
+        uint32_t dss_node_id,  // = 0,
+        uint32_t ng_id,        // = 0,
+        const std::unordered_map<uint32_t, std::vector<txservice::NodeConfig>>
+            &ng_configs,
+        uint32_t dss_leader_node_id,  // if no leader,set uint32t_max
+        DataStoreServiceClusterManager &cluster_manager);
+
     void ConnectToLocalDataStoreService(
         std::unique_ptr<DataStoreService> ds_serv);
 
@@ -377,7 +390,12 @@ class DataStoreServiceClient : public txservice::store::DataStoreHandler
 
     bool OnLeaderStart(uint32_t *next_leader_node) override;
 
-    void OnStartFollowing() override;
+    bool OnLeaderStop(int64_t term) override;
+
+    void OnStartFollowing(uint32_t leader_node_id,
+                          int64_t term,
+                          int64_t standby_term,
+                          bool resubscribe) override;
 
     void OnShutdown() override;
 

eloq_data_store_service/data_store_service.cpp

@@ -237,7 +237,9 @@ DataStoreService::~DataStoreService()
     }
 }
 
-bool DataStoreService::StartService(bool create_db_if_missing)
+bool DataStoreService::StartService(bool create_db_if_missing,
+                                    uint32_t dss_leader_node_id,
+                                    uint32_t dss_node_id)
 {
     if (server_ != nullptr)
     {
@@ -252,15 +254,25 @@ bool DataStoreService::StartService(bool create_db_if_missing)
     {
         shard_id_ = dss_shards.at(0);
         auto open_mode = cluster_manager_.FetchDSShardStatus(shard_id_);
+        DLOG(INFO) << "StartService data store shard id:" << shard_id_
+                   << ", open_mode:" << static_cast<int>(open_mode)
+                   << ", create_db_if_missing:" << create_db_if_missing
+                   << ", dss_leader_node_id:" << dss_leader_node_id
+                   << ", dss_node_id:" << dss_node_id;
         if (open_mode == DSShardStatus::ReadOnly ||
             open_mode == DSShardStatus::ReadWrite)
         {
             auto expect_status = DSShardStatus::Closed;
             if (shard_status_.compare_exchange_strong(expect_status,
                                                       DSShardStatus::Starting))
             {
+                // start underling db if this dss node is the
+                // leader dss node
                 data_store_ = data_store_factory_->CreateDataStore(
-                    create_db_if_missing, shard_id_, this, true);
+                    create_db_if_missing,
+                    shard_id_,
+                    this,
+                    dss_leader_node_id == dss_node_id);
                 if (data_store_ == nullptr)
                 {
                     LOG(ERROR) << "Failed to create data store on starting "
@@ -277,7 +289,7 @@ bool DataStoreService::StartService(bool create_db_if_missing)
         }
 
         DLOG(INFO) << "Created data store shard id:" << shard_id_
-                   << ", shard_status:" << shard_status_;
+                   << ", shard_status:" << static_cast<int>(shard_status_);
     }
 
     server_ = std::make_unique<brpc::Server>();
@@ -311,7 +323,7 @@ bool DataStoreService::ConnectAndStartDataStore(uint32_t data_shard_id,
     {
         return true;
     }
-    assert(open_mode == DSShardStatus::ReadOnly);
+    // assert(open_mode == DSShardStatus::ReadOnly);
 
     DSShardStatus expect_status = DSShardStatus::Closed;
     if (!shard_status_.compare_exchange_strong(expect_status,
@@ -329,6 +341,11 @@ bool DataStoreService::ConnectAndStartDataStore(uint32_t data_shard_id,
         return expect_status == open_mode;
     }
 
+    DLOG(INFO) << "Connecting and starting data store for shard id:"
+              << data_shard_id
+              << ", open_mode:" << static_cast<int>(open_mode)
+              << ", create_db_if_missing:" << create_db_if_missing
+              << ", data_store_ is null:" << (data_store_ == nullptr);
     assert(data_store_factory_ != nullptr);
     if (data_store_ == nullptr)
     {
@@ -358,10 +375,18 @@ bool DataStoreService::ConnectAndStartDataStore(uint32_t data_shard_id,
             return false;
         }
     }
-
-    data_store_->SwitchToReadOnly();
-    cluster_manager_.SwitchShardToReadOnly(data_shard_id,
-                                           DSShardStatus::Closed);
+    if (open_mode == DSShardStatus::ReadOnly)
+    {
+        data_store_->SwitchToReadOnly();
+        cluster_manager_.SwitchShardToReadOnly(data_shard_id,
+                                               DSShardStatus::Closed);
+    }
+    else
+    {
+        assert(open_mode == DSShardStatus::ReadWrite);
+        cluster_manager_.SwitchShardToReadWrite(data_shard_id,
+                                                DSShardStatus::Closed);
+    }
 
     expect_status = DSShardStatus::Starting;
     shard_status_.compare_exchange_strong(
@@ -1786,6 +1811,48 @@ bool DataStoreService::FetchConfigFromPeer(
     return true;
 }
 
+void DataStoreService::CloseDataStore(uint32_t shard_id)
+{
+    assert(shard_id == shard_id_);
+    if (!IsOwnerOfShard(shard_id))
+    {
+        return;
+    }
+    if (shard_status_.load() == DSShardStatus::ReadWrite)
+    {
+        SwitchReadWriteToReadOnly(shard_id);
+    }
+
+    if (shard_status_.load() == DSShardStatus::ReadOnly)
+    {
+        SwitchReadOnlyToClosed(shard_id);
+    }
+}
+
+void DataStoreService::OpenDataStore(uint32_t shard_id)
+{
+    // no-op if this DSS does not own any shard
+    if (shard_id_ == UINT32_MAX)
+    {
+        DLOG(INFO) << "OpenDataStore no-op for non-owner DSS"
+                   << ", shard " << shard_id
+                   << ", shard_id_: " << shard_id_;
+        return;
+    }
+
+    assert(shard_id == shard_id_);
+
+    DLOG(INFO) << "OpenDataStore for shard " << shard_id
+               << ", current status: " << static_cast<int>(shard_status_.load());
+    if (shard_status_.load() != DSShardStatus::Closed)
+    {
+        return;
+    }
+    DSShardStatus open_mode = DSShardStatus::ReadWrite;
+    bool create_db_if_missing = false;
+    ConnectAndStartDataStore(shard_id, open_mode, create_db_if_missing);
+}
+
 std::pair<remote::ShardMigrateError, std::string>
 DataStoreService::NewMigrateTask(const std::string &event_id,
                                  int data_shard_id,
@@ -2459,4 +2526,4 @@ void DataStoreService::CleanupOldMigrateLogs()
     }
 }
 
-}  // namespace EloqDS
+}  // namespace EloqDS