Skip to main content

cmd/
frontend.rs

1// Copyright 2023 Greptime Team
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::fmt::Debug;
16use std::path::Path;
17use std::sync::Arc;
18use std::time::Duration;
19
20use async_trait::async_trait;
21use cache::{build_fundamental_cache_registry, with_default_composite_cache_registry};
22use catalog::information_extension::DistributedInformationExtension;
23use catalog::information_schema::InformationExtensionRef;
24use catalog::kvbackend::{
25    CachedKvBackendBuilder, CatalogManagerConfiguratorRef, KvBackendCatalogManagerBuilder,
26    MetaKvBackend,
27};
28use catalog::process_manager::ProcessManager;
29use clap::Parser;
30use client::client_manager::NodeClients;
31use common_base::Plugins;
32use common_config::{Configurable, DEFAULT_DATA_HOME};
33use common_error::ext::BoxedError;
34use common_grpc::channel_manager::ChannelConfig;
35use common_meta::cache::{CacheRegistryBuilder, LayeredCacheRegistryBuilder};
36use common_meta::heartbeat::handler::HandlerGroupExecutor;
37use common_meta::heartbeat::handler::invalidate_table_cache::InvalidateCacheHandler;
38use common_meta::heartbeat::handler::parse_mailbox_message::ParseMailboxMessageHandler;
39use common_meta::heartbeat::handler::suspend::SuspendHandler;
40use common_query::prelude::set_default_prefix;
41use common_stat::ResourceStatImpl;
42use common_telemetry::info;
43use common_telemetry::logging::{DEFAULT_LOGGING_DIR, TracingOptions};
44use common_time::timezone::set_default_timezone;
45use common_version::{short_version, verbose_version};
46use frontend::frontend::Frontend;
47use frontend::heartbeat::HeartbeatTask;
48use frontend::instance::builder::FrontendBuilder;
49use frontend::server::Services;
50use meta_client::{MetaClientOptions, MetaClientRef, MetaClientType};
51use plugins::PluginOptions;
52use plugins::frontend::context::{
53    CatalogManagerConfigureContext, DistributedCatalogManagerConfigureContext,
54};
55use plugins::frontend::setup_frontend_dynamic_plugins;
56use plugins::options::PluginOptionsDeserializerImpl;
57use servers::addrs;
58use servers::grpc::GrpcOptions;
59use servers::tls::{TlsMode, TlsOption, merge_tls_option};
60use snafu::{OptionExt, ResultExt};
61use tracing_appender::non_blocking::WorkerGuard;
62
63use crate::error::{self, OtherSnafu, Result};
64use crate::options::{GlobalOptions, GreptimeOptions};
65use crate::{App, create_resource_limit_metrics, log_versions, maybe_activate_heap_profile};
66
67type FrontendOptions = GreptimeOptions<frontend::frontend::FrontendOptions>;
68
69pub struct Instance {
70    frontend: Frontend,
71    // Keep the logging guard to prevent the worker from being dropped.
72    _guard: Vec<WorkerGuard>,
73}
74
75pub const APP_NAME: &str = "greptime-frontend";
76
77impl Instance {
78    pub fn new(frontend: Frontend, _guard: Vec<WorkerGuard>) -> Self {
79        Self { frontend, _guard }
80    }
81
82    pub fn inner(&self) -> &Frontend {
83        &self.frontend
84    }
85
86    pub fn mut_inner(&mut self) -> &mut Frontend {
87        &mut self.frontend
88    }
89}
90
91#[async_trait]
92impl App for Instance {
93    fn name(&self) -> &str {
94        APP_NAME
95    }
96
97    async fn start(&mut self) -> Result<()> {
98        let plugins = self.frontend.instance.plugins().clone();
99        plugins::start_frontend_plugins(plugins)
100            .await
101            .context(error::StartFrontendSnafu)?;
102
103        self.frontend
104            .start()
105            .await
106            .context(error::StartFrontendSnafu)
107    }
108
109    async fn stop(&mut self) -> Result<()> {
110        self.frontend
111            .shutdown()
112            .await
113            .context(error::ShutdownFrontendSnafu)
114    }
115}
116
117#[derive(Parser)]
118pub struct Command {
119    #[clap(subcommand)]
120    pub subcmd: SubCommand,
121}
122
123impl Command {
124    pub async fn build(&self, opts: FrontendOptions) -> Result<Instance> {
125        self.subcmd.build(opts).await
126    }
127
128    pub fn load_options(&self, global_options: &GlobalOptions) -> Result<FrontendOptions> {
129        self.subcmd.load_options(global_options)
130    }
131}
132
133#[derive(Parser)]
134pub enum SubCommand {
135    Start(StartCommand),
136}
137
138impl SubCommand {
139    async fn build(&self, opts: FrontendOptions) -> Result<Instance> {
140        match self {
141            SubCommand::Start(cmd) => cmd.build(opts).await,
142        }
143    }
144
145    fn load_options(&self, global_options: &GlobalOptions) -> Result<FrontendOptions> {
146        match self {
147            SubCommand::Start(cmd) => cmd.load_options(global_options),
148        }
149    }
150}
151
152#[derive(Debug, Default, Parser)]
153pub struct StartCommand {
154    /// The address to bind the gRPC server.
155    #[clap(long = "grpc-bind-addr", alias = "rpc-bind-addr", alias = "rpc-addr")]
156    grpc_bind_addr: Option<String>,
157    /// The address advertised to the metasrv, and used for connections from outside the host.
158    /// If left empty or unset, the server will automatically use the IP address of the first network interface
159    /// on the host, with the same port number as the one specified in `grpc_bind_addr`.
160    #[clap(
161        long = "grpc-server-addr",
162        alias = "rpc-server-addr",
163        alias = "rpc-hostname"
164    )]
165    grpc_server_addr: Option<String>,
166    /// The address to bind the internal gRPC server.
167    #[clap(
168        long = "internal-grpc-bind-addr",
169        alias = "internal-rpc-bind-addr",
170        alias = "internal-rpc-addr"
171    )]
172    internal_grpc_bind_addr: Option<String>,
173    /// The address advertised to the metasrv, and used for connections from outside the host.
174    /// If left empty or unset, the server will automatically use the IP address of the first network interface
175    /// on the host, with the same port number as the one specified in `internal_grpc_bind_addr`.
176    #[clap(
177        long = "internal-grpc-server-addr",
178        alias = "internal-rpc-server-addr",
179        alias = "internal-rpc-hostname"
180    )]
181    internal_grpc_server_addr: Option<String>,
182    #[clap(long)]
183    http_addr: Option<String>,
184    #[clap(long)]
185    http_timeout: Option<u64>,
186    #[clap(long)]
187    mysql_addr: Option<String>,
188    #[clap(long)]
189    postgres_addr: Option<String>,
190    #[clap(short, long)]
191    pub config_file: Option<String>,
192    #[clap(short, long)]
193    influxdb_enable: Option<bool>,
194    #[clap(long, value_delimiter = ',', num_args = 1..)]
195    metasrv_addrs: Option<Vec<String>>,
196    #[clap(long)]
197    tls_mode: Option<TlsMode>,
198    #[clap(long)]
199    tls_cert_path: Option<String>,
200    #[clap(long)]
201    tls_key_path: Option<String>,
202    #[clap(long)]
203    tls_watch: bool,
204    #[clap(long)]
205    user_provider: Option<String>,
206    #[clap(long)]
207    disable_dashboard: Option<bool>,
208    #[clap(long, default_value = "GREPTIMEDB_FRONTEND")]
209    pub env_prefix: String,
210}
211
212impl StartCommand {
213    fn load_options(&self, global_options: &GlobalOptions) -> Result<FrontendOptions> {
214        let mut opts = FrontendOptions::load_layered_options(
215            self.config_file.as_deref(),
216            self.env_prefix.as_ref(),
217        )
218        .context(error::LoadLayeredConfigSnafu)?;
219
220        self.merge_with_cli_options(global_options, &mut opts)?;
221
222        Ok(opts)
223    }
224
225    // The precedence order is: cli > config file > environment variables > default values.
226    fn merge_with_cli_options(
227        &self,
228        global_options: &GlobalOptions,
229        opts: &mut FrontendOptions,
230    ) -> Result<()> {
231        let opts = &mut opts.component;
232
233        if let Some(dir) = &global_options.log_dir {
234            opts.logging.dir.clone_from(dir);
235        }
236
237        // If the logging dir is not set, use the default logs dir in the data home.
238        if opts.logging.dir.is_empty() {
239            opts.logging.dir = Path::new(DEFAULT_DATA_HOME)
240                .join(DEFAULT_LOGGING_DIR)
241                .to_string_lossy()
242                .to_string();
243        }
244
245        if global_options.log_level.is_some() {
246            opts.logging.level.clone_from(&global_options.log_level);
247        }
248
249        opts.tracing = TracingOptions {
250            #[cfg(feature = "tokio-console")]
251            tokio_console_addr: global_options.tokio_console_addr.clone(),
252        };
253
254        let tls_opts = TlsOption::new(
255            self.tls_mode,
256            self.tls_cert_path.clone(),
257            self.tls_key_path.clone(),
258            self.tls_watch,
259        );
260
261        if let Some(addr) = &self.http_addr {
262            opts.http.addr.clone_from(addr);
263        }
264
265        if let Some(http_timeout) = self.http_timeout {
266            opts.http.timeout = Duration::from_secs(http_timeout)
267        }
268
269        if let Some(disable_dashboard) = self.disable_dashboard {
270            opts.http.disable_dashboard = disable_dashboard;
271        }
272
273        if let Some(addr) = &self.grpc_bind_addr {
274            opts.grpc.bind_addr.clone_from(addr);
275            opts.grpc.tls = merge_tls_option(&opts.grpc.tls, tls_opts.clone());
276        }
277
278        if let Some(addr) = &self.grpc_server_addr {
279            opts.grpc.server_addr.clone_from(addr);
280        }
281
282        if let Some(addr) = &self.internal_grpc_bind_addr {
283            if let Some(internal_grpc) = &mut opts.internal_grpc {
284                internal_grpc.bind_addr = addr.clone();
285            } else {
286                let grpc_options = GrpcOptions {
287                    bind_addr: addr.clone(),
288                    ..Default::default()
289                };
290
291                opts.internal_grpc = Some(grpc_options);
292            }
293        }
294
295        if let Some(addr) = &self.internal_grpc_server_addr {
296            if let Some(internal_grpc) = &mut opts.internal_grpc {
297                internal_grpc.server_addr = addr.clone();
298            } else {
299                let grpc_options = GrpcOptions {
300                    server_addr: addr.clone(),
301                    ..Default::default()
302                };
303                opts.internal_grpc = Some(grpc_options);
304            }
305        }
306
307        if let Some(addr) = &self.mysql_addr {
308            opts.mysql.enable = true;
309            opts.mysql.addr.clone_from(addr);
310            opts.mysql.tls = merge_tls_option(&opts.mysql.tls, tls_opts.clone());
311        }
312
313        if let Some(addr) = &self.postgres_addr {
314            opts.postgres.enable = true;
315            opts.postgres.addr.clone_from(addr);
316            opts.postgres.tls = merge_tls_option(&opts.postgres.tls, tls_opts.clone());
317        }
318
319        if let Some(enable) = self.influxdb_enable {
320            opts.influxdb.enable = enable;
321        }
322
323        if let Some(metasrv_addrs) = &self.metasrv_addrs {
324            opts.meta_client
325                .get_or_insert_with(MetaClientOptions::default)
326                .metasrv_addrs
327                .clone_from(metasrv_addrs);
328        }
329
330        if let Some(user_provider) = &self.user_provider {
331            opts.user_provider = Some(user_provider.clone());
332        }
333
334        Ok(())
335    }
336
337    async fn build(&self, opts: FrontendOptions) -> Result<Instance> {
338        common_runtime::init_global_runtimes(&opts.runtime);
339
340        let guard = common_telemetry::init_global_logging(
341            APP_NAME,
342            &opts.component.logging,
343            &opts.component.tracing,
344            opts.component.node_id.clone(),
345            Some(&opts.component.slow_query),
346        );
347
348        log_versions(verbose_version(), short_version(), APP_NAME);
349        maybe_activate_heap_profile(&opts.component.memory);
350        create_resource_limit_metrics(APP_NAME);
351
352        info!("Frontend start command: {:#?}", self);
353        info!("Frontend options: {:#?}", opts);
354
355        let plugin_opts = opts.plugins;
356        let mut opts = opts.component;
357        opts.grpc.detect_server_addr();
358        let mut plugins = Plugins::new();
359        plugins::setup_frontend_plugins(&mut plugins, &plugin_opts, &opts)
360            .await
361            .context(error::StartFrontendSnafu)?;
362
363        set_default_timezone(opts.default_timezone.as_deref()).context(error::InitTimezoneSnafu)?;
364        set_default_prefix(opts.default_column_prefix.as_deref())
365            .map_err(BoxedError::new)
366            .context(error::BuildCliSnafu)?;
367
368        let meta_client_options = opts
369            .meta_client
370            .as_ref()
371            .context(error::MissingConfigSnafu {
372                msg: "'meta_client'",
373            })?;
374
375        let cache_max_capacity = meta_client_options.metadata_cache_max_capacity;
376        let cache_ttl = meta_client_options.metadata_cache_ttl;
377        let cache_tti = meta_client_options.metadata_cache_tti;
378
379        let meta_client = meta_client::create_meta_client(
380            MetaClientType::Frontend,
381            meta_client_options,
382            Some(&plugins),
383            None,
384        )
385        .await
386        .context(error::MetaClientInitSnafu)?;
387
388        let meta_config: Vec<PluginOptions> = meta_client
389            .pull_config(PluginOptionsDeserializerImpl)
390            .await
391            .context(error::MetaClientInitSnafu)?;
392        setup_frontend_dynamic_plugins(meta_config, &mut plugins)
393            .await
394            .context(error::StartFrontendSnafu)?;
395
396        // TODO(discord9): add helper function to ease the creation of cache registry&such
397        let cached_meta_backend =
398            CachedKvBackendBuilder::new(Arc::new(MetaKvBackend::new(meta_client.clone())))
399                .cache_max_capacity(cache_max_capacity)
400                .cache_ttl(cache_ttl)
401                .cache_tti(cache_tti)
402                .build();
403        let cached_meta_backend = Arc::new(cached_meta_backend);
404
405        // Builds cache registry
406        let layered_cache_builder = LayeredCacheRegistryBuilder::default().add_cache_registry(
407            CacheRegistryBuilder::default()
408                .add_cache(cached_meta_backend.clone())
409                .build(),
410        );
411        let fundamental_cache_registry =
412            build_fundamental_cache_registry(Arc::new(MetaKvBackend::new(meta_client.clone())));
413        let layered_cache_registry = Arc::new(
414            with_default_composite_cache_registry(
415                layered_cache_builder.add_cache_registry(fundamental_cache_registry),
416            )
417            .context(error::BuildCacheRegistrySnafu)?
418            .build(),
419        );
420
421        // frontend to datanode need not timeout.
422        // Some queries are expected to take long time.
423        let mut channel_config = ChannelConfig {
424            timeout: None,
425            tcp_nodelay: opts.datanode.client.tcp_nodelay,
426            connect_timeout: Some(opts.datanode.client.connect_timeout),
427            ..Default::default()
428        };
429        if opts.grpc.flight_compression.transport_compression() {
430            channel_config.accept_compression = true;
431            channel_config.send_compression = true;
432        }
433        let client = Arc::new(NodeClients::new(channel_config));
434
435        let information_extension = Arc::new(DistributedInformationExtension::new(
436            meta_client.clone(),
437            client.clone(),
438        ));
439        plugins.insert::<InformationExtensionRef>(information_extension.clone());
440
441        let process_manager = Arc::new(ProcessManager::new(
442            addrs::resolve_addr(&opts.grpc.bind_addr, Some(&opts.grpc.server_addr)),
443            Some(meta_client.clone()),
444        ));
445
446        let builder = KvBackendCatalogManagerBuilder::new(
447            information_extension,
448            cached_meta_backend.clone(),
449            layered_cache_registry.clone(),
450        )
451        .with_process_manager(process_manager.clone());
452        let builder = if let Some(configurator) =
453            plugins.get::<CatalogManagerConfiguratorRef<CatalogManagerConfigureContext>>()
454        {
455            let ctx = DistributedCatalogManagerConfigureContext {
456                meta_client: meta_client.clone(),
457            };
458            let ctx = CatalogManagerConfigureContext::Distributed(ctx);
459
460            configurator
461                .configure(builder, ctx)
462                .await
463                .context(OtherSnafu)?
464        } else {
465            builder
466        };
467        let catalog_manager = builder.build();
468
469        let instance = FrontendBuilder::new(
470            opts.clone(),
471            cached_meta_backend.clone(),
472            layered_cache_registry.clone(),
473            catalog_manager,
474            client,
475            meta_client.clone(),
476            process_manager,
477        )
478        .with_plugin(plugins.clone())
479        .with_local_cache_invalidator(layered_cache_registry)
480        .try_build()
481        .await
482        .context(error::StartFrontendSnafu)?;
483
484        let heartbeat_task = Some(create_heartbeat_task(&opts, meta_client, &instance));
485
486        let instance = Arc::new(instance);
487
488        let servers = Services::new(opts, instance.clone(), plugins)
489            .build()
490            .context(error::StartFrontendSnafu)?;
491
492        let frontend = Frontend {
493            instance,
494            servers,
495            heartbeat_task,
496        };
497
498        Ok(Instance::new(frontend, guard))
499    }
500}
501
502pub fn create_heartbeat_task(
503    options: &frontend::frontend::FrontendOptions,
504    meta_client: MetaClientRef,
505    instance: &frontend::instance::Instance,
506) -> HeartbeatTask {
507    let executor = Arc::new(HandlerGroupExecutor::new(vec![
508        Arc::new(ParseMailboxMessageHandler),
509        Arc::new(SuspendHandler::new(instance.suspend_state())),
510        Arc::new(InvalidateCacheHandler::new(
511            instance.cache_invalidator().clone(),
512        )),
513    ]));
514
515    let stat = {
516        let mut stat = ResourceStatImpl::default();
517        stat.start_collect_cpu_usage();
518        Arc::new(stat)
519    };
520
521    HeartbeatTask::new(options, meta_client, executor, stat)
522}
523
524#[cfg(test)]
525mod tests {
526    use std::io::Write;
527    use std::time::Duration;
528
529    use auth::{Identity, Password, UserProviderRef};
530    use clap::{CommandFactory, Parser};
531    use common_base::readable_size::ReadableSize;
532    use common_config::ENV_VAR_SEP;
533    use common_test_util::temp_dir::create_named_temp_file;
534    use servers::grpc::GrpcOptions;
535    use servers::http::HttpOptions;
536
537    use super::*;
538    use crate::options::GlobalOptions;
539
540    #[test]
541    fn test_try_from_start_command() {
542        let command = StartCommand {
543            http_addr: Some("127.0.0.1:1234".to_string()),
544            mysql_addr: Some("127.0.0.1:5678".to_string()),
545            postgres_addr: Some("127.0.0.1:5432".to_string()),
546            internal_grpc_bind_addr: Some("127.0.0.1:4010".to_string()),
547            internal_grpc_server_addr: Some("10.0.0.24:4010".to_string()),
548            influxdb_enable: Some(false),
549            disable_dashboard: Some(false),
550            ..Default::default()
551        };
552
553        let opts = command.load_options(&Default::default()).unwrap().component;
554
555        assert_eq!(opts.http.addr, "127.0.0.1:1234");
556        assert_eq!(ReadableSize::mb(64), opts.http.body_limit);
557        assert_eq!(opts.mysql.addr, "127.0.0.1:5678");
558        assert_eq!(opts.postgres.addr, "127.0.0.1:5432");
559
560        let internal_grpc = opts.internal_grpc.as_ref().unwrap();
561        assert_eq!(internal_grpc.bind_addr, "127.0.0.1:4010");
562        assert_eq!(internal_grpc.server_addr, "10.0.0.24:4010");
563
564        let default_opts = FrontendOptions::default().component;
565
566        assert_eq!(opts.grpc.bind_addr, default_opts.grpc.bind_addr);
567        assert!(opts.mysql.enable);
568        assert_eq!(opts.mysql.runtime_size, default_opts.mysql.runtime_size);
569        assert!(opts.postgres.enable);
570        assert_eq!(
571            opts.postgres.runtime_size,
572            default_opts.postgres.runtime_size
573        );
574        assert!(opts.opentsdb.enable);
575
576        assert!(!opts.influxdb.enable);
577    }
578
579    #[test]
580    fn test_read_from_config_file() {
581        let mut file = create_named_temp_file();
582        let toml_str = r#"
583            [http]
584            addr = "127.0.0.1:4000"
585            timeout = "0s"
586            body_limit = "2GB"
587
588            [opentsdb]
589            enable = false
590
591            [logging]
592            level = "debug"
593            dir = "./greptimedb_data/test/logs"
594        "#;
595        write!(file, "{}", toml_str).unwrap();
596
597        let command = StartCommand {
598            config_file: Some(file.path().to_str().unwrap().to_string()),
599            disable_dashboard: Some(false),
600            ..Default::default()
601        };
602
603        let fe_opts = command.load_options(&Default::default()).unwrap().component;
604
605        assert_eq!("127.0.0.1:4000".to_string(), fe_opts.http.addr);
606        assert_eq!(Duration::from_secs(0), fe_opts.http.timeout);
607
608        assert_eq!(ReadableSize::gb(2), fe_opts.http.body_limit);
609
610        assert_eq!("debug", fe_opts.logging.level.as_ref().unwrap());
611        assert_eq!(
612            "./greptimedb_data/test/logs".to_string(),
613            fe_opts.logging.dir
614        );
615        assert!(!fe_opts.opentsdb.enable);
616    }
617
618    #[tokio::test]
619    async fn test_try_from_start_command_to_anymap() {
620        let fe_opts = frontend::frontend::FrontendOptions {
621            http: HttpOptions {
622                disable_dashboard: false,
623                ..Default::default()
624            },
625            user_provider: Some("static_user_provider:cmd:test=test".to_string()),
626            ..Default::default()
627        };
628
629        let mut plugins = Plugins::new();
630        plugins::setup_frontend_plugins(&mut plugins, &[], &fe_opts)
631            .await
632            .unwrap();
633
634        let provider = plugins.get::<UserProviderRef>().unwrap();
635        let result = provider
636            .authenticate(
637                Identity::UserId("test", None),
638                Password::PlainText("test".to_string().into()),
639            )
640            .await;
641        let _ = result.unwrap();
642    }
643
644    #[test]
645    fn test_load_log_options_from_cli() {
646        let cmd = StartCommand {
647            disable_dashboard: Some(false),
648            ..Default::default()
649        };
650
651        let options = cmd
652            .load_options(&GlobalOptions {
653                log_dir: Some("./greptimedb_data/test/logs".to_string()),
654                log_level: Some("debug".to_string()),
655
656                #[cfg(feature = "tokio-console")]
657                tokio_console_addr: None,
658            })
659            .unwrap()
660            .component;
661
662        let logging_opt = options.logging;
663        assert_eq!("./greptimedb_data/test/logs", logging_opt.dir);
664        assert_eq!("debug", logging_opt.level.as_ref().unwrap());
665    }
666
667    #[test]
668    fn test_config_precedence_order() {
669        let mut file = create_named_temp_file();
670        let toml_str = r#"
671            [http]
672            addr = "127.0.0.1:4000"
673
674            [meta_client]
675            timeout = "3s"
676            connect_timeout = "5s"
677            tcp_nodelay = true
678
679            [mysql]
680            addr = "127.0.0.1:4002"
681        "#;
682        write!(file, "{}", toml_str).unwrap();
683
684        let env_prefix = "FRONTEND_UT";
685        temp_env::with_vars(
686            [
687                (
688                    // mysql.addr = 127.0.0.1:14002
689                    [
690                        env_prefix.to_string(),
691                        "mysql".to_uppercase(),
692                        "addr".to_uppercase(),
693                    ]
694                    .join(ENV_VAR_SEP),
695                    Some("127.0.0.1:14002"),
696                ),
697                (
698                    // mysql.runtime_size = 11
699                    [
700                        env_prefix.to_string(),
701                        "mysql".to_uppercase(),
702                        "runtime_size".to_uppercase(),
703                    ]
704                    .join(ENV_VAR_SEP),
705                    Some("11"),
706                ),
707                (
708                    // http.addr = 127.0.0.1:24000
709                    [
710                        env_prefix.to_string(),
711                        "http".to_uppercase(),
712                        "addr".to_uppercase(),
713                    ]
714                    .join(ENV_VAR_SEP),
715                    Some("127.0.0.1:24000"),
716                ),
717                (
718                    // meta_client.metasrv_addrs = 127.0.0.1:3001,127.0.0.1:3002,127.0.0.1:3003
719                    [
720                        env_prefix.to_string(),
721                        "meta_client".to_uppercase(),
722                        "metasrv_addrs".to_uppercase(),
723                    ]
724                    .join(ENV_VAR_SEP),
725                    Some("127.0.0.1:3001,127.0.0.1:3002,127.0.0.1:3003"),
726                ),
727            ],
728            || {
729                let command = StartCommand {
730                    config_file: Some(file.path().to_str().unwrap().to_string()),
731                    http_addr: Some("127.0.0.1:14000".to_string()),
732                    env_prefix: env_prefix.to_string(),
733                    ..Default::default()
734                };
735
736                let fe_opts = command.load_options(&Default::default()).unwrap().component;
737
738                // Should be read from env, env > default values.
739                assert_eq!(fe_opts.mysql.runtime_size, 11);
740                assert_eq!(
741                    fe_opts.meta_client.unwrap().metasrv_addrs,
742                    vec![
743                        "127.0.0.1:3001".to_string(),
744                        "127.0.0.1:3002".to_string(),
745                        "127.0.0.1:3003".to_string()
746                    ]
747                );
748
749                // Should be read from config file, config file > env > default values.
750                assert_eq!(fe_opts.mysql.addr, "127.0.0.1:4002");
751
752                // Should be read from cli, cli > config file > env > default values.
753                assert_eq!(fe_opts.http.addr, "127.0.0.1:14000");
754
755                // Should be default value.
756                assert_eq!(fe_opts.grpc.bind_addr, GrpcOptions::default().bind_addr);
757            },
758        );
759    }
760
761    #[test]
762    fn test_parse_grpc_cli_aliases() {
763        let command = StartCommand::try_parse_from([
764            "frontend",
765            "--grpc-bind-addr",
766            "127.0.0.1:14001",
767            "--grpc-server-addr",
768            "10.0.0.1:14001",
769            "--internal-grpc-bind-addr",
770            "127.0.0.1:14010",
771            "--internal-grpc-server-addr",
772            "10.0.0.1:14010",
773        ])
774        .unwrap();
775        assert_eq!(command.grpc_bind_addr.as_deref(), Some("127.0.0.1:14001"));
776        assert_eq!(command.grpc_server_addr.as_deref(), Some("10.0.0.1:14001"));
777        assert_eq!(
778            command.internal_grpc_bind_addr.as_deref(),
779            Some("127.0.0.1:14010")
780        );
781        assert_eq!(
782            command.internal_grpc_server_addr.as_deref(),
783            Some("10.0.0.1:14010")
784        );
785
786        let command = StartCommand::try_parse_from([
787            "frontend",
788            "--rpc-bind-addr",
789            "127.0.0.1:24001",
790            "--rpc-server-addr",
791            "10.0.0.2:24001",
792            "--internal-rpc-bind-addr",
793            "127.0.0.1:24010",
794            "--internal-rpc-server-addr",
795            "10.0.0.2:24010",
796        ])
797        .unwrap();
798        assert_eq!(command.grpc_bind_addr.as_deref(), Some("127.0.0.1:24001"));
799        assert_eq!(command.grpc_server_addr.as_deref(), Some("10.0.0.2:24001"));
800        assert_eq!(
801            command.internal_grpc_bind_addr.as_deref(),
802            Some("127.0.0.1:24010")
803        );
804        assert_eq!(
805            command.internal_grpc_server_addr.as_deref(),
806            Some("10.0.0.2:24010")
807        );
808
809        let command = StartCommand::try_parse_from([
810            "frontend",
811            "--rpc-addr",
812            "127.0.0.1:34001",
813            "--rpc-hostname",
814            "10.0.0.3:34001",
815            "--internal-rpc-addr",
816            "127.0.0.1:34010",
817            "--internal-rpc-hostname",
818            "10.0.0.3:34010",
819        ])
820        .unwrap();
821        assert_eq!(command.grpc_bind_addr.as_deref(), Some("127.0.0.1:34001"));
822        assert_eq!(command.grpc_server_addr.as_deref(), Some("10.0.0.3:34001"));
823        assert_eq!(
824            command.internal_grpc_bind_addr.as_deref(),
825            Some("127.0.0.1:34010")
826        );
827        assert_eq!(
828            command.internal_grpc_server_addr.as_deref(),
829            Some("10.0.0.3:34010")
830        );
831    }
832
833    #[test]
834    fn test_help_uses_grpc_option_names() {
835        let mut cmd = StartCommand::command();
836        let mut help = Vec::new();
837        cmd.write_long_help(&mut help).unwrap();
838        let help = String::from_utf8(help).unwrap();
839
840        assert!(help.contains("--grpc-bind-addr"));
841        assert!(help.contains("--grpc-server-addr"));
842        assert!(help.contains("--internal-grpc-bind-addr"));
843        assert!(help.contains("--internal-grpc-server-addr"));
844        assert!(!help.contains("--rpc-bind-addr"));
845        assert!(!help.contains("--rpc-server-addr"));
846        assert!(!help.contains("--rpc-addr"));
847        assert!(!help.contains("--rpc-hostname"));
848        assert!(!help.contains("--internal-rpc-bind-addr"));
849        assert!(!help.contains("--internal-rpc-server-addr"));
850        assert!(!help.contains("--internal-rpc-addr"));
851        assert!(!help.contains("--internal-rpc-hostname"));
852    }
853}