risingwave_meta/manager/
env.rs

1// Copyright 2025 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::ops::Deref;
16use std::path::PathBuf;
17use std::sync::Arc;
18
19use anyhow::Context;
20use risingwave_common::config::{
21    CompactionConfig, DefaultParallelism, ObjectStoreConfig, RpcClientConfig,
22};
23use risingwave_common::session_config::SessionConfig;
24use risingwave_common::system_param::reader::SystemParamsReader;
25use risingwave_common::{bail, system_param};
26use risingwave_meta_model::prelude::Cluster;
27use risingwave_pb::meta::SystemParams;
28use risingwave_rpc_client::{
29    FrontendClientPool, FrontendClientPoolRef, StreamClientPool, StreamClientPoolRef,
30};
31use risingwave_sqlparser::ast::RedactSqlOptionKeywordsRef;
32use sea_orm::EntityTrait;
33
34use crate::MetaResult;
35use crate::controller::SqlMetaStore;
36use crate::controller::id::{
37    IdGeneratorManager as SqlIdGeneratorManager, IdGeneratorManagerRef as SqlIdGeneratorManagerRef,
38};
39use crate::controller::session_params::{SessionParamsController, SessionParamsControllerRef};
40use crate::controller::system_param::{SystemParamsController, SystemParamsControllerRef};
41use crate::hummock::sequence::SequenceGenerator;
42use crate::manager::event_log::{EventLogManagerRef, start_event_log_manager};
43use crate::manager::{IdleManager, IdleManagerRef, NotificationManager, NotificationManagerRef};
44use crate::model::ClusterId;
45
46/// [`MetaSrvEnv`] is the global environment in Meta service. The instance will be shared by all
47/// kind of managers inside Meta.
48#[derive(Clone)]
49pub struct MetaSrvEnv {
50    /// id generator manager.
51    id_gen_manager_impl: SqlIdGeneratorManagerRef,
52
53    /// system param manager.
54    system_param_manager_impl: SystemParamsControllerRef,
55
56    /// session param manager.
57    session_param_manager_impl: SessionParamsControllerRef,
58
59    /// meta store.
60    meta_store_impl: SqlMetaStore,
61
62    /// notification manager.
63    notification_manager: NotificationManagerRef,
64
65    /// stream client pool memorization.
66    stream_client_pool: StreamClientPoolRef,
67
68    /// rpc client pool for frontend nodes.
69    frontend_client_pool: FrontendClientPoolRef,
70
71    /// idle status manager.
72    idle_manager: IdleManagerRef,
73
74    event_log_manager: EventLogManagerRef,
75
76    /// Unique identifier of the cluster.
77    cluster_id: ClusterId,
78
79    pub hummock_seq: Arc<SequenceGenerator>,
80
81    /// The await-tree registry of the current meta node.
82    await_tree_reg: await_tree::Registry,
83
84    /// options read by all services
85    pub opts: Arc<MetaOpts>,
86}
87
88/// Options shared by all meta service instances
89#[derive(Clone, serde::Serialize)]
90pub struct MetaOpts {
91    /// Whether to enable the recovery of the cluster. If disabled, the meta service will exit on
92    /// abnormal cases.
93    pub enable_recovery: bool,
94    /// Whether to disable the auto-scaling feature.
95    pub disable_automatic_parallelism_control: bool,
96    /// The number of streaming jobs per scaling operation.
97    pub parallelism_control_batch_size: usize,
98    /// The period of parallelism control trigger.
99    pub parallelism_control_trigger_period_sec: u64,
100    /// The first delay of parallelism control.
101    pub parallelism_control_trigger_first_delay_sec: u64,
102    /// The maximum number of barriers in-flight in the compute nodes.
103    pub in_flight_barrier_nums: usize,
104    /// After specified seconds of idle (no mview or flush), the process will be exited.
105    /// 0 for infinite, process will never be exited due to long idle time.
106    pub max_idle_ms: u64,
107    /// Whether run in compaction detection test mode
108    pub compaction_deterministic_test: bool,
109    /// Default parallelism of units for all streaming jobs.
110    pub default_parallelism: DefaultParallelism,
111
112    /// Interval of invoking a vacuum job, to remove stale metadata from meta store and objects
113    /// from object store.
114    pub vacuum_interval_sec: u64,
115    /// The spin interval inside a vacuum job. It avoids the vacuum job monopolizing resources of
116    /// meta node.
117    pub vacuum_spin_interval_ms: u64,
118    pub time_travel_vacuum_interval_sec: u64,
119    /// Interval of hummock version checkpoint.
120    pub hummock_version_checkpoint_interval_sec: u64,
121    pub enable_hummock_data_archive: bool,
122    pub hummock_time_travel_snapshot_interval: u64,
123    pub hummock_time_travel_sst_info_fetch_batch_size: usize,
124    pub hummock_time_travel_sst_info_insert_batch_size: usize,
125    pub hummock_time_travel_epoch_version_insert_batch_size: usize,
126    pub hummock_gc_history_insert_batch_size: usize,
127    pub hummock_time_travel_filter_out_objects_batch_size: usize,
128    pub hummock_time_travel_filter_out_objects_v1: bool,
129    pub hummock_time_travel_filter_out_objects_list_version_batch_size: usize,
130    pub hummock_time_travel_filter_out_objects_list_delta_batch_size: usize,
131    /// The minimum delta log number a new checkpoint should compact, otherwise the checkpoint
132    /// attempt is rejected. Greater value reduces object store IO, meanwhile it results in
133    /// more loss of in memory `HummockVersionCheckpoint::stale_objects` state when meta node is
134    /// restarted.
135    pub min_delta_log_num_for_hummock_version_checkpoint: u64,
136    /// Objects within `min_sst_retention_time_sec` won't be deleted by hummock full GC, even they
137    /// are dangling.
138    pub min_sst_retention_time_sec: u64,
139    /// Interval of automatic hummock full GC.
140    pub full_gc_interval_sec: u64,
141    /// Max number of object per full GC job can fetch.
142    pub full_gc_object_limit: u64,
143    /// Duration in seconds to retain garbage collection history data.
144    pub gc_history_retention_time_sec: u64,
145    /// Max number of inflight time travel query.
146    pub max_inflight_time_travel_query: u64,
147    /// Enable sanity check when SSTs are committed
148    pub enable_committed_sst_sanity_check: bool,
149    /// Schedule compaction for all compaction groups with this interval.
150    pub periodic_compaction_interval_sec: u64,
151    /// Interval of reporting the number of nodes in the cluster.
152    pub node_num_monitor_interval_sec: u64,
153    /// Whether to protect the drop table operation with incoming sink.
154    pub protect_drop_table_with_incoming_sink: bool,
155    /// The Prometheus endpoint for Meta Dashboard Service.
156    /// The Dashboard service uses this in the following ways:
157    /// 1. Query Prometheus for relevant metrics to find Stream Graph Bottleneck, and display it.
158    /// 2. Provide cluster diagnostics, at `/api/monitor/diagnose` to troubleshoot cluster.
159    ///    These are just examples which show how the Meta Dashboard Service queries Prometheus.
160    pub prometheus_endpoint: Option<String>,
161
162    /// The additional selector used when querying Prometheus.
163    pub prometheus_selector: Option<String>,
164
165    /// The VPC id of the cluster.
166    pub vpc_id: Option<String>,
167
168    /// A usable security group id to assign to a vpc endpoint
169    pub security_group_id: Option<String>,
170
171    /// Default tag for the endpoint created when creating a privatelink connection.
172    /// Will be appended to the tags specified in the `tags` field in with clause in `create
173    /// connection`.
174    pub privatelink_endpoint_default_tags: Option<Vec<(String, String)>>,
175
176    /// Schedule `space_reclaim_compaction` for all compaction groups with this interval.
177    pub periodic_space_reclaim_compaction_interval_sec: u64,
178
179    /// telemetry enabled in config file or not
180    pub telemetry_enabled: bool,
181    /// Schedule `ttl_reclaim_compaction` for all compaction groups with this interval.
182    pub periodic_ttl_reclaim_compaction_interval_sec: u64,
183
184    /// Schedule `tombstone_reclaim_compaction` for all compaction groups with this interval.
185    pub periodic_tombstone_reclaim_compaction_interval_sec: u64,
186
187    /// Schedule `periodic_scheduling_compaction_group_split_interval_sec` for all compaction groups with this interval.
188    pub periodic_scheduling_compaction_group_split_interval_sec: u64,
189
190    /// Whether config object storage bucket lifecycle to purge stale data.
191    pub do_not_config_object_storage_lifecycle: bool,
192
193    pub partition_vnode_count: u32,
194
195    /// threshold of high write throughput of state-table, unit: B/sec
196    pub table_high_write_throughput_threshold: u64,
197    /// threshold of low write throughput of state-table, unit: B/sec
198    pub table_low_write_throughput_threshold: u64,
199
200    pub compaction_task_max_heartbeat_interval_secs: u64,
201    pub compaction_task_max_progress_interval_secs: u64,
202    pub compaction_config: Option<CompactionConfig>,
203
204    /// hybird compaction group config
205    ///
206    /// `hybrid_partition_vnode_count` determines the granularity of vnodes in the hybrid compaction group for SST alignment.
207    /// When `hybrid_partition_vnode_count` > 0, in hybrid compaction group
208    /// - Tables with high write throughput will be split at vnode granularity
209    /// - Tables with high size tables will be split by table granularity
210    ///   When `hybrid_partition_vnode_count` = 0,no longer be special alignment operations for the hybird compaction group
211    pub hybrid_partition_node_count: u32,
212
213    pub event_log_enabled: bool,
214    pub event_log_channel_max_size: u32,
215    pub advertise_addr: String,
216    /// The number of traces to be cached in-memory by the tracing collector
217    /// embedded in the meta node.
218    pub cached_traces_num: u32,
219    /// The maximum memory usage in bytes for the tracing collector embedded
220    /// in the meta node.
221    pub cached_traces_memory_limit_bytes: usize,
222
223    /// l0 picker whether to select trivial move task
224    pub enable_trivial_move: bool,
225
226    /// l0 multi level picker whether to check the overlap accuracy between sub levels
227    pub enable_check_task_level_overlap: bool,
228    pub enable_dropped_column_reclaim: bool,
229
230    /// Whether to split the compaction group when the size of the group exceeds the threshold.
231    pub split_group_size_ratio: f64,
232
233    /// To split the compaction group when the high throughput statistics of the group exceeds the threshold.
234    pub table_stat_high_write_throughput_ratio_for_split: f64,
235
236    /// To merge the compaction group when the low throughput statistics of the group exceeds the threshold.
237    pub table_stat_low_write_throughput_ratio_for_merge: f64,
238
239    /// The window seconds of table throughput statistic history for split compaction group.
240    pub table_stat_throuput_window_seconds_for_split: usize,
241
242    /// The window seconds of table throughput statistic history for merge compaction group.
243    pub table_stat_throuput_window_seconds_for_merge: usize,
244
245    /// The configuration of the object store
246    pub object_store_config: ObjectStoreConfig,
247
248    /// The maximum number of trivial move tasks to be picked in a single loop
249    pub max_trivial_move_task_count_per_loop: usize,
250
251    /// The maximum number of times to probe for `PullTaskEvent`
252    pub max_get_task_probe_times: usize,
253
254    pub compact_task_table_size_partition_threshold_low: u64,
255    pub compact_task_table_size_partition_threshold_high: u64,
256
257    pub periodic_scheduling_compaction_group_merge_interval_sec: u64,
258
259    pub compaction_group_merge_dimension_threshold: f64,
260
261    // The private key for the secret store, used when the secret is stored in the meta.
262    pub secret_store_private_key: Option<Vec<u8>>,
263    /// The path of the temp secret file directory.
264    pub temp_secret_file_dir: String,
265
266    // Cluster limits
267    pub actor_cnt_per_worker_parallelism_hard_limit: usize,
268    pub actor_cnt_per_worker_parallelism_soft_limit: usize,
269
270    pub license_key_path: Option<PathBuf>,
271
272    pub compute_client_config: RpcClientConfig,
273    pub stream_client_config: RpcClientConfig,
274    pub frontend_client_config: RpcClientConfig,
275    pub redact_sql_option_keywords: RedactSqlOptionKeywordsRef,
276}
277
278impl MetaOpts {
279    /// Default opts for testing. Some tests need `enable_recovery=true`
280    pub fn test(enable_recovery: bool) -> Self {
281        Self {
282            enable_recovery,
283            disable_automatic_parallelism_control: false,
284            parallelism_control_batch_size: 1,
285            parallelism_control_trigger_period_sec: 10,
286            parallelism_control_trigger_first_delay_sec: 30,
287            in_flight_barrier_nums: 40,
288            max_idle_ms: 0,
289            compaction_deterministic_test: false,
290            default_parallelism: DefaultParallelism::Full,
291            vacuum_interval_sec: 30,
292            time_travel_vacuum_interval_sec: 30,
293            vacuum_spin_interval_ms: 0,
294            hummock_version_checkpoint_interval_sec: 30,
295            enable_hummock_data_archive: false,
296            hummock_time_travel_snapshot_interval: 0,
297            hummock_time_travel_sst_info_fetch_batch_size: 10_000,
298            hummock_time_travel_sst_info_insert_batch_size: 10,
299            hummock_time_travel_epoch_version_insert_batch_size: 1000,
300            hummock_gc_history_insert_batch_size: 1000,
301            hummock_time_travel_filter_out_objects_batch_size: 1000,
302            hummock_time_travel_filter_out_objects_v1: false,
303            hummock_time_travel_filter_out_objects_list_version_batch_size: 10,
304            hummock_time_travel_filter_out_objects_list_delta_batch_size: 1000,
305            min_delta_log_num_for_hummock_version_checkpoint: 1,
306            min_sst_retention_time_sec: 3600 * 24 * 7,
307            full_gc_interval_sec: 3600 * 24 * 7,
308            full_gc_object_limit: 100_000,
309            gc_history_retention_time_sec: 3600 * 24 * 7,
310            max_inflight_time_travel_query: 1000,
311            enable_committed_sst_sanity_check: false,
312            periodic_compaction_interval_sec: 60,
313            node_num_monitor_interval_sec: 10,
314            protect_drop_table_with_incoming_sink: false,
315            prometheus_endpoint: None,
316            prometheus_selector: None,
317            vpc_id: None,
318            security_group_id: None,
319            privatelink_endpoint_default_tags: None,
320            periodic_space_reclaim_compaction_interval_sec: 60,
321            telemetry_enabled: false,
322            periodic_ttl_reclaim_compaction_interval_sec: 60,
323            periodic_tombstone_reclaim_compaction_interval_sec: 60,
324            periodic_scheduling_compaction_group_split_interval_sec: 60,
325            compact_task_table_size_partition_threshold_low: 128 * 1024 * 1024,
326            compact_task_table_size_partition_threshold_high: 512 * 1024 * 1024,
327            table_high_write_throughput_threshold: 128 * 1024 * 1024,
328            table_low_write_throughput_threshold: 64 * 1024 * 1024,
329            do_not_config_object_storage_lifecycle: true,
330            partition_vnode_count: 32,
331            compaction_task_max_heartbeat_interval_secs: 0,
332            compaction_task_max_progress_interval_secs: 1,
333            compaction_config: None,
334            hybrid_partition_node_count: 4,
335            event_log_enabled: false,
336            event_log_channel_max_size: 1,
337            advertise_addr: "".to_owned(),
338            cached_traces_num: 1,
339            cached_traces_memory_limit_bytes: usize::MAX,
340            enable_trivial_move: true,
341            enable_check_task_level_overlap: true,
342            enable_dropped_column_reclaim: false,
343            object_store_config: ObjectStoreConfig::default(),
344            max_trivial_move_task_count_per_loop: 256,
345            max_get_task_probe_times: 5,
346            secret_store_private_key: Some(
347                hex::decode("0123456789abcdef0123456789abcdef").unwrap(),
348            ),
349            temp_secret_file_dir: "./secrets".to_owned(),
350            actor_cnt_per_worker_parallelism_hard_limit: usize::MAX,
351            actor_cnt_per_worker_parallelism_soft_limit: usize::MAX,
352            split_group_size_ratio: 0.9,
353            table_stat_high_write_throughput_ratio_for_split: 0.5,
354            table_stat_low_write_throughput_ratio_for_merge: 0.7,
355            table_stat_throuput_window_seconds_for_split: 60,
356            table_stat_throuput_window_seconds_for_merge: 240,
357            periodic_scheduling_compaction_group_merge_interval_sec: 60 * 10,
358            compaction_group_merge_dimension_threshold: 1.2,
359            license_key_path: None,
360            compute_client_config: RpcClientConfig::default(),
361            stream_client_config: RpcClientConfig::default(),
362            frontend_client_config: RpcClientConfig::default(),
363            redact_sql_option_keywords: Arc::new(Default::default()),
364        }
365    }
366}
367
368impl MetaSrvEnv {
369    pub async fn new(
370        opts: MetaOpts,
371        mut init_system_params: SystemParams,
372        init_session_config: SessionConfig,
373        meta_store_impl: SqlMetaStore,
374    ) -> MetaResult<Self> {
375        let idle_manager = Arc::new(IdleManager::new(opts.max_idle_ms));
376        let stream_client_pool =
377            Arc::new(StreamClientPool::new(1, opts.stream_client_config.clone())); // typically no need for plural clients
378        let frontend_client_pool = Arc::new(FrontendClientPool::new(
379            1,
380            opts.frontend_client_config.clone(),
381        ));
382        let event_log_manager = Arc::new(start_event_log_manager(
383            opts.event_log_enabled,
384            opts.event_log_channel_max_size,
385        ));
386
387        // When license key path is specified, license key from system parameters can be easily
388        // overwritten. So we simply reject this case.
389        if opts.license_key_path.is_some()
390            && init_system_params.license_key
391                != system_param::default::license_key_opt().map(Into::into)
392        {
393            bail!(
394                "argument `--license-key-path` (or env var `RW_LICENSE_KEY_PATH`) and \
395                 system parameter `license_key` (or env var `RW_LICENSE_KEY`) may not \
396                 be set at the same time"
397            );
398        }
399
400        let cluster_first_launch = meta_store_impl.up().await.context(
401            "Failed to initialize the meta store, \
402            this may happen if there's existing metadata incompatible with the current version of RisingWave, \
403            e.g., downgrading from a newer release or a nightly build to an older one. \
404            For a single-node deployment, you may want to reset all data by deleting the data directory, \
405            typically located at `~/.risingwave`.",
406        )?;
407
408        let notification_manager =
409            Arc::new(NotificationManager::new(meta_store_impl.clone()).await);
410        let cluster_id = Cluster::find()
411            .one(&meta_store_impl.conn)
412            .await?
413            .map(|c| c.cluster_id.to_string().into())
414            .unwrap();
415
416        // For new clusters:
417        // - the name of the object store needs to be prefixed according to the object id.
418        //
419        // For old clusters
420        // - the prefix is not divided for the sake of compatibility.
421        init_system_params.use_new_object_prefix_strategy = Some(cluster_first_launch);
422
423        let system_param_controller = Arc::new(
424            SystemParamsController::new(
425                meta_store_impl.clone(),
426                notification_manager.clone(),
427                init_system_params,
428            )
429            .await?,
430        );
431        let session_param_controller = Arc::new(
432            SessionParamsController::new(
433                meta_store_impl.clone(),
434                notification_manager.clone(),
435                init_session_config,
436            )
437            .await?,
438        );
439        Ok(Self {
440            id_gen_manager_impl: Arc::new(SqlIdGeneratorManager::new(&meta_store_impl.conn).await?),
441            system_param_manager_impl: system_param_controller,
442            session_param_manager_impl: session_param_controller,
443            meta_store_impl: meta_store_impl.clone(),
444            notification_manager,
445            stream_client_pool,
446            frontend_client_pool,
447            idle_manager,
448            event_log_manager,
449            cluster_id,
450            hummock_seq: Arc::new(SequenceGenerator::new(meta_store_impl.conn.clone())),
451            opts: opts.into(),
452            // Await trees on the meta node is lightweight, thus always enabled.
453            await_tree_reg: await_tree::Registry::new(Default::default()),
454        })
455    }
456
457    pub fn meta_store(&self) -> SqlMetaStore {
458        self.meta_store_impl.clone()
459    }
460
461    pub fn meta_store_ref(&self) -> &SqlMetaStore {
462        &self.meta_store_impl
463    }
464
465    pub fn id_gen_manager(&self) -> &SqlIdGeneratorManagerRef {
466        &self.id_gen_manager_impl
467    }
468
469    pub fn notification_manager_ref(&self) -> NotificationManagerRef {
470        self.notification_manager.clone()
471    }
472
473    pub fn notification_manager(&self) -> &NotificationManager {
474        self.notification_manager.deref()
475    }
476
477    pub fn idle_manager_ref(&self) -> IdleManagerRef {
478        self.idle_manager.clone()
479    }
480
481    pub fn idle_manager(&self) -> &IdleManager {
482        self.idle_manager.deref()
483    }
484
485    pub async fn system_params_reader(&self) -> SystemParamsReader {
486        self.system_param_manager_impl.get_params().await
487    }
488
489    pub fn system_params_manager_impl_ref(&self) -> SystemParamsControllerRef {
490        self.system_param_manager_impl.clone()
491    }
492
493    pub fn session_params_manager_impl_ref(&self) -> SessionParamsControllerRef {
494        self.session_param_manager_impl.clone()
495    }
496
497    pub fn stream_client_pool_ref(&self) -> StreamClientPoolRef {
498        self.stream_client_pool.clone()
499    }
500
501    pub fn stream_client_pool(&self) -> &StreamClientPool {
502        self.stream_client_pool.deref()
503    }
504
505    pub fn frontend_client_pool(&self) -> &FrontendClientPool {
506        self.frontend_client_pool.deref()
507    }
508
509    pub fn cluster_id(&self) -> &ClusterId {
510        &self.cluster_id
511    }
512
513    pub fn event_log_manager_ref(&self) -> EventLogManagerRef {
514        self.event_log_manager.clone()
515    }
516
517    pub fn await_tree_reg(&self) -> &await_tree::Registry {
518        &self.await_tree_reg
519    }
520}
521
522#[cfg(any(test, feature = "test"))]
523impl MetaSrvEnv {
524    // Instance for test.
525    pub async fn for_test() -> Self {
526        Self::for_test_opts(MetaOpts::test(false), |_| ()).await
527    }
528
529    pub async fn for_test_opts(
530        opts: MetaOpts,
531        on_test_system_params: impl FnOnce(&mut risingwave_pb::meta::PbSystemParams),
532    ) -> Self {
533        let mut system_params = risingwave_common::system_param::system_params_for_test();
534        on_test_system_params(&mut system_params);
535        Self::new(
536            opts,
537            system_params,
538            Default::default(),
539            SqlMetaStore::for_test().await,
540        )
541        .await
542        .unwrap()
543    }
544}
risingwave_meta/manager/env.rs

risingwave_meta/manager/
env.rs
