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

risingwave_meta/manager/
env.rs
