1use 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 new_read_only_meta_kv_backend,
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 _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 #[clap(long = "grpc-bind-addr", alias = "rpc-bind-addr", alias = "rpc-addr")]
156 grpc_bind_addr: Option<String>,
157 #[clap(
161 long = "grpc-server-addr",
162 alias = "rpc-server-addr",
163 alias = "rpc-hostname"
164 )]
165 grpc_server_addr: Option<String>,
166 #[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 #[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 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 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 let readonly_meta_backend = new_read_only_meta_kv_backend(meta_client.clone());
397
398 let cached_meta_backend = CachedKvBackendBuilder::new(readonly_meta_backend.clone())
400 .cache_max_capacity(cache_max_capacity)
401 .cache_ttl(cache_ttl)
402 .cache_tti(cache_tti)
403 .build();
404 let cached_meta_backend = Arc::new(cached_meta_backend);
405
406 let layered_cache_builder = LayeredCacheRegistryBuilder::default().add_cache_registry(
408 CacheRegistryBuilder::default()
409 .add_cache(cached_meta_backend.clone())
410 .build(),
411 );
412 let fundamental_cache_registry =
413 build_fundamental_cache_registry(readonly_meta_backend.clone());
414 let mut layered_cache_builder = with_default_composite_cache_registry(
415 layered_cache_builder.add_cache_registry(fundamental_cache_registry),
416 )
417 .context(error::BuildCacheRegistrySnafu)?;
418
419 if let Some(plugin_cache_builder) = plugins::frontend::configure_cache_registry(&plugins) {
420 layered_cache_builder =
421 layered_cache_builder.add_cache_registry(plugin_cache_builder.build());
422 }
423
424 let layered_cache_registry = Arc::new(layered_cache_builder.build());
425
426 let mut channel_config = ChannelConfig {
429 timeout: None,
430 tcp_nodelay: opts.datanode.client.tcp_nodelay,
431 connect_timeout: Some(opts.datanode.client.connect_timeout),
432 ..Default::default()
433 };
434 if opts.grpc.flight_compression.transport_compression() {
435 channel_config.accept_compression = true;
436 channel_config.send_compression = true;
437 }
438 let client = Arc::new(NodeClients::new(channel_config));
439
440 let information_extension = Arc::new(DistributedInformationExtension::new(
441 meta_client.clone(),
442 client.clone(),
443 ));
444 plugins.insert::<InformationExtensionRef>(information_extension.clone());
445
446 let process_manager = Arc::new(ProcessManager::new(
447 addrs::resolve_addr(&opts.grpc.bind_addr, Some(&opts.grpc.server_addr)),
448 Some(meta_client.clone()),
449 ));
450
451 let builder = KvBackendCatalogManagerBuilder::new(
452 information_extension,
453 cached_meta_backend.clone(),
454 layered_cache_registry.clone(),
455 )
456 .with_process_manager(process_manager.clone());
457 let builder = if let Some(configurator) =
458 plugins.get::<CatalogManagerConfiguratorRef<CatalogManagerConfigureContext>>()
459 {
460 let ctx = DistributedCatalogManagerConfigureContext {
461 meta_client: meta_client.clone(),
462 };
463 let ctx = CatalogManagerConfigureContext::Distributed(ctx);
464
465 configurator
466 .configure(builder, ctx)
467 .await
468 .context(OtherSnafu)?
469 } else {
470 builder
471 };
472 let catalog_manager = builder.build();
473
474 let instance = FrontendBuilder::new(
475 opts.clone(),
476 cached_meta_backend.clone(),
477 layered_cache_registry.clone(),
478 catalog_manager,
479 client,
480 meta_client.clone(),
481 process_manager,
482 )
483 .with_plugin(plugins.clone())
484 .with_local_cache_invalidator(layered_cache_registry)
485 .try_build()
486 .await
487 .context(error::StartFrontendSnafu)?;
488
489 let heartbeat_task = Some(create_heartbeat_task(&opts, meta_client, &instance));
490
491 let instance = Arc::new(instance);
492
493 let servers = Services::new(opts, instance.clone(), plugins)
494 .build()
495 .context(error::StartFrontendSnafu)?;
496
497 let frontend = Frontend {
498 instance,
499 servers,
500 heartbeat_task,
501 };
502
503 Ok(Instance::new(frontend, guard))
504 }
505}
506
507pub fn create_heartbeat_task(
508 options: &frontend::frontend::FrontendOptions,
509 meta_client: MetaClientRef,
510 instance: &frontend::instance::Instance,
511) -> HeartbeatTask {
512 let executor = Arc::new(HandlerGroupExecutor::new(vec![
513 Arc::new(ParseMailboxMessageHandler),
514 Arc::new(SuspendHandler::new(instance.suspend_state())),
515 Arc::new(InvalidateCacheHandler::new(
516 instance.cache_invalidator().clone(),
517 )),
518 ]));
519
520 let stat = {
521 let mut stat = ResourceStatImpl::default();
522 stat.start_collect_cpu_usage();
523 Arc::new(stat)
524 };
525
526 HeartbeatTask::new(
527 instance.frontend_peer_addr().to_string(),
528 options,
529 meta_client,
530 executor,
531 stat,
532 )
533}
534
535#[cfg(test)]
536mod tests {
537 use std::io::Write;
538 use std::time::Duration;
539
540 use auth::{Identity, Password, UserProviderRef};
541 use clap::{CommandFactory, Parser};
542 use common_base::readable_size::ReadableSize;
543 use common_config::ENV_VAR_SEP;
544 use common_test_util::temp_dir::create_named_temp_file;
545 use servers::grpc::GrpcOptions;
546 use servers::http::HttpOptions;
547
548 use super::*;
549 use crate::options::GlobalOptions;
550
551 #[test]
552 fn test_try_from_start_command() {
553 let command = StartCommand {
554 http_addr: Some("127.0.0.1:1234".to_string()),
555 mysql_addr: Some("127.0.0.1:5678".to_string()),
556 postgres_addr: Some("127.0.0.1:5432".to_string()),
557 internal_grpc_bind_addr: Some("127.0.0.1:4010".to_string()),
558 internal_grpc_server_addr: Some("10.0.0.24:4010".to_string()),
559 influxdb_enable: Some(false),
560 disable_dashboard: Some(false),
561 ..Default::default()
562 };
563
564 let opts = command.load_options(&Default::default()).unwrap().component;
565
566 assert_eq!(opts.http.addr, "127.0.0.1:1234");
567 assert_eq!(ReadableSize::mb(64), opts.http.body_limit);
568 assert_eq!(opts.mysql.addr, "127.0.0.1:5678");
569 assert_eq!(opts.postgres.addr, "127.0.0.1:5432");
570
571 let internal_grpc = opts.internal_grpc.as_ref().unwrap();
572 assert_eq!(internal_grpc.bind_addr, "127.0.0.1:4010");
573 assert_eq!(internal_grpc.server_addr, "10.0.0.24:4010");
574
575 let default_opts = FrontendOptions::default().component;
576
577 assert_eq!(opts.grpc.bind_addr, default_opts.grpc.bind_addr);
578 assert!(opts.mysql.enable);
579 assert_eq!(opts.mysql.runtime_size, default_opts.mysql.runtime_size);
580 assert!(opts.postgres.enable);
581 assert_eq!(
582 opts.postgres.runtime_size,
583 default_opts.postgres.runtime_size
584 );
585 assert!(opts.opentsdb.enable);
586
587 assert!(!opts.influxdb.enable);
588 }
589
590 #[test]
591 fn test_read_from_config_file() {
592 let mut file = create_named_temp_file();
593 let toml_str = r#"
594 [http]
595 addr = "127.0.0.1:4000"
596 timeout = "0s"
597 body_limit = "2GB"
598
599 [opentsdb]
600 enable = false
601
602 [logging]
603 level = "debug"
604 dir = "./greptimedb_data/test/logs"
605 "#;
606 write!(file, "{}", toml_str).unwrap();
607
608 let command = StartCommand {
609 config_file: Some(file.path().to_str().unwrap().to_string()),
610 disable_dashboard: Some(false),
611 ..Default::default()
612 };
613
614 let fe_opts = command.load_options(&Default::default()).unwrap().component;
615
616 assert_eq!("127.0.0.1:4000".to_string(), fe_opts.http.addr);
617 assert_eq!(Duration::from_secs(0), fe_opts.http.timeout);
618
619 assert_eq!(ReadableSize::gb(2), fe_opts.http.body_limit);
620
621 assert_eq!("debug", fe_opts.logging.level.as_ref().unwrap());
622 assert_eq!(
623 "./greptimedb_data/test/logs".to_string(),
624 fe_opts.logging.dir
625 );
626 assert!(!fe_opts.opentsdb.enable);
627 }
628
629 #[tokio::test]
630 async fn test_try_from_start_command_to_anymap() {
631 let fe_opts = frontend::frontend::FrontendOptions {
632 http: HttpOptions {
633 disable_dashboard: false,
634 ..Default::default()
635 },
636 meta_client: Some(MetaClientOptions::default()),
637 user_provider: Some("static_user_provider:cmd:test=test".to_string()),
638 ..Default::default()
639 };
640
641 let mut plugins = Plugins::new();
642 plugins::setup_frontend_plugins(&mut plugins, &[], &fe_opts)
643 .await
644 .unwrap();
645
646 let provider = plugins.get::<UserProviderRef>().unwrap();
647 let result = provider
648 .authenticate(
649 Identity::UserId("test", None),
650 Password::PlainText("test".to_string().into()),
651 )
652 .await;
653 let _ = result.unwrap();
654 }
655
656 #[test]
657 fn test_load_log_options_from_cli() {
658 let cmd = StartCommand {
659 disable_dashboard: Some(false),
660 ..Default::default()
661 };
662
663 let options = cmd
664 .load_options(&GlobalOptions {
665 log_dir: Some("./greptimedb_data/test/logs".to_string()),
666 log_level: Some("debug".to_string()),
667
668 #[cfg(feature = "tokio-console")]
669 tokio_console_addr: None,
670 })
671 .unwrap()
672 .component;
673
674 let logging_opt = options.logging;
675 assert_eq!("./greptimedb_data/test/logs", logging_opt.dir);
676 assert_eq!("debug", logging_opt.level.as_ref().unwrap());
677 }
678
679 #[test]
680 fn test_config_precedence_order() {
681 let mut file = create_named_temp_file();
682 let toml_str = r#"
683 [http]
684 addr = "127.0.0.1:4000"
685
686 [meta_client]
687 timeout = "3s"
688 connect_timeout = "5s"
689 tcp_nodelay = true
690
691 [mysql]
692 addr = "127.0.0.1:4002"
693 "#;
694 write!(file, "{}", toml_str).unwrap();
695
696 let env_prefix = "FRONTEND_UT";
697 temp_env::with_vars(
698 [
699 (
700 [
702 env_prefix.to_string(),
703 "mysql".to_uppercase(),
704 "addr".to_uppercase(),
705 ]
706 .join(ENV_VAR_SEP),
707 Some("127.0.0.1:14002"),
708 ),
709 (
710 [
712 env_prefix.to_string(),
713 "mysql".to_uppercase(),
714 "runtime_size".to_uppercase(),
715 ]
716 .join(ENV_VAR_SEP),
717 Some("11"),
718 ),
719 (
720 [
722 env_prefix.to_string(),
723 "http".to_uppercase(),
724 "addr".to_uppercase(),
725 ]
726 .join(ENV_VAR_SEP),
727 Some("127.0.0.1:24000"),
728 ),
729 (
730 [
732 env_prefix.to_string(),
733 "meta_client".to_uppercase(),
734 "metasrv_addrs".to_uppercase(),
735 ]
736 .join(ENV_VAR_SEP),
737 Some("127.0.0.1:3001,127.0.0.1:3002,127.0.0.1:3003"),
738 ),
739 ],
740 || {
741 let command = StartCommand {
742 config_file: Some(file.path().to_str().unwrap().to_string()),
743 http_addr: Some("127.0.0.1:14000".to_string()),
744 env_prefix: env_prefix.to_string(),
745 ..Default::default()
746 };
747
748 let fe_opts = command.load_options(&Default::default()).unwrap().component;
749
750 assert_eq!(fe_opts.mysql.runtime_size, 11);
752 assert_eq!(
753 fe_opts.meta_client.unwrap().metasrv_addrs,
754 vec![
755 "127.0.0.1:3001".to_string(),
756 "127.0.0.1:3002".to_string(),
757 "127.0.0.1:3003".to_string()
758 ]
759 );
760
761 assert_eq!(fe_opts.mysql.addr, "127.0.0.1:4002");
763
764 assert_eq!(fe_opts.http.addr, "127.0.0.1:14000");
766
767 assert_eq!(fe_opts.grpc.bind_addr, GrpcOptions::default().bind_addr);
769 },
770 );
771 }
772
773 #[test]
774 fn test_parse_grpc_cli_aliases() {
775 let command = StartCommand::try_parse_from([
776 "frontend",
777 "--grpc-bind-addr",
778 "127.0.0.1:14001",
779 "--grpc-server-addr",
780 "10.0.0.1:14001",
781 "--internal-grpc-bind-addr",
782 "127.0.0.1:14010",
783 "--internal-grpc-server-addr",
784 "10.0.0.1:14010",
785 ])
786 .unwrap();
787 assert_eq!(command.grpc_bind_addr.as_deref(), Some("127.0.0.1:14001"));
788 assert_eq!(command.grpc_server_addr.as_deref(), Some("10.0.0.1:14001"));
789 assert_eq!(
790 command.internal_grpc_bind_addr.as_deref(),
791 Some("127.0.0.1:14010")
792 );
793 assert_eq!(
794 command.internal_grpc_server_addr.as_deref(),
795 Some("10.0.0.1:14010")
796 );
797
798 let command = StartCommand::try_parse_from([
799 "frontend",
800 "--rpc-bind-addr",
801 "127.0.0.1:24001",
802 "--rpc-server-addr",
803 "10.0.0.2:24001",
804 "--internal-rpc-bind-addr",
805 "127.0.0.1:24010",
806 "--internal-rpc-server-addr",
807 "10.0.0.2:24010",
808 ])
809 .unwrap();
810 assert_eq!(command.grpc_bind_addr.as_deref(), Some("127.0.0.1:24001"));
811 assert_eq!(command.grpc_server_addr.as_deref(), Some("10.0.0.2:24001"));
812 assert_eq!(
813 command.internal_grpc_bind_addr.as_deref(),
814 Some("127.0.0.1:24010")
815 );
816 assert_eq!(
817 command.internal_grpc_server_addr.as_deref(),
818 Some("10.0.0.2:24010")
819 );
820
821 let command = StartCommand::try_parse_from([
822 "frontend",
823 "--rpc-addr",
824 "127.0.0.1:34001",
825 "--rpc-hostname",
826 "10.0.0.3:34001",
827 "--internal-rpc-addr",
828 "127.0.0.1:34010",
829 "--internal-rpc-hostname",
830 "10.0.0.3:34010",
831 ])
832 .unwrap();
833 assert_eq!(command.grpc_bind_addr.as_deref(), Some("127.0.0.1:34001"));
834 assert_eq!(command.grpc_server_addr.as_deref(), Some("10.0.0.3:34001"));
835 assert_eq!(
836 command.internal_grpc_bind_addr.as_deref(),
837 Some("127.0.0.1:34010")
838 );
839 assert_eq!(
840 command.internal_grpc_server_addr.as_deref(),
841 Some("10.0.0.3:34010")
842 );
843 }
844
845 #[test]
846 fn test_help_uses_grpc_option_names() {
847 let mut cmd = StartCommand::command();
848 let mut help = Vec::new();
849 cmd.write_long_help(&mut help).unwrap();
850 let help = String::from_utf8(help).unwrap();
851
852 assert!(help.contains("--grpc-bind-addr"));
853 assert!(help.contains("--grpc-server-addr"));
854 assert!(help.contains("--internal-grpc-bind-addr"));
855 assert!(help.contains("--internal-grpc-server-addr"));
856 assert!(!help.contains("--rpc-bind-addr"));
857 assert!(!help.contains("--rpc-server-addr"));
858 assert!(!help.contains("--rpc-addr"));
859 assert!(!help.contains("--rpc-hostname"));
860 assert!(!help.contains("--internal-rpc-bind-addr"));
861 assert!(!help.contains("--internal-rpc-server-addr"));
862 assert!(!help.contains("--internal-rpc-addr"));
863 assert!(!help.contains("--internal-rpc-hostname"));
864 }
865}