rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-29 15:06:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: refactor grpc options of metasrv (#6275)

Browse Source 
* feat: refactor grpc options of metasrv

* fix: unit test

* feat: config update
This commit is contained in:
jeremyhi
2025-06-10 01:35:57 +00:00
committed by GitHub
parent 2979aa048e
commit 12648f388a
12 changed files with 99 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+10 -10
View File
@@ -4511,9 +4511,9 @@ dependencies = [
[[package]]
name = "flate2"
version = "1.1.1"
version = "1.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7ced92e76e966ca2fd84c8f7aa01a4aea65b0eb6648d72f7c8f3e2764a67fece"
checksum = "4a3d7db9596fecd151c5f638c0ee5d5bd487b6e0ea232e5dc96d5250f6f94b1d"
dependencies = [
"crc32fast",
"libz-rs-sys",
@@ -5146,9 +5146,9 @@ dependencies = [
[[package]]
name = "grok"
version = "2.0.0"
version = "2.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "273797968160270573071022613fc4aa28b91fe68f3eef6c96a1b2a1947ddfbd"
checksum = "6c52724b609896f661a3f4641dd3a44dc602958ef615857c12d00756b4e9355b"
dependencies = [
"glob",
"onig",
@@ -6716,9 +6716,9 @@ dependencies = [
[[package]]
name = "libz-rs-sys"
version = "0.5.0"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6489ca9bd760fe9642d7644e827b0c9add07df89857b0416ee15c1cc1a3b8c5a"
checksum = "172a788537a2221661b480fee8dc5f96c580eb34fa88764d3205dc356c7e4221"
dependencies = [
"zlib-rs",
]
@@ -9664,9 +9664,9 @@ dependencies = [
[[package]]
name = "psl"
version = "2.1.112"
version = "2.1.119"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1c6b4c497a0c6bfb466f75167c728b1a861b0cdc39de9c35b877208a270a9590"
checksum = "d0e49aa528239f2ca13ad87387977c208e59c3fb8c437609f95f1b3898ec6ef1"
dependencies = [
"psl-types",
]
@@ -14701,9 +14701,9 @@ dependencies = [
[[package]]
name = "zlib-rs"
version = "0.5.0"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "868b928d7949e09af2f6086dfc1e01936064cc7a819253bce650d4e2a2d63ba8"
checksum = "626bd9fa9734751fc50d6060752170984d7053f5a39061f524cda68023d4db8a"
[[package]]
name = "zstd"
config/config.md
+7 -3
View File
@@ -313,11 +313,9 @@
| Key | Type | Default | Descriptions |
| --- | -----| ------- | ----------- |
| `data_home` | String | `./greptimedb_data` | The working home directory. |
| `bind_addr` | String | `127.0.0.1:3002` | The bind address of metasrv. |
| `server_addr` | String | `127.0.0.1:3002` | The communication server address for the frontend and datanode to connect to metasrv.<br/>If left empty or unset, the server will automatically use the IP address of the first network interface<br/>on the host, with the same port number as the one specified in `bind_addr`. |
| `store_addrs` | Array | -- | Store server address default to etcd store.<br/>For postgres store, the format is:<br/>"password=password dbname=postgres user=postgres host=localhost port=5432"<br/>For etcd store, the format is:<br/>"127.0.0.1:2379" |
| `store_key_prefix` | String | `""` | If it's not empty, the metasrv will store all data with this key prefix. |
| `backend` | String | `etcd_store` | The datastore for meta server.<br/>Available values:<br/>- `etcd_store` (default value)<br/>- `memory_store`<br/>- `postgres_store` |
| `backend` | String | `etcd_store` | The datastore for meta server.<br/>Available values:<br/>- `etcd_store` (default value)<br/>- `memory_store`<br/>- `postgres_store`<br/>- `mysql_store` |
| `meta_table_name` | String | `greptime_metakv` | Table name in RDS to store metadata. Effect when using a RDS kvbackend.<br/>**Only used when backend is `postgres_store`.** |
| `meta_election_lock_id` | Integer | `1` | Advisory lock id in PostgreSQL for election. Effect when using PostgreSQL as kvbackend<br/>Only used when backend is `postgres_store`. |
| `selector` | String | `round_robin` | Datanode selector type.<br/>- `round_robin` (default value)<br/>- `lease_based`<br/>- `load_based`<br/>For details, please see "https://docs.greptime.com/developer-guide/metasrv/selector". |
@@ -329,6 +327,12 @@
| `runtime` | -- | -- | The runtime options. |
| `runtime.global_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
| `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute the runtime for global write operations. |
| `grpc` | -- | -- | The gRPC server options. |
| `grpc.bind_addr` | String | `127.0.0.1:3002` | The address to bind the gRPC server. |
| `grpc.server_addr` | String | `127.0.0.1:3002` | The communication server address for the frontend and datanode to connect to metasrv.<br/>If left empty or unset, the server will automatically use the IP address of the first network interface<br/>on the host, with the same port number as the one specified in `bind_addr`. |
| `grpc.runtime_size` | Integer | `8` | The number of server worker threads. |
| `grpc.max_recv_message_size` | String | `512MB` | The maximum receive message size for gRPC server. |
| `grpc.max_send_message_size` | String | `512MB` | The maximum send message size for gRPC server. |
| `http` | -- | -- | The HTTP server options. |
| `http.addr` | String | `127.0.0.1:4000` | The address to bind the HTTP server. |
| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout. |
config/metasrv.example.toml
+16 -8
View File
@@ -1,14 +1,6 @@
## The working home directory.
data_home = "./greptimedb_data"
## The bind address of metasrv.
bind_addr = "127.0.0.1:3002"
## The communication server address for the frontend and datanode to connect to metasrv.
## If left empty or unset, the server will automatically use the IP address of the first network interface
## on the host, with the same port number as the one specified in `bind_addr`.
server_addr = "127.0.0.1:3002"
## Store server address default to etcd store.
## For postgres store, the format is:
## "password=password dbname=postgres user=postgres host=localhost port=5432"
@@ -24,6 +16,7 @@ store_key_prefix = ""
## - `etcd_store` (default value)
## - `memory_store`
## - `postgres_store`
## - `mysql_store`
backend = "etcd_store"
## Table name in RDS to store metadata. Effect when using a RDS kvbackend.
@@ -67,6 +60,21 @@ node_max_idle_time = "24hours"
## The number of threads to execute the runtime for global write operations.
#+ compact_rt_size = 4
## The gRPC server options.
[grpc]
## The address to bind the gRPC server.
bind_addr = "127.0.0.1:3002"
## The communication server address for the frontend and datanode to connect to metasrv.
## If left empty or unset, the server will automatically use the IP address of the first network interface
## on the host, with the same port number as the one specified in `bind_addr`.
server_addr = "127.0.0.1:3002"
## The number of server worker threads.
runtime_size = 8
## The maximum receive message size for gRPC server.
max_recv_message_size = "512MB"
## The maximum send message size for gRPC server.
max_send_message_size = "512MB"
## The HTTP server options.
[http]
## The address to bind the HTTP server.
src/cmd/src/metasrv.rs
+14 -6
View File
@@ -237,12 +237,20 @@ impl StartCommand {
tokio_console_addr: global_options.tokio_console_addr.clone(),
};
#[allow(deprecated)]
if let Some(addr) = &self.rpc_bind_addr {
opts.bind_addr.clone_from(addr);
opts.grpc.bind_addr.clone_from(addr);
} else if !opts.bind_addr.is_empty() {
opts.grpc.bind_addr.clone_from(&opts.bind_addr);
}
#[allow(deprecated)]
if let Some(addr) = &self.rpc_server_addr {
opts.server_addr.clone_from(addr);
opts.grpc.server_addr.clone_from(addr);
} else if !opts.server_addr.is_empty() {
opts.grpc.server_addr.clone_from(&opts.server_addr);
}
if let Some(addrs) = &self.store_addrs {
@@ -319,7 +327,7 @@ impl StartCommand {
let plugin_opts = opts.plugins;
let mut opts = opts.component;
opts.detect_server_addr();
opts.grpc.detect_server_addr();
info!("Metasrv options: {:#?}", opts);
@@ -363,7 +371,7 @@ mod tests {
};
let options = cmd.load_options(&Default::default()).unwrap().component;
assert_eq!("127.0.0.1:3002".to_string(), options.bind_addr);
assert_eq!("127.0.0.1:3002".to_string(), options.grpc.bind_addr);
assert_eq!(vec!["127.0.0.1:2380".to_string()], options.store_addrs);
assert_eq!(SelectorType::LoadBased, options.selector);
}
@@ -396,8 +404,8 @@ mod tests {
};
let options = cmd.load_options(&Default::default()).unwrap().component;
assert_eq!("127.0.0.1:3002".to_string(), options.bind_addr);
assert_eq!("127.0.0.1:3002".to_string(), options.server_addr);
assert_eq!("127.0.0.1:3002".to_string(), options.grpc.bind_addr);
assert_eq!("127.0.0.1:3002".to_string(), options.grpc.server_addr);
assert_eq!(vec!["127.0.0.1:2379".to_string()], options.store_addrs);
assert_eq!(SelectorType::LeaseBased, options.selector);
assert_eq!("debug", options.logging.level.as_ref().unwrap());
@@ -509,10 +517,10 @@ mod tests {
let opts = command.load_options(&Default::default()).unwrap().component;
// Should be read from env, env > default values.
assert_eq!(opts.bind_addr, "127.0.0.1:14002");
assert_eq!(opts.grpc.bind_addr, "127.0.0.1:14002");
// Should be read from config file, config file > env > default values.
assert_eq!(opts.server_addr, "127.0.0.1:3002");
assert_eq!(opts.grpc.server_addr, "127.0.0.1:3002");
// Should be read from cli, cli > config file > env > default values.
assert_eq!(opts.http.addr, "127.0.0.1:14000");
src/cmd/tests/load_config_test.rs
+5 -1
View File
@@ -176,7 +176,11 @@ fn test_load_metasrv_example_config() {
component: MetasrvOptions {
selector: SelectorType::default(),
data_home: DEFAULT_DATA_HOME.to_string(),
server_addr: "127.0.0.1:3002".to_string(),
grpc: GrpcOptions {
bind_addr: "127.0.0.1:3002".to_string(),
server_addr: "127.0.0.1:3002".to_string(),
..Default::default()
},
logging: LoggingOptions {
dir: Path::new(DEFAULT_DATA_HOME)
.join(DEFAULT_LOGGING_DIR)
src/meta-srv/src/bootstrap.rs
+5 -4
View File
@@ -142,7 +142,8 @@ impl MetasrvInstance {
let (serve_state_tx, serve_state_rx) = oneshot::channel();
let socket_addr =
bootstrap_metasrv_with_router(&self.opts.bind_addr, router, serve_state_tx, rx).await?;
bootstrap_metasrv_with_router(&self.opts.grpc.bind_addr, router, serve_state_tx, rx)
.await?;
self.bind_addr = Some(socket_addr);
let addr = self.opts.http.addr.parse().context(error::ParseAddrSnafu {
@@ -258,7 +259,7 @@ pub async fn metasrv_builder(
let etcd_client = create_etcd_client(&opts.store_addrs).await?;
let kv_backend = EtcdStore::with_etcd_client(etcd_client.clone(), opts.max_txn_ops);
let election = EtcdElection::with_etcd_client(
&opts.server_addr,
&opts.grpc.server_addr,
etcd_client,
opts.store_key_prefix.clone(),
)
@@ -288,7 +289,7 @@ pub async fn metasrv_builder(
let election_client =
ElectionPgClient::new(pool, execution_timeout, meta_lease_ttl, statement_timeout)?;
let election = PgElection::with_pg_client(
opts.server_addr.clone(),
opts.grpc.server_addr.clone(),
election_client,
opts.store_key_prefix.clone(),
candidate_lease_ttl,
@@ -316,7 +317,7 @@ pub async fn metasrv_builder(
let election_table_name = opts.meta_table_name.clone() + "_election";
let election_client = create_mysql_client(opts).await?;
let election = MySqlElection::with_mysql_client(
opts.server_addr.clone(),
opts.grpc.server_addr.clone(),
election_client,
opts.store_key_prefix.clone(),
CANDIDATE_LEASE_SECS,
src/meta-srv/src/metasrv.rs
+14 -37
View File
@@ -46,6 +46,7 @@ use common_telemetry::{error, info, warn};
use common_wal::config::MetasrvWalConfig;
use serde::{Deserialize, Serialize};
use servers::export_metrics::ExportMetricsOption;
use servers::grpc::GrpcOptions;
use servers::http::HttpOptions;
use snafu::{OptionExt, ResultExt};
use store_api::storage::RegionId;
@@ -96,8 +97,10 @@ pub enum BackendImpl {
#[serde(default)]
pub struct MetasrvOptions {
/// The address the server listens on.
#[deprecated(note = "Use grpc.bind_addr instead")]
pub bind_addr: String,
/// The address the server advertises to the clients.
#[deprecated(note = "Use grpc.server_addr instead")]
pub server_addr: String,
/// The address of the store, e.g., etcd.
pub store_addrs: Vec<String>,
@@ -112,6 +115,7 @@ pub struct MetasrvOptions {
/// If it's true, the region failover will be allowed even if the local WAL is used.
/// Note that this option is not recommended to be set to true, because it may lead to data loss during failover.
pub allow_region_failover_on_local_wal: bool,
pub grpc: GrpcOptions,
/// The HTTP server options.
pub http: HttpOptions,
/// The logging options.
@@ -166,8 +170,6 @@ impl fmt::Debug for MetasrvOptions {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let mut debug_struct = f.debug_struct("MetasrvOptions");
debug_struct
.field("bind_addr", &self.bind_addr)
.field("server_addr", &self.server_addr)
.field("store_addrs", &self.sanitize_store_addrs())
.field("selector", &self.selector)
.field("use_memory_store", &self.use_memory_store)
@@ -176,6 +178,7 @@ impl fmt::Debug for MetasrvOptions {
"allow_region_failover_on_local_wal",
&self.allow_region_failover_on_local_wal,
)
.field("grpc", &self.grpc)
.field("http", &self.http)
.field("logging", &self.logging)
.field("procedure", &self.procedure)
@@ -208,14 +211,19 @@ const DEFAULT_METASRV_ADDR_PORT: &str = "3002";
impl Default for MetasrvOptions {
fn default() -> Self {
Self {
bind_addr: format!("127.0.0.1:{}", DEFAULT_METASRV_ADDR_PORT),
// If server_addr is not set, the server will use the local ip address as the server address.
#[allow(deprecated)]
bind_addr: String::new(),
#[allow(deprecated)]
server_addr: String::new(),
store_addrs: vec!["127.0.0.1:2379".to_string()],
selector: SelectorType::default(),
use_memory_store: false,
enable_region_failover: false,
allow_region_failover_on_local_wal: false,
grpc: GrpcOptions {
bind_addr: format!("127.0.0.1:{}", DEFAULT_METASRV_ADDR_PORT),
..Default::default()
},
http: HttpOptions::default(),
logging: LoggingOptions::default(),
procedure: ProcedureConfig {
@@ -253,37 +261,6 @@ impl Configurable for MetasrvOptions {
}
impl MetasrvOptions {
/// Detect server address.
#[cfg(not(target_os = "android"))]
pub fn detect_server_addr(&mut self) {
if self.server_addr.is_empty() {
match local_ip_address::local_ip() {
Ok(ip) => {
let detected_addr = format!(
"{}:{}",
ip,
self.bind_addr
.split(':')
.nth(1)
.unwrap_or(DEFAULT_METASRV_ADDR_PORT)
);
info!("Using detected: {} as server address", detected_addr);
self.server_addr = detected_addr;
}
Err(e) => {
error!("Failed to detect local ip address: {}", e);
}
}
}
}
#[cfg(target_os = "android")]
pub fn detect_server_addr(&mut self) {
if self.server_addr.is_empty() {
common_telemetry::debug!("detect local IP is not supported on Android");
}
}
fn sanitize_store_addrs(&self) -> Vec<String> {
self.store_addrs
.iter()
@@ -639,7 +616,7 @@ impl Metasrv {
pub fn node_info(&self) -> MetasrvNodeInfo {
let build_info = common_version::build_info();
MetasrvNodeInfo {
addr: self.options().server_addr.clone(),
addr: self.options().grpc.server_addr.clone(),
version: build_info.version.to_string(),
git_commit: build_info.commit_short.to_string(),
start_time_ms: self.start_time_ms(),
@@ -731,7 +708,7 @@ impl Metasrv {
#[inline]
pub fn new_ctx(&self) -> Context {
let server_addr = self.options().server_addr.clone();
let server_addr = self.options().grpc.server_addr.clone();
let in_memory = self.in_memory.clone();
let kv_backend = self.kv_backend.clone();
let leader_cached_kv_backend = self.leader_cached_kv_backend.clone();
src/meta-srv/src/metasrv/builder.rs
+6 -6
View File
@@ -179,8 +179,8 @@ impl MetasrvBuilder {
let in_memory = in_memory.unwrap_or_else(|| Arc::new(MemoryKvBackend::new()));
let state = Arc::new(RwLock::new(match election {
None => State::leader(options.server_addr.to_string(), true),
Some(_) => State::follower(options.server_addr.to_string()),
None => State::leader(options.grpc.server_addr.to_string(), true),
Some(_) => State::follower(options.grpc.server_addr.to_string()),
}));
let leader_cached_kv_backend = Arc::new(LeaderCachedKvBackend::new(
@@ -203,7 +203,7 @@ impl MetasrvBuilder {
));
let maintenance_mode_manager = Arc::new(MaintenanceModeManager::new(kv_backend.clone()));
let selector_ctx = SelectorContext {
server_addr: options.server_addr.clone(),
server_addr: options.grpc.server_addr.clone(),
datanode_lease_secs: distributed_time_constants::DATANODE_LEASE_SECS,
flownode_lease_secs: distributed_time_constants::FLOWNODE_LEASE_SECS,
kv_backend: kv_backend.clone(),
@@ -272,7 +272,7 @@ impl MetasrvBuilder {
let cache_invalidator = Arc::new(MetasrvCacheInvalidator::new(
mailbox.clone(),
MetasrvInfo {
server_addr: options.server_addr.clone(),
server_addr: options.grpc.server_addr.clone(),
},
));
let peer_lookup_service = Arc::new(MetaPeerLookupService::new(meta_peer_client.clone()));
@@ -315,7 +315,7 @@ impl MetasrvBuilder {
memory_region_keeper.clone(),
region_failure_detector_controller.clone(),
mailbox.clone(),
options.server_addr.clone(),
options.grpc.server_addr.clone(),
cache_invalidator.clone(),
),
));
@@ -390,7 +390,7 @@ impl MetasrvBuilder {
client: Arc::new(kafka_client),
table_metadata_manager: table_metadata_manager.clone(),
leader_region_registry: leader_region_registry.clone(),
server_addr: options.server_addr.clone(),
server_addr: options.grpc.server_addr.clone(),
mailbox: mailbox.clone(),
};
let wal_prune_manager = WalPruneManager::new(
src/meta-srv/src/mocks.rs
+10 -3
View File
@@ -26,6 +26,7 @@ use common_meta::kv_backend::etcd::EtcdStore;
use common_meta::kv_backend::memory::MemoryKvBackend;
use common_meta::kv_backend::{KvBackendRef, ResettableKvBackendRef};
use hyper_util::rt::TokioIo;
use servers::grpc::GrpcOptions;
use tonic::codec::CompressionEncoding;
use tower::service_fn;
@@ -47,7 +48,10 @@ pub async fn mock_with_memstore() -> MockInfo {
let in_memory = Arc::new(MemoryKvBackend::new());
mock(
MetasrvOptions {
server_addr: "127.0.0.1:3002".to_string(),
grpc: GrpcOptions {
server_addr: "127.0.0.1:3002".to_string(),
..Default::default()
},
..Default::default()
},
kv_backend,
@@ -62,7 +66,10 @@ pub async fn mock_with_etcdstore(addr: &str) -> MockInfo {
let kv_backend = EtcdStore::with_endpoints([addr], 128).await.unwrap();
mock(
MetasrvOptions {
server_addr: "127.0.0.1:3002".to_string(),
grpc: GrpcOptions {
server_addr: "127.0.0.1:3002".to_string(),
..Default::default()
},
..Default::default()
},
kv_backend,
@@ -80,7 +87,7 @@ pub async fn mock(
datanode_clients: Option<Arc<NodeClients>>,
in_memory: Option<ResettableKvBackendRef>,
) -> MockInfo {
let server_addr = opts.server_addr.clone();
let server_addr = opts.grpc.server_addr.clone();
let table_metadata_manager = Arc::new(TableMetadataManager::new(kv_backend.clone()));
table_metadata_manager.init().await.unwrap();
src/meta-srv/src/service/cluster.rs
+1 -1
View File
@@ -88,7 +88,7 @@ impl cluster_server::Cluster for Metasrv {
return Ok(Response::new(resp));
}
let leader_addr = &self.options().server_addr;
let leader_addr = &self.options().grpc.server_addr;
let (leader, followers) = match self.election() {
Some(election) => {
let nodes = election.all_candidates().await?;
src/meta-srv/src/service/heartbeat.rs
+6 -2
View File
@@ -190,6 +190,7 @@ mod tests {
use api::v1::meta::*;
use common_meta::kv_backend::memory::MemoryKvBackend;
use common_telemetry::tracing_context::W3cTrace;
use servers::grpc::GrpcOptions;
use tonic::IntoRequest;
use super::get_node_id;
@@ -203,7 +204,10 @@ mod tests {
let metasrv = MetasrvBuilder::new()
.kv_backend(kv_backend)
.options(MetasrvOptions {
server_addr: "127.0.0.1:3002".to_string(),
grpc: GrpcOptions {
server_addr: "127.0.0.1:3002".to_string(),
..Default::default()
},
..Default::default()
})
.build()
@@ -216,7 +220,7 @@ mod tests {
let res = metasrv.ask_leader(req.into_request()).await.unwrap();
let res = res.into_inner();
assert_eq!(metasrv.options().bind_addr, res.leader.unwrap().addr);
assert_eq!(metasrv.options().grpc.server_addr, res.leader.unwrap().addr);
}
#[test]
tests-integration/src/cluster.rs
+5 -1
View File
@@ -56,6 +56,7 @@ use meta_srv::metasrv::{Metasrv, MetasrvOptions, SelectorRef};
use meta_srv::mocks::MockInfo;
use servers::grpc::flight::FlightCraftWrapper;
use servers::grpc::region_server::RegionServerRequestHandler;
use servers::grpc::GrpcOptions;
use servers::heartbeat_options::HeartbeatOptions;
use servers::server::ServerHandlers;
use tempfile::TempDir;
@@ -190,7 +191,10 @@ impl GreptimeDbClusterBuilder {
max_running_procedures: 128,
},
wal: self.metasrv_wal_config.clone(),
server_addr: "127.0.0.1:3002".to_string(),
grpc: GrpcOptions {
server_addr: "127.0.0.1:3002".to_string(),
..Default::default()
},
..Default::default()
};