You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by db...@apache.org on 2023/10/25 10:51:59 UTC
[impala] branch master updated (c244aadcf -> c9c5fb89b)
This is an automated email from the ASF dual-hosted git repository.
dbecker pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
from c244aadcf IMPALA-12402: Make CatalogdMetaProvider's cache concurrency level configurable
new 141e3f174 IMPALA-12481: Fix sorted-run-merger-ir.cc make dependency
new c9c5fb89b IMPALA-12156: Support High Availability for Statestore
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
be/generated-sources/gen-cpp/CMakeLists.txt | 1 +
be/src/catalog/catalog-server.cc | 10 +-
be/src/common/global-flags.cc | 14 +
be/src/rpc/thrift-server-test.cc | 3 +
be/src/runtime/CMakeLists.txt | 2 +-
be/src/runtime/exec-env.cc | 17 +-
be/src/runtime/exec-env.h | 4 +-
be/src/scheduling/admissiond-env.cc | 6 +-
.../statestore/statestore-service-client-wrapper.h | 45 ++
be/src/statestore/statestore-subscriber-catalog.cc | 5 +-
be/src/statestore/statestore-subscriber-catalog.h | 3 +
.../statestore-subscriber-client-wrapper.h | 8 +
be/src/statestore/statestore-subscriber.cc | 349 +++++++++--
be/src/statestore/statestore-subscriber.h | 79 ++-
be/src/statestore/statestore-test.cc | 6 +-
be/src/statestore/statestore.cc | 682 ++++++++++++++++++++-
be/src/statestore/statestore.h | 202 +++++-
be/src/statestore/statestored-main.cc | 19 +-
bin/start-impala-cluster.py | 179 +++++-
common/thrift/StatestoreService.thrift | 151 +++++
common/thrift/metrics.json | 196 ++++++
tests/common/impala_cluster.py | 43 +-
tests/common/impala_service.py | 6 +-
tests/custom_cluster/test_statestored_ha.py | 627 +++++++++++++++++++
24 files changed, 2542 insertions(+), 115 deletions(-)
create mode 100644 tests/custom_cluster/test_statestored_ha.py
[impala] 01/02: IMPALA-12481: Fix sorted-run-merger-ir.cc make dependency
Posted by db...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dbecker pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 141e3f1748822fa2083840bb96fecf53b5c58582
Author: Noemi Pap-Takacs <np...@cloudera.com>
AuthorDate: Fri Oct 20 15:02:38 2023 +0200
IMPALA-12481: Fix sorted-run-merger-ir.cc make dependency
IMPALA-11477 added sorted-run-merger-ir.cc, and incorrectly
added it to the Runtime target, not RuntimeIr.
This patch fixes the issue by adding sorted-run-merger-ir.cc
to RuntimeIr instead in CMakeLists.txt.
Change-Id: Id39cc9613bec60f6308db05dde99f375c68166e0
Reviewed-on: http://gerrit.cloudera.org:8080/20604
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/runtime/CMakeLists.txt | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/be/src/runtime/CMakeLists.txt b/be/src/runtime/CMakeLists.txt
index faf7b2cd5..c5c254a61 100644
--- a/be/src/runtime/CMakeLists.txt
+++ b/be/src/runtime/CMakeLists.txt
@@ -32,6 +32,7 @@ add_library(RuntimeIr
raw-value-ir.cc
runtime-filter-ir.cc
sorter-ir.cc
+ sorted-run-merger-ir.cc
tuple-ir.cc
)
add_dependencies(RuntimeIr gen-deps)
@@ -81,7 +82,6 @@ add_library(Runtime
runtime-state.cc
scanner-mem-limiter.cc
sorted-run-merger.cc
- sorted-run-merger-ir.cc
sorter.cc
spillable-row-batch-queue.cc
string-value.cc
[impala] 02/02: IMPALA-12156: Support High Availability for Statestore
Posted by db...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dbecker pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit c9c5fb89b5679dcb8e41f61529a65b7648500741
Author: wzhou-code <wz...@cloudera.com>
AuthorDate: Thu Jun 22 18:06:08 2023 -0700
IMPALA-12156: Support High Availability for Statestore
To support statestore HA, we allow two statestored instances in an
Active-Passive HA pair to be added to an Impala cluster. We add the
preemptive behavior for statestored. When HA is enabled, the preemptive
behavior allows the statestored with the higher priority to become
active and the paired statestored becomes standby. The active
statestored acts as the owner of Impala cluster and provides statestore
service for the cluster members.
To enable catalog HA for a cluster, two statestoreds in the HA pair and
all subscribers must be started with starting flag
"enable_statestored_ha" as true.
This patch makes following changes:
- Defined new service for Statestore HA.
- Statestored negotiates the role for HA with its peer statestore
instance on startup.
- Create HA monitor thread:
Active statestored sends heartbeat to standby statestored.
Standby statestored monitors peer's connection states with their
subscribers.
- Standby statestored sends heartbeat to subscribers with request
for connection state between active statestore and subscribers.
Standby statestored saves the connection state as failure detecter.
- When standby statestored lost connection with active statestore,
it checks the connection states for active statestore, and takes over
active role if majority of subscribers lost connections with active
statestore.
- New active statestored sends RPC notification to all subscribers
for new active statestored and active catalogd elected by the new
active statestored.
- New active statestored starts sending heartbeat to its peer when it
receives handshake from its peer.
- Active statestored enters recovery mode if it lost connections with
its peer statestored and all subscribers. It keeps sending HA
handshake to its peer until receiving response.
- All subscribers (impalad/catalogd/admissiond) register to two
statestoreds.
- Subscribers report connection state for the requests from standby
statestore.
- Subscribers switch to new active statestore when receiving RPC
notifications from new active statestored.
- Only active statestored sends topic update messages to subscribers.
- Add a new option "enable_statestored_ha" in script
bin/start-impala-cluster.py for starting Impala mini-cluster with
statestored HA enabled.
- Add a new Thrift API in statestore service to disable network
for statestored. It's only used for unit-test to simulate network
failure. For safety, it's only working when the debug action is set
in starting flags.
Testings:
- Added end-to-end unit tests for statestored HA.
- Passed core tests
Change-Id: Ibd2c814bbad5c04c1d50c2edaa5b910c82a6fd87
Reviewed-on: http://gerrit.cloudera.org:8080/20372
Reviewed-by: Michael Smith <mi...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/generated-sources/gen-cpp/CMakeLists.txt | 1 +
be/src/catalog/catalog-server.cc | 10 +-
be/src/common/global-flags.cc | 14 +
be/src/rpc/thrift-server-test.cc | 3 +
be/src/runtime/exec-env.cc | 17 +-
be/src/runtime/exec-env.h | 4 +-
be/src/scheduling/admissiond-env.cc | 6 +-
.../statestore/statestore-service-client-wrapper.h | 45 ++
be/src/statestore/statestore-subscriber-catalog.cc | 5 +-
be/src/statestore/statestore-subscriber-catalog.h | 3 +
.../statestore-subscriber-client-wrapper.h | 8 +
be/src/statestore/statestore-subscriber.cc | 349 +++++++++--
be/src/statestore/statestore-subscriber.h | 79 ++-
be/src/statestore/statestore-test.cc | 6 +-
be/src/statestore/statestore.cc | 682 ++++++++++++++++++++-
be/src/statestore/statestore.h | 202 +++++-
be/src/statestore/statestored-main.cc | 19 +-
bin/start-impala-cluster.py | 179 +++++-
common/thrift/StatestoreService.thrift | 151 +++++
common/thrift/metrics.json | 196 ++++++
tests/common/impala_cluster.py | 43 +-
tests/common/impala_service.py | 6 +-
tests/custom_cluster/test_statestored_ha.py | 627 +++++++++++++++++++
23 files changed, 2541 insertions(+), 114 deletions(-)
diff --git a/be/generated-sources/gen-cpp/CMakeLists.txt b/be/generated-sources/gen-cpp/CMakeLists.txt
index 165e2b9dd..2f1a23940 100644
--- a/be/generated-sources/gen-cpp/CMakeLists.txt
+++ b/be/generated-sources/gen-cpp/CMakeLists.txt
@@ -70,6 +70,7 @@ set(SRC_FILES
ResourceProfile_types.cpp
RuntimeProfile_types.cpp
SqlConstraints_types.cpp
+ StatestoreHaService.cpp
StatestoreService.cpp
StatestoreService_types.cpp
StatestoreSubscriber.cpp
diff --git a/be/src/catalog/catalog-server.cc b/be/src/catalog/catalog-server.cc
index 30ae739dd..d1f17ce3d 100644
--- a/be/src/catalog/catalog-server.cc
+++ b/be/src/catalog/catalog-server.cc
@@ -149,8 +149,10 @@ DEFINE_bool(enable_skipping_older_events, false, "This configuration is used to
"quickly catch with the events of HMS");
DECLARE_string(state_store_host);
-DECLARE_int32(state_store_subscriber_port);
DECLARE_int32(state_store_port);
+DECLARE_string(state_store_2_host);
+DECLARE_int32(state_store_2_port);
+DECLARE_int32(state_store_subscriber_port);
DECLARE_string(hostname);
DECLARE_bool(compact_catalog_topic);
DECLARE_bool(enable_catalogd_ha);
@@ -372,6 +374,8 @@ Status CatalogServer::Start() {
MakeNetworkAddress(FLAGS_hostname, FLAGS_state_store_subscriber_port);
TNetworkAddress statestore_address =
MakeNetworkAddress(FLAGS_state_store_host, FLAGS_state_store_port);
+ TNetworkAddress statestore2_address =
+ MakeNetworkAddress(FLAGS_state_store_2_host, FLAGS_state_store_2_port);
TNetworkAddress server_address = MakeNetworkAddress(FLAGS_hostname,
FLAGS_catalog_service_port);
@@ -391,8 +395,8 @@ Status CatalogServer::Start() {
active_catalogd_version_checker_.reset(new ActiveCatalogdVersionChecker());
statestore_subscriber_.reset(new StatestoreSubscriberCatalog(
Substitute("catalog-server@$0", TNetworkAddressToString(server_address)),
- subscriber_address, statestore_address, metrics_, protocol_version_,
- server_address));
+ subscriber_address, statestore_address, statestore2_address, metrics_,
+ protocol_version_, server_address));
StatestoreSubscriber::UpdateCallback cb =
bind<void>(mem_fn(&CatalogServer::UpdateCatalogTopicCallback), this, _1, _2);
diff --git a/be/src/common/global-flags.cc b/be/src/common/global-flags.cc
index 144f03647..b7190ff48 100644
--- a/be/src/common/global-flags.cc
+++ b/be/src/common/global-flags.cc
@@ -415,6 +415,20 @@ DEFINE_int32(iceberg_reload_new_files_threshold, 100, "(Advanced) If during a ta
"reload all file metadata. If number of new files are less or equal to this, "
"catalogd will only load the metadata of the newly added files.");
+// Host and port of Statestore Service
+DEFINE_string(state_store_host, "localhost",
+ "hostname where StatestoreService is running");
+DEFINE_int32(state_store_port, 24000, "port where StatestoreService is running");
+
+// Starting flags for Statestore HA
+DEFINE_string(state_store_2_host, "localhost",
+ "hostname where second StatestoreService instance is running");
+DEFINE_int32(state_store_2_port, 24001,
+ "port where second StatestoreService instance is running");
+// Port for RPC communication between two statestore instances for Statestore HA
+DEFINE_int32(state_store_ha_port, 24020,
+ "port where StatestoreHaService is running");
+
// TGeospatialLibrary's values are mapped here as constants
static const string geo_lib_none = "NONE";
static const string geo_lib_hive_esri = "HIVE_ESRI";
diff --git a/be/src/rpc/thrift-server-test.cc b/be/src/rpc/thrift-server-test.cc
index b9f82ec0c..425155c94 100644
--- a/be/src/rpc/thrift-server-test.cc
+++ b/be/src/rpc/thrift-server-test.cc
@@ -84,6 +84,9 @@ class DummyStatestoreService : public StatestoreServiceIf {
virtual void GetProtocolVersion(TGetProtocolVersionResponse& response,
const TGetProtocolVersionRequest& request) {
}
+ virtual void SetStatestoreDebugAction(TSetStatestoreDebugActionResponse& response,
+ const TSetStatestoreDebugActionRequest& params) {
+ }
};
std::shared_ptr<TProcessor> MakeProcessor() {
diff --git a/be/src/runtime/exec-env.cc b/be/src/runtime/exec-env.cc
index 3931f5f44..8f2768e49 100644
--- a/be/src/runtime/exec-env.cc
+++ b/be/src/runtime/exec-env.cc
@@ -84,8 +84,6 @@ DEFINE_bool(enable_webserver, true, "If true, debug webserver is enabled");
DEFINE_bool(ping_expose_webserver_url, true,
"If true, debug webserver url is exposed via PingImpalaService/PingImpalaHS2Service "
"RPC calls");
-DEFINE_string(state_store_host, "localhost",
- "hostname where StatestoreService is running");
DEFINE_int32(state_store_subscriber_port, 23000,
"port where StatestoreSubscriberService should be exported");
DEFINE_int32(num_hdfs_worker_threads, 16,
@@ -128,7 +126,6 @@ DEFINE_int32(local_catalog_cache_concurrency_level, 4,
"level to avoid lock contention, the default value 4 is consistent with the "
"default value of the original cache.");
-DECLARE_int32(state_store_port);
DECLARE_int32(num_threads_per_core);
DECLARE_int32(num_cores);
DECLARE_int32(krpc_port);
@@ -146,6 +143,10 @@ DECLARE_string(admission_service_host);
DECLARE_int32(admission_service_port);
DECLARE_string(catalog_service_host);
DECLARE_int32(catalog_service_port);
+DECLARE_string(state_store_host);
+DECLARE_int32(state_store_port);
+DECLARE_string(state_store_2_host);
+DECLARE_int32(state_store_2_port);
DECLARE_string(ssl_client_ca_certificate);
@@ -208,10 +209,12 @@ ExecEnv* ExecEnv::exec_env_ = nullptr;
ExecEnv::ExecEnv(bool external_fe)
: ExecEnv(FLAGS_krpc_port, FLAGS_state_store_subscriber_port, FLAGS_webserver_port,
- FLAGS_state_store_host, FLAGS_state_store_port, external_fe) {}
+ FLAGS_state_store_host, FLAGS_state_store_port, FLAGS_state_store_2_host,
+ FLAGS_state_store_2_port, external_fe) {}
ExecEnv::ExecEnv(int krpc_port, int subscriber_port, int webserver_port,
- const string& statestore_host, int statestore_port, bool external_fe)
+ const string& statestore_host, int statestore_port,
+ const std::string& statestore2_host, int statestore2_port, bool external_fe)
: obj_pool_(new ObjectPool),
metrics_(new MetricGroup("impala-metrics")),
// Create the CatalogServiceClientCache with num_retries = 1 and wait_ms = 0.
@@ -251,6 +254,8 @@ ExecEnv::ExecEnv(int krpc_port, int subscriber_port, int webserver_port,
MakeNetworkAddress(FLAGS_hostname, subscriber_port);
TNetworkAddress statestore_address =
MakeNetworkAddress(statestore_host, statestore_port);
+ TNetworkAddress statestore2_address =
+ MakeNetworkAddress(statestore2_host, statestore2_port);
catalogd_address_ = std::make_shared<const TNetworkAddress>(
MakeNetworkAddress(FLAGS_catalog_service_host, FLAGS_catalog_service_port));
@@ -266,7 +271,7 @@ ExecEnv::ExecEnv(int krpc_port, int subscriber_port, int webserver_port,
// Set StatestoreSubscriber::subscriber_id as hostname + krpc_port.
statestore_subscriber_.reset(new StatestoreSubscriber(
Substitute("impalad@$0:$1", FLAGS_hostname, FLAGS_krpc_port), subscriber_address,
- statestore_address, metrics_.get(), subscriber_type));
+ statestore_address, statestore2_address, metrics_.get(), subscriber_type));
if (FLAGS_is_coordinator) {
hdfs_op_thread_pool_.reset(
diff --git a/be/src/runtime/exec-env.h b/be/src/runtime/exec-env.h
index e71878fc3..53f8f68bc 100644
--- a/be/src/runtime/exec-env.h
+++ b/be/src/runtime/exec-env.h
@@ -92,7 +92,9 @@ class ExecEnv {
ExecEnv(bool external_fe = false);
ExecEnv(int krpc_port, int subscriber_port, int webserver_port,
- const std::string& statestore_host, int statestore_port, bool external_fe = false);
+ const std::string& statestore_host, int statestore_port,
+ const std::string& statestore2_host = "", int statestore2_port = 0,
+ bool external_fe = false);
/// Returns the most recently created exec env instance. In a normal impalad, this is
/// the only instance. In test setups with multiple ExecEnv's per process,
diff --git a/be/src/scheduling/admissiond-env.cc b/be/src/scheduling/admissiond-env.cc
index a7b226049..8efa5c9d6 100644
--- a/be/src/scheduling/admissiond-env.cc
+++ b/be/src/scheduling/admissiond-env.cc
@@ -38,6 +38,8 @@ DEFINE_int32(
DECLARE_string(state_store_host);
DECLARE_int32(state_store_port);
+DECLARE_string(state_store_2_host);
+DECLARE_int32(state_store_2_port);
DECLARE_int32(state_store_subscriber_port);
DECLARE_string(hostname);
@@ -58,9 +60,11 @@ AdmissiondEnv::AdmissiondEnv()
MakeNetworkAddress(FLAGS_hostname, FLAGS_state_store_subscriber_port);
TNetworkAddress statestore_address =
MakeNetworkAddress(FLAGS_state_store_host, FLAGS_state_store_port);
+ TNetworkAddress statestore2_address =
+ MakeNetworkAddress(FLAGS_state_store_2_host, FLAGS_state_store_2_port);
statestore_subscriber_.reset(new StatestoreSubscriber(
Substitute("admissiond@$0", TNetworkAddressToString(admission_service_addr)),
- subscriber_address, statestore_address, metrics,
+ subscriber_address, statestore_address, statestore2_address, metrics,
TStatestoreSubscriberType::ADMISSIOND));
scheduler_.reset(new Scheduler(metrics, request_pool_service()));
diff --git a/be/src/statestore/statestore-service-client-wrapper.h b/be/src/statestore/statestore-service-client-wrapper.h
index e085929c4..d75d93196 100644
--- a/be/src/statestore/statestore-service-client-wrapper.h
+++ b/be/src/statestore/statestore-service-client-wrapper.h
@@ -18,6 +18,7 @@
#ifndef STATESTORE_STATESTORE_SERVICE_CLIENT_WRAPPER_H
#define STATESTORE_STATESTORE_SERVICE_CLIENT_WRAPPER_H
+#include "gen-cpp/StatestoreHaService.h"
#include "gen-cpp/StatestoreService.h"
namespace impala {
@@ -56,6 +57,50 @@ class StatestoreServiceClientWrapper : public StatestoreServiceClient {
recv_GetProtocolVersion(_return);
}
+ void SetStatestoreDebugAction(TSetStatestoreDebugActionResponse& _return,
+ const TSetStatestoreDebugActionRequest& params, bool* send_done) {
+ DCHECK(!*send_done);
+ send_SetStatestoreDebugAction(params);
+ *send_done = true;
+ recv_SetStatestoreDebugAction(_return);
+ }
+#pragma clang diagnostic pop
+};
+
+class StatestoreHaServiceClientWrapper : public StatestoreHaServiceClient {
+ public:
+ StatestoreHaServiceClientWrapper(
+ std::shared_ptr<::apache::thrift::protocol::TProtocol> prot)
+ : StatestoreHaServiceClient(prot) {
+ }
+
+ StatestoreHaServiceClientWrapper(
+ std::shared_ptr<::apache::thrift::protocol::TProtocol> iprot,
+ std::shared_ptr<::apache::thrift::protocol::TProtocol> oprot)
+ : StatestoreHaServiceClient(iprot, oprot) {
+ }
+
+/// We intentionally disable this clang warning as we intend to hide the
+/// the same-named functions defined in the base class.
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Woverloaded-virtual"
+
+ void StatestoreHaHandshake(TStatestoreHaHandshakeResponse& _return,
+ const TStatestoreHaHandshakeRequest& params, bool* send_done) {
+ DCHECK(!*send_done);
+ send_StatestoreHaHandshake(params);
+ *send_done = true;
+ recv_StatestoreHaHandshake(_return);
+ }
+
+ void StatestoreHaHeartbeat(TStatestoreHaHeartbeatResponse& _return,
+ const TStatestoreHaHeartbeatRequest& params, bool* send_done) {
+ DCHECK(!*send_done);
+ send_StatestoreHaHeartbeat(params);
+ *send_done = true;
+ recv_StatestoreHaHeartbeat(_return);
+ }
+
#pragma clang diagnostic pop
};
diff --git a/be/src/statestore/statestore-subscriber-catalog.cc b/be/src/statestore/statestore-subscriber-catalog.cc
index 25ed30fb6..04055ee14 100644
--- a/be/src/statestore/statestore-subscriber-catalog.cc
+++ b/be/src/statestore/statestore-subscriber-catalog.cc
@@ -27,11 +27,12 @@ StatestoreSubscriberCatalog::StatestoreSubscriberCatalog(
const std::string& subscriber_id,
const TNetworkAddress& heartbeat_address,
const TNetworkAddress& statestore_address,
+ const TNetworkAddress& statestore2_address,
MetricGroup* metrics,
CatalogServiceVersion::type catalog_protocol_version,
const TNetworkAddress& catalogd_address)
- : StatestoreSubscriber(subscriber_id, heartbeat_address, statestore_address, metrics,
- TStatestoreSubscriberType::CATALOGD) {
+ : StatestoreSubscriber(subscriber_id, heartbeat_address, statestore_address,
+ statestore2_address, metrics, TStatestoreSubscriberType::CATALOGD) {
catalogd_registration_.__set_protocol(catalog_protocol_version);
catalogd_registration_.__set_address(catalogd_address);
catalogd_registration_.__set_enable_catalogd_ha(FLAGS_enable_catalogd_ha);
diff --git a/be/src/statestore/statestore-subscriber-catalog.h b/be/src/statestore/statestore-subscriber-catalog.h
index 867f8e1ea..e10248059 100644
--- a/be/src/statestore/statestore-subscriber-catalog.h
+++ b/be/src/statestore/statestore-subscriber-catalog.h
@@ -30,11 +30,14 @@ class StatestoreSubscriberCatalog : public StatestoreSubscriber {
/// heartbeat_address - the local address on which the heartbeat service which
/// communicates with the statestore should be started.
/// statestore_address - the address of the statestore to register with
+ /// statestore2_address - the address of the second statestore instance which is
+ /// launched when statestored HA is enabled.
/// catalogd_address - address of catalog service.
/// catalog_protocol_version - protocol version of Catalog service.
StatestoreSubscriberCatalog(const std::string& subscriber_id,
const TNetworkAddress& heartbeat_address,
const TNetworkAddress& statestore_address,
+ const TNetworkAddress& statestore2_address,
MetricGroup* metrics,
CatalogServiceVersion::type catalog_protocol_version,
const TNetworkAddress& catalogd_address);
diff --git a/be/src/statestore/statestore-subscriber-client-wrapper.h b/be/src/statestore/statestore-subscriber-client-wrapper.h
index 0360243a0..5e771080a 100644
--- a/be/src/statestore/statestore-subscriber-client-wrapper.h
+++ b/be/src/statestore/statestore-subscriber-client-wrapper.h
@@ -64,6 +64,14 @@ class StatestoreSubscriberClientWrapper : public StatestoreSubscriberClient {
recv_UpdateCatalogd(_return);
}
+ void UpdateStatestoredRole(TUpdateStatestoredRoleResponse& _return,
+ const TUpdateStatestoredRoleRequest& params, bool* send_done) {
+ DCHECK(!*send_done);
+ send_UpdateStatestoredRole(params);
+ *send_done = true;
+ recv_UpdateStatestoredRole(_return);
+ }
+
#pragma clang diagnostic pop
};
diff --git a/be/src/statestore/statestore-subscriber.cc b/be/src/statestore/statestore-subscriber.cc
index 78582e0e3..4cb281f2f 100644
--- a/be/src/statestore/statestore-subscriber.cc
+++ b/be/src/statestore/statestore-subscriber.cc
@@ -69,6 +69,7 @@ DEFINE_int64_hidden(statestore_subscriber_recovery_grace_period_ms, 30000L, "Per
DEFINE_int32(statestore_client_rpc_timeout_ms, 300000, "(Advanced) The underlying "
"TSocket send/recv timeout in milliseconds for a catalog client RPC.");
+DECLARE_bool(enable_statestored_ha);
DECLARE_bool(tolerate_statestore_startup_delay);
DECLARE_string(debug_actions);
DECLARE_string(ssl_client_ca_certificate);
@@ -120,6 +121,10 @@ class StatestoreSubscriberThriftIf : public StatestoreSubscriberIf {
.ToThrift(&response.status);
// Make sure Thrift thinks the field is set.
response.__set_skipped(response.skipped);
+ } else {
+ LOG(WARNING) << Substitute("Receive UpdateState RPC request from incompatible "
+ "statestored (protocol version $0), the request is ignored.",
+ params.protocol_version);
}
TStatus thrift_status;
status.ToThrift(&thrift_status);
@@ -130,13 +135,29 @@ class StatestoreSubscriberThriftIf : public StatestoreSubscriberIf {
Status status = CheckProtocolVersion(request.protocol_version);
if (status.ok()) {
TUniqueId registration_id, statestore_id;
+ bool request_statestore_conn_state = false;
+ TStatestoreConnState::type active_statestore_conn_state =
+ TStatestoreConnState::UNKNOWN;
if (request.__isset.registration_id) {
registration_id = request.registration_id;
}
if (request.__isset.statestore_id) {
statestore_id = request.statestore_id;
}
- subscriber_->Heartbeat(registration_id, statestore_id);
+ if (request.__isset.request_statestore_conn_state) {
+ request_statestore_conn_state = request.request_statestore_conn_state;
+ }
+ subscriber_->Heartbeat(registration_id, statestore_id,
+ request_statestore_conn_state, &active_statestore_conn_state);
+ if (request_statestore_conn_state) {
+ // Send connection state with active statestore instance to standby statestore
+ // instance.
+ response.__set_statestore_conn_state(active_statestore_conn_state);
+ }
+ } else {
+ LOG(WARNING) << Substitute("Receive Heartbeat RPC request from incompatible "
+ "statestored (protocol version $0), the request is ignored.",
+ request.protocol_version);
}
TStatus thrift_status;
status.ToThrift(&thrift_status);
@@ -152,6 +173,38 @@ class StatestoreSubscriberThriftIf : public StatestoreSubscriberIf {
request.registration_id, request.statestore_id, request.catalogd_version,
&update_skipped);
response.__set_skipped(update_skipped);
+ } else {
+ LOG(WARNING) << Substitute("Receive UpdateCatalogd RPC request from incompatible "
+ "statestored (protocol version $0), the request is ignored.",
+ request.protocol_version);
+ }
+ TStatus thrift_status;
+ status.ToThrift(&thrift_status);
+ response.__set_status(thrift_status);
+ }
+
+ virtual void UpdateStatestoredRole(TUpdateStatestoredRoleResponse& response,
+ const TUpdateStatestoredRoleRequest& request) {
+ bool update_skipped = false;
+ Status status = CheckProtocolVersion(request.protocol_version);
+ if (status.ok()) {
+ bool update_active_catalogd = false;
+ int64_t active_catalogd_version = 0;
+ TCatalogRegistration catalogd_registration;
+ if (request.__isset.catalogd_version) {
+ update_active_catalogd = true;
+ active_catalogd_version = request.catalogd_version;
+ catalogd_registration = request.catalogd_registration;
+ }
+ subscriber_->UpdateStatestoredRole(request.is_active, request.registration_id,
+ request.statestore_id, request.active_statestored_version,
+ update_active_catalogd, &catalogd_registration, active_catalogd_version,
+ &update_skipped);
+ response.__set_skipped(update_skipped);
+ } else {
+ LOG(WARNING) << Substitute("Receive UpdateStatestoredRole RPC request from "
+ "incompatible statestored (protocol version $0), the request is ignored.",
+ request.protocol_version);
}
TStatus thrift_status;
status.ToThrift(&thrift_status);
@@ -173,7 +226,8 @@ class StatestoreSubscriberThriftIf : public StatestoreSubscriberIf {
StatestoreSubscriber::StatestoreSubscriber(const string& subscriber_id,
const TNetworkAddress& heartbeat_address, const TNetworkAddress& statestore_address,
- MetricGroup* metrics, TStatestoreSubscriberType::type subscriber_type)
+ const TNetworkAddress& statestore2_address, MetricGroup* metrics,
+ TStatestoreSubscriberType::type subscriber_type)
: subscriber_id_(subscriber_id),
protocol_version_(StatestoreServiceVersion::V2),
catalog_protocol_version_(CatalogServiceVersion::V2),
@@ -183,34 +237,58 @@ StatestoreSubscriber::StatestoreSubscriber(const string& subscriber_id,
client_cache_(new StatestoreClientCache(1, 0, FLAGS_statestore_client_rpc_timeout_ms,
FLAGS_statestore_client_rpc_timeout_ms, "",
!FLAGS_ssl_client_ca_certificate.empty())),
- metrics_(metrics->GetOrCreateChildGroup("statestore-subscriber")) {
- statestore_ = new StatestoreStub(this, statestore_address, metrics);
+ metrics_(metrics->GetOrCreateChildGroup("statestore-subscriber")),
+ enable_statestored_ha_(FLAGS_enable_statestored_ha) {
+ statestore_ =
+ new StatestoreStub(this, /* first_statestore */true, statestore_address, metrics);
+ if (!enable_statestored_ha_) {
+ statestore_->SetStatestoreActive(true, 1);
+ active_statestore_ = statestore_;
+ } else {
+ DCHECK(!statestore2_address.hostname.empty() && statestore2_address.port != 0);
+ statestore2_ = new StatestoreStub(
+ this, /* first_statestore */false, statestore2_address, metrics);
+ }
client_cache_->InitMetrics(metrics, "statestore-subscriber.statestore");
}
Status StatestoreSubscriber::AddTopic(const Statestore::TopicId& topic_id,
bool is_transient, bool populate_min_subscriber_topic_version,
string filter_prefix, const UpdateCallback& callback) {
- return statestore_->AddTopic(topic_id, is_transient,
- populate_min_subscriber_topic_version, filter_prefix, callback);
+ RETURN_IF_ERROR(statestore_->AddTopic(topic_id, is_transient,
+ populate_min_subscriber_topic_version, filter_prefix, callback));
+ if (statestore2_ != nullptr) {
+ RETURN_IF_ERROR(statestore2_->AddTopic(topic_id, is_transient,
+ populate_min_subscriber_topic_version, filter_prefix, callback));
+ }
+ return Status::OK();
}
void StatestoreSubscriber::AddUpdateCatalogdTopic(
const UpdateCatalogdCallback& callback) {
statestore_->AddUpdateCatalogdTopic(callback);
+ if (statestore2_ != nullptr) {
+ statestore2_->AddUpdateCatalogdTopic(callback);
+ }
}
void StatestoreSubscriber::AddCompleteRegistrationTopic(
const CompleteRegistrationCallback& callback) {
statestore_->AddCompleteRegistrationTopic(callback);
+ if (statestore2_ != nullptr) {
+ statestore2_->AddCompleteRegistrationTopic(callback);
+ }
}
-bool StatestoreSubscriber::IsInPostRecoveryGracePeriod() const {
- return statestore_->IsInPostRecoveryGracePeriod();
+bool StatestoreSubscriber::IsInPostRecoveryGracePeriod() {
+ StatestoreStub* active_statestore = GetActiveStatestore();
+ return (active_statestore != nullptr) ?
+ active_statestore->IsInPostRecoveryGracePeriod() : false;
}
-bool StatestoreSubscriber::IsRegistered() const {
- return statestore_->IsRegistered();
+bool StatestoreSubscriber::IsRegistered() {
+ StatestoreStub* active_statestore = GetActiveStatestore();
+ return (active_statestore != nullptr) ? active_statestore->IsRegistered() : false;
}
Status StatestoreSubscriber::Start() {
@@ -246,7 +324,30 @@ Status StatestoreSubscriber::Start() {
// Specify the port which the heartbeat server is listening on.
heartbeat_address_.port = heartbeat_server_->port();
- return statestore_->Start();
+ if (!enable_statestored_ha_) {
+ return statestore_->Start();
+ } else {
+ DCHECK(statestore2_ != nullptr);
+ bool statestore_is_active = false;
+ bool statestore2_is_active = false;
+ RETURN_IF_ERROR(statestore_->Start(&statestore_is_active));
+ RETURN_IF_ERROR(statestore2_->Start(&statestore2_is_active));
+ lock_guard<mutex> r(statestore_ha_lock_);
+ DCHECK(!statestore_is_active || !statestore2_is_active);
+ if (statestore_is_active) {
+ active_statestore_ = statestore_;
+ standby_statestore_ = statestore2_;
+ } else if (statestore2_is_active) {
+ active_statestore_ = statestore2_;
+ standby_statestore_ = statestore_;
+ } else {
+ // Both statestoreds are not ready. If FLAGS_tolerate_statestore_startup_delay is
+ // as true, subscriber enter recovery mode.
+ active_statestore_ = nullptr;
+ standby_statestore_ = nullptr;
+ }
+ }
+ return Status::OK();
}
/// Set Register Request
@@ -265,15 +366,34 @@ Status StatestoreSubscriber::SetRegisterRequest(
request->__set_subscriber_location(heartbeat_address);
request->__set_subscriber_id(subscriber_id_);
request->__set_subscriber_type(subscriber_type_);
- request->__set_subscribe_catalogd_change(statestore_->IsSubscribedCatalogdChange());
return Status::OK();
}
-void StatestoreSubscriber::Heartbeat(
- const RegistrationId& registration_id, const TUniqueId& statestore_id) {
+void StatestoreSubscriber::Heartbeat(const RegistrationId& registration_id,
+ const TUniqueId& statestore_id, bool request_active_conn_state,
+ TStatestoreConnState::type* active_statestore_conn_state) {
// It's possible the heartbeat is received for previous registration.
- if (statestore_->IsMatchingStatestoreId(statestore_id)) {
+ DCHECK(statestore_ != nullptr);
+ if (statestore_->IsMatchingStatestoreId(statestore_id) || statestore2_ == nullptr) {
+ // Due to race, subscriber could receive heartbeat before receiving registration
+ // response. "statestore2_ == nullptr" means HA is not enabled. In that case,
+ // we still handle the heartbeat request from statestore even registration response
+ // is not received.
statestore_->Heartbeat(registration_id);
+ // Report connection state with active statestore instance for the request from
+ // standby statestore.
+ if (request_active_conn_state && !statestore_->IsStatestoreActive()
+ && statestore2_ != nullptr) {
+ *active_statestore_conn_state = statestore2_->GetStatestoreConnState();
+ }
+ } else if (statestore2_ != nullptr
+ && statestore2_->IsMatchingStatestoreId(statestore_id)) {
+ statestore2_->Heartbeat(registration_id);
+ // Report connection state with active statestore instance for the request from
+ // standby statestore.
+ if (request_active_conn_state && !statestore2_->IsStatestoreActive()) {
+ *active_statestore_conn_state = statestore_->GetStatestoreConnState();
+ }
} else {
VLOG(3) << "Ignore heartbeat message from unknown statestored: " << statestore_id;
}
@@ -283,46 +403,103 @@ void StatestoreSubscriber::UpdateCatalogd(
const TCatalogRegistration& catalogd_registration,
const RegistrationId& registration_id, const TUniqueId& statestore_id,
int64_t active_catalogd_version, bool* update_skipped) {
- if (statestore_->IsMatchingStatestoreId(statestore_id)) {
- statestore_->UpdateCatalogd(
- catalogd_registration, registration_id, active_catalogd_version, update_skipped);
+ // Accept UpdateCatalogd RPC from active statestore
+ StatestoreStub* active_statestore = GetActiveStatestore();
+ if (active_statestore != nullptr
+ && active_statestore->IsMatchingStatestoreId(statestore_id)) {
+ active_statestore->UpdateCatalogd(catalogd_registration, registration_id,
+ active_catalogd_version, /* statestore_failover */false, update_skipped);
} else {
// It's possible the catalogd update RPC is received before the registration response
// is received. Skip this update so that the statestore will retry this update in
// the future.
*update_skipped = true;
- LOG(INFO) << "Skipped updating catalogd message from unknown statestored: "
+ LOG(INFO) << "Skipped updating catalogd message from unknown or inactive "
+ << "statestored: " << statestore_id;
+ }
+}
+
+void StatestoreSubscriber::UpdateStatestoredRole(bool is_active,
+ const RegistrationId& registration_id, const TUniqueId& statestore_id,
+ int64_t active_statestored_version, bool update_active_catalogd,
+ const TCatalogRegistration* catalogd_registration, int64_t active_catalogd_version,
+ bool* update_skipped) {
+ DCHECK(enable_statestored_ha_);
+ // Accept UpdateStatestoredRole RPC from standby statestored
+ StatestoreStub* standby_statestore = GetStandbyStatestore();
+ if (standby_statestore != nullptr
+ && standby_statestore_->IsMatchingStatestoreId(statestore_id)) {
+ // Receive notification of statestore service fail over, switch active and standby
+ // statestoreds.
+ standby_statestore->IncCountForUpdateStatestoredRoleRPC();
+ DCHECK(is_active);
+ {
+ lock_guard<mutex> r(statestore_ha_lock_);
+ StatestoreStub* tmp = active_statestore_;
+ active_statestore_ = standby_statestore_;
+ standby_statestore_ = tmp;
+ active_statestore_->SetStatestoreActive(is_active, active_statestored_version);
+ standby_statestore_->SetStatestoreActive(!is_active, active_statestored_version);
+ }
+ if (update_active_catalogd) {
+ StatestoreStub* active_statestore = GetActiveStatestore();
+ DCHECK(active_statestore != nullptr);
+ active_statestore->UpdateCatalogd(*catalogd_registration, registration_id,
+ active_catalogd_version, /* statestore_failover */true, update_skipped);
+ DCHECK(!(*update_skipped));
+ }
+ } else {
+ // It's possible the statestored update RPC is received before the registration
+ // response is received. Skip this update so that the statestore will retry this
+ // update in the future.
+ *update_skipped = true;
+ LOG(INFO) << "Skipped updating statestored message from unknown statestored: "
<< statestore_id;
}
}
+StatestoreSubscriber::StatestoreStub* StatestoreSubscriber::GetActiveStatestore() {
+ lock_guard<mutex> r(statestore_ha_lock_);
+ return active_statestore_;
+}
+
+StatestoreSubscriber::StatestoreStub* StatestoreSubscriber::GetStandbyStatestore() {
+ lock_guard<mutex> r(statestore_ha_lock_);
+ DCHECK(standby_statestore_ != nullptr);
+ return standby_statestore_;
+}
+
Status StatestoreSubscriber::UpdateState(const TopicDeltaMap& incoming_topic_deltas,
const RegistrationId& registration_id, const TUniqueId& statestore_id,
vector<TTopicDelta>* subscriber_topic_updates, bool* skipped) {
- if (statestore_->IsMatchingStatestoreId(statestore_id)) {
- return statestore_->UpdateState(
+ // Accept UpdateState RPC from active statestore.
+ StatestoreStub* active_statestore = GetActiveStatestore();
+ if (active_statestore != nullptr
+ && active_statestore->IsMatchingStatestoreId(statestore_id)) {
+ return active_statestore->UpdateState(
incoming_topic_deltas, registration_id, subscriber_topic_updates, skipped);
} else {
// It's possible the topic update is received before the registration response is
// received. Skip this update so that the statestore will retry this update in the
// future.
*skipped = true;
- VLOG(3) << "Skipped topic update message from unknown statestored: " << statestore_id;
+ VLOG(3) << "Skipped topic update message from unknown or inactive statestored: "
+ << statestore_id;
return Status::OK();
}
}
StatestoreSubscriber::StatestoreStub::StatestoreStub(StatestoreSubscriber* subscriber,
- const TNetworkAddress& statestore_address, MetricGroup* metrics)
+ bool first_statestore, const TNetworkAddress& statestore_address,
+ MetricGroup* metrics)
: subscriber_(subscriber),
statestore_address_(statestore_address),
failure_detector_(
new TimeoutFailureDetector(seconds(FLAGS_statestore_subscriber_timeout_seconds),
seconds(FLAGS_statestore_subscriber_timeout_seconds / 2))),
is_registered_(false) {
- std::string metrics_name_prefix = "";
- metrics_ = metrics->GetOrCreateChildGroup(
- metrics_name_prefix + "statestore-subscriber");
+ std::string name_prefix = first_statestore ? "statestore" : "statestore2";
+ metrics_ = metrics->GetOrCreateChildGroup(name_prefix + "-subscriber");
connected_to_statestore_metric_ = metrics_->AddProperty(
"statestore-subscriber.connected", false);
connection_failure_metric_ = metrics_->AddCounter(
@@ -343,6 +520,10 @@ StatestoreSubscriber::StatestoreStub::StatestoreStub(StatestoreSubscriber* subsc
"statestore-subscriber.statestore-id", "N/A");
update_catalogd_rpc_metric_ = metrics_->AddCounter(
"statestore-subscriber.num-update-catalogd-rpc", 0);
+ update_statestored_role_rpc_metric_ = metrics_->AddCounter(
+ "statestore-subscriber.num-update-statestored-role-rpc", 0);
+ active_status_metric_ = metrics_->AddProperty(
+ name_prefix + "-subscriber.statestore-active-status", is_active_);
re_registr_attempt_metric_ = metrics_->AddCounter(
"statestore-subscriber.num-re-register-attempt", 0);
}
@@ -425,6 +606,7 @@ Status StatestoreSubscriber::StatestoreStub::Register(bool* has_active_catalogd,
thrift_topic.__set_filter_prefix(registration.second.filter_prefix);
request.topic_registrations.push_back(thrift_topic);
}
+ request.__set_subscribe_catalogd_change(IsSubscribedCatalogdChange());
{
// Reset registration_id_ and statestore_id_ before registering with statestore
@@ -480,6 +662,15 @@ Status StatestoreSubscriber::StatestoreStub::Register(bool* has_active_catalogd,
} else {
VLOG(1) << "No statestore ID received from statestore";
}
+ if (status.ok() && response.__isset.statestore_is_active) {
+ is_active_ = response.statestore_is_active;
+ if (is_active_) {
+ DCHECK(response.active_statestored_version > 0
+ && active_statestored_version_ <= response.active_statestored_version);
+ }
+ active_statestored_version_ = response.active_statestored_version;
+ active_status_metric_->SetValue(is_active_);
+ }
if (status.ok() && response.__isset.catalogd_registration) {
VLOG(1) << "Active catalogd address: "
<< TNetworkAddressToString(response.catalogd_registration.address);
@@ -496,7 +687,7 @@ Status StatestoreSubscriber::StatestoreStub::Register(bool* has_active_catalogd,
return status;
}
-Status StatestoreSubscriber::StatestoreStub::Start() {
+Status StatestoreSubscriber::StatestoreStub::Start(bool* startstore_is_active) {
Status status;
{
bool has_active_catalogd = false;
@@ -515,6 +706,7 @@ Status StatestoreSubscriber::StatestoreStub::Start() {
}
if (status.ok()) {
is_registered_ = true;
+ if (startstore_is_active != nullptr) *startstore_is_active = is_active_;
LOG(INFO) << "statestore registration successful on startup";
if (has_active_catalogd) {
DCHECK(active_catalogd_version >= 0);
@@ -557,6 +749,7 @@ void StatestoreSubscriber::StatestoreStub::RecoveryModeChecker() {
// that the statestore has failed from our perspective. If so, enter recovery
// mode and try to reconnect, followed by reregistering all subscriptions.
while (true) {
+ bool updateStatestoredRole = false;
FailureDetector::PeerState state = failure_detector_->GetPeerState(STATESTORE_ID);
if (state == FailureDetector::FAILED
|| (state == FailureDetector::UNKNOWN && !is_registered)) {
@@ -588,6 +781,11 @@ void StatestoreSubscriber::StatestoreStub::RecoveryModeChecker() {
callback();
}
}
+ if (is_active_) {
+ updateStatestoredRole = true;
+ LOG(INFO) << "Statestored " << TNetworkAddressToString(statestore_address_)
+ << " is active.";
+ }
// Make sure to update failure detector so that we don't immediately fail on
// the next loop while we're waiting for heartbeat messages to resume.
failure_detector_->UpdateHeartbeat(STATESTORE_ID, true);
@@ -624,6 +822,19 @@ void StatestoreSubscriber::StatestoreStub::RecoveryModeChecker() {
last_recovery_time_metric_->SetValue(CurrentTimeString());
}
+ if (FLAGS_enable_statestored_ha && updateStatestoredRole) {
+ LOG(INFO) << "Update the roles of statestoreds";
+ bool update_skipped = false;
+ bool is_active = false;
+ RegistrationId registration_id;
+ TUniqueId statestore_id;
+ int64_t active_statestored_version = GetActiveVersion(&is_active);
+ GetRegistrationIdAndStatestoreId(®istration_id, &statestore_id);
+ subscriber_->UpdateStatestoredRole(is_active, registration_id,
+ statestore_id, active_statestored_version, /* update_active_catalogd */false,
+ nullptr, 0, &update_skipped);
+ DCHECK(!update_skipped);
+ }
SleepForMs(SLEEP_INTERVAL_MS);
}
}
@@ -667,20 +878,32 @@ void StatestoreSubscriber::StatestoreStub::Heartbeat(
void StatestoreSubscriber::StatestoreStub::UpdateCatalogd(
const TCatalogRegistration& catalogd_registration,
const RegistrationId& registration_id, int64_t active_catalogd_version,
- bool* update_skipped) {
+ bool statestore_failover, bool* update_skipped) {
const Status& status = CheckRegistrationId(registration_id);
if (status.ok()) {
- // Try to acquire lock to avoid race with updating catalogd from registration thread.
- shared_lock<shared_mutex> l(lock_, boost::try_to_lock);
- if (!l.owns_lock()) {
- LOG(INFO) << "Unable to acquire the lock, skip UpdateCatalogd RPC notification.";
- *update_skipped = true;
- return;
- }
- update_catalogd_rpc_metric_->Increment(1);
- DCHECK(active_catalogd_version >= 0);
- for (const UpdateCatalogdCallback& callback : update_catalogd_callbacks_) {
- callback(false, active_catalogd_version, catalogd_registration);
+ if (statestore_failover) {
+ lock_guard<shared_mutex> exclusive_lock(lock_);
+ DCHECK(active_catalogd_version >= 0);
+ for (const UpdateCatalogdCallback& callback : update_catalogd_callbacks_) {
+ // Set is_registration_reply as true like new registration if statestore service
+ // fail over to standby statestored.
+ callback(/* is_registration_reply */true, active_catalogd_version,
+ catalogd_registration);
+ }
+ } else {
+ // Try to acquire lock to avoid race with updating catalogd from registration
+ // thread.
+ shared_lock<shared_mutex> l(lock_, boost::try_to_lock);
+ if (!l.owns_lock()) {
+ LOG(INFO) << "Unable to acquire the lock, skip UpdateCatalogd RPC notification.";
+ *update_skipped = true;
+ return;
+ }
+ update_catalogd_rpc_metric_->Increment(1);
+ DCHECK(active_catalogd_version >= 0);
+ for (const UpdateCatalogdCallback& callback : update_catalogd_callbacks_) {
+ callback(false, active_catalogd_version, catalogd_registration);
+ }
}
} else {
// It's possible the registration is not completed.
@@ -831,4 +1054,52 @@ bool StatestoreSubscriber::StatestoreStub::IsRegistered() {
return is_registered_;
}
+void StatestoreSubscriber::StatestoreStub::SetStatestoreActive(
+ bool is_active, int64_t active_statestored_version) {
+ lock_guard<shared_mutex> exclusive_lock(lock_);
+ is_active_ = is_active;
+ DCHECK(active_statestored_version_ <= active_statestored_version);
+ active_statestored_version_ = active_statestored_version;
+ active_status_metric_->SetValue(is_active);
+}
+
+bool StatestoreSubscriber::StatestoreStub::IsStatestoreActive() {
+ lock_guard<shared_mutex> exclusive_lock(lock_);
+ return is_active_;
+}
+
+int64_t StatestoreSubscriber::StatestoreStub::GetActiveVersion(bool* is_active) {
+ lock_guard<shared_mutex> exclusive_lock(lock_);
+ *is_active = is_active_;
+ return active_statestored_version_;
+}
+
+void StatestoreSubscriber::StatestoreStub::GetRegistrationIdAndStatestoreId(
+ RegistrationId* registration_id, TUniqueId* statestore_id) {
+ lock_guard<shared_mutex> exclusive_lock(lock_);
+ *registration_id = registration_id_;
+ *statestore_id = statestore_id_;
+}
+
+void StatestoreSubscriber::StatestoreStub::IncCountForUpdateStatestoredRoleRPC() {
+ lock_guard<shared_mutex> exclusive_lock(lock_);
+ update_statestored_role_rpc_metric_->Increment(1);
+}
+
+TStatestoreConnState::type
+StatestoreSubscriber::StatestoreStub::GetStatestoreConnState() {
+ lock_guard<shared_mutex> exclusive_lock(lock_);
+ FailureDetector::PeerState state = failure_detector_->GetPeerState(STATESTORE_ID);
+ switch (state) {
+ case FailureDetector::FAILED:
+ return TStatestoreConnState::FAILED;
+ case FailureDetector::UNKNOWN:
+ return TStatestoreConnState::UNKNOWN;
+ case FailureDetector::SUSPECTED:
+ case FailureDetector::OK:
+ default:
+ return TStatestoreConnState::OK;
+ }
+}
+
}
diff --git a/be/src/statestore/statestore-subscriber.h b/be/src/statestore/statestore-subscriber.h
index eea999d3b..5f406b567 100644
--- a/be/src/statestore/statestore-subscriber.h
+++ b/be/src/statestore/statestore-subscriber.h
@@ -73,10 +73,12 @@ class StatestoreSubscriber {
/// heartbeat_address - the local address on which the heartbeat service which
/// communicates with the statestore should be started.
/// statestore_address - the address of the statestore to register with
+ /// statestore2_address - the address of the second statestore to register with
/// subscriber_type - subscriber type
StatestoreSubscriber(const std::string& subscriber_id,
const TNetworkAddress& heartbeat_address,
const TNetworkAddress& statestore_address,
+ const TNetworkAddress& statestore2_address,
MetricGroup* metrics,
TStatestoreSubscriberType::type subscriber_type);
@@ -166,10 +168,10 @@ class StatestoreSubscriber {
/// Returns true if the statestore has recovered and the configurable post-recovery
/// grace period has not yet elapsed.
- bool IsInPostRecoveryGracePeriod() const;
+ bool IsInPostRecoveryGracePeriod();
/// Returns true if the registration with statestore is completed.
- bool IsRegistered() const;
+ bool IsRegistered();
private:
/// Unique, but opaque, identifier for this subscriber.
@@ -247,11 +249,16 @@ class StatestoreSubscriber {
/// tracks variety of bookkeeping information.
class StatestoreStub {
public:
- StatestoreStub(StatestoreSubscriber* subscriber,
+ StatestoreStub(StatestoreSubscriber* subscriber, bool first_statestore,
const TNetworkAddress& statestore_address, MetricGroup* metrics);
+ /// Start to register with the statestore server.
/// Returns OK unless some error occurred, like a failure to connect.
- Status Start();
+ /// Note: If the statestore server is not ready, the registration will fail and the
+ /// subscriber enter recovery mode. In recovery mode, the class-wide lock_ is taken
+ /// to ensure mutual exclusion with any operations in flight. To avoid accessing
+ /// member variable is_active_, return its value in the given output parameter.
+ Status Start(bool* startstore_is_active = nullptr);
/// Adds a topic to the set of topics that updates will be received
/// for. When a topic update is received, the supplied UpdateCallback
@@ -302,7 +309,7 @@ class StatestoreSubscriber {
/// Called when the catalogd has been updated.
void UpdateCatalogd(const TCatalogRegistration& catalogd_registration,
const RegistrationId& registration_id, int64_t active_catalogd_version,
- bool* update_skipped);
+ bool statestore_failover, bool* update_skipped);
/// Run in a separate thread. In a loop, check failure_detector_ to see if the
/// statestore is still sending heartbeat messages. If not, enter 'recovery mode'
@@ -343,6 +350,23 @@ class StatestoreSubscriber {
/// Returns true if the registration with statestore is completed.
bool IsRegistered();
+ /// Get/set the active state of the registered statestore instance.
+ bool IsStatestoreActive();
+ void SetStatestoreActive(bool is_active, int64_t active_statestored_version);
+
+ /// Return the version of active statestore.
+ int64_t GetActiveVersion(bool* is_active);
+
+ /// Return registration-Id and statestore-Id.
+ void GetRegistrationIdAndStatestoreId(
+ RegistrationId* registration_id, TUniqueId* statestore_id);
+
+ /// Increase count for receiving UpdateStatestoredRole RPC.
+ void IncCountForUpdateStatestoredRoleRPC();
+
+ /// Get connection state with the registered statestore instance.
+ TStatestoreConnState::type GetStatestoreConnState();
+
private:
/// Pointer to parent StatestoreSubscriber object
StatestoreSubscriber* subscriber_;
@@ -358,6 +382,12 @@ class StatestoreSubscriber {
/// comments.
boost::shared_mutex lock_;
+ /// True if the registered statestore instance is active.
+ bool is_active_ = false;
+
+ /// The version of active statestored.
+ int64_t active_statestored_version_ = 0;
+
/// Failure detector that tracks heartbeat messages from the statestore.
boost::scoped_ptr<impala::TimeoutFailureDetector> failure_detector_;
@@ -399,6 +429,13 @@ class StatestoreSubscriber {
/// Metric to count the total number of UpdateCatalogd RPCs received by subscriber.
IntCounter* update_catalogd_rpc_metric_;
+ /// Metric that tracks if the statestored is active when statestored HA is enabled.
+ BooleanProperty* active_status_metric_;
+
+ /// Metric to count the total number of UpdateStatestoredRole RPCs received by
+ /// subscriber.
+ IntCounter* update_statestored_role_rpc_metric_;
+
/// Metric to count the total number of re-registration attempt.
IntCounter* re_registr_attempt_metric_;
@@ -446,8 +483,22 @@ class StatestoreSubscriber {
AtomicInt64 last_registration_ms_{0};
};
- // statestore instances
+ /// Set to true if statestored HA is enabled.
+ bool enable_statestored_ha_ = false;
+
+ /// Pointers of two StatestoreStub objects
StatestoreStub* statestore_ = nullptr;
+ StatestoreStub* statestore2_ = nullptr;
+
+ /// Lock to protect statestore HA related variables: active_statestore_,
+ /// standby_statestore_
+ std::mutex statestore_ha_lock_;
+
+ /// Pointers for active and standby StatestoreStub objects. Subscriber find active
+ /// statestore during registration, or get notification when statestore service
+ /// fail over to standby statestore.
+ StatestoreStub* active_statestore_ = nullptr;
+ StatestoreStub* standby_statestore_ = nullptr;
/// Called when the statestore sends a topic update.
Status UpdateState(const TopicDeltaMap& incoming_topic_deltas,
@@ -456,13 +507,25 @@ class StatestoreSubscriber {
/// Called when the statestore sends a heartbeat message.
void Heartbeat(const RegistrationId& registration_id,
- const TUniqueId& statestore_id);
+ const TUniqueId& statestore_id, bool request_active_conn_state,
+ TStatestoreConnState::type* active_statestore_conn_state);
- /// Called when the catalogd has been updated.
+ /// Called when the active catalogd has been updated.
void UpdateCatalogd(const TCatalogRegistration& catalogd_registration,
const RegistrationId& registration_id, const TUniqueId& statestore_id,
int64_t active_catalogd_version, bool* update_skipped);
+ /// Called when the active statestore has been updated.
+ void UpdateStatestoredRole(bool is_active,
+ const RegistrationId& registration_id, const TUniqueId& statestore_id,
+ int64_t active_statestored_version, bool update_active_catalogd,
+ const TCatalogRegistration* catalogd_registration, int64_t active_catalogd_version,
+ bool* update_skipped);
+
+ /// Return the pointer of the active/standby StatestoreStub.
+ StatestoreStub* GetActiveStatestore();
+ StatestoreStub* GetStandbyStatestore();
+
/// Subscriber thrift implementation, needs to access UpdateState
friend class StatestoreSubscriberThriftIf;
};
diff --git a/be/src/statestore/statestore-test.cc b/be/src/statestore/statestore-test.cc
index beed00df6..2a13d05f8 100644
--- a/be/src/statestore/statestore-test.cc
+++ b/be/src/statestore/statestore-test.cc
@@ -57,14 +57,14 @@ TEST(StatestoreTest, SmokeTest) {
StatestoreSubscriber* sub_will_start = perm_objects->Add(
new StatestoreSubscriber("sub1", MakeNetworkAddress("localhost", 0),
- MakeNetworkAddress("localhost", statestore->port()),
+ MakeNetworkAddress("localhost", statestore->port()), MakeNetworkAddress("", 0),
new MetricGroup(""), TStatestoreSubscriberType::COORDINATOR_EXECUTOR));
ASSERT_OK(sub_will_start->Start());
// Confirm that a subscriber trying to use an in-use port will fail to start.
StatestoreSubscriber* sub_will_not_start = perm_objects->Add(new StatestoreSubscriber(
"sub3", MakeNetworkAddress("localhost", sub_will_start->heartbeat_port()),
- MakeNetworkAddress("localhost", statestore->port()),
+ MakeNetworkAddress("localhost", statestore->port()), MakeNetworkAddress("", 0),
new MetricGroup(""), TStatestoreSubscriberType::COORDINATOR_EXECUTOR));
ASSERT_FALSE(sub_will_not_start->Start().ok());
@@ -93,7 +93,7 @@ void SslSmokeTestHelper(const string& server_ca_certificate,
StatestoreSubscriber* sub = perm_objects->Add(
new StatestoreSubscriber("smoke_sub", MakeNetworkAddress("localhost", 0),
- MakeNetworkAddress("localhost", statestore->port()),
+ MakeNetworkAddress("localhost", statestore->port()), MakeNetworkAddress("", 0),
new MetricGroup(""), TStatestoreSubscriberType::COORDINATOR_EXECUTOR));
Status sub_status = sub->Start();
ASSERT_EQ(sub_should_start, sub_status.ok());
diff --git a/be/src/statestore/statestore.cc b/be/src/statestore/statestore.cc
index f946e54b5..4a598f610 100644
--- a/be/src/statestore/statestore.cc
+++ b/be/src/statestore/statestore.cc
@@ -21,6 +21,7 @@
#include <tuple>
#include <utility>
+#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/lexical_cast.hpp>
#include <thrift/Thrift.h>
#include <thrift/protocol/TProtocolException.h>
@@ -47,6 +48,7 @@
#include "common/names.h"
+using boost::posix_time::seconds;
using boost::shared_lock;
using boost::shared_mutex;
using boost::upgrade_lock;
@@ -83,8 +85,6 @@ DEFINE_int32(statestore_heartbeat_frequency_ms, 1000, "(Advanced) Frequency (in
DEFINE_double_hidden(heartbeat_monitoring_frequency_ms, 60000, "(Advanced) Frequency (in "
"ms) with which the statestore monitors heartbeats from a subscriber.");
-DEFINE_int32(state_store_port, 24000, "port where StatestoreService is running");
-
DEFINE_int32(statestore_heartbeat_tcp_timeout_seconds, 3, "(Advanced) The time after "
"which a heartbeat RPC to a subscriber will timeout. This setting protects against "
"badly hung machines that are not able to respond to the heartbeat RPC in short "
@@ -112,6 +112,52 @@ DEFINE_int32(statestore_update_catalogd_tcp_timeout_seconds, 3, "(Advanced) The
"protects against badly hung machines that are not able to respond to the "
"UpdateCatalogd RPC in short order");
+// Flags for Statestore HA
+// Host and port of peer's StatestoreHaService for Statestore HA.
+DEFINE_string(state_store_peer_host, "localhost",
+ "hostname where peer's StatestoreHaService is running");
+DEFINE_int32(state_store_peer_ha_port, 24021,
+ "port where peer's StatestoreHaService is running");
+DEFINE_bool(enable_statestored_ha, false, "Set to true to enable Statestore HA");
+DEFINE_bool(statestore_force_active, false, "Set to true to force this statestored "
+ "instance to take active role. It's used to perform manual fail over for statestore "
+ "service.");
+// Use network address as priority value of statestored instance when designating active
+// statestored. The lower network address corresponds to a higher priority.
+// This is mainly used in unit-test for predictable results.
+DEFINE_bool(use_network_address_as_statestore_priority, false, "Network address is "
+ "used as priority value of statestored instance if this is set as true. Otherwise, "
+ "statestored_id which is generated as random number will be used as priority value "
+ "of statestored instance.");
+// Waiting period in ms for HA preemption. It should be set with proper value based on the
+// time to take for bringing a statestored instance in-line in the deployment environment.
+DEFINE_int64(statestore_ha_preemption_wait_period_ms, 10000, "(Advanced) The time after "
+ "which statestored designates itself as active role if the statestore does not "
+ "receive HA handshake request/response from peer statestored.");
+DEFINE_double_hidden(statestore_ha_heartbeat_monitoring_frequency_ms, 1000, "(Advanced) "
+ "Frequency (in ms) with which the statestore monitors HA heartbeats from active "
+ "statestore.");
+DEFINE_int32(statestore_update_statestore_tcp_timeout_seconds, 3, "(Advanced) The "
+ "time after which a UpdateStatestoredRole RPC to a subscriber will timeout. This "
+ "setting protects against badly hung machines that are not able to respond to the "
+ "UpdateStatestoredRole RPC in short order.");
+DEFINE_int32(statestore_peer_timeout_seconds, 30, "The amount of time (in seconds) that "
+ "may elapse before the connection with the peer statestore is considered lost.");
+DEFINE_int32(statestore_peer_cnxn_attempts, 10, "The number of times to retry an "
+ "RPC connection to the peer statestore. A setting of 0 means retry indefinitely");
+DEFINE_int32(statestore_peer_cnxn_retry_interval_ms, 1000, "The interval, in ms, "
+ "to wait between attempts to make an RPC connection to the peer statestore. "
+ "It's set as statestore_ha_preemption_wait_period_ms/statestore_peer_cnxn_attempts "
+ "if statestore_peer_cnxn_attempts > 0, default value if "
+ "statestore_peer_cnxn_attempts == 0.");
+DEFINE_int32(statestore_ha_client_rpc_timeout_ms, 300000, "(Advanced) The underlying "
+ "TSocket send/recv timeout in milliseconds for a client RPC of Statestore HA "
+ "service.");
+DEFINE_int64(update_statestore_rpc_resend_interval_ms, 100, "(Advanced) Interval "
+ "(in ms) with which the statestore resends the RPCs of updating statestored's role "
+ "to subscribers if the statestore has failed to send the RPCs to the subscribers.");
+
+DECLARE_string(hostname);
DECLARE_bool(enable_catalogd_ha);
DECLARE_int64(active_catalogd_designation_monitoring_interval_ms);
DECLARE_int64(update_catalogd_rpc_resend_interval_ms);
@@ -125,6 +171,9 @@ DECLARE_string(ssl_minimum_version);
DECLARE_int32(stress_statestore_startup_delay_ms);
#endif
+// Used to identify the statestore in the failure detector
+const string STATESTORE_ID = "STATESTORE";
+
// Metric keys
// TODO: Replace 'backend' with 'subscriber' when we can coordinate a change with CM
const string STATESTORE_LIVE_SUBSCRIBERS = "statestore.live-backends";
@@ -140,9 +189,17 @@ const string STATESTORE_SUCCESSFUL_UPDATE_CATALOGD_RPC_NUM =
"statestore.num-successful-update-catalogd-rpc";
const string STATESTORE_FAILED_UPDATE_CATALOGD_RPC_NUM =
"statestore.num-failed-update-catalogd-rpc";
+const string STATESTORE_SUCCESSFUL_UPDATE_STATESTORED_ROLE_RPC_NUM =
+ "statestore.num-successful-update-statestored-role-rpc";
+const string STATESTORE_FAILED_UPDATE_STATESTORED_ROLE_RPC_NUM =
+ "statestore.num-failed-update-statestored-role-rpc";
const string STATESTORE_CLEAR_TOPIC_ENTRIES_NUM =
"statestore.num-clear-topic-entries-requests";
const string STATESTORE_ACTIVE_CATALOGD_ADDRESS = "statestore.active-catalogd-address";
+const string STATESTORE_ACTIVE_STATUS = "statestore.active-status";
+const string STATESTORE_SERVICE_STARTED = "statestore.service-started";
+const string STATESTORE_IN_HA_RECOVERY = "statestore.in-ha-recovery-mode";
+const string STATESTORE_CONNECTED_PEER = "statestore.connected-with-peer-statestored";
// Initial version for each Topic registered by a Subscriber. Generally, the Topic will
// have a Version that is the MAX() of all entries in the Topic, but this initial
@@ -158,6 +215,7 @@ const char* Statestore::IMPALA_MEMBERSHIP_TOPIC = "impala-membership";
const char* Statestore::IMPALA_REQUEST_QUEUE_TOPIC = "impala-request-queue";
typedef ClientConnection<StatestoreSubscriberClientWrapper> StatestoreSubscriberConn;
+typedef ClientConnection<StatestoreHaServiceClientWrapper> StatestoreHaServiceConn;
namespace impala {
@@ -233,7 +291,12 @@ class StatestoreThriftIf : public StatestoreServiceIf {
response.__set_registration_id(registration_id);
response.__set_statestore_id(statestore_->GetStateStoreId());
response.__set_protocol_version(statestore_->GetProtocolVersion());
- if (has_active_catalogd) {
+ bool is_active_statestored = false;
+ int64_t active_statestored_version =
+ statestore_->GetActiveVersion(&is_active_statestored);
+ response.__set_statestore_is_active(is_active_statestored);
+ response.__set_active_statestored_version(active_statestored_version);
+ if (is_active_statestored && has_active_catalogd) {
response.__set_catalogd_registration(active_catalogd_registration);
response.__set_catalogd_version(active_catalogd_version);
}
@@ -249,6 +312,73 @@ class StatestoreThriftIf : public StatestoreServiceIf {
return;
}
+ virtual void SetStatestoreDebugAction(TSetStatestoreDebugActionResponse& response,
+ const TSetStatestoreDebugActionRequest& params) {
+ if (params.__isset.disable_network) {
+ bool disable_network = params.disable_network;
+ LOG(INFO) << (disable_network ? "Disable" : "Enable")
+ << " statestored network";
+ statestore_->SetStatestoreDebugAction(disable_network);
+ }
+ Status status = Status::OK();
+ status.ToThrift(&response.status);
+ return;
+ }
+
+ private:
+ Statestore* statestore_;
+};
+
+class StatestoreHaThriftIf : public StatestoreHaServiceIf {
+ public:
+ StatestoreHaThriftIf(Statestore* statestore) : statestore_(statestore) {
+ DCHECK(statestore_ != NULL);
+ }
+
+ // Receive HA handshake request from peer statestore instance.
+ // Each statestore instance start this StatestoreHaService server and each has client
+ // cache. They use client to negotiate active-standby role with peer. If retry failed
+ // for 3 times, assume its peer is not started and the instance take the active role.
+ // Otherwise, designate the role based on the priorities.
+ virtual void StatestoreHaHandshake(TStatestoreHaHandshakeResponse& response,
+ const TStatestoreHaHandshakeRequest& params) {
+ if (params.src_protocol_version < statestore_->GetProtocolVersion()) {
+ // Refuse old version of statestore
+ Status status = Status(TErrorCode::STATESTORE_INCOMPATIBLE_PROTOCOL,
+ statestore_->GetPeerStatestoreHaAddress(), params.src_protocol_version + 1,
+ statestore_->GetProtocolVersion() + 1);
+ status.ToThrift(&response.status);
+ return;
+ } else if (params.src_statestore_id == statestore_->GetStateStoreId()) {
+ // Ignore request sent by itself.
+ return;
+ }
+ bool dst_statestore_active = false;
+ Status status = statestore_->ReceiveHaHandshakeRequest(params.src_statestore_id,
+ params.src_statestore_address, params.src_force_active, &dst_statestore_active);
+ status.ToThrift(&response.status);
+ response.__set_dst_protocol_version(statestore_->GetProtocolVersion());
+ response.__set_dst_statestore_id(statestore_->GetStateStoreId());
+ response.__set_dst_statestore_active(dst_statestore_active);
+ }
+
+ // Receive HA heartbeat from peer statestore instance.
+ virtual void StatestoreHaHeartbeat(TStatestoreHaHeartbeatResponse& response,
+ const TStatestoreHaHeartbeatRequest& params) {
+ if (params.protocol_version < statestore_->GetProtocolVersion()) {
+ // Refuse old version of statestore
+ Status status =
+ Status(TErrorCode::STATESTORE_INCOMPATIBLE_PROTOCOL,
+ statestore_->GetPeerStatestoreHaAddress(), params.protocol_version + 1,
+ statestore_->GetProtocolVersion() + 1);
+ status.ToThrift(&response.status);
+ return;
+ }
+ statestore_->HaHeartbeatRequest(params.dst_statestore_id, params.src_statestore_id);
+ Status status = Status::OK();
+ status.ToThrift(&response.status);
+ }
+
private:
Statestore* statestore_;
};
@@ -539,6 +669,7 @@ Statestore::Statestore(MetricGroup* metrics)
FLAGS_statestore_update_catalogd_tcp_timeout_seconds * 1000, "",
IsInternalTlsConfigured())),
thrift_iface_(new StatestoreThriftIf(this)),
+ ha_thrift_iface_(new StatestoreHaThriftIf(this)),
failure_detector_(new MissedHeartbeatFailureDetector(
FLAGS_statestore_max_missed_heartbeats,
FLAGS_statestore_max_missed_heartbeats / 2)) {
@@ -562,15 +693,50 @@ Statestore::Statestore(MetricGroup* metrics)
metrics->AddCounter(STATESTORE_SUCCESSFUL_UPDATE_CATALOGD_RPC_NUM, 0);
failed_update_catalogd_rpc_metric_ =
metrics->AddCounter(STATESTORE_FAILED_UPDATE_CATALOGD_RPC_NUM, 0);
+ successful_update_statestored_role_rpc_metric_ =
+ metrics->AddCounter(STATESTORE_SUCCESSFUL_UPDATE_STATESTORED_ROLE_RPC_NUM, 0);
+ failed_update_statestored_role_rpc_metric_ =
+ metrics->AddCounter(STATESTORE_FAILED_UPDATE_STATESTORED_ROLE_RPC_NUM, 0);
clear_topic_entries_metric_ =
metrics->AddCounter(STATESTORE_CLEAR_TOPIC_ENTRIES_NUM, 0);
active_catalogd_address_metric_ = metrics->AddProperty<string>(
STATESTORE_ACTIVE_CATALOGD_ADDRESS, "");
+ active_status_metric_ = metrics->AddProperty(STATESTORE_ACTIVE_STATUS, true);
+ service_started_metric_ = metrics->AddProperty(STATESTORE_SERVICE_STARTED, false);
update_state_client_cache_->InitMetrics(metrics, "subscriber-update-state");
heartbeat_client_cache_->InitMetrics(metrics, "subscriber-heartbeat");
update_catalogd_client_cache_->InitMetrics(
metrics, "subscriber-update-catalogd");
+ if (!FLAGS_enable_statestored_ha) {
+ is_active_ = true;
+ active_status_metric_->SetValue(is_active_);
+ active_version_ = UnixMicros();
+ } else {
+ is_active_ = false;
+ active_status_metric_->SetValue(is_active_);
+ active_version_ = 0;
+ update_statestored_client_cache_.reset(new StatestoreSubscriberClientCache(
+ 1, 0, FLAGS_statestore_update_statestore_tcp_timeout_seconds * 1000,
+ FLAGS_statestore_update_statestore_tcp_timeout_seconds * 1000, "",
+ IsInternalTlsConfigured()));
+ update_statestored_client_cache_->InitMetrics(
+ metrics, "subscriber-update-statestored");
+ ha_client_cache_.reset(new StatestoreHaClientCache(1, 0,
+ FLAGS_statestore_ha_client_rpc_timeout_ms,
+ FLAGS_statestore_ha_client_rpc_timeout_ms, "",
+ IsInternalTlsConfigured()));
+ ha_client_cache_->InitMetrics(metrics, "statestored-ha");
+ ha_standby_ss_failure_detector_.reset(new MissedHeartbeatFailureDetector(
+ FLAGS_statestore_max_missed_heartbeats,
+ FLAGS_statestore_max_missed_heartbeats / 2));
+ ha_active_ss_failure_detector_.reset(new TimeoutFailureDetector(
+ seconds(FLAGS_statestore_peer_timeout_seconds),
+ seconds(FLAGS_statestore_peer_timeout_seconds / 2)));
+
+ in_ha_recovery_mode_metric_ = metrics->AddProperty(STATESTORE_IN_HA_RECOVERY, false);
+ connected_peer_metric_ = metrics->AddProperty(STATESTORE_CONNECTED_PEER, false);
+ }
}
Statestore::~Statestore() {
@@ -613,7 +779,7 @@ Status Statestore::Init(int32_t state_store_port) {
RETURN_IF_ERROR(Thread::Create("statestore-update-catalogd", "update-catalogd-thread",
&Statestore::MonitorUpdateCatalogd, this, &update_catalogd_thread_));
service_started_ = true;
-
+ service_started_metric_->SetValue(true);
return Status::OK();
}
@@ -720,6 +886,16 @@ Status Statestore::OfferUpdate(const ScheduledSubscriberUpdate& update,
SubscriberMap::iterator subscriber_it = subscribers_.find(update.subscriber_id);
DCHECK(subscriber_it != subscribers_.end());
subscribers_.erase(subscriber_it);
+ if (FLAGS_enable_statestored_ha) {
+ ActiveConnStateMap::iterator conn_states_it =
+ active_conn_states_.find(update.subscriber_id);
+ if (conn_states_it != active_conn_states_.end()) {
+ if (conn_states_it->second == TStatestoreConnState::FAILED) {
+ --failed_conn_state_count_;
+ }
+ active_conn_states_.erase(conn_states_it);
+ }
+ }
LOG(ERROR) << ss.str();
return Status(ss.str());
}
@@ -743,6 +919,8 @@ Status Statestore::RegisterSubscriber(const SubscriberId& subscriber_id,
} else if (is_catalogd
&& FLAGS_enable_catalogd_ha != catalogd_registration.enable_catalogd_ha) {
return Status("CalaogD HA enabling flag from catalogd does not match.");
+ } else if (disable_network_.Load()) {
+ return Status("Reject registration since network is disabled.");
}
// Create any new topics first, so that when the subscriber is first sent a topic update
@@ -797,6 +975,9 @@ Status Statestore::RegisterSubscriber(const SubscriberId& subscriber_id,
subscriber_id, *registration_id, location, topic_registrations,
subscriber_type, subscribe_catalogd_change));
subscribers_.emplace(subscriber_id, current_registration);
+ if (FLAGS_enable_statestored_ha) {
+ active_conn_states_.emplace(subscriber_id, TStatestoreConnState::OK);
+ }
failure_detector_->UpdateHeartbeat(subscriber_id, true);
num_subscribers_metric_->SetValue(subscribers_.size());
subscriber_set_metric_->Add(subscriber_id);
@@ -814,6 +995,12 @@ Status Statestore::RegisterSubscriber(const SubscriberId& subscriber_id,
return Status::OK();
}
+void Statestore::SetStatestoreDebugAction(bool disable_network) {
+ if (FLAGS_debug_actions == "DISABLE_STATESTORE_NETWORK") {
+ disable_network_.Store(disable_network);
+ }
+}
+
bool Statestore::FindSubscriber(const SubscriberId& subscriber_id,
const RegistrationId& registration_id, shared_ptr<Subscriber>* subscriber) {
DCHECK(subscriber != nullptr);
@@ -827,9 +1014,12 @@ bool Statestore::FindSubscriber(const SubscriberId& subscriber_id,
Status Statestore::SendTopicUpdate(Subscriber* subscriber, UpdateKind update_kind,
bool* update_skipped) {
- // Don't send topic update to inactive catalogd.
- if (FLAGS_enable_catalogd_ha && subscriber->IsCatalogd()
+ if (!IsActive()) {
+ // Don't send topic update if the statestored is not active.
+ return Status::OK();
+ } else if (FLAGS_enable_catalogd_ha && subscriber->IsCatalogd()
&& !catalog_manager_.IsActiveCatalogd(subscriber->id())) {
+ // Don't send topic update to inactive catalogd.
VLOG(3) << "Skip sending topic update to inactive catalogd";
return Status::OK();
}
@@ -1040,6 +1230,10 @@ ThreadPool<Statestore::ScheduledSubscriberUpdate>* Statestore::GetThreadPool(
}
Status Statestore::SendHeartbeat(Subscriber* subscriber) {
+ if (disable_network_.Load()) {
+ return Status("Don't send heartbeat since network is disabled.");
+ }
+
MonotonicStopWatch sw;
sw.Start();
@@ -1052,10 +1246,39 @@ Status Statestore::SendHeartbeat(Subscriber* subscriber) {
THeartbeatResponse response;
request.__set_registration_id(subscriber->registration_id());
request.__set_statestore_id(statestore_id_);
+ if (FLAGS_enable_statestored_ha && !IsActive()) {
+ // Send heartbeat to subscriber with request for connection state between active
+ // statestore and subscriber.
+ request.__set_request_statestore_conn_state(true);
+ }
RETURN_IF_ERROR(
client.DoRpc(&StatestoreSubscriberClientWrapper::Heartbeat, request, &response));
heartbeat_duration_metric_->Update(sw.ElapsedTime() / (1000.0 * 1000.0 * 1000.0));
+ if (FLAGS_enable_statestored_ha && !IsActive()
+ && response.__isset.statestore_conn_state) {
+ TStatestoreConnState::type conn_state = response.statestore_conn_state;
+ lock_guard<mutex> l(subscribers_lock_);
+ if (active_conn_states_.find(subscriber->id()) != active_conn_states_.end()
+ && active_conn_states_[subscriber->id()] != conn_state) {
+ if (conn_state == TStatestoreConnState::FAILED) {
+ DCHECK(active_conn_states_[subscriber->id()] == TStatestoreConnState::OK
+ || active_conn_states_[subscriber->id()] == TStatestoreConnState::UNKNOWN);
+ ++failed_conn_state_count_;
+ } else if (active_conn_states_[subscriber->id()] == TStatestoreConnState::FAILED) {
+ DCHECK(conn_state == TStatestoreConnState::OK
+ || conn_state == TStatestoreConnState::UNKNOWN);
+ --failed_conn_state_count_;
+ } else {
+ DCHECK((active_conn_states_[subscriber->id()] == TStatestoreConnState::OK &&
+ conn_state == TStatestoreConnState::UNKNOWN)
+ || (conn_state == TStatestoreConnState::OK &&
+ active_conn_states_[subscriber->id()] == TStatestoreConnState::UNKNOWN));
+ }
+ // Save the connection state between active statestore and subscriber.
+ active_conn_states_[subscriber->id()] = conn_state;
+ }
+ }
return Status::OK();
}
@@ -1252,6 +1475,9 @@ void Statestore::DoSubscriberUpdate(UpdateKind update_kind, int thread_id,
void Statestore::SendUpdateCatalogdNotification(int64_t* last_active_catalogd_version,
vector<std::shared_ptr<Subscriber>>& rpc_receivers) {
+ // Don't send UpdateCatalogd RPC if the statestore is not active.
+ if (!IsActive()) return;
+
bool has_active_catalogd;
int64_t active_catalogd_version = 0;
TCatalogRegistration catalogd_registration =
@@ -1357,6 +1583,125 @@ void Statestore::SendUpdateCatalogdNotification(int64_t* last_active_catalogd_ve
}
}
+[[noreturn]] void Statestore::MonitorUpdateStatestoredRole() {
+ // rpc_receivers is used to track subscribers to which statestore need to send RPCs
+ // when there is a change in the elected active statestored. It is updated from
+ // subscribers_, and the subscribers will be removed from this list if the RPCs are
+ // successfully sent to them.
+ vector<std::shared_ptr<Subscriber>> rpc_receivers;
+ // Wait for notification. If new statestored leader is elected, send notification
+ // to all subscribers.
+ int64_t timeout_us = FLAGS_update_statestore_rpc_resend_interval_ms * MICROS_PER_MILLI;
+ int64_t last_active_statestored_version = 0;
+ while (1) {
+ {
+ unique_lock<mutex> l(ha_lock_);
+ update_statestored_cv_.WaitFor(l, timeout_us);
+ }
+ SendUpdateStatestoredRoleNotification(
+ &last_active_statestored_version, rpc_receivers);
+ }
+}
+
+void Statestore::SendUpdateStatestoredRoleNotification(
+ int64_t* last_active_statestored_version,
+ vector<std::shared_ptr<Subscriber>>& rpc_receivers) {
+ bool is_active_statestored = false;
+ int64_t active_statestored_version = GetActiveVersion(&is_active_statestored);
+ if (!is_active_statestored ||
+ (active_statestored_version == *last_active_statestored_version
+ && rpc_receivers.empty())) {
+ // Don't resend RPCs if there is no change in active statestored and no RPC failure
+ // in last round.
+ return;
+ }
+ bool has_active_catalogd;
+ int64_t active_catalogd_version = 0;
+ TCatalogRegistration catalogd_registration =
+ catalog_manager_.GetActiveCatalogRegistration(
+ &has_active_catalogd, &active_catalogd_version);
+
+ bool resend_rpc = false;
+ if (active_statestored_version > *last_active_statestored_version) {
+ // Send notification for the latest elected active statestored.
+ LOG(INFO) << "Send notification for active statestored version: "
+ << active_statestored_version;
+ // statestored_active_status_metric_->SetValue(true);
+ rpc_receivers.clear();
+ {
+ lock_guard<mutex> l(subscribers_lock_);
+ for (const auto& subscriber : subscribers_) {
+ rpc_receivers.push_back(subscriber.second);
+ }
+ }
+ *last_active_statestored_version = active_statestored_version;
+ } else {
+ DCHECK(!rpc_receivers.empty());
+ lock_guard<mutex> l(subscribers_lock_);
+ for (std::vector<std::shared_ptr<Subscriber>>::iterator it = rpc_receivers.begin();
+ it != rpc_receivers.end();) {
+ // Don't resend RPC to subscribers which have been removed from subscriber list.
+ std::shared_ptr<Subscriber> subscriber = *it;
+ if (subscribers_.find(subscriber->id()) == subscribers_.end()) {
+ it = rpc_receivers.erase(it);
+ } else {
+ ++it;
+ }
+ }
+ if (rpc_receivers.empty()) return;
+ resend_rpc = true;
+ }
+
+ for (std::vector<std::shared_ptr<Subscriber>>::iterator it = rpc_receivers.begin();
+ it != rpc_receivers.end();) {
+ std::shared_ptr<Subscriber> subscriber = *it;
+ Status status;
+ if (!resend_rpc) {
+ status = DebugAction(
+ FLAGS_debug_actions, "SEND_UPDATE_STATESTORED_RPC_FIRST_ATTEMPT");
+ }
+ if (status.ok()) {
+ StatestoreSubscriberConn client(update_statestored_client_cache_.get(),
+ subscriber->network_address(), &status);
+ if (status.ok()) {
+ TUpdateStatestoredRoleRequest request;
+ TUpdateStatestoredRoleResponse response;
+ request.__set_registration_id(subscriber->registration_id());
+ request.__set_statestore_id(statestore_id_);
+ request.__set_active_statestored_version(active_statestored_version);
+ request.__set_is_active(true);
+ if (has_active_catalogd && subscriber->IsSubscribedCatalogdChange()) {
+ request.__set_catalogd_version(active_catalogd_version);
+ request.__set_catalogd_registration(catalogd_registration);
+ }
+ status = client.DoRpc(&StatestoreSubscriberClientWrapper::UpdateStatestoredRole,
+ request, &response);
+ if (!status.ok() && status.code() == TErrorCode::RPC_RECV_TIMEOUT) {
+ // Add details to status to make it more useful, while preserving the stack
+ status.AddDetail(Substitute(
+ "Subscriber $0 timed-out during update catalogd RPC. Timeout is $1s.",
+ subscriber->id(), FLAGS_statestore_update_catalogd_tcp_timeout_seconds));
+ }
+ }
+ }
+ if (status.ok()) {
+ // Remove the subscriber from the receiver list so that Statestore will not resend
+ // RPC to it in next round.
+ successful_update_statestored_role_rpc_metric_->Increment(1);
+ it = rpc_receivers.erase(it);
+ } else {
+ LOG(ERROR) << "Couldn't send UpdateStatestoredRole RPC, " << status.GetDetail();
+ failed_update_statestored_role_rpc_metric_->Increment(1);
+ // Leave the subscriber in the receiver list. Statestore will resend RPC to it in
+ // next round.
+ ++it;
+ }
+ }
+ if (rpc_receivers.empty()) {
+ LOG(INFO) << "Successfully sent UpdateStatestoredRole RPCs to all subscribers";
+ }
+}
+
void Statestore::UnregisterSubscriber(Subscriber* subscriber) {
SubscriberMap::const_iterator it = subscribers_.find(subscriber->id());
if (it == subscribers_.end() ||
@@ -1369,6 +1714,9 @@ void Statestore::UnregisterSubscriber(Subscriber* subscriber) {
update_state_client_cache_->CloseConnections(subscriber->network_address());
heartbeat_client_cache_->CloseConnections(subscriber->network_address());
update_catalogd_client_cache_->CloseConnections(subscriber->network_address());
+ if (FLAGS_enable_statestored_ha) {
+ update_statestored_client_cache_->CloseConnections(subscriber->network_address());
+ }
// Prevent the failure detector from growing without bound
failure_detector_->EvictPeer(subscriber->id());
@@ -1382,6 +1730,16 @@ void Statestore::UnregisterSubscriber(Subscriber* subscriber) {
num_subscribers_metric_->Increment(-1L);
subscriber_set_metric_->Remove(subscriber->id());
subscribers_.erase(subscriber->id());
+ if (FLAGS_enable_statestored_ha) {
+ ActiveConnStateMap::iterator conn_states_it =
+ active_conn_states_.find(subscriber->id());
+ if (conn_states_it != active_conn_states_.end()) {
+ if (conn_states_it->second == TStatestoreConnState::FAILED) {
+ --failed_conn_state_count_;
+ }
+ active_conn_states_.erase(conn_states_it);
+ }
+ }
}
void Statestore::MainLoop() {
@@ -1414,3 +1772,315 @@ void Statestore::HealthzHandler(
*(data) << "Not Available";
*response = HttpStatusCode::ServiceUnavailable;
}
+
+Status Statestore::InitStatestoreHa(
+ int32_t statestore_ha_port, const TNetworkAddress& peer_statestore_ha_addr) {
+ local_statestore_ha_addr_ = MakeNetworkAddress(FLAGS_hostname, statestore_ha_port);
+ peer_statestore_ha_addr_ = peer_statestore_ha_addr;
+
+ std::shared_ptr<TProcessor> processor(
+ new StatestoreHaServiceProcessor(ha_thrift_iface()));
+ std::shared_ptr<TProcessorEventHandler> event_handler(
+ new RpcEventHandler("StatestoreHa", metrics_));
+ processor->setEventHandler(event_handler);
+ ThriftServerBuilder builder("StatestoreHaService", processor, statestore_ha_port);
+ if (IsInternalTlsConfigured()) {
+ SSLProtocol ssl_version;
+ RETURN_IF_ERROR(
+ SSLProtoVersions::StringToProtocol(FLAGS_ssl_minimum_version, &ssl_version));
+ LOG(INFO) << "Enabling SSL for Statestore";
+ builder.ssl(FLAGS_ssl_server_certificate, FLAGS_ssl_private_key)
+ .pem_password_cmd(FLAGS_ssl_private_key_password_cmd)
+ .ssl_version(ssl_version)
+ .cipher_list(FLAGS_ssl_cipher_list);
+ }
+ ThriftServer* ha_server;
+ RETURN_IF_ERROR(builder.metrics(metrics_).Build(&ha_server));
+ ha_thrift_server_.reset(ha_server);
+ RETURN_IF_ERROR(ha_thrift_server_->Start());
+ // Wait till Thrift server is ready.
+ RETURN_IF_ERROR(WaitForLocalServer(
+ *ha_thrift_server_, /* num_retries */ 10, /* retry_interval_ms */ 1000));
+
+ RETURN_IF_ERROR(Thread::Create("statestore-ha-heartbeat",
+ "ha-heartbeat-monitoring-thread", &Statestore::MonitorStatestoredHaHeartbeat,
+ this, &ha_monitoring_thread_));
+ RETURN_IF_ERROR(Thread::Create("statestore-update-statestored",
+ "update-statestored-thread",&Statestore::MonitorUpdateStatestoredRole, this,
+ &update_statestored_thread_));
+
+ // Negotiate role for HA with peer statestore instance on startup.
+ TStatestoreHaHandshakeResponse response;
+ Status status = SendHaHandshake(&response);
+ if (!status.ok()) {
+ // statestored designates itself as active if the statestore can not connect to the
+ // peer statestored.
+ lock_guard<mutex> l(ha_lock_);
+ is_active_ = true;
+ active_status_metric_->SetValue(is_active_);
+ active_version_ = UnixMicros();
+ LOG(INFO) << "Set Statestore as active since it does not receive handshake "
+ << "response in HA preemption waiting period";
+ found_peer_ = false;
+ connected_peer_metric_->SetValue(found_peer_);
+ } else {
+ status = Status(response.status);
+ DCHECK(status.ok());
+ lock_guard<mutex> l(ha_lock_);
+ peer_statestore_id_ = response.dst_statestore_id;
+ is_active_ = !response.dst_statestore_active;
+ active_status_metric_->SetValue(is_active_);
+ if (is_active_) active_version_ = UnixMicros();
+ found_peer_ = true;
+ connected_peer_metric_->SetValue(found_peer_);
+ // connected_to_peer_statestore_metric_->SetValue(true);
+ LOG(INFO) << "Receive Statestore HA handshake response, set the statestore as "
+ << (is_active_ ? "active" : "standby");
+ }
+ return Status::OK();
+}
+
+bool Statestore::IsActive() {
+ lock_guard<mutex> l(ha_lock_);
+ return is_active_;
+}
+
+int64_t Statestore::GetActiveVersion(bool* is_active) {
+ lock_guard<mutex> l(ha_lock_);
+ *is_active = is_active_;
+ return active_version_;
+}
+
+Status Statestore::SendHaHandshake(TStatestoreHaHandshakeResponse* response) {
+ if (disable_network_.Load()) {
+ return Status("Don't send HA handshake since network is disabled.");
+ }
+ // Negotiate the role for HA with peer statestore instance in client mode.
+ LOG(INFO) << "Send Statestore HA handshake request";
+ TStatestoreHaHandshakeRequest request;
+ request.__set_src_statestore_id(statestore_id_);
+ request.__set_src_statestore_address(
+ TNetworkAddressToString(local_statestore_ha_addr_));
+ request.__set_src_force_active(FLAGS_statestore_force_active);
+ int attempt = 0; // Used for debug action only.
+ if (FLAGS_statestore_peer_cnxn_attempts > 0) {
+ FLAGS_statestore_peer_cnxn_retry_interval_ms =
+ FLAGS_statestore_ha_preemption_wait_period_ms /
+ FLAGS_statestore_peer_cnxn_attempts;
+ }
+ StatestoreHaServiceConn::RpcStatus rpc_status =
+ StatestoreHaServiceConn::DoRpcWithRetry(ha_client_cache_.get(),
+ peer_statestore_ha_addr_,
+ &StatestoreHaServiceClientWrapper::StatestoreHaHandshake,
+ request,
+ FLAGS_statestore_peer_cnxn_attempts,
+ FLAGS_statestore_peer_cnxn_retry_interval_ms,
+ [&attempt]() {
+ return attempt++ == 0 ?
+ DebugAction(FLAGS_debug_actions, "STATESTORE_HA_HANDSHAKE_FIRST_ATTEMPT")
+ : Status::OK();
+ },
+ response);
+ return rpc_status.status;
+}
+
+Status Statestore::ReceiveHaHandshakeRequest(const TUniqueId& peer_statestore_id,
+ const string& peer_statestore_address, bool peer_force_active,
+ bool* statestore_active) {
+ // Process HA handshake request from peer statstore
+ LOG(INFO) << "Receive Statestore HA handshake request";
+ lock_guard<mutex> l(ha_lock_);
+ peer_statestore_id_ = peer_statestore_id;
+ if (peer_force_active) {
+ is_active_ = false;
+ active_status_metric_->SetValue(is_active_);
+ ha_active_ss_failure_detector_->UpdateHeartbeat(STATESTORE_ID, true);
+ LOG(INFO) << "Set the statestored as standby since the peer is started with force "
+ << "active flag";
+ } else if (!is_active_) {
+ if (FLAGS_statestore_force_active) {
+ is_active_ = true;
+ active_status_metric_->SetValue(is_active_);
+ active_version_ = UnixMicros();
+ ha_standby_ss_failure_detector_->UpdateHeartbeat(STATESTORE_ID, true);
+ LOG(INFO) << "Set the statestored as active since it's started with force active "
+ << "flag";
+ } else {
+ // Compare priority and assign the statestored with high priority as active
+ // statestored.
+ is_active_ = FLAGS_use_network_address_as_statestore_priority ?
+ TNetworkAddressToString(local_statestore_ha_addr_) < peer_statestore_address :
+ statestore_id_ < peer_statestore_id;
+ active_status_metric_->SetValue(is_active_);
+ if (is_active_) {
+ active_version_ = UnixMicros();
+ ha_standby_ss_failure_detector_->UpdateHeartbeat(STATESTORE_ID, true);
+ } else {
+ ha_active_ss_failure_detector_->UpdateHeartbeat(STATESTORE_ID, true);
+ }
+ LOG(INFO) << "Set the statestored as " << (is_active_ ? "active" : "standby");
+ }
+ }
+ *statestore_active = is_active_;
+ if (!found_peer_) {
+ found_peer_ = true;
+ connected_peer_metric_->SetValue(found_peer_);
+ }
+ return Status::OK();
+}
+
+void Statestore::HaHeartbeatRequest(const TUniqueId& dst_statestore_id,
+ const TUniqueId& src_statestore_id) {
+ // Don't process HA heartbeat if network is disabled.
+ if (disable_network_.Load()) return;
+ lock_guard<mutex> l(ha_lock_);
+ if (!is_active_) {
+ // process HA heartbeat from active statestore
+ ha_active_ss_failure_detector_->UpdateHeartbeat(STATESTORE_ID, true);
+ } else {
+ // Receive heartbeat from its peer statestored. That means both statestored designate
+ // themselves as active. Enter recovery mode to restart negotiation.
+ LOG(WARNING)
+ << "Both statestoreds designate themselves as active, restart negotiation.";
+ in_recovery_mode_ = true;
+ in_ha_recovery_mode_metric_->SetValue(in_recovery_mode_);
+ is_active_ = false;
+ active_status_metric_->SetValue(is_active_);
+ LOG(WARNING) << "Enter HA recovery mode.";
+ }
+}
+
+[[noreturn]] void Statestore::MonitorStatestoredHaHeartbeat() {
+ while (1) {
+ SleepForMs(FLAGS_statestore_ha_heartbeat_monitoring_frequency_ms);
+ lock_guard<mutex> l(ha_lock_);
+ if (in_recovery_mode_) {
+ // Keep sending HA handshake request to its peer periodically until receiving
+ // response.
+ TStatestoreHaHandshakeResponse response;
+ Status status = SendHaHandshake(&response);
+ if (!status.ok()) continue;
+ status = Status(response.status);
+ DCHECK(status.ok());
+
+ // Exit "recovery" mode.
+ in_recovery_mode_ = false;
+ in_ha_recovery_mode_metric_->SetValue(in_recovery_mode_);
+ peer_statestore_id_ = response.dst_statestore_id;
+ is_active_ = !response.dst_statestore_active;
+ active_status_metric_->SetValue(is_active_);
+ found_peer_ = true;
+ connected_peer_metric_->SetValue(found_peer_);
+ LOG(INFO) << "Receive Statestore HA handshake response, exit HA recovery mode. "
+ << "Set the statestored as " << (is_active_ ? "active" : "standby");
+ if (is_active_) {
+ active_version_ = UnixMicros();
+ // Send notification to all subscribers.
+ update_statestored_cv_.NotifyAll();
+ }
+ } else if (is_active_) {
+ // Statestored in active state
+ // Send HA heartbeat to standby statestored.
+ if (found_peer_) {
+ Status status = SendHaHeartbeat();
+ if (status.ok()) continue;
+ }
+ // Check if standby statestored is reachable.
+ FailureDetector::PeerState state =
+ ha_standby_ss_failure_detector_->GetPeerState(STATESTORE_ID);
+ if (state != FailureDetector::FAILED) {
+ continue;
+ } else if (found_peer_) {
+ // Stop sending HA heartbeat.
+ found_peer_ = false;
+ connected_peer_metric_->SetValue(found_peer_);
+ LOG(INFO) << "Statestored lost connection with peer statestored";
+ }
+
+ lock_guard<mutex> l(subscribers_lock_);
+ if (subscribers_.size() == 0) {
+ // To avoid race with new active statestored, original active statestored enter
+ // "recovery" mode if it does not receive heartbeat responses from standby
+ // statestored and all subscribers.
+ in_recovery_mode_ = true;
+ in_ha_recovery_mode_metric_->SetValue(in_recovery_mode_);
+ is_active_ = false;
+ active_status_metric_->SetValue(is_active_);
+ LOG(WARNING) << "Enter HA recovery mode.";
+ }
+ // TODO: IMPALA-12507 Need better approach to handle split-brain in network.
+ // In the scenario, active statestored still can reach some subscribers.
+ } else {
+ // Statestored in standby state
+ // Monitor connection state with its peer statestored.
+ FailureDetector::PeerState state =
+ ha_active_ss_failure_detector_->GetPeerState(STATESTORE_ID);
+ // Check if the majority of subscribers lost connection with active statestored.
+ int failed_conn_state_count = 0;
+ int total_subscribers = 0;
+ {
+ lock_guard<mutex> l(subscribers_lock_);
+ failed_conn_state_count = failed_conn_state_count_;
+ total_subscribers = active_conn_states_.size();
+ }
+ bool majority_failed = failed_conn_state_count > 0 &&
+ failed_conn_state_count > total_subscribers / 2;
+ if (state != FailureDetector::FAILED) {
+ if (majority_failed) {
+ LOG(WARNING) << "Active statestored may have network issue. "
+ << failed_conn_state_count << " out of " << total_subscribers
+ << " subscribers lost connections with active statestored.";
+ }
+ continue;
+ } else if (majority_failed) {
+ // When standby statestored lost connection with active statestored, take over
+ // active role if the majority of subscribers lost connections with active
+ // statestored.
+ LOG(INFO) << "Statestore change to active state, " << failed_conn_state_count
+ << " out of " << total_subscribers
+ << " subscribers lost connections with active statestored.";
+ is_active_ = true;
+ active_status_metric_->SetValue(is_active_);
+ active_version_ = UnixMicros();
+ found_peer_ = false;
+ connected_peer_metric_->SetValue(found_peer_);
+ // Send notification to all subscribers.
+ update_statestored_cv_.NotifyAll();
+ } else if (total_subscribers == 0) {
+ // If there is no subscriber, it means this statestored lost connection with
+ // other nodes in the cluster, enter "recovery" mode.
+ in_recovery_mode_ = true;
+ in_ha_recovery_mode_metric_->SetValue(in_recovery_mode_);
+ LOG(WARNING) << "Enter HA recovery mode.";
+ } else {
+ VLOG(3) << "Standby statestored missed HA heartbeat from active statestored, "
+ << failed_conn_state_count << " out of " << total_subscribers
+ << " subscribers lost connections with active statestored.";
+ }
+ }
+ }
+}
+
+Status Statestore::SendHaHeartbeat() {
+ if (disable_network_.Load()) {
+ ha_standby_ss_failure_detector_->UpdateHeartbeat(STATESTORE_ID, false);
+ return Status("Don't send HA heartbeat since network is disabled.");
+ }
+
+ Status status;
+ StatestoreHaServiceConn client(ha_client_cache_.get(),
+ peer_statestore_ha_addr_, &status);
+ RETURN_IF_ERROR(status);
+
+ TStatestoreHaHeartbeatRequest request;
+ TStatestoreHaHeartbeatResponse response;
+ request.__set_dst_statestore_id(peer_statestore_id_);
+ request.__set_src_statestore_id(statestore_id_);
+ status = client.DoRpc(&StatestoreHaServiceClientWrapper::StatestoreHaHeartbeat,
+ request, &response);
+ ha_standby_ss_failure_detector_->UpdateHeartbeat(STATESTORE_ID, status.ok());
+ if (status.ok()) {
+ status = Status(response.status);
+ }
+ return status;
+}
diff --git a/be/src/statestore/statestore.h b/be/src/statestore/statestore.h
index 62b24384a..c55cc227b 100644
--- a/be/src/statestore/statestore.h
+++ b/be/src/statestore/statestore.h
@@ -31,6 +31,7 @@
#include "common/atomic.h"
#include "common/status.h"
+#include "gen-cpp/StatestoreHaService.h"
#include "gen-cpp/StatestoreService.h"
#include "gen-cpp/StatestoreService_types.h"
#include "gen-cpp/StatestoreSubscriber.h"
@@ -39,6 +40,7 @@
#include "runtime/client-cache.h"
#include "statestore/failure-detector.h"
#include "statestore/statestore-catalogd-mgr.h"
+#include "statestore/statestore-service-client-wrapper.h"
#include "statestore/statestore-subscriber-client-wrapper.h"
#include "util/aligned-new.h"
#include "util/condition-variable.h"
@@ -53,6 +55,7 @@ namespace impala {
class Status;
typedef ClientCache<StatestoreSubscriberClientWrapper> StatestoreSubscriberClientCache;
+typedef ClientCache<StatestoreHaServiceClientWrapper> StatestoreHaClientCache;
std::string SubscriberTypeToString(TStatestoreSubscriberType::type t);
@@ -115,6 +118,42 @@ std::string SubscriberTypeToString(TStatestoreSubscriberType::type t);
/// versions. This ensure all of components in a cluster could communicate with compatible
/// APIs and message format in the wire.
///
+/// Statestore HA:
+/// To support statestore HA, we add the preemptive behavior for statestored. When HA is
+/// enabled, the preemptive behavior allows the statestored with the higher priority to
+/// become active and the paired statestored becomes standby. The active statestored acts
+/// as the owner of the Impala cluster and provides statestore service for the cluster
+/// members.
+/// By default, preemption is disabled on the statestored which is running as single
+/// statestore instance in an Impala cluster. For deployment with statestore HA enabled,
+/// the preemption must be enabled with starting flag "enable_statestored_ha" on two
+/// statestoreds in the HA pair and all the subscribers.
+/// To avoid introduce external dependency like Apache ZooKeeper, standby statestored is
+/// used to monitor the health of active statestored. The basic idea is to have two
+/// statestore instances as HA pair in Active-Passive mode. The both primary and standby
+/// statestored send heartbeats to all subscribers. Subscribers send heartbeat response
+/// to standby statestored with their connection state with active statestored so that
+/// standby statestored could monitor the connection state between active statestore and
+/// the subscribers. Active statestored also sends heartbeats to standby statestored.
+/// Automatic fail-over will be triggered if standby statestored does not receive
+/// heartbeats from active statestored and the majority of the subscribers do not receive
+/// heartbeats from active statestored.
+/// In the scenario where standby statestored is still receiving heartbeats from active
+/// statestored, but majority of subscribers are unable to reach the active statestored,
+/// warning messages will be logged on standby statestored.
+/// The statestored in an Active-Passive HA pair can be assigned an instance priority
+/// value to indicate a preference for which statestored should assume the active role.
+/// The statestore ID which is generated for each statestored can be used as instance
+/// priority value. The lower numerical value in statestore ID corresponds to a higher
+/// priority. The statestored with the higher priority is designated as active, the other
+/// statestored is designated as standby after initial HA handshake. Only the active
+/// statestored propagates the IMPALA_CATALOG_TOPIC and elected active catalogd to the
+/// cluster members.
+/// To make a specific statestored in the HA pair as active instance, the statestored must
+/// be started with starting flag "statestore_force_active" so that the statestored will
+/// be assigned with active role in HA handshake. This allows administrator to manually
+/// perform statestore service fail over.
+///
/// +================+
/// | Implementation |
/// +================+
@@ -122,10 +161,11 @@ std::string SubscriberTypeToString(TStatestoreSubscriberType::type t);
/// Locking:
/// --------
/// The lock acquisition order is:
-/// 1. 'subscribers_lock_'
-/// 2. 'topics_map_lock_'
-/// 3. Subscriber::transient_entry_lock_
-/// 4. Topic::lock_ (terminal)
+/// 1. 'ha_lock_'
+/// 2. 'subscribers_lock_'
+/// 3. 'topics_map_lock_'
+/// 4. Subscriber::transient_entry_lock_
+/// 5. Topic::lock_ (terminal)
class Statestore : public CacheLineAligned {
public:
/// A TopicId uniquely identifies a single topic
@@ -185,6 +225,12 @@ class Statestore : public CacheLineAligned {
return thrift_iface_;
}
+ /// Returns the Thrift API interface that proxies requests onto the local Statestore
+ /// HA Service.
+ const std::shared_ptr<StatestoreHaServiceIf>& ha_thrift_iface() const {
+ return ha_thrift_iface_;
+ }
+
/// Names of prioritized topics that are handled in a separate threadpool. The topic
/// names are hardcoded here for expediency. Ideally we would have a more generic
/// interface for specifying prioritized topics, but for now we only have a small
@@ -200,12 +246,52 @@ class Statestore : public CacheLineAligned {
/// after it stops responding to heartbeats.
static int64_t FailedExecutorDetectionTimeMs();
+ /// Return statestore-ID.
const TUniqueId& GetStateStoreId() const { return statestore_id_; }
+ /// Return protocol version of statestore service.
StatestoreServiceVersion::type GetProtocolVersion() const {
return protocol_version_;
}
+ /// Disable network if the input parameter disable_network is set as true.
+ /// statestored will not send heartbeat and HA handshake messages when network is
+ /// disabled. This is used to simulate network failure for unit-test.
+ void SetStatestoreDebugAction(bool disable_network);
+
+ /// Initialize statestore HA.
+ Status InitStatestoreHa(
+ int32_t state_store_ha_port, const TNetworkAddress& peer_statestore_addr);
+
+ // Return true if this statestore instance is in active role.
+ bool IsActive();
+
+ // Return the version of active status.
+ int64_t GetActiveVersion(bool* is_active);
+
+ /// Send RPC to negotiate the role with peer statestored for Statestore HA.
+ Status SendHaHandshake(TStatestoreHaHandshakeResponse* response);
+
+ /// Called to process HA handshake request from peer statstore
+ Status ReceiveHaHandshakeRequest(const TUniqueId& subscriber_id,
+ const string& src_statestore_address, bool src_force_active,
+ bool* dst_statestore_active);
+
+ /// Called to process HA heartbeat from active statestore
+ void HaHeartbeatRequest(const TUniqueId& dst_statestore_id,
+ const TUniqueId& src_statestore_id);
+
+ /// Sends and monitors the HA heartbeats between two statestore instances.
+ [[noreturn]] void MonitorStatestoredHaHeartbeat();
+
+ /// Send HA heartbeat to peer statestore instance.
+ Status SendHaHeartbeat();
+
+ /// Return the network address of peer statestore instance.
+ std::string GetPeerStatestoreHaAddress() {
+ return TNetworkAddressToString(peer_statestore_ha_addr_);
+ }
+
private:
/// A TopicEntry is a single entry in a topic, and logically is a <string, byte string>
/// pair.
@@ -553,6 +639,9 @@ class Statestore : public CacheLineAligned {
/// Condition variable for sending the notifications of updating catalogd.
ConditionVariable update_catalod_cv_;
+ /// Condition variable for sending the notifications of updating role of statestored.
+ ConditionVariable update_statestored_cv_;
+
/// Used to generated unique IDs for each new registration.
boost::uuids::random_generator subscriber_uuid_generator_;
@@ -619,6 +708,9 @@ class Statestore : public CacheLineAligned {
/// Thread to send notification of updating catalogd.
std::unique_ptr<Thread> update_catalogd_thread_;
+ /// Thread to send notification of updating role of statestored.
+ std::unique_ptr<Thread> update_statestored_thread_;
+
/// Indicates whether the statestore has been initialized and the service is ready.
std::atomic_bool service_started_{false};
@@ -637,15 +729,33 @@ class Statestore : public CacheLineAligned {
/// failure.
boost::scoped_ptr<StatestoreSubscriberClientCache> update_catalogd_client_cache_;
+ /// Cache of subscriber clients used for UpdateStatestoredRole() RPCs. Only one client
+ /// per subscriber should be used, but the cache helps with the client lifecycle on
+ /// failure.
+ boost::scoped_ptr<StatestoreSubscriberClientCache> update_statestored_client_cache_;
+
+ /// statestore client cache - only one client is ever used. Initialized in constructor.
+ /// The StatestoreHaClientCache is created with num_retries = 1 and wait_ms = 0.
+ /// Connections are still retried, but the retry mechanism is driven by DoRpcWithRetry.
+ /// Clients should always use DoRpcWithRetry rather than DoRpc to ensure that both RPCs
+ /// and connections are retried.
+ boost::scoped_ptr<StatestoreHaClientCache> ha_client_cache_;
+
/// Container for the internal statestore service.
boost::scoped_ptr<ThriftServer> thrift_server_;
+ /// Container for the internal statestore HA service.
+ boost::scoped_ptr<ThriftServer> ha_thrift_server_;
+
/// Pointer to the MetricGroup for this statestore. Not owned.
MetricGroup* metrics_;
- /// Thrift API implementation which proxies requests onto this Statestore
+ /// Thrift API implementation which proxies requests onto this Statestore Service
std::shared_ptr<StatestoreServiceIf> thrift_iface_;
+ /// Thrift API implementation which proxies requests onto this Statestore HA Service
+ std::shared_ptr<StatestoreHaServiceIf> ha_thrift_iface_;
+
/// Failure detector for subscribers. If a subscriber misses a configurable number of
/// consecutive heartbeat messages, it is considered failed and a) its transient topic
/// entries are removed and b) its entry in the subscriber map is erased. The
@@ -655,6 +765,60 @@ class Statestore : public CacheLineAligned {
/// carry over to any new registrations of the previous subscriber.
boost::scoped_ptr<MissedHeartbeatFailureDetector> failure_detector_;
+ /// Lock to protect statestored HA related variables:
+ std::mutex ha_lock_;
+
+ /// Network address of peer statestore instance for StatestoreHaService.
+ TNetworkAddress peer_statestore_ha_addr_;
+
+ /// Unique identifier for peer statestore instance.
+ TUniqueId peer_statestore_id_;
+
+ /// Local network address for StatestoreHaService.
+ TNetworkAddress local_statestore_ha_addr_;
+
+ /// True if the statestore instance is active.
+ bool is_active_ = false;
+
+ /// The version of active statestore. It's set as the number of microseconds that have
+ /// passed since the Unix epoch.
+ /// The HA notifications could be received by subscribers out of order due to network
+ /// delays. This variable is sent in HA notifications to guarantee that only latest
+ /// notification takes effect on subscribers. We can not use an incrementing counter
+ /// for this purpose, otherwise the version will be reset when statestored is restarted.
+ int64_t active_version_ = 0;
+
+ /// True if the peer statestore instance has been detected.
+ bool found_peer_ = false;
+
+ /// True if the statestore instance is in recovery mode.
+ bool in_recovery_mode_ = false;
+
+ /// Disable network if this variable is set as true by statestore service API.
+ /// This is only used for unit-test.
+ AtomicBool disable_network_{false};
+
+ /// Failure detector for active statestore. If the standby statestore misses a
+ /// configurable number of consecutive heartbeat messages, it is considered failed and
+ /// if it does not receive heartbeat responses from more than 50% of subscribers,
+ /// active statestore enter "looking" state .
+ boost::scoped_ptr<MissedHeartbeatFailureDetector> ha_standby_ss_failure_detector_;
+
+ /// Failure detector for standby statestore. It tracks heartbeat messages from the
+ /// active statestore.
+ boost::scoped_ptr<impala::TimeoutFailureDetector> ha_active_ss_failure_detector_;
+
+ /// Record of connection state between active statestore and subscribers.
+ typedef boost::unordered_map<SubscriberId, TStatestoreConnState::type>
+ ActiveConnStateMap;
+ ActiveConnStateMap active_conn_states_;
+
+ /// Number of subscriber which lost connection with active statestored.
+ int failed_conn_state_count_ = 0;
+
+ /// Thread that sends and monitors the HA heartbeats between two statestore instances.
+ std::unique_ptr<Thread> ha_monitoring_thread_;
+
/// Metric that track the registered, non-failed subscribers.
IntGauge* num_subscribers_metric_;
SetMetric<std::string>* subscriber_set_metric_;
@@ -680,6 +844,14 @@ class Statestore : public CacheLineAligned {
/// Metric to count the total number of failed UpdateCatalogd RPCs sent by statestore.
IntCounter* failed_update_catalogd_rpc_metric_;
+ /// Metric to count the total number of successful UpdateCStatestoredRole RPCs sent by
+ /// statestore.
+ IntCounter* successful_update_statestored_role_rpc_metric_;
+
+ /// Metric to count the total number of failed UpdateStatestoredRole RPCs sent by
+ /// statestore.
+ IntCounter* failed_update_statestored_role_rpc_metric_;
+
/// Metric to count the total number of requests for clearing topic entries from
/// catalogd. Catalogd indicates to clear topic entries when it is restarted or its
/// role has been changed from standby to active.
@@ -688,6 +860,19 @@ class Statestore : public CacheLineAligned {
/// Metric that tracks the address of active catalogd for catalogd HA
StringProperty* active_catalogd_address_metric_;
+ /// Metric that tracks if the statestored is active when statestored HA is enabled.
+ BooleanProperty* active_status_metric_;
+
+ /// Metric that tracks if the statestore service is started and ready to accept
+ /// connections.
+ BooleanProperty* service_started_metric_;
+
+ /// Metric that tracks if the statestored is in HA recovery mode.
+ BooleanProperty* in_ha_recovery_mode_metric_;
+
+ /// Metric that tracks if the statestored is connected with peer statestored.
+ BooleanProperty* connected_peer_metric_;
+
/// Utility method to add an update to the given thread pool, and to fail if the thread
/// pool is already at capacity. Assumes that subscribers_lock_ is held by the caller.
Status OfferUpdate(const ScheduledSubscriberUpdate& update,
@@ -800,10 +985,17 @@ class Statestore : public CacheLineAligned {
/// Monitors the notification of updating catalogd.
[[noreturn]] void MonitorUpdateCatalogd();
+ /// Monitors the notification of updating statestored's role.
+ [[noreturn]] void MonitorUpdateStatestoredRole();
+
/// Send notification of updating catalogd to the coordinators.
void SendUpdateCatalogdNotification(int64_t* last_active_catalogd_version,
vector<std::shared_ptr<Subscriber>>& receivers);
+ /// Send notification of updating statestored's role to all subscribers.
+ void SendUpdateStatestoredRoleNotification(int64_t* last_active_statestored_version,
+ vector<std::shared_ptr<Subscriber>>& receivers);
+
/// Raw callback to indicate whether the service is ready.
void HealthzHandler(const Webserver::WebRequest& req, std::stringstream* data,
HttpStatusCode* response);
diff --git a/be/src/statestore/statestored-main.cc b/be/src/statestore/statestored-main.cc
index 9e65bf827..763c59514 100644
--- a/be/src/statestore/statestored-main.cc
+++ b/be/src/statestore/statestored-main.cc
@@ -19,6 +19,7 @@
// This file contains the main() function for the state store process,
// which exports the Thrift service StatestoreService.
+#include <gflags/gflags.h>
#include <iostream>
#include <string>
@@ -37,20 +38,34 @@
DECLARE_int32(metrics_webserver_port);
DECLARE_int32(state_store_port);
DECLARE_int32(webserver_port);
+DECLARE_bool(enable_statestored_ha);
+DECLARE_int32(state_store_ha_port);
+DECLARE_int32(state_store_peer_ha_port);
+DECLARE_string(state_store_peer_host);
#include "common/names.h"
using namespace impala;
int StatestoredMain(int argc, char** argv) {
- // Override default for webserver port
- FLAGS_webserver_port = 25010;
+ // Override default for webserver port if and only if this port had not been set
+ // explicitly in command line.
+ if (google::GetCommandLineFlagInfoOrDie("webserver_port").is_default) {
+ FLAGS_webserver_port = 25010;
+ }
InitCommonRuntime(argc, argv, false);
DaemonEnv daemon_env("statestore");
ABORT_IF_ERROR(daemon_env.Init(/* init_jvm */ false));
Statestore statestore(daemon_env.metrics());
+ if (FLAGS_enable_statestored_ha) {
+ // Negotiate role with peer statestore instance if Statestore HA is enabled.
+ TNetworkAddress peer_statestore_addr =
+ MakeNetworkAddress(FLAGS_state_store_peer_host, FLAGS_state_store_peer_ha_port);
+ ABORT_IF_ERROR(
+ statestore.InitStatestoreHa(FLAGS_state_store_ha_port, peer_statestore_addr));
+ }
ABORT_IF_ERROR(statestore.Init(FLAGS_state_store_port));
statestore.RegisterWebpages(daemon_env.webserver(), false);
if (FLAGS_metrics_webserver_port > 0) {
diff --git a/bin/start-impala-cluster.py b/bin/start-impala-cluster.py
index b2a5a48b4..4fb86f224 100755
--- a/bin/start-impala-cluster.py
+++ b/bin/start-impala-cluster.py
@@ -43,6 +43,8 @@ from tests.common.impala_cluster import (ImpalaCluster, DEFAULT_BEESWAX_PORT,
DEFAULT_ADMISSIOND_WEBSERVER_PORT, DEFAULT_CATALOGD_JVM_DEBUG_PORT,
DEFAULT_CATALOG_SERVICE_PORT, DEFAULT_CATALOGD_STATE_STORE_SUBSCRIBER_PORT,
DEFAULT_EXTERNAL_FE_PORT, DEFAULT_IMPALAD_JVM_DEBUG_PORT,
+ DEFAULT_STATESTORE_SERVICE_PORT, DEFAULT_STATESTORE_HA_SERVICE_PORT,
+ DEFAULT_PEER_STATESTORE_HA_SERVICE_PORT,
find_user_processes, run_daemon)
LOG = logging.getLogger(os.path.splitext(os.path.basename(__file__))[0])
@@ -158,6 +160,10 @@ parser.add_option("--enable_catalogd_ha", dest="enable_catalogd_ha",
parser.add_option("--jni_frontend_class", dest="jni_frontend_class",
action="store", default="org/apache/impala/service/JniFrontend",
help="Use a custom java frontend interface.")
+parser.add_option("--enable_statestored_ha", dest="enable_statestored_ha",
+ action="store_true", default=False,
+ help="If true, enables StatestoreD HA - the cluster will be launched "
+ "with two statestored instances as Active-Passive HA pair.")
# For testing: list of comma-separated delays, in milliseconds, that delay impalad catalog
# replica initialization. The ith delay is applied to the ith impalad.
@@ -318,6 +324,48 @@ def catalogd_service_name(i):
return "catalogd_node{node_num}".format(node_num=i)
+def choose_statestored_ports(enable_statestored_ha, instance_num):
+ """Compute the ports for statestored instance num 'instance_num', returning as a map
+ from the argument name to the port number."""
+ if not enable_statestored_ha:
+ return {'state_store_port': DEFAULT_STATESTORE_SERVICE_PORT + instance_num,
+ 'webserver_port': DEFAULT_STATESTORED_WEBSERVER_PORT + instance_num}
+ else:
+ # Assume two statestore instances will be launched when statestore HA is enabled
+ state_store_peer_ha_port =\
+ DEFAULT_STATESTORE_HA_SERVICE_PORT + ((instance_num + 1) % 2)
+ return {'state_store_port': DEFAULT_STATESTORE_SERVICE_PORT + instance_num,
+ 'state_store_ha_port': DEFAULT_STATESTORE_HA_SERVICE_PORT + instance_num,
+ 'state_store_peer_ha_port': state_store_peer_ha_port,
+ 'webserver_port': DEFAULT_STATESTORED_WEBSERVER_PORT + instance_num}
+
+
+def build_statestored_port_args(enable_statestored_ha, instance_num):
+ if not enable_statestored_ha:
+ STATESTORED_PORTS = (
+ "-state_store_port={state_store_port} "
+ "-webserver_port={webserver_port}")
+ return STATESTORED_PORTS.format(
+ **choose_statestored_ports(enable_statestored_ha, instance_num))
+ else:
+ STATESTORED_PORTS = (
+ "-state_store_port={state_store_port} "
+ "-state_store_ha_port={state_store_ha_port} "
+ "-state_store_peer_ha_port={state_store_peer_ha_port} "
+ "-webserver_port={webserver_port}")
+ return STATESTORED_PORTS.format(
+ **choose_statestored_ports(enable_statestored_ha, instance_num))
+
+
+def statestored_service_name(i):
+ """Return the name to use for the ith statestore daemon in the cluster."""
+ if i == 0:
+ # The first statestored always logs to statestored.INFO
+ return "statestored"
+ else:
+ return "statestored_node{node_num}".format(node_num=i)
+
+
def combine_arg_list_opts(opt_args):
"""Helper for processing arguments like impalad_args. The input is a list of strings,
each of which is the string passed into one instance of the argument, e.g. for
@@ -328,17 +376,29 @@ def combine_arg_list_opts(opt_args):
return list(itertools.chain(*[shlex.split(arg) for arg in opt_args]))
-def build_statestored_arg_list(enable_catalogd_ha):
- """Build a list of command line arguments to pass to the statestored."""
- args = (build_logging_args("statestored") + build_kerberos_args("statestored")
- + combine_arg_list_opts(options.state_store_args))
- if (enable_catalogd_ha):
- args.extend(["-enable_catalogd_ha=true"])
- return args
+def build_statestored_arg_list(num_statestored, remap_ports):
+ """Build a list of lists of command line arguments to pass to each statestored
+ instance. Build args for two statestored instances if statestored HA is enabled."""
+ statestored_arg_list = []
+ for i in range(num_statestored):
+ service_name = statestored_service_name(i)
+ args = (build_logging_args(service_name)
+ + build_kerberos_args("statestored")
+ + combine_arg_list_opts(options.state_store_args))
+ if remap_ports:
+ statestored_port_args =\
+ build_statestored_port_args(options.enable_statestored_ha, i)
+ args.extend(shlex.split(statestored_port_args))
+ if options.enable_catalogd_ha:
+ args.extend(["-enable_catalogd_ha=true"])
+ if options.enable_statestored_ha:
+ args.extend(["-enable_statestored_ha=true"])
+ statestored_arg_list.append(args)
+ return statestored_arg_list
-def build_catalogd_arg_list(num_catalogd, enable_catalogd_ha, remap_ports):
- """Build a list of command line arguments to pass to the catalogd.
+def build_catalogd_arg_list(num_catalogd, remap_ports):
+ """Build a list of lists of command line arguments to pass to each catalogd instance.
Build args for two catalogd instances if catalogd HA is enabled."""
catalogd_arg_list = []
for i in range(num_catalogd):
@@ -350,16 +410,32 @@ def build_catalogd_arg_list(num_catalogd, enable_catalogd_ha, remap_ports):
if remap_ports:
catalogd_port_args = build_catalogd_port_args(i)
args.extend(shlex.split(catalogd_port_args))
- if enable_catalogd_ha:
+ if options.enable_catalogd_ha:
args.extend(["-enable_catalogd_ha=true"])
+ if options.enable_statestored_ha:
+ args.extend(["-enable_statestored_ha=true"])
+ state_store_port = DEFAULT_STATESTORE_SERVICE_PORT
+ args.extend(
+ ["-state_store_port={0}".format(state_store_port)])
+ args.extend(
+ ["-state_store_2_port={0}".format(state_store_port + 1)])
catalogd_arg_list.append(args)
return catalogd_arg_list
def build_admissiond_arg_list():
"""Build a list of command line arguments to pass to the admissiond."""
- return (build_logging_args("admissiond") + build_kerberos_args("admissiond") +
- combine_arg_list_opts(options.admissiond_args))
+ args = (build_logging_args("admissiond")
+ + build_kerberos_args("admissiond")
+ + combine_arg_list_opts(options.admissiond_args))
+ if options.enable_statestored_ha:
+ args.extend(["-enable_statestored_ha=true"])
+ state_store_port = DEFAULT_STATESTORE_SERVICE_PORT
+ args.extend(
+ ["-state_store_port={0}".format(state_store_port)])
+ args.extend(
+ ["-state_store_2_port={0}".format(state_store_port + 1)])
+ return args
def build_impalad_arg_lists(cluster_size, num_coordinators, use_exclusive_coordinators,
@@ -484,6 +560,14 @@ def build_impalad_arg_lists(cluster_size, num_coordinators, use_exclusive_coordi
args = "{args} -admission_service_host={host}".format(
args=args, host=admissiond_host)
+ if options.enable_statestored_ha:
+ state_store_port = DEFAULT_STATESTORE_SERVICE_PORT
+ state_store_2_port = DEFAULT_STATESTORE_SERVICE_PORT + 1
+ args = "{args} -enable_statestored_ha=true -state_store_port={state_store_port} "\
+ "-state_store_2_port={state_store_2_port}".format(
+ args=args, state_store_port=state_store_port,
+ state_store_2_port=state_store_2_port)
+
if "geospatial_library" not in args:
args = "{args} -geospatial_library={geospatial_library}".format(
args=args, geospatial_library=options.geospatial_library)
@@ -557,29 +641,36 @@ class MiniClusterOperations(object):
def kill_all_catalogds(self, force=False):
kill_matching_processes(["catalogd"], force=force)
- def kill_statestored(self, force=False):
+ def kill_all_statestoreds(self, force=False):
kill_matching_processes(["statestored"], force=force)
def kill_admissiond(self, force=False):
kill_matching_processes(["admissiond"], force=force)
def start_statestore(self):
- LOG.info("Starting State Store logging to {log_dir}/statestored.INFO".format(
- log_dir=options.log_dir))
- output_file = os.path.join(options.log_dir, "statestore-out.log")
- run_daemon_with_options("statestored",
- build_statestored_arg_list(options.enable_catalogd_ha), output_file)
- if not check_process_exists("statestored", 10):
- raise RuntimeError("Unable to start statestored. Check log or file permissions"
- " for more details.")
+ if options.enable_statestored_ha:
+ num_statestored = 2
+ else:
+ num_statestored = 1
+ statestored_arg_lists = build_statestored_arg_list(num_statestored, remap_ports=True)
+ for i in range(num_statestored):
+ service_name = statestored_service_name(i)
+ LOG.info(
+ "Starting State Store logging to {log_dir}/{service_name}.INFO".format(
+ log_dir=options.log_dir, service_name=service_name))
+ output_file = os.path.join(
+ options.log_dir, "{service_name}-out.log".format(service_name=service_name))
+ run_daemon_with_options("statestored", statestored_arg_lists[i], output_file)
+ if not check_process_exists("statestored", 10):
+ raise RuntimeError("Unable to start statestored. Check log or file permissions"
+ " for more details.")
def start_catalogd(self):
if options.enable_catalogd_ha:
num_catalogd = 2
else:
num_catalogd = 1
- catalogd_arg_lists = build_catalogd_arg_list(
- num_catalogd, options.enable_catalogd_ha, remap_ports=True)
+ catalogd_arg_lists = build_catalogd_arg_list(num_catalogd, remap_ports=True)
for i in range(num_catalogd):
service_name = catalogd_service_name(i)
LOG.info(
@@ -654,7 +745,7 @@ class DockerMiniClusterOperations(object):
use_admission_service=options.enable_admission_service)
def kill_all_daemons(self, force=False):
- self.kill_statestored(force=force)
+ self.kill_all_statestoreds(force=force)
self.kill_all_catalogds(force=force)
self.kill_admissiond(force=force)
self.kill_all_impalads(force=force)
@@ -679,24 +770,46 @@ class DockerMiniClusterOperations(object):
LOG.info("Stopping container {0}".format(container_name))
check_call(["docker", "stop", container_name])
- def kill_statestored(self, force=False):
- self.__stop_container__("statestored")
+ def kill_all_statestoreds(self, force=False):
+ # List all running containers on the network and kill those with the statestored name
+ # prefix to make sure that no running container are left over from previous clusters.
+ container_name_prefix = self.__gen_container_name__("statestored")
+ for container_id, info in self.__get_network_info__()["Containers"].items():
+ container_name = info["Name"]
+ if container_name.startswith(container_name_prefix):
+ LOG.info("Stopping container {0}".format(container_name))
+ check_call(["docker", "stop", container_name])
def kill_admissiond(self, force=False):
self.__stop_container__("admissiond")
def start_statestore(self):
- self.__run_container__("statestored",
- build_statestored_arg_list(options.enable_catalogd_ha),
- {DEFAULT_STATESTORED_WEBSERVER_PORT: DEFAULT_STATESTORED_WEBSERVER_PORT})
+ if not options.enable_statestored_ha:
+ statestored_arg_lists =\
+ build_statestored_arg_list(num_statestored=1, remap_ports=False)
+ self.__run_container__("statestored", statestored_arg_lists[0],
+ {DEFAULT_STATESTORED_WEBSERVER_PORT: DEFAULT_STATESTORED_WEBSERVER_PORT})
+ else:
+ num_statestored = 2
+ statestored_arg_lists =\
+ build_statestored_arg_list(num_statestored, remap_ports=False)
+ for i in range(num_statestored):
+ chosen_ports = choose_statestored_ports(
+ enable_statestored_ha=True, instance_num=i)
+ port_map = {
+ DEFAULT_STATESTORE_SERVICE_PORT: chosen_ports['state_store_port'],
+ DEFAULT_STATESTORE_HA_SERVICE_PORT: chosen_ports['state_store_ha_port'],
+ DEFAULT_PEER_STATESTORE_HA_SERVICE_PORT:
+ chosen_ports['state_store_peer_ha_port'],
+ DEFAULT_STATESTORED_WEBSERVER_PORT: chosen_ports['webserver_port']}
+ self.__run_container__("statestored", statestored_arg_lists[i], port_map, i)
def start_catalogd(self):
if options.enable_catalogd_ha:
num_catalogd = 2
else:
num_catalogd = 1
- catalogd_arg_lists = build_catalogd_arg_list(
- num_catalogd, options.enable_catalogd_ha, remap_ports=False)
+ catalogd_arg_lists = build_catalogd_arg_list(num_catalogd, remap_ports=False)
for i in range(num_catalogd):
chosen_ports = choose_catalogd_ports(i)
port_map = {DEFAULT_CATALOG_SERVICE_PORT: chosen_ports['catalog_service_port'],
@@ -705,7 +818,7 @@ class DockerMiniClusterOperations(object):
def start_admissiond(self):
self.__run_container__("admissiond", build_admissiond_arg_list(),
- {DEFAULT_ADMISSIOND_WEBSERVER_PORT: DEFAULT_ADMISSIOND_WEBSERVER_PORT})
+ {DEFAULT_ADMISSIOND_WEBSERVER_PORT: DEFAULT_ADMISSIOND_WEBSERVER_PORT})
def start_impalads(self, cluster_size, num_coordinators, use_exclusive_coordinators):
impalad_arg_lists = build_impalad_arg_lists(cluster_size, num_coordinators,
@@ -884,7 +997,7 @@ if __name__ == "__main__":
elif options.restart_catalogd_only:
cluster_ops.kill_all_catalogds(force=options.force_kill)
elif options.restart_statestored_only:
- cluster_ops.kill_statestored(force=options.force_kill)
+ cluster_ops.kill_all_statestoreds(force=options.force_kill)
elif options.add_executors:
pass
else:
diff --git a/common/thrift/StatestoreService.thrift b/common/thrift/StatestoreService.thrift
index d752ac099..69b73a26d 100644
--- a/common/thrift/StatestoreService.thrift
+++ b/common/thrift/StatestoreService.thrift
@@ -194,6 +194,16 @@ enum TStatestoreSubscriberType {
COORDINATOR_EXECUTOR = 5
}
+// To support statestore HA, the standby statestore instance monitor the state of active
+// statestore instance by collecting connection states between active statestore instance
+// and subscribers. This enum define the connection states reported by subscribers in
+// heartbeat response message.
+enum TStatestoreConnState {
+ OK = 0
+ FAILED = 1
+ UNKNOWN = 2
+}
+
// Additional registration info for catalog daemon.
struct TCatalogRegistration {
// Protocol version of Catalog service.
@@ -257,6 +267,12 @@ struct TRegisterSubscriberResponse {
// The version of active catalogd
6: optional i64 catalogd_version;
+
+ // Active state of this Statestore instance.
+ 7: optional bool statestore_is_active;
+
+ // The version of active statestored
+ 8: optional i64 active_statestored_version;
}
struct TGetProtocolVersionRequest {
@@ -276,6 +292,22 @@ struct TGetProtocolVersionResponse {
3: optional Types.TUniqueId statestore_id;
}
+struct TSetStatestoreDebugActionRequest {
+ // Protocol version of the subscriber
+ 1: required StatestoreServiceVersion protocol_version =
+ StatestoreServiceVersion.V2
+
+ // Disable network for statestored if it's set as true so that statestored does not
+ // sent heartbeat and HA handshake messages. Enable network if it's set as false.
+ // This is used to simulate network failure for unit-test.
+ 2: optional bool disable_network;
+}
+
+struct TSetStatestoreDebugActionResponse {
+ // whether the call was executed correctly at the application level
+ 1: required Status.TStatus status;
+}
+
service StatestoreService {
// Register a single subscriber. Note that after a subscriber is registered, no new
// topics may be added.
@@ -283,6 +315,11 @@ service StatestoreService {
// Get protocol version of the statestore
TGetProtocolVersionResponse GetProtocolVersion(1: TGetProtocolVersionRequest params);
+
+ // Set debug action for the statestored.
+ // This API is only used for unit-test.
+ TSetStatestoreDebugActionResponse SetStatestoreDebugAction(
+ 1: TSetStatestoreDebugActionRequest params);
}
struct TUpdateStateRequest {
@@ -323,11 +360,19 @@ struct THeartbeatRequest {
// Unique identifier for the statestore instance.
3: optional Types.TUniqueId statestore_id;
+
+ // Flag set by standby Statestore instance to request the connection state between
+ // active statestore and the subscriber.
+ 4: optional bool request_statestore_conn_state;
}
struct THeartbeatResponse {
// Whether the call was executed correctly at the application level
1: required Status.TStatus status;
+
+ // Connection state between active statestore and the subscriber reported by
+ // the subscriber.
+ 2: optional TStatestoreConnState statestore_conn_state;
}
struct TUpdateCatalogdRequest {
@@ -358,6 +403,40 @@ struct TUpdateCatalogdResponse {
2: optional bool skipped;
}
+struct TUpdateStatestoredRoleRequest {
+ // Protocol version of the statestore
+ 1: required StatestoreServiceVersion protocol_version =
+ StatestoreServiceVersion.V2;
+
+ // registration_id of the receiver
+ 2: required Types.TUniqueId registration_id;
+
+ // Unique identifier for the statestore instance.
+ 3: required Types.TUniqueId statestore_id;
+
+ // Monotonously increasing number
+ 4: required i64 active_statestored_version;
+
+ // Active state of Statestore instance
+ 5: required bool is_active;
+
+ // Send elected active catalogd along with the role of active statestore.
+ // The version of active catalogd. Don't set this catalogd_version and
+ // catalogd_registration if active catalogd has not been elected.
+ 6: optional i64 catalogd_version;
+
+ // Catalog registration info.
+ 7: optional TCatalogRegistration catalogd_registration;
+}
+
+struct TUpdateStatestoredRoleResponse {
+ // Whether the call was executed correctly
+ 1: required Status.TStatus status;
+
+ // True if this update was skipped by the subscriber.
+ 2: optional bool skipped;
+}
+
service StatestoreSubscriber {
// Called when the statestore sends a topic update. The request contains a map of
// topic names to TTopicDelta updates, sent from the statestore to the subscriber. Each
@@ -369,6 +448,12 @@ service StatestoreSubscriber {
// received an unexpected delta update version range, they can request a new delta
// update based off a specific version from the statestore. The next statestore
// delta update will be based off of the version the subscriber requested.
+ //
+ // When Statestore HA is enabled, two statestore instances will be created in an Impala
+ // cluster as Active-Passive HA pair. All subscribers should register to both
+ // statestore instances so that two statestore instances have same set of subscribers.
+ // Both Statestore instances send heartbeat to their subscribers, but only active
+ // statestore instance sends topic update to its subscribers.
TUpdateStateResponse UpdateState(1: TUpdateStateRequest params);
// Called when the statestore sends a heartbeat.
@@ -376,4 +461,70 @@ service StatestoreSubscriber {
// Called when active catalogd has been updated.
TUpdateCatalogdResponse UpdateCatalogd(1: TUpdateCatalogdRequest params);
+
+ // Called by standby statestore when the Statestore service failover happens, which
+ // causes the active states of statestoreds are changed.
+ TUpdateStatestoredRoleResponse UpdateStatestoredRole(
+ 1: TUpdateStatestoredRoleRequest params);
+}
+
+struct TStatestoreHaHandshakeRequest {
+ // Service's version of source Statestore instance
+ 1: required StatestoreServiceVersion src_protocol_version =
+ StatestoreServiceVersion.V2;
+
+ // Instance id of source Statestore instance
+ 2: required Types.TUniqueId src_statestore_id;
+
+ // Address of the Statestore that source statestore instance runs
+ 3: required string src_statestore_address;
+
+ // True if starting flag statestore_force_active is set as true for source statestore
+ // instance.
+ 4: required bool src_force_active;
+}
+
+struct TStatestoreHaHandshakeResponse {
+ // Whether the call was executed correctly
+ 1: required Status.TStatus status;
+
+ // Service's version of destination Statestore instance
+ 2: required StatestoreServiceVersion dst_protocol_version =
+ StatestoreServiceVersion.V2;
+
+ // Instance id of destination Statestore instance
+ 3: required Types.TUniqueId dst_statestore_id;
+
+ // Indication if the destination Statestore is active for preemption. Source
+ // statestore should set itself as inactive if dst_statestore_active is set as true.
+ // Otherwise, source statestore sets itself as active.
+ 4: required bool dst_statestore_active;
+}
+
+struct TStatestoreHaHeartbeatRequest {
+ // Service's version of source Statestore instance
+ 1: required StatestoreServiceVersion protocol_version =
+ StatestoreServiceVersion.V2;
+
+ // Destination Statestore instance id
+ 2: required Types.TUniqueId dst_statestore_id;
+
+ // Source Statestore instance id
+ 3: required Types.TUniqueId src_statestore_id;
+}
+
+struct TStatestoreHaHeartbeatResponse {
+ 1: required Status.TStatus status;
+}
+
+// Service for communication between two statestore instances in an Active-Passive
+// HA (High Availability) pair.
+service StatestoreHaService {
+ // StatestoreHaHandshake: negotiate active role between two instances during startup.
+ TStatestoreHaHandshakeResponse StatestoreHaHandshake(
+ 1: TStatestoreHaHandshakeRequest params);
+
+ // StatestoreHaHeartbeat: active statestore sends heartbeat to standby statestore.
+ TStatestoreHaHeartbeatResponse StatestoreHaHeartbeat(
+ 1: TStatestoreHaHeartbeatRequest params);
}
diff --git a/common/thrift/metrics.json b/common/thrift/metrics.json
index ae53d06d9..55de02f44 100644
--- a/common/thrift/metrics.json
+++ b/common/thrift/metrics.json
@@ -1171,6 +1171,66 @@
"kind": "GAUGE",
"key": "impala.thrift-server.StatestoreService.timedout-cnxn-requests"
},
+ {
+ "description": "The number of active connections to this StateStore's HA service.",
+ "contexts": [
+ "STATESTORE"
+ ],
+ "label": "StateStore HA Service Active Connections",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "impala.thrift-server.StatestoreHaService.connections-in-use"
+ },
+ {
+ "description": "Amount of time clients of StateStore HA Service spent waiting for connection to be set up",
+ "contexts": [
+ "STATESTORE"
+ ],
+ "label": "StateStore HA Service client connection setup time",
+ "units": "TIME_US",
+ "kind": "HISTOGRAM",
+ "key": "impala.thrift-server.StatestoreHaService.connection-setup-time"
+ },
+ {
+ "description": "Amount of time clients of StateStore HA Service spent waiting for service threads",
+ "contexts": [
+ "STATESTORE"
+ ],
+ "label": "StateStore HA Service clients' wait time for service threads",
+ "units": "TIME_US",
+ "kind": "HISTOGRAM",
+ "key": "impala.thrift-server.StatestoreHaService.svc-thread-wait-time"
+ },
+ {
+ "description": "The total number of connections made to this StateStore's StateStore HA service over its lifetime.",
+ "contexts": [
+ "STATESTORE"
+ ],
+ "label": "StateStore HA Service Connections",
+ "units": "UNIT",
+ "kind": "COUNTER",
+ "key": "impala.thrift-server.StatestoreHaService.total-connections"
+ },
+ {
+ "description": "The number of connections to the Statestore HA Service that have been accepted and are waiting to be setup.",
+ "contexts": [
+ "STATESTORE"
+ ],
+ "label": "Statestore HA Service Connections Queued for Setup",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "impala.thrift-server.StatestoreHaService.connection-setup-queue-size"
+ },
+ {
+ "description": "The number of connection requests to the Statestore HA Service that have been timed out waiting to be setup.",
+ "contexts": [
+ "STATESTORE"
+ ],
+ "label": "Statestore HA Service Connection Requests Timed Out",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "impala.thrift-server.StatestoreHaService.timedout-cnxn-requests"
+ },
{
"description": "The number of active Impala Backend client connections to this Impala Daemon.",
"contexts": [
@@ -1782,6 +1842,18 @@
"kind": "COUNTER",
"key": "statestore-subscriber.num-update-catalogd-rpc"
},
+ {
+ "description": "The number of RPCs received for updating role of statestored.",
+ "contexts": [
+ "ADMISSIOND",
+ "CATALOGSERVER",
+ "IMPALAD"
+ ],
+ "label": "Number of UpdateStatestoredRole RPCs",
+ "units": "UNIT",
+ "kind": "COUNTER",
+ "key": "statestore-subscriber.num-update-statestored-role-rpc"
+ },
{
"description": "The number of Re-Registration Attempt.",
"contexts": [
@@ -1860,6 +1932,30 @@
"kind": "STATS",
"key": "statestore-subscriber.topic-update-interval-time"
},
+ {
+ "description": "Current active status of the Statestore daemon.",
+ "contexts": [
+ "ADMISSIOND",
+ "CATALOGSERVER",
+ "IMPALAD"
+ ],
+ "label": "The active status of Statestore daemon",
+ "units": "NONE",
+ "kind": "PROPERTY",
+ "key": "statestore-subscriber.statestore-active-status"
+ },
+ {
+ "description": "Current active status of the Statestore daemon.",
+ "contexts": [
+ "ADMISSIOND",
+ "CATALOGSERVER",
+ "IMPALAD"
+ ],
+ "label": "The active status of Statestore daemon",
+ "units": "NONE",
+ "kind": "PROPERTY",
+ "key": "statestore2-subscriber.statestore-active-status"
+ },
{
"description": "The time (sec) spent sending heartbeat RPCs. Includes subscriber-side processing time and network transmission time.",
"contexts": [
@@ -1890,6 +1986,26 @@
"kind": "COUNTER",
"key": "statestore.num-failed-update-catalogd-rpc"
},
+ {
+ "description": "The number of successful RPCs for updating role of statestored.",
+ "contexts": [
+ "STATESTORE"
+ ],
+ "label": "Number of successful UpdateStatestoredRole RPCs sent by statestore",
+ "units": "UNIT",
+ "kind": "COUNTER",
+ "key": "statestore.num-successful-update-statestored-role-rpc"
+ },
+ {
+ "description": "The number of failed RPCs for updating role of statestored.",
+ "contexts": [
+ "STATESTORE"
+ ],
+ "label": "Number of failed UpdatetatestoredRole RPCs sent by statestore",
+ "units": "UNIT",
+ "kind": "COUNTER",
+ "key": "statestore.num-failed-update-statestored-role-rpc"
+ },
{
"description": "The number of requests for clearing topic entries from catalogd.",
"contexts": [
@@ -1990,6 +2106,46 @@
"kind": "PROPERTY",
"key": "statestore.version"
},
+ {
+ "description": "Current active status of this Statestore daemon.",
+ "contexts": [
+ "STATESTORE"
+ ],
+ "label": "The active status of Statestore daemon",
+ "units": "NONE",
+ "kind": "PROPERTY",
+ "key": "statestore.active-status"
+ },
+ {
+ "description": "Statestore service is started and ready to accept connections.",
+ "contexts": [
+ "STATESTORE"
+ ],
+ "label": "Statestore service started",
+ "units": "NONE",
+ "kind": "PROPERTY",
+ "key": "statestore.service-started"
+ },
+ {
+ "description": "Statestored instance is in HA recovery mode.",
+ "contexts": [
+ "STATESTORE"
+ ],
+ "label": "Statestored in HA recovery mode",
+ "units": "NONE",
+ "kind": "PROPERTY",
+ "key": "statestore.in-ha-recovery-mode"
+ },
+ {
+ "description": "Statestored instance connected with peer statestored.",
+ "contexts": [
+ "STATESTORE"
+ ],
+ "label": "Statestored connected with peer statestored",
+ "units": "NONE",
+ "kind": "PROPERTY",
+ "key": "statestore.connected-with-peer-statestored"
+ },
{
"description": "The number of clients in use by the Statestore heartbeat client cache.",
"contexts": [
@@ -2050,6 +2206,46 @@
"kind": "GAUGE",
"key": "subscriber-update-catalogd.client-cache.total-clients"
},
+ {
+ "description": "The number of clients in use by the Statestore update statestored client cache.",
+ "contexts": [
+ "STATESTORE"
+ ],
+ "label": "Subscriber Update role of Statestored Client Cache Clients In Use",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "subscriber-update-statestored.client-cache.clients-in-use"
+ },
+ {
+ "description": "The total number of clients in the Statestore update statestored client cache.",
+ "contexts": [
+ "STATESTORE"
+ ],
+ "label": "Subscriber Update role of Statestored Client Cache Total Clients",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "subscriber-update-statestored.client-cache.total-clients"
+ },
+ {
+ "description": "The number of clients in use by the Statestore update statestored client cache.",
+ "contexts": [
+ "STATESTORE"
+ ],
+ "label": "Subscriber Update role of Statestored Client Cache Clients In Use",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "statestored-ha.client-cache.clients-in-use"
+ },
+ {
+ "description": "The total number of clients in the Statestore update statestored client cache.",
+ "contexts": [
+ "STATESTORE"
+ ],
+ "label": "Subscriber Update role of Statestored Client Cache Total Clients",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "statestored-ha.client-cache.total-clients"
+ },
{
"description": "Number of bytes used by the application. This will not typically match the memory use reported by the OS, because it does not include TCMalloc overhead or memory fragmentation.",
"contexts": [
diff --git a/tests/common/impala_cluster.py b/tests/common/impala_cluster.py
index 6989123f6..8e1d0a70f 100644
--- a/tests/common/impala_cluster.py
+++ b/tests/common/impala_cluster.py
@@ -55,6 +55,9 @@ DEFAULT_KRPC_PORT = 27000
DEFAULT_CATALOG_SERVICE_PORT = 26000
DEFAULT_STATE_STORE_SUBSCRIBER_PORT = 23000
DEFAULT_CATALOGD_STATE_STORE_SUBSCRIBER_PORT = 23020
+DEFAULT_STATESTORE_SERVICE_PORT = 24000
+DEFAULT_STATESTORE_HA_SERVICE_PORT = 24020
+DEFAULT_PEER_STATESTORE_HA_SERVICE_PORT = 24021
DEFAULT_IMPALAD_WEBSERVER_PORT = 25000
DEFAULT_STATESTORED_WEBSERVER_PORT = 25010
DEFAULT_CATALOGD_WEBSERVER_PORT = 25020
@@ -117,6 +120,13 @@ class ImpalaCluster(object):
# If no statestored process exists, return None.
return self.__statestoreds[0] if len(self.__statestoreds) > 0 else None
+ def statestoreds(self):
+ """Returns a list of the known statestored processes"""
+ return self.__statestoreds
+
+ def get_first_statestored(self):
+ return self.statestoreds[0]
+
@property
def impalads(self):
"""Returns a list of the known impalad processes"""
@@ -238,7 +248,7 @@ class ImpalaCluster(object):
environment this would need to enumerate each machine in the cluster.
"""
impalads = list()
- statestored = list()
+ statestoreds = list()
catalogds = list()
admissiond = None
daemons = ['impalad', 'catalogd', 'statestored', 'admissiond']
@@ -259,15 +269,16 @@ class ImpalaCluster(object):
if binary == 'impalad':
impalads.append(ImpaladProcess(cmdline))
elif binary == 'statestored':
- statestored.append(StateStoreProcess(cmdline))
+ statestoreds.append(StateStoreProcess(cmdline))
elif binary == 'catalogd':
catalogds.append(CatalogdProcess(cmdline))
elif binary == 'admissiond':
admissiond = AdmissiondProcess(cmdline)
self.__sort_impalads(impalads)
+ self.__sort_statestoreds(statestoreds)
self.__sort_catalogds(catalogds)
- return impalads, statestored, catalogds, admissiond
+ return impalads, statestoreds, catalogds, admissiond
def __find_docker_containers(self):
"""
@@ -308,6 +319,7 @@ class ImpalaCluster(object):
admissiond = AdmissiondProcess(args, container_id=container_id,
port_map=port_map)
self.__sort_impalads(impalads)
+ self.__sort_statestoreds(statestoreds)
self.__sort_catalogds(catalogds)
return impalads, statestoreds, catalogds, admissiond
@@ -318,6 +330,13 @@ class ImpalaCluster(object):
the containerised cluster."""
impalads.sort(key=lambda i: i.service.hs2_port)
+ def __sort_statestoreds(self, statestoreds):
+ """Does an in-place sort of a list of StateStoredService objects into a canonical
+ order. We order them by their service port, so that get_first_statestored() always
+ returns the first one. We need to use a port that is exposed and mapped to a host
+ port for the containerised cluster."""
+ statestoreds.sort(key=lambda i: i.service.service_port)
+
def __sort_catalogds(self, catalogds):
"""Does an in-place sort of a list of CatalogdProcess objects into a canonical order.
We order them by their service port, so that get_first_catalogd() always returns the
@@ -569,11 +588,27 @@ class StateStoreProcess(BaseImpalaProcess):
def __init__(self, cmd, container_id=None, port_map=None):
super(StateStoreProcess, self).__init__(cmd, container_id, port_map)
self.service = StateStoredService(self.hostname, self.webserver_interface,
- self.get_webserver_port(), self._get_webserver_certificate_file())
+ self.get_webserver_port(), self._get_webserver_certificate_file(),
+ self.__get_port())
def _get_default_webserver_port(self):
return DEFAULT_STATESTORED_WEBSERVER_PORT
+ def __get_port(self):
+ return int(self._get_port('state_store_port', DEFAULT_STATESTORE_SERVICE_PORT))
+
+ def start(self, wait_until_ready=True, additional_args=None):
+ """Starts statestored and waits until the service is started and ready to accept
+ connections."""
+ restart_args = self.cmd[1:]
+ if additional_args:
+ restart_args = restart_args + [additional_args]
+ LOG.info("Starting Statestored process: {0}".format(restart_args))
+ run_daemon("statestored", restart_args)
+ if wait_until_ready:
+ self.service.wait_for_metric_value('statestore.service-started',
+ expected_value=1, timeout=30)
+
# Represents a catalogd process
class CatalogdProcess(BaseImpalaProcess):
diff --git a/tests/common/impala_service.py b/tests/common/impala_service.py
index a412133c7..d46f67256 100644
--- a/tests/common/impala_service.py
+++ b/tests/common/impala_service.py
@@ -473,14 +473,18 @@ class ImpaladService(BaseImpalaService):
# accessing the debug webpage.
class StateStoredService(BaseImpalaService):
def __init__(self, hostname, webserver_interface, webserver_port,
- webserver_certificate_file):
+ webserver_certificate_file, service_port):
super(StateStoredService, self).__init__(
hostname, webserver_interface, webserver_port, webserver_certificate_file)
+ self.service_port = service_port
def wait_for_live_subscribers(self, num_subscribers, timeout=15, interval=1):
self.wait_for_metric_value('statestore.live-backends', num_subscribers,
timeout=timeout, interval=interval)
+ def get_statestore_service_port(self):
+ return self.service_port
+
# Allows for interacting with the Catalog service to perform operations such as
# accessing the debug webpage.
diff --git a/tests/custom_cluster/test_statestored_ha.py b/tests/custom_cluster/test_statestored_ha.py
new file mode 100644
index 000000000..1707fd950
--- /dev/null
+++ b/tests/custom_cluster/test_statestored_ha.py
@@ -0,0 +1,627 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import absolute_import, division, print_function
+import logging
+import pytest
+
+from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
+from tests.common.environ import build_flavor_timeout
+from tests.common.impala_cluster import (
+ DEFAULT_CATALOG_SERVICE_PORT, DEFAULT_STATESTORE_SERVICE_PORT)
+from time import sleep
+
+from thrift.protocol import TBinaryProtocol
+from thrift.transport import TSocket, TTransport
+
+import StatestoreService.StatestoreSubscriber as Subscriber
+import StatestoreService.StatestoreService as Statestore
+
+LOG = logging.getLogger('statestored_ha_test')
+
+
+class TestStatestoredHA(CustomClusterTestSuite):
+ """A simple wrapper class to launch a cluster with Statestored HA enabled.
+ The cluster will be launched with two statestored instances as Active-Passive HA pair.
+ All cluster components are started with starting flag FLAGS_enable_statestored_ha
+ as true."""
+
+ def get_workload(self):
+ return 'functional-query'
+
+ def __disable_statestored_network(self,
+ service_port=DEFAULT_STATESTORE_SERVICE_PORT, disable_network=False):
+ request = Subscriber.TSetStatestoreDebugActionRequest(
+ protocol_version=Subscriber.StatestoreServiceVersion.V2,
+ disable_network=disable_network)
+ client_transport = TTransport.TBufferedTransport(
+ TSocket.TSocket('localhost', service_port))
+ trans_protocol = TBinaryProtocol.TBinaryProtocol(client_transport)
+ client = Statestore.Client(trans_protocol)
+ client_transport.open()
+ try:
+ return client.SetStatestoreDebugAction(request)
+ except Exception as e:
+ assert False, str(e)
+
+ # Return port of the active catalogd of statestore
+ def __get_active_catalogd_port(self, statestore_service):
+ active_catalogd_address = \
+ statestore_service.get_metric_value("statestore.active-catalogd-address")
+ _, catalog_service_port = active_catalogd_address.split(":")
+ return int(catalog_service_port)
+
+ # Verify port of the active catalogd of impalad is matching with the catalog
+ # service port of the given catalogd service.
+ def __verify_impalad_active_catalogd_port(self, impalad_index, catalogd_service):
+ impalad_service = self.cluster.impalads[impalad_index].service
+ active_catalogd_address = \
+ impalad_service.get_metric_value("catalog.active-catalogd-address")
+ _, catalog_service_port = active_catalogd_address.split(":")
+ assert(int(catalog_service_port) == catalogd_service.get_catalog_service_port())
+
+ # Verify the active statestored on impalad is matching with the current active
+ # statestored.
+ def __verify_impalad_active_statestored(self, impalad_index, active_statestored_index):
+ impalad_service = self.cluster.impalads[impalad_index].service
+ if active_statestored_index == 0:
+ impalad_service.wait_for_metric_value(
+ "statestore-subscriber.statestore-active-status",
+ expected_value=True, timeout=5)
+ assert(impalad_service.get_metric_value(
+ "statestore-subscriber.statestore-active-status"))
+ assert(not impalad_service.get_metric_value(
+ "statestore2-subscriber.statestore-active-status"))
+ else:
+ impalad_service.wait_for_metric_value(
+ "statestore2-subscriber.statestore-active-status",
+ expected_value=True, timeout=5)
+ assert(not impalad_service.get_metric_value(
+ "statestore-subscriber.statestore-active-status"))
+ assert(impalad_service.get_metric_value(
+ "statestore2-subscriber.statestore-active-status"))
+
+ # Verify the active statestored on catalogd is matching with the current active
+ # statestored.
+ def __verify_catalogd_active_statestored(
+ self, catalogd_index, active_statestored_index):
+ catalogds = self.cluster.catalogds()
+ catalogd_service = catalogds[catalogd_index].service
+ if active_statestored_index == 0:
+ catalogd_service.wait_for_metric_value(
+ "statestore-subscriber.statestore-active-status",
+ expected_value=True, timeout=5)
+ assert(catalogd_service.get_metric_value(
+ "statestore-subscriber.statestore-active-status"))
+ assert(not catalogd_service.get_metric_value(
+ "statestore2-subscriber.statestore-active-status"))
+ else:
+ catalogd_service.wait_for_metric_value(
+ "statestore2-subscriber.statestore-active-status",
+ expected_value=True, timeout=5)
+ assert(not catalogd_service.get_metric_value(
+ "statestore-subscriber.statestore-active-status"))
+ assert(catalogd_service.get_metric_value(
+ "statestore2-subscriber.statestore-active-status"))
+
+ def __run_simple_queries(self, sync_ddl=False):
+ try:
+ if sync_ddl:
+ self.execute_query_expect_success(self.client, "set SYNC_DDL=1")
+ self.execute_query_expect_success(
+ self.client, "drop table if exists test_statestored_ha")
+ self.execute_query_expect_success(
+ self.client, "create table if not exists test_statestored_ha (id int)")
+ self.execute_query_expect_success(
+ self.client, "insert into table test_statestored_ha values(1), (2), (3)")
+ self.execute_query_expect_success(
+ self.client, "select count(*) from test_statestored_ha")
+ finally:
+ self.execute_query_expect_success(
+ self.client, "drop table if exists test_statestored_ha")
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ statestored_args="--use_network_address_as_statestore_priority=true",
+ start_args="--enable_statestored_ha")
+ def test_statestored_ha_with_two_statestored(self):
+ """The test case for cluster started with statestored HA enabled."""
+ # Verify two statestored instances are created with one as active.
+ statestoreds = self.cluster.statestoreds()
+ assert(len(statestoreds) == 2)
+ assert(statestoreds[0].service.get_metric_value("statestore.active-status"))
+ assert(not statestoreds[1].service.get_metric_value("statestore.active-status"))
+
+ # Verify the active statestored on catalogd and impalad are matching with
+ # the current active statestored.
+ self.__verify_catalogd_active_statestored(0, 0)
+ self.__verify_impalad_active_statestored(0, 0)
+ self.__verify_impalad_active_statestored(1, 0)
+ self.__verify_impalad_active_statestored(2, 0)
+
+ # Verify simple queries are ran successfully.
+ self.__run_simple_queries()
+ # Verify simple queries with sync_ddl as 1.
+ self.__run_simple_queries(sync_ddl=True)
+
+ # Restart one coordinator. Verify it has right active statestored.
+ self.cluster.impalads[0].restart()
+ self.cluster.impalads[0].service.wait_for_metric_value('impala-server.ready',
+ expected_value=1, timeout=30)
+ self.__verify_impalad_active_statestored(0, 0)
+
+ # Restart standby statestore. Verify that the roles are not changed.
+ statestoreds[1].kill()
+ statestoreds[1].start(wait_until_ready=True)
+ sleep(1)
+ assert(statestoreds[0].service.get_metric_value("statestore.active-status"))
+ assert(not statestoreds[1].service.get_metric_value("statestore.active-status"))
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ statestored_args="--enable_statestored_ha=true "
+ "--use_network_address_as_statestore_priority=true "
+ "--statestore_ha_preemption_wait_period_ms=200",
+ catalogd_args="--enable_statestored_ha=true",
+ impalad_args="--enable_statestored_ha=true")
+ def test_statestored_ha_with_one_statestored(self):
+ """The test case for cluster with only one statestored when statestored HA is
+ enabled."""
+ # Verify the statestored instances is created as active.
+ statestoreds = self.cluster.statestoreds()
+ assert(len(statestoreds) == 1)
+ assert(statestoreds[0].service.get_metric_value("statestore.active-status"))
+
+ # Verify the active statestored on catalogd and impalad are matching with
+ # the current active statestored.
+ self.__verify_catalogd_active_statestored(0, 0)
+ self.__verify_impalad_active_statestored(0, 0)
+ self.__verify_impalad_active_statestored(1, 0)
+ self.__verify_impalad_active_statestored(2, 0)
+
+ # Verify simple queries are ran successfully.
+ self.__run_simple_queries()
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ statestored_args="--use_network_address_as_statestore_priority=true",
+ start_args="--enable_statestored_ha --enable_catalogd_ha")
+ def test_statestored_ha_with_catalogd_ha(self):
+ """The test case for cluster started with statestored HA enabled."""
+ # Verify two statestored instances are created with one as active.
+ statestoreds = self.cluster.statestoreds()
+ assert(len(statestoreds) == 2)
+ assert(statestoreds[0].service.get_metric_value("statestore.active-status"))
+ assert(not statestoreds[1].service.get_metric_value("statestore.active-status"))
+
+ # Verify the active statestored on catalogd and impalad are matching with
+ # the current active statestored.
+ self.__verify_catalogd_active_statestored(0, 0)
+ self.__verify_catalogd_active_statestored(1, 0)
+ self.__verify_impalad_active_statestored(0, 0)
+ self.__verify_impalad_active_statestored(1, 0)
+ self.__verify_impalad_active_statestored(2, 0)
+
+ # Verify simple queries are ran successfully.
+ self.__run_simple_queries()
+ # Verify simple queries with sync_ddl as 1.
+ self.__run_simple_queries(sync_ddl=True)
+
+ def __test_statestored_auto_failover(
+ self, catalogd_ha_enabled=False, no_catalogd_failover=True):
+ """Stop active statestored and verify standby statestored becomes active.
+ Restart original active statestored. Verify that it does not resume its active
+ role."""
+ # Verify two statestored instances are created with one as active.
+ statestoreds = self.cluster.statestoreds()
+ assert(len(statestoreds) == 2)
+ statestore_service_0 = statestoreds[0].service
+ statestore_service_1 = statestoreds[1].service
+ assert(statestore_service_0.get_metric_value("statestore.active-status"))
+ assert(not statestore_service_1.get_metric_value("statestore.active-status"))
+
+ # Get active catalogd port from active statestored
+ original_active_catalogd_port = self.__get_active_catalogd_port(statestore_service_0)
+
+ if catalogd_ha_enabled:
+ self.__verify_catalogd_active_statestored(0, 0)
+ self.__verify_catalogd_active_statestored(1, 0)
+ else:
+ self.__verify_catalogd_active_statestored(0, 0)
+ self.__verify_impalad_active_statestored(0, 0)
+ self.__verify_impalad_active_statestored(1, 0)
+ self.__verify_impalad_active_statestored(2, 0)
+
+ # Kill active statestored
+ statestoreds[0].kill()
+
+ # Wait for long enough for the standby statestored to detect the failure of active
+ # statestored and assign itself with active role.
+ statestore_service_1.wait_for_metric_value(
+ "statestore.active-status", expected_value=True, timeout=120)
+ assert(statestore_service_1.get_metric_value("statestore.active-status"))
+ sleep(1)
+
+ # Verify the active statestored on catalogd and impalad are matching with
+ # the current active statestored.
+ if catalogd_ha_enabled:
+ self.__verify_catalogd_active_statestored(0, 1)
+ self.__verify_catalogd_active_statestored(1, 1)
+ else:
+ self.__verify_catalogd_active_statestored(0, 1)
+ self.__verify_impalad_active_statestored(0, 1)
+ self.__verify_impalad_active_statestored(1, 1)
+ self.__verify_impalad_active_statestored(2, 1)
+
+ # Get active catalogd port from current active statestored
+ current_active_catalogd_port = self.__get_active_catalogd_port(statestore_service_1)
+
+ if catalogd_ha_enabled:
+ if no_catalogd_failover:
+ assert(original_active_catalogd_port == current_active_catalogd_port)
+ else:
+ # Check if there is event of catalogd fail over
+ if original_active_catalogd_port != current_active_catalogd_port:
+ catalogds = self.cluster.catalogds()
+ if current_active_catalogd_port == DEFAULT_CATALOG_SERVICE_PORT:
+ catalogd_service = catalogds[0].service
+ else:
+ catalogd_service = catalogds[1].service
+ # Wait for long enough for the current active catalogd to receive notification
+ # of catalogd failover
+ catalogd_service.wait_for_metric_value(
+ "catalog-server.active-status", expected_value=True, timeout=60)
+ # Wait catalog topics are propagated from new active catalogd through the
+ # active statestored.
+ sleep(2)
+ self.__verify_impalad_active_catalogd_port(0, catalogd_service)
+ self.__verify_impalad_active_catalogd_port(1, catalogd_service)
+ self.__verify_impalad_active_catalogd_port(2, catalogd_service)
+
+ # Verify simple queries are ran successfully.
+ self.__run_simple_queries()
+ # Verify simple queries with sync_ddl as 1.
+ self.__run_simple_queries(sync_ddl=True)
+
+ # Restart original active statestored. Verify that the statestored does not resume
+ # its active role to avoid flip-flop.
+ statestoreds[0].start(wait_until_ready=True)
+ sleep(1)
+ statestore_service_0 = statestoreds[0].service
+ assert(not statestore_service_0.get_metric_value("statestore.active-status"))
+ assert(statestore_service_1.get_metric_value("statestore.active-status"))
+
+ # Verify the active statestored on catalogd and impalad are matching with
+ # the current active statestored.
+ if catalogd_ha_enabled:
+ self.__verify_catalogd_active_statestored(0, 1)
+ self.__verify_catalogd_active_statestored(1, 1)
+ else:
+ self.__verify_catalogd_active_statestored(0, 1)
+ self.__verify_impalad_active_statestored(0, 1)
+ self.__verify_impalad_active_statestored(1, 1)
+ self.__verify_impalad_active_statestored(2, 1)
+
+ successful_update_statestored_rpc_num_0 = statestore_service_0.get_metric_value(
+ "statestore.num-successful-update-statestored-role-rpc")
+ successful_update_statestored_rpc_num_1 = statestore_service_1.get_metric_value(
+ "statestore.num-successful-update-statestored-role-rpc")
+ assert(successful_update_statestored_rpc_num_0 == 0)
+ assert(successful_update_statestored_rpc_num_1 >= 4)
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ statestored_args="--use_network_address_as_statestore_priority=true "
+ "--statestore_ha_heartbeat_monitoring_frequency_ms=50 "
+ "--statestore_peer_timeout_seconds=2",
+ impalad_args="--statestore_subscriber_timeout_seconds=2",
+ catalogd_args="--statestore_subscriber_timeout_seconds=2",
+ start_args="--enable_statestored_ha")
+ def test_statestored_auto_failover(self):
+ """Tests for Statestore Service auto fail over without failed RPCs."""
+ self.__test_statestored_auto_failover()
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ statestored_args="--use_network_address_as_statestore_priority=true "
+ "--statestore_ha_heartbeat_monitoring_frequency_ms=50 "
+ "--statestore_peer_timeout_seconds=2 "
+ "--debug_actions=SEND_UPDATE_STATESTORED_RPC_FIRST_ATTEMPT:FAIL@1.0",
+ impalad_args="--statestore_subscriber_timeout_seconds=2",
+ catalogd_args="--statestore_subscriber_timeout_seconds=2",
+ start_args="--enable_statestored_ha")
+ def test_statestored_auto_failover_with_failed_rpc(self):
+ """Tests for Statestore Service auto fail over with failed RPCs."""
+ self.__test_statestored_auto_failover()
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ statestored_args="--use_network_address_as_statestore_priority=true "
+ "--statestore_ha_heartbeat_monitoring_frequency_ms=50 "
+ "--statestore_peer_timeout_seconds=2 "
+ "--use_subscriber_id_as_catalogd_priority=true",
+ impalad_args="--statestore_subscriber_timeout_seconds=2",
+ catalogd_args="--statestore_subscriber_timeout_seconds=2",
+ start_args="--enable_statestored_ha --enable_catalogd_ha")
+ def test_statestored_auto_failover_without_catalogd_failover(self):
+ """Tests for Statestore Service auto fail over with CatalogD HA enabled.
+ use_subscriber_id_as_catalogd_priority is set as true so that both statestored
+ elect same catalogd as active, and there is no catalogd fail over when Statestore
+ servier fail over to standby statestored.
+ """
+ self.__test_statestored_auto_failover(
+ catalogd_ha_enabled=True, no_catalogd_failover=True)
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ statestored_args="--use_network_address_as_statestore_priority=true "
+ "--statestore_ha_heartbeat_monitoring_frequency_ms=50 "
+ "--statestore_peer_timeout_seconds=2 ",
+ impalad_args="--statestore_subscriber_timeout_seconds=2",
+ catalogd_args="--statestore_subscriber_timeout_seconds=2",
+ start_args="--enable_statestored_ha --enable_catalogd_ha")
+ def test_statestored_auto_failover_with_possible_catalogd_failover(self):
+ """Tests for Statestore Service auto fail over with CatalogD HA enabled.
+ use_subscriber_id_as_catalogd_priority is not set so that both statestored could
+ elect different catalogd as active. It's possible there is catalogd fail over when
+ Statestore servier fail over to standby statestored.
+ """
+ self.__test_statestored_auto_failover(
+ catalogd_ha_enabled=True, no_catalogd_failover=False)
+
+ def __test_statestored_manual_failover(self, second_failover=False):
+ """Stop active statestored and verify standby statestored becomes active.
+ Restart original active statestored with statestore_force_active as true. Verify
+ that the statestored resumes its active role.
+ """
+ # Verify two statestored instances are created with one as active.
+ statestoreds = self.cluster.statestoreds()
+ assert(len(statestoreds) == 2)
+ statestore_service_0 = statestoreds[0].service
+ statestore_service_1 = statestoreds[1].service
+ assert(statestore_service_0.get_metric_value("statestore.active-status"))
+ assert(not statestore_service_1.get_metric_value("statestore.active-status"))
+
+ self.__verify_catalogd_active_statestored(0, 0)
+ self.__verify_impalad_active_statestored(0, 0)
+ self.__verify_impalad_active_statestored(1, 0)
+ self.__verify_impalad_active_statestored(2, 0)
+
+ # Kill active statestored
+ statestoreds[0].kill()
+
+ # Wait for long enough for the standby statestored to detect the failure of active
+ # statestored and assign itself with active role.
+ statestore_service_1.wait_for_metric_value(
+ "statestore.active-status", expected_value=True, timeout=120)
+ assert(statestore_service_1.get_metric_value("statestore.active-status"))
+ sleep(1)
+
+ # Verify the active statestored on catalogd and impalad are matching with
+ # the current active statestored.
+ self.__verify_catalogd_active_statestored(0, 1)
+ self.__verify_impalad_active_statestored(0, 1)
+ self.__verify_impalad_active_statestored(1, 1)
+ self.__verify_impalad_active_statestored(2, 1)
+
+ # Verify simple queries are ran successfully.
+ self.__run_simple_queries()
+ # Verify simple queries with sync_ddl as 1.
+ self.__run_simple_queries(sync_ddl=True)
+
+ # Restart original active statestored with statestore_force_active as true.
+ # Verify that the statestored resumes its active role.
+ statestoreds[0].start(wait_until_ready=True,
+ additional_args="--statestore_force_active=true")
+ sleep(1)
+ statestore_service_0 = statestoreds[0].service
+ assert(statestore_service_0.get_metric_value("statestore.active-status"))
+ assert(not statestore_service_1.get_metric_value("statestore.active-status"))
+
+ # Verify the active statestored on catalogd is matching with the current active
+ # statestored.
+ catalogds = self.cluster.catalogds()
+ catalogds[0].service.wait_for_metric_value(
+ "statestore-subscriber.statestore-active-status", expected_value=True, timeout=60)
+ self.__verify_catalogd_active_statestored(0, 0)
+
+ successful_update_statestored_rpc_num_1 = statestore_service_1.get_metric_value(
+ "statestore.num-successful-update-statestored-role-rpc")
+ assert(successful_update_statestored_rpc_num_1 >= 4)
+
+ # Trigger second fail over by disabling the network of active statestored.
+ if second_failover:
+ self.__disable_statestored_network(disable_network=True)
+ # Wait for long enough for the standby statestored to detect the failure of active
+ # statestored and assign itself with active role.
+ statestore_service_1.wait_for_metric_value(
+ "statestore.active-status", expected_value=True, timeout=120)
+ assert(statestore_service_1.get_metric_value("statestore.active-status"))
+ # Verify that original active statestored is in HA recovery mode and is not active.
+ statestore_service_0.wait_for_metric_value(
+ "statestore.in-ha-recovery-mode", expected_value=True, timeout=60)
+ assert(not statestore_service_0.get_metric_value("statestore.active-status"))
+ sleep(1)
+
+ # Verify the active statestored on catalogd and impalad are matching with
+ # the current active statestored.
+ self.__verify_catalogd_active_statestored(0, 1)
+ self.__verify_impalad_active_statestored(0, 1)
+ self.__verify_impalad_active_statestored(1, 1)
+ self.__verify_impalad_active_statestored(2, 1)
+
+ # Verify simple queries are ran successfully.
+ self.__run_simple_queries()
+ # Verify simple queries with sync_ddl as 1.
+ self.__run_simple_queries(sync_ddl=True)
+
+ # Re-enable network for original active statestored. Verify that the statestored
+ # exits HA recovery mode, and resume its active role since it was started with
+ # statestore_force_active as true.
+ self.__disable_statestored_network(disable_network=False)
+ statestore_service_0.wait_for_metric_value(
+ "statestore.in-ha-recovery-mode", expected_value=False, timeout=60)
+ sleep(1)
+ assert(statestore_service_0.get_metric_value("statestore.active-status"))
+ assert(not statestore_service_1.get_metric_value("statestore.active-status"))
+
+ # Verify the active statestored on catalogd and impalad are matching with
+ # the current active statestored.
+ self.__verify_catalogd_active_statestored(0, 0)
+ self.__verify_impalad_active_statestored(0, 0)
+ self.__verify_impalad_active_statestored(1, 0)
+ self.__verify_impalad_active_statestored(2, 0)
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ statestored_args="--use_network_address_as_statestore_priority=true "
+ "--statestore_ha_heartbeat_monitoring_frequency_ms=100 "
+ "--statestore_peer_timeout_seconds=2 "
+ "--debug_actions=DISABLE_STATESTORE_NETWORK",
+ impalad_args="--statestore_subscriber_timeout_seconds=2",
+ catalogd_args="--statestore_subscriber_timeout_seconds=2",
+ start_args="--enable_statestored_ha")
+ def test_statestored_manual_failover(self):
+ """Tests for Statestore Service manual fail over without failed RPCs."""
+ self.__test_statestored_manual_failover(second_failover=True)
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ statestored_args="--use_network_address_as_statestore_priority=true "
+ "--statestore_ha_heartbeat_monitoring_frequency_ms=100 "
+ "--statestore_peer_timeout_seconds=2 "
+ "--debug_actions=SEND_UPDATE_STATESTORED_RPC_FIRST_ATTEMPT:FAIL@1.0",
+ impalad_args="--statestore_subscriber_timeout_seconds=2",
+ catalogd_args="--statestore_subscriber_timeout_seconds=2",
+ start_args="--enable_statestored_ha")
+ def test_statestored_manual_failover_with_failed_rpc(self):
+ """Tests for Statestore Service manual fail over with failed RPCs."""
+ self.__test_statestored_manual_failover(second_failover=False)
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ statestored_args="--statestore_force_active=true",
+ start_args="--enable_statestored_ha")
+ def test_two_statestored_with_force_active(self):
+ """The test case for cluster started with Statestored HA enabled and
+ both statestoreds started with 'statestore_force_active' as true.
+ Verify that one and only one statestored is active."""
+ statestoreds = self.cluster.statestoreds()
+ assert(len(statestoreds) == 2)
+ sleep_time_s = build_flavor_timeout(2, slow_build_timeout=5)
+ sleep(sleep_time_s)
+ statestore_service_1 = statestoreds[0].service
+ statestore_service_2 = statestoreds[1].service
+ assert(statestore_service_1.get_metric_value("statestore.active-status")
+ != statestore_service_2.get_metric_value("statestore.active-status"))
+
+ # Verify the active statestored on catalogd and impalad are matching with
+ # the current active statestored.
+ if statestore_service_1.get_metric_value("statestore.active-status"):
+ self.__verify_catalogd_active_statestored(0, 0)
+ self.__verify_impalad_active_statestored(0, 0)
+ self.__verify_impalad_active_statestored(1, 0)
+ self.__verify_impalad_active_statestored(2, 0)
+ else:
+ self.__verify_catalogd_active_statestored(0, 1)
+ self.__verify_impalad_active_statestored(0, 1)
+ self.__verify_impalad_active_statestored(1, 1)
+ self.__verify_impalad_active_statestored(2, 1)
+
+ # Verify simple queries are ran successfully.
+ self.__run_simple_queries()
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ statestored_args="--use_network_address_as_statestore_priority=true "
+ "--statestore_ha_heartbeat_monitoring_frequency_ms=50 "
+ "--statestore_peer_timeout_seconds=2 "
+ "--debug_actions=DISABLE_STATESTORE_NETWORK",
+ impalad_args="--statestore_subscriber_timeout_seconds=2",
+ catalogd_args="--statestore_subscriber_timeout_seconds=2",
+ start_args="--enable_statestored_ha")
+ def test_statestored_auto_failover_with_disabling_network(self):
+ """Tests for Statestore Service auto fail over when active statestored lost
+ connections with other nodes in the cluster.
+ Disable network for active statestored and verify standby statestored becomes
+ active. When original active statestored recover, it does not resume its active
+ role."""
+ # Verify two statestored instances are created with one as active.
+ statestoreds = self.cluster.statestoreds()
+ assert(len(statestoreds) == 2)
+ statestore_service_0 = statestoreds[0].service
+ statestore_service_1 = statestoreds[1].service
+ assert(statestore_service_0.get_metric_value("statestore.active-status"))
+ assert(not statestore_service_1.get_metric_value("statestore.active-status"))
+
+ self.__verify_catalogd_active_statestored(0, 0)
+ self.__verify_impalad_active_statestored(0, 0)
+ self.__verify_impalad_active_statestored(1, 0)
+ self.__verify_impalad_active_statestored(2, 0)
+
+ # Disable the network of active statestored.
+ self.__disable_statestored_network(disable_network=True)
+
+ # Wait for long enough for the standby statestored to detect the failure of active
+ # statestored and assign itself with active role.
+ statestore_service_1.wait_for_metric_value(
+ "statestore.active-status", expected_value=True, timeout=120)
+ assert(statestore_service_1.get_metric_value("statestore.active-status"))
+ # Verify that original active statestored is in HA recovery mode.
+ statestore_service_0.wait_for_metric_value(
+ "statestore.in-ha-recovery-mode", expected_value=True, timeout=60)
+ assert(not statestore_service_0.get_metric_value("statestore.active-status"))
+ sleep(1)
+
+ # Verify the active statestored on catalogd and impalad are matching with
+ # the current active statestored.
+ self.__verify_catalogd_active_statestored(0, 1)
+ self.__verify_impalad_active_statestored(0, 1)
+ self.__verify_impalad_active_statestored(1, 1)
+ self.__verify_impalad_active_statestored(2, 1)
+
+ # Verify simple queries are ran successfully.
+ self.__run_simple_queries()
+ # Verify simple queries with sync_ddl as 1.
+ self.__run_simple_queries(sync_ddl=True)
+
+ # Re-enable network for original active statestored. Verify that the statestored
+ # exits HA recovery mode, but does not resume its active role.
+ self.__disable_statestored_network(disable_network=False)
+ statestore_service_0.wait_for_metric_value(
+ "statestore.in-ha-recovery-mode", expected_value=False, timeout=60)
+ assert(not statestore_service_0.get_metric_value("statestore.active-status"))
+ assert(statestore_service_1.get_metric_value("statestore.active-status"))
+
+ # Verify simple queries are ran successfully.
+ self.__run_simple_queries()
+ # Verify simple queries with sync_ddl as 1.
+ self.__run_simple_queries(sync_ddl=True)
+
+ # Disable the network of current standby statestored.
+ self.__disable_statestored_network(disable_network=True)
+ # Verify that the standby statestored enters HA recovery mode.
+ statestore_service_0.wait_for_metric_value(
+ "statestore.in-ha-recovery-mode", expected_value=True, timeout=120)
+ assert(not statestore_service_0.get_metric_value("statestore.active-status"))
+
+ # Re-enable network for standby statestored. Verify that the statestored exits
+ # HA recovery mode.
+ self.__disable_statestored_network(disable_network=False)
+ statestore_service_0.wait_for_metric_value(
+ "statestore.in-ha-recovery-mode", expected_value=False, timeout=120)
+ assert(not statestore_service_0.get_metric_value("statestore.active-status"))