risedev/task/
meta_node_service.rs

1// Copyright 2022 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::env;
16use std::path::{Path, PathBuf};
17use std::process::Command;
18use std::sync::LazyLock;
19
20use anyhow::{Context, Result, anyhow, bail};
21use itertools::Itertools;
22use sqlx::{ConnectOptions, Database};
23use url::Url;
24
25use super::{ExecuteContext, Task};
26use crate::util::{get_program_args, get_program_env_cmd, get_program_name, is_env_set};
27use crate::{
28    Application, HummockInMemoryStrategy, MetaBackend, MetaNodeConfig, add_hummock_backend,
29    add_tempo_endpoint,
30};
31
32/// URL for connecting to the SQL meta store, retrieved from the env var `RISEDEV_SQL_ENDPOINT`.
33/// If it is not set, a temporary sqlite file is created and used.
34///
35/// # Examples
36///
37/// - `mysql://root:my-secret-pw@127.0.0.1:3306/metastore`
38/// - `postgresql://localhost:5432/metastore`
39/// - `sqlite:///path/to/file.db`
40/// - `sqlite::memory:`
41fn sql_endpoint_from_env() -> String {
42    static SQL_ENDPOINT: LazyLock<String> = LazyLock::new(|| {
43        if let Ok(endpoint) = env::var("RISEDEV_SQL_ENDPOINT") {
44            tracing::info!(
45                "sql endpoint from env RISEDEV_SQL_ENDPOINT resolved to `{}`",
46                endpoint
47            );
48            endpoint
49        } else {
50            // `meta-backend: env` is specified, but env var is not set.
51            // Act as if `meta-backend: sqlite` is specified.
52            // Not using a temporary file because we want to persist the data across restarts.
53            let prefix_data = env::var("PREFIX_DATA").unwrap();
54            let dir = PathBuf::from(&prefix_data).join("meta-backend-env-fallback-sqlite");
55            fs_err::create_dir_all(&dir).unwrap();
56
57            let path = dir.join("metadata.db");
58            let sqlite_endpoint = format!("sqlite://{}?mode=rwc", path.to_string_lossy());
59            tracing::warn!(
60                "env RISEDEV_SQL_ENDPOINT not set, use fallback sqlite `{}`",
61                sqlite_endpoint
62            );
63            sqlite_endpoint
64        }
65    });
66
67    SQL_ENDPOINT.to_owned()
68}
69
70pub struct MetaNodeService {
71    config: MetaNodeConfig,
72}
73
74impl MetaNodeService {
75    pub fn new(config: MetaNodeConfig) -> Result<Self> {
76        Ok(Self { config })
77    }
78
79    /// Apply command args according to config
80    pub fn apply_command_args(
81        cmd: &mut Command,
82        config: &MetaNodeConfig,
83        hummock_in_memory_strategy: HummockInMemoryStrategy,
84    ) -> Result<()> {
85        cmd.arg("--listen-addr")
86            .arg(format!("{}:{}", config.listen_address, config.port))
87            .arg("--advertise-addr")
88            .arg(format!("{}:{}", config.address, config.port))
89            .arg("--dashboard-host")
90            .arg(format!(
91                "{}:{}",
92                config.listen_address, config.dashboard_port
93            ));
94
95        cmd.arg("--prometheus-host").arg(format!(
96            "{}:{}",
97            config.listen_address, config.exporter_port
98        ));
99
100        match config.provide_prometheus.as_ref().unwrap().as_slice() {
101            [] => {}
102            [prometheus] => {
103                cmd.arg("--prometheus-endpoint")
104                    .arg(format!("http://{}:{}", prometheus.address, prometheus.port));
105            }
106            _ => {
107                return Err(anyhow!(
108                    "unexpected prometheus config {:?}, only 1 instance is supported",
109                    config.provide_prometheus
110                ));
111            }
112        }
113
114        let mut is_persistent_meta_store = false;
115
116        match &config.meta_backend {
117            MetaBackend::Memory => {
118                cmd.arg("--backend").arg("mem");
119            }
120            MetaBackend::Sqlite => {
121                let sqlite_config = config.provide_sqlite_backend.as_ref().unwrap();
122                assert_eq!(
123                    sqlite_config.len(),
124                    1,
125                    "should have exactly 1 sqlite config"
126                );
127                is_persistent_meta_store = true;
128
129                let prefix_data = env::var("PREFIX_DATA")?;
130                let file_path = PathBuf::from(&prefix_data)
131                    .join(&sqlite_config[0].id)
132                    .join(&sqlite_config[0].file);
133                cmd.arg("--backend")
134                    .arg("sqlite")
135                    .arg("--sql-endpoint")
136                    .arg(file_path);
137            }
138            MetaBackend::Postgres => {
139                let pg_config = config.provide_postgres_backend.as_ref().unwrap();
140                let pg_store_config = Itertools::exactly_one(
141                    pg_config
142                        .iter()
143                        .filter(|c| c.application == Application::Metastore),
144                )
145                .expect("more than one or no pg store config found for metastore");
146                is_persistent_meta_store = true;
147
148                cmd.arg("--backend")
149                    .arg("postgres")
150                    .arg("--sql-endpoint")
151                    .arg(format!(
152                        "{}:{}",
153                        pg_store_config.address, pg_store_config.port,
154                    ))
155                    .arg("--sql-username")
156                    .arg(&pg_store_config.user)
157                    .arg("--sql-password")
158                    .arg(&pg_store_config.password)
159                    .arg("--sql-database")
160                    .arg(&pg_store_config.database);
161            }
162            MetaBackend::Mysql => {
163                let mysql_config = config.provide_mysql_backend.as_ref().unwrap();
164                let mysql_store_config = Itertools::exactly_one(
165                    mysql_config
166                        .iter()
167                        .filter(|c| c.application == Application::Metastore),
168                )
169                .expect("more than one or no mysql store config found for metastore");
170                is_persistent_meta_store = true;
171
172                cmd.arg("--backend")
173                    .arg("mysql")
174                    .arg("--sql-endpoint")
175                    .arg(format!(
176                        "{}:{}",
177                        mysql_store_config.address, mysql_store_config.port,
178                    ))
179                    .arg("--sql-username")
180                    .arg(&mysql_store_config.user)
181                    .arg("--sql-password")
182                    .arg(&mysql_store_config.password)
183                    .arg("--sql-database")
184                    .arg(&mysql_store_config.database);
185            }
186            MetaBackend::Env => {
187                let endpoint = sql_endpoint_from_env();
188                is_persistent_meta_store = true;
189
190                cmd.arg("--backend")
191                    .arg("sql")
192                    .arg("--sql-endpoint")
193                    .arg(endpoint);
194            }
195        }
196
197        let provide_minio = config.provide_minio.as_ref().unwrap();
198        let provide_opendal = config.provide_opendal.as_ref().unwrap();
199        let provide_aws_s3 = config.provide_aws_s3.as_ref().unwrap();
200        let provide_moat = config.provide_moat.as_ref().unwrap();
201
202        let provide_compute_node = config.provide_compute_node.as_ref().unwrap();
203        let provide_compactor = config.provide_compactor.as_ref().unwrap();
204
205        let (is_shared_backend, is_persistent_backend) = add_hummock_backend(
206            &config.id,
207            provide_opendal,
208            provide_minio,
209            provide_aws_s3,
210            provide_moat,
211            hummock_in_memory_strategy,
212            cmd,
213        )?;
214
215        if (provide_compute_node.len() > 1 || !provide_compactor.is_empty()) && !is_shared_backend {
216            return Err(anyhow!(
217                "Hummock storage may behave incorrectly with in-memory backend for multiple compute-node or compactor-enabled configuration. Should use a shared backend (e.g. MinIO) instead. Consider adding `use: minio` in risedev config."
218            ));
219        }
220
221        let provide_compactor = config.provide_compactor.as_ref().unwrap();
222        if is_shared_backend && provide_compactor.is_empty() {
223            return Err(anyhow!(
224                "When using a shared backend (minio, aws-s3, or shared in-memory with `risedev playground`), at least one compactor is required. Consider adding `use: compactor` in risedev config."
225            ));
226        }
227        if is_persistent_meta_store && !is_persistent_backend {
228            return Err(anyhow!(
229                "When using a persistent meta store (sql), a persistent state store is required (e.g. minio, aws-s3, etc.)."
230            ));
231        }
232
233        cmd.arg("--data-directory").arg("hummock_001");
234
235        let provide_tempo = config.provide_tempo.as_ref().unwrap();
236        add_tempo_endpoint(provide_tempo, cmd)?;
237
238        Ok(())
239    }
240}
241
242impl Task for MetaNodeService {
243    fn execute(&mut self, ctx: &mut ExecuteContext<impl std::io::Write>) -> anyhow::Result<()> {
244        ctx.service(self);
245        ctx.pb.set_message("starting...");
246
247        let mut cmd = ctx.risingwave_cmd("meta-node")?;
248
249        if crate::util::is_env_set("RISEDEV_ENABLE_PROFILE") {
250            cmd.env(
251                "RW_PROFILE_PATH",
252                Path::new(&env::var("PREFIX_LOG")?).join(format!("profile-{}", self.id())),
253            );
254        }
255
256        if crate::util::is_env_set("RISEDEV_ENABLE_HEAP_PROFILE") {
257            // See https://linux.die.net/man/3/jemalloc for the descriptions of profiling options
258            let conf = "prof:true,lg_prof_interval:32,lg_prof_sample:19,prof_prefix:meta-node";
259            cmd.env("_RJEM_MALLOC_CONF", conf); // prefixed for macos
260            cmd.env("MALLOC_CONF", conf); // unprefixed for linux
261        }
262
263        Self::apply_command_args(&mut cmd, &self.config, HummockInMemoryStrategy::Allowed)?;
264
265        if let MetaBackend::Env = self.config.meta_backend
266            && is_env_set("RISEDEV_CLEAN_START")
267        {
268            ctx.pb.set_message("initializing meta store from env...");
269            initialize_meta_store()?;
270        }
271
272        if !self.config.user_managed {
273            ctx.run_command(ctx.tmux_run(cmd)?)?;
274            ctx.pb.set_message("started");
275        } else {
276            ctx.pb.set_message("user managed");
277            writeln!(
278                &mut ctx.log,
279                "Please use the following parameters to start the meta:\n{}\n{} {}\n\n",
280                get_program_env_cmd(&cmd),
281                get_program_name(&cmd),
282                get_program_args(&cmd)
283            )?;
284        }
285
286        Ok(())
287    }
288
289    fn id(&self) -> String {
290        self.config.id.clone()
291    }
292}
293
294fn initialize_meta_store() -> Result<(), anyhow::Error> {
295    let rt = tokio::runtime::Builder::new_current_thread()
296        .enable_all()
297        .build()?;
298
299    let endpoint: Url = sql_endpoint_from_env()
300        .parse()
301        .context("invalid url for SQL endpoint")?;
302    let scheme = endpoint.scheme();
303
304    // Retrieve the database name to use for the meta store.
305    // Modify the URL to establish a temporary connection to initialize that database.
306    let (db, init_url) = if sqlx::Postgres::URL_SCHEMES.contains(&scheme) {
307        let options = sqlx::postgres::PgConnectOptions::from_url(&endpoint)
308            .context("invalid database url for Postgres meta backend")?;
309
310        let db = options
311            .get_database()
312            .unwrap_or_else(|| options.get_username()) // PG defaults to username if no database is specified
313            .to_owned();
314        // https://www.postgresql.org/docs/current/manage-ag-templatedbs.html
315        let init_options = options.database("template1");
316        let init_url = init_options.to_url_lossy();
317
318        (db, init_url)
319    } else if sqlx::MySql::URL_SCHEMES.contains(&scheme) {
320        let options = sqlx::mysql::MySqlConnectOptions::from_url(&endpoint)
321            .context("invalid database url for MySQL meta backend")?;
322
323        let db = options
324            .get_database()
325            .context("database not specified for MySQL meta backend")?
326            .to_owned();
327        // Effectively unset the database field when converting back to URL, meaning connect to no database.
328        let init_options = options.database("");
329        let init_url = init_options.to_url_lossy();
330
331        (db, init_url)
332    } else if sqlx::Sqlite::URL_SCHEMES.contains(&scheme) {
333        // For SQLite, simply empty the file.
334        let options = sqlx::sqlite::SqliteConnectOptions::from_url(&endpoint)
335            .context("invalid database url for SQLite meta backend")?;
336
337        if endpoint.as_str().contains(":memory:") || endpoint.as_str().contains("mode=memory") {
338            // SQLite in-memory database does not need initialization.
339        } else {
340            let filename = options.get_filename();
341            if std::fs::exists(filename)? {
342                fs_err::write(filename, b"").context("failed to empty SQLite file")?;
343            }
344        }
345
346        return Ok(());
347    } else {
348        bail!("unsupported SQL scheme for meta backend: {}", scheme);
349    };
350
351    rt.block_on(async move {
352        use sqlx::any::*;
353        install_default_drivers();
354
355        let options = sqlx::any::AnyConnectOptions::from_url(&init_url)?
356            .log_statements(log::LevelFilter::Debug);
357
358        let mut conn = options
359            .connect()
360            .await
361            .context("failed to connect to a template database for meta store")?;
362
363        // Intentionally not executing in a transaction because Postgres does not allow it.
364        sqlx::raw_sql(&format!("DROP DATABASE IF EXISTS {};", db))
365            .execute(&mut conn)
366            .await?;
367        sqlx::raw_sql(&format!("CREATE DATABASE {};", db))
368            .execute(&mut conn)
369            .await?;
370
371        Ok::<_, anyhow::Error>(())
372    })
373    .context("failed to initialize database for meta store")?;
374
375    Ok(())
376}
risedev/task/meta_node_service.rs

risedev/task/
meta_node_service.rs
