From 96c76cae54de990d310d243018dfd4b054118e3e Mon Sep 17 00:00:00 2001 From: August Date: Wed, 3 Apr 2024 16:44:28 +0800 Subject: [PATCH] feat(risedev): support starting cluster with sqlite meta backend (#15693) (#16128) --- risedev.yml | 12 ++++- src/risedevtool/src/bin/risedev-compose.rs | 1 + src/risedevtool/src/bin/risedev-dev.rs | 33 ++++++++++++- src/risedevtool/src/config.rs | 1 + src/risedevtool/src/service_config.rs | 14 ++++++ src/risedevtool/src/task/meta_node_service.rs | 47 ++++++++++++------- 6 files changed, 89 insertions(+), 19 deletions(-) diff --git a/risedev.yml b/risedev.yml index 82be3912842a..5282e2913586 100644 --- a/risedev.yml +++ b/risedev.yml @@ -70,7 +70,7 @@ profile: - use: frontend - use: compactor - # `dev-compute-node` have the same settings as default except the the compute node will be started by user. + # `dev-compute-node` have the same settings as default except the compute node will be started by user. dev-compute-node: steps: - use: meta-node @@ -1043,6 +1043,13 @@ template: # Other etcd nodes provide-etcd: "etcd*" + sqlite: + # Id of this instance + id: sqlite + + # File name of the sqlite database + file: metadata.db + compute-node: # Compute-node advertise address address: "127.0.0.1" @@ -1117,6 +1124,9 @@ template: # Etcd backend config provide-etcd-backend: "etcd*" + # Sqlite backend config + provide-sqlite-backend: "sqlite*" + # Prometheus nodes used by dashboard service provide-prometheus: "prometheus*" diff --git a/src/risedevtool/src/bin/risedev-compose.rs b/src/risedevtool/src/bin/risedev-compose.rs index fe0e6833a352..0f3cb5b020fc 100644 --- a/src/risedevtool/src/bin/risedev-compose.rs +++ b/src/risedevtool/src/bin/risedev-compose.rs @@ -123,6 +123,7 @@ fn main() -> Result<()> { volumes.insert(c.id.clone(), ComposeVolume::default()); (c.address.clone(), c.compose(&compose_config)?) } + ServiceConfig::Sqlite(_) => continue, ServiceConfig::Prometheus(c) => { volumes.insert(c.id.clone(), ComposeVolume::default()); (c.address.clone(), c.compose(&compose_config)?) diff --git a/src/risedevtool/src/bin/risedev-dev.rs b/src/risedevtool/src/bin/risedev-dev.rs index 9723ee89fbd5..a9bc78135564 100644 --- a/src/risedevtool/src/bin/risedev-dev.rs +++ b/src/risedevtool/src/bin/risedev-dev.rs @@ -14,7 +14,7 @@ use std::env; use std::fmt::Write; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -27,7 +27,7 @@ use risedev::{ generate_risedev_env, preflight_check, AwsS3Config, CompactorService, ComputeNodeService, ConfigExpander, ConfigureTmuxTask, EnsureStopService, ExecuteContext, FrontendService, GrafanaService, KafkaService, MetaNodeService, MinioService, OpendalConfig, PrometheusService, - PubsubService, RedisService, ServiceConfig, Task, TempoService, ZooKeeperService, + PubsubService, RedisService, ServiceConfig, SqliteConfig, Task, TempoService, ZooKeeperService, RISEDEV_SESSION_NAME, }; use tempfile::tempdir; @@ -101,6 +101,7 @@ fn task_main( let listen_info = match service { ServiceConfig::Minio(c) => Some((c.port, c.id.clone())), ServiceConfig::Etcd(c) => Some((c.port, c.id.clone())), + ServiceConfig::Sqlite(_) => None, ServiceConfig::Prometheus(c) => Some((c.port, c.id.clone())), ServiceConfig::ComputeNode(c) => Some((c.port, c.id.clone())), ServiceConfig::MetaNode(c) => Some((c.port, c.id.clone())), @@ -158,6 +159,34 @@ fn task_main( risedev::ConfigureGrpcNodeTask::new(c.address.clone(), c.port, false)?; task.execute(&mut ctx)?; } + ServiceConfig::Sqlite(c) => { + let mut ctx = + ExecuteContext::new(&mut logger, manager.new_progress(), status_dir.clone()); + + struct SqliteService(SqliteConfig); + impl Task for SqliteService { + fn execute( + &mut self, + _ctx: &mut ExecuteContext, + ) -> anyhow::Result<()> { + Ok(()) + } + + fn id(&self) -> String { + self.0.id.clone() + } + } + + let prefix_data = env::var("PREFIX_DATA")?; + let file_dir = PathBuf::from(&prefix_data).join(&c.id); + std::fs::create_dir_all(&file_dir)?; + let file_path = file_dir.join(&c.file); + + ctx.service(&SqliteService(c.clone())); + ctx.complete_spin(); + ctx.pb + .set_message(format!("using local sqlite: {:?}", file_path)); + } ServiceConfig::Prometheus(c) => { let mut ctx = ExecuteContext::new(&mut logger, manager.new_progress(), status_dir.clone()); diff --git a/src/risedevtool/src/config.rs b/src/risedevtool/src/config.rs index 219b1d2d7948..d5fe2d8c220a 100644 --- a/src/risedevtool/src/config.rs +++ b/src/risedevtool/src/config.rs @@ -159,6 +159,7 @@ impl ConfigExpander { let result = match use_type.as_str() { "minio" => ServiceConfig::Minio(serde_yaml::from_str(&out_str)?), "etcd" => ServiceConfig::Etcd(serde_yaml::from_str(&out_str)?), + "sqlite" => ServiceConfig::Sqlite(serde_yaml::from_str(&out_str)?), "frontend" => ServiceConfig::Frontend(serde_yaml::from_str(&out_str)?), "compactor" => ServiceConfig::Compactor(serde_yaml::from_str(&out_str)?), "compute-node" => ServiceConfig::ComputeNode(serde_yaml::from_str(&out_str)?), diff --git a/src/risedevtool/src/service_config.rs b/src/risedevtool/src/service_config.rs index f6312b0fba17..97996bbeb510 100644 --- a/src/risedevtool/src/service_config.rs +++ b/src/risedevtool/src/service_config.rs @@ -61,6 +61,7 @@ pub struct MetaNodeConfig { pub user_managed: bool, pub provide_etcd_backend: Option>, + pub provide_sqlite_backend: Option>, pub provide_prometheus: Option>, pub provide_compute_node: Option>, @@ -168,6 +169,17 @@ pub struct EtcdConfig { pub provide_etcd: Option>, } +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +#[serde(deny_unknown_fields)] +pub struct SqliteConfig { + #[serde(rename = "use")] + phantom_use: Option, + pub id: String, + + pub file: String, +} + #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] #[serde(rename_all = "kebab-case")] #[serde(deny_unknown_fields)] @@ -331,6 +343,7 @@ pub enum ServiceConfig { Compactor(CompactorConfig), Minio(MinioConfig), Etcd(EtcdConfig), + Sqlite(SqliteConfig), Prometheus(PrometheusConfig), Grafana(GrafanaConfig), Tempo(TempoConfig), @@ -352,6 +365,7 @@ impl ServiceConfig { Self::Compactor(c) => &c.id, Self::Minio(c) => &c.id, Self::Etcd(c) => &c.id, + Self::Sqlite(c) => &c.id, Self::Prometheus(c) => &c.id, Self::Grafana(c) => &c.id, Self::Tempo(c) => &c.id, diff --git a/src/risedevtool/src/task/meta_node_service.rs b/src/risedevtool/src/task/meta_node_service.rs index 91288538303a..4a121bb630d3 100644 --- a/src/risedevtool/src/task/meta_node_service.rs +++ b/src/risedevtool/src/task/meta_node_service.rs @@ -13,7 +13,7 @@ // limitations under the License. use std::env; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::process::Command; use anyhow::{anyhow, Result}; @@ -76,21 +76,36 @@ impl MetaNodeService { } let mut is_persistent_meta_store = false; - match config.provide_etcd_backend.as_ref().unwrap().as_slice() { - [] => { - cmd.arg("--backend").arg("mem"); - } - etcds => { - is_persistent_meta_store = true; - cmd.arg("--backend") - .arg("etcd") - .arg("--etcd-endpoints") - .arg( - etcds - .iter() - .map(|etcd| format!("{}:{}", etcd.address, etcd.port)) - .join(","), - ); + + if let Some(sqlite_config) = &config.provide_sqlite_backend + && !sqlite_config.is_empty() + { + is_persistent_meta_store = true; + let prefix_data = env::var("PREFIX_DATA")?; + let file_path = PathBuf::from(&prefix_data) + .join(&sqlite_config[0].id) + .join(&sqlite_config[0].file); + cmd.arg("--backend") + .arg("sql") + .arg("--sql-endpoint") + .arg(format!("sqlite://{}?mode=rwc", file_path.display())); + } else { + match config.provide_etcd_backend.as_ref().unwrap().as_slice() { + [] => { + cmd.arg("--backend").arg("mem"); + } + etcds => { + is_persistent_meta_store = true; + cmd.arg("--backend") + .arg("etcd") + .arg("--etcd-endpoints") + .arg( + etcds + .iter() + .map(|etcd| format!("{}:{}", etcd.address, etcd.port)) + .join(","), + ); + } } }